From 12ee2a7ae7304e80436632d2ced950c515478f8b Mon Sep 17 00:00:00 2001 From: yuqei Date: Thu, 8 Oct 2026 22:13:58 +0800 Subject: [PATCH] fix(infra): preserve inventory isolation through lazy imports Keep asset snapshot assembly inside the candidate import context so caller-preloaded modules cannot supply lazy dependencies. Verify original modules and paths are restored on both success and rejection. --- .../scripts/build-rc06-publication-batch.py | 111 +++++++++--------- .../tests/test_rc06_publication_batch.py | 23 +++- 2 files changed, 78 insertions(+), 56 deletions(-) diff --git a/yuqei-infra/scripts/build-rc06-publication-batch.py b/yuqei-infra/scripts/build-rc06-publication-batch.py index 08c2c595..3da80bfe 100644 --- a/yuqei-infra/scripts/build-rc06-publication-batch.py +++ b/yuqei-infra/scripts/build-rc06-publication-batch.py @@ -20,9 +20,10 @@ import stat import subprocess import sys import tempfile +from contextlib import contextmanager from datetime import UTC, datetime, timedelta from pathlib import Path, PurePath -from typing import Any, Iterable, Mapping +from typing import Any, Iterable, Iterator, Mapping from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives import hashes, serialization @@ -736,7 +737,8 @@ def _project_imports(repo: Path) -> tuple[list[str], dict[str, Any]]: return inserted, previous -def _load_inventory_symbols(repo: Path) -> tuple[Any, Any, Any, Any, Any]: +@contextmanager +def _load_inventory_symbols(repo: Path) -> Iterator[tuple[Any, Any, Any, Any, Any]]: inserted, previous = _project_imports(repo) try: from yuqei_contract_api.contract_templates import ( # type: ignore @@ -748,7 +750,7 @@ def _load_inventory_symbols(repo: Path) -> tuple[Any, Any, Any, Any, Any]: REVIEW_ROUTING_VERSION, review_prompt_routes, ) - return ( + yield ( _load_template_body, list_contract_templates, template_document_kind, @@ -927,64 +929,63 @@ def load_runtime_inventory( frozen_source_paths: list[str] | None = None, ) -> dict[str, Any]: """Load the real source catalog without consulting a service or database.""" - ( + with _load_inventory_symbols(repo) as ( _load_template_body, list_contract_templates, template_document_kind, REVIEW_ROUTING_VERSION, review_prompt_routes, - ) = _load_inventory_symbols(repo) - - page = list_contract_templates() - if page.total != EXPECTED_ASSET_COUNT or len(page.items) != EXPECTED_ASSET_COUNT: - raise PublicationBatchError( - f"source catalog must contain exactly {EXPECTED_ASSET_COUNT} templates; got {page.total}" - ) - assets: list[dict[str, Any]] = [] - seen_assets: set[str] = set() - for template in page.items: - if template.id in seen_assets: - raise PublicationBatchError(f"duplicate template ID: {template.id}") - seen_assets.add(template.id) - body = _load_template_body(template.id) - schema = template.input_schema.model_dump(mode="json") if template.input_schema else None - scene = schema.get("scene") if isinstance(schema, dict) else None - document_kind = ( - scene.get("document_kind") - if isinstance(scene, dict) and isinstance(scene.get("document_kind"), str) - else template_document_kind(template.contract_type) - ) - required = [item.model_dump(mode="json") for item in template.required_variables] - optional = [item.model_dump(mode="json") for item in template.optional_variables] - routes = list(review_prompt_routes(template.contract_type, template.id)) - review_identity = { - "routing_version": REVIEW_ROUTING_VERSION, - "prompt_routes": routes, - "review_rule_ids": list(template.review_rule_ids), - } - assets.append( - _finish_asset( - { - "template_id": template.id, - "name": template.name, - "version": template.lifecycle_version or template.version, - "document_kind": document_kind, - "contract_type": template.contract_type, - "source": template.source, - "source_lifecycle_status": template.lifecycle_status, - "body_content": body, - "content_sha256": template.content_sha256, - "input_schema": schema, - "required_variables": required, - "optional_variables": optional, - "validation_rules": list(template.validation_rules), - "clause_ids": list(template.clause_ids), - "review_rule_ids": list(template.review_rule_ids), - "review_identity": review_identity, - } + ): + page = list_contract_templates() + if page.total != EXPECTED_ASSET_COUNT or len(page.items) != EXPECTED_ASSET_COUNT: + raise PublicationBatchError( + f"source catalog must contain exactly {EXPECTED_ASSET_COUNT} templates; got {page.total}" ) - ) - assets.sort(key=lambda item: item["template_id"]) + assets: list[dict[str, Any]] = [] + seen_assets: set[str] = set() + for template in page.items: + if template.id in seen_assets: + raise PublicationBatchError(f"duplicate template ID: {template.id}") + seen_assets.add(template.id) + body = _load_template_body(template.id) + schema = template.input_schema.model_dump(mode="json") if template.input_schema else None + scene = schema.get("scene") if isinstance(schema, dict) else None + document_kind = ( + scene.get("document_kind") + if isinstance(scene, dict) and isinstance(scene.get("document_kind"), str) + else template_document_kind(template.contract_type) + ) + required = [item.model_dump(mode="json") for item in template.required_variables] + optional = [item.model_dump(mode="json") for item in template.optional_variables] + routes = list(review_prompt_routes(template.contract_type, template.id)) + review_identity = { + "routing_version": REVIEW_ROUTING_VERSION, + "prompt_routes": routes, + "review_rule_ids": list(template.review_rule_ids), + } + assets.append( + _finish_asset( + { + "template_id": template.id, + "name": template.name, + "version": template.lifecycle_version or template.version, + "document_kind": document_kind, + "contract_type": template.contract_type, + "source": template.source, + "source_lifecycle_status": template.lifecycle_status, + "body_content": body, + "content_sha256": template.content_sha256, + "input_schema": schema, + "required_variables": required, + "optional_variables": optional, + "validation_rules": list(template.validation_rules), + "clause_ids": list(template.clause_ids), + "review_rule_ids": list(template.review_rule_ids), + "review_identity": review_identity, + } + ) + ) + assets.sort(key=lambda item: item["template_id"]) drafting_raw, _ = secure_read_json(repo / DRAFTING_MODULE_PATH, "drafting module source") review_raw, _ = secure_read_json(repo / REVIEW_MODULE_PATH, "review module source") diff --git a/yuqei-infra/tests/test_rc06_publication_batch.py b/yuqei-infra/tests/test_rc06_publication_batch.py index 4e11f9f9..e6ddfda9 100644 --- a/yuqei-infra/tests/test_rc06_publication_batch.py +++ b/yuqei-infra/tests/test_rc06_publication_batch.py @@ -3687,26 +3687,47 @@ def test_frozen_import_closure_is_derived_from_commit() -> None: ) in paths +@pytest.mark.parametrize("preload_inputs", [False, True]) def test_runtime_inventory_does_not_reuse_preloaded_project_modules( monkeypatch: pytest.MonkeyPatch, + preload_inputs: bool, ) -> None: poisoned_package = type(sys)("yuqei_contract_api") poisoned_package.__path__ = [] poisoned_templates = type(sys)("yuqei_contract_api.contract_templates") + poisoned_inputs = type(sys)("yuqei_contract_api.template_inputs") + original_path = list(sys.path) - def poisoned_loader(): + def poisoned_loader(*_args, **_kwargs): raise AssertionError("preloaded mutable project module was executed") poisoned_templates.list_contract_templates = poisoned_loader + poisoned_inputs.TemplateInputRule = poisoned_loader monkeypatch.setitem(sys.modules, "yuqei_contract_api", poisoned_package) monkeypatch.setitem( sys.modules, "yuqei_contract_api.contract_templates", poisoned_templates, ) + if preload_inputs: + monkeypatch.setitem(sys.modules, "yuqei_contract_api.template_inputs", poisoned_inputs) isolated = batch.load_runtime_inventory(REPO_ROOT) assert len(isolated["assets"]) == batch.EXPECTED_ASSET_COUNT + assert sys.modules["yuqei_contract_api"] is poisoned_package + assert sys.modules["yuqei_contract_api.contract_templates"] is poisoned_templates + if preload_inputs: + assert sys.modules["yuqei_contract_api.template_inputs"] is poisoned_inputs + assert sys.path == original_path + + monkeypatch.setattr(batch, "EXPECTED_ASSET_COUNT", batch.EXPECTED_ASSET_COUNT + 1) + with pytest.raises(batch.PublicationBatchError, match="source catalog must contain exactly"): + batch.load_runtime_inventory(REPO_ROOT) + assert sys.modules["yuqei_contract_api"] is poisoned_package + assert sys.modules["yuqei_contract_api.contract_templates"] is poisoned_templates + if preload_inputs: + assert sys.modules["yuqei_contract_api.template_inputs"] is poisoned_inputs + assert sys.path == original_path def test_source_manifest_embedded_hash_and_clean_candidate_are_mandatory() -> None: