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.
2026-10-08 22:13:58 +08:00
parent dc6b739e22
commit 12ee2a7ae7
2 changed files with 78 additions and 56 deletions
@@ -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")
@@ -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: