test: Verify locked shared import provenance (ARCH-06A)

.github/workflows/ci.yml

@@ -147,6 +147,12 @@ jobs:
           assert resolve_canonical_profile("russell_top50_leader_rotation") == "russell_top50_leader_rotation"
           PY
 
+      - name: Verify installed shared test imports
+        run: |
+          set -euo pipefail
+          uv run --no-sync python -m pytest -q -p test_support.pytest_provenance \
+            --shared-import-mode installed-lock tests/test_shared_import_contract.py
+
       - name: Verify Python dependencies
         run: uv pip check
 

docs/shared-test-imports.md

@@ -0,0 +1,58 @@
+# Shared dependency test imports
+
+The ordinary pytest command keeps its existing import order and test doubles.
+It does not certify shared dependency provenance. Select the separate strict lane
+explicitly after installing the unchanged `uv.lock` with `uv sync --frozen --extra test`.
+Dependency installation is preparation; run tests without network or credential access.
+
+The installed-lock lane checks VCS URL, exact locked revision and version together
+with every loaded shared module's actual file/spec origin, package search path,
+distribution ownership and RECORD SHA-256. Editable installs and sibling shadows fail.
+The existing sibling entries in pytest's `pythonpath` remain unchanged.
+
+```bash
+uv run --no-sync python -m pytest -q -p test_support.pytest_provenance \
+  --shared-import-mode installed-lock tests/test_shared_import_contract.py
+```
+
+The audited selection is exactly the three tests in
+`tests/test_shared_import_contract.py`: safe pinned URI/profile parsers and one
+explicit, restored test double. Checks run in that pytest process at session start,
+after collection, before and after setup/call/teardown, and at successful completion.
+Session start rejects any preloaded shared `sys.modules` entry as unverified:
+current disk bytes cannot establish what an earlier module object executed, even
+when its path/spec metadata agrees. No module is reloaded or evicted. Each boundary
+also requires a standard Python source loader and compares its code with a fresh compile of the verified
+source. Stale timestamp caches (including equal-size/mtime collisions) and
+unchecked-hash bytecode cannot pass merely because the `.py` file is correct.
+Later checks cover all shared modules present at each boundary, including parser implementations. No shared namespace is exempted.
+A scoped double replaces one previously verified module, is counted separately,
+cannot count toward real-source completion, and must restore the original identity.
+An undeclared or leaked double fails the strict lane. The lane never imports broker
+or cloud entry points, fetches remote artifacts or reads real credentials.
+
+Completion requires the exact full collection, all three calls passed, no failed or
+skipped phases, final restored state and the complete required real-module inventory.
+Collection-only, deselection, interrupted runs and failed checks cannot emit
+`SHARED_IMPORT_PROVENANCE_COMPLETE`. Receipts contain public source revisions,
+module names and declared origin classes, without absolute paths or environment dumps.
+They are engineering observations, not authenticated review or deployment attestations.
+
+Pinned-checkout validation requires an explicit root, exact locked Git HEAD, a clean
+worktree, tracked source bytes equal to the Git object, and origins inside that root.
+An unexpected same-named root, dirty content, ignored/untracked source, unknown origin,
+file/spec disagreement or symlink escape fails. Installed-lock mode rejects editable
+sources even if revision metadata appears correct.
+
+Run synthetic negative controls with
+`uv run --no-sync python -m pytest -q tests/test_import_provenance.py`.
+They use disposable modules/metadata/Git repositories, never modify real dependencies,
+and observe a child process successfully importing a sibling before the guard rejects
+it with a nonzero exit. Repeat/reorder the strict selection to check cleanup.
+
+This is a bounded parser lane. It does not certify the ordinary whole suite or
+transient modules loaded and removed between boundaries by arbitrary third-party hooks.
+Existing test-specific path edits and mocks remain in place, so legacy whole-suite
+source ambiguity remains characterized rather than eliminated. The test support is
+excluded from the setuptools production package selection; pins, production source,
+existing scanner/completion checks and coverage configuration are unchanged.

test_support/__init__.py

@@ -0,0 +1 @@
+"""Repository-local test support; excluded from production packages."""

test_support/provenance.py

@@ -0,0 +1,198 @@
+"""Test-only shared import checks. Receipts describe a bounded pytest process."""
+from __future__ import annotations
+
+import base64
+from contextlib import contextmanager
+import hashlib
+from importlib import metadata
+from importlib.machinery import SourceFileLoader
+import json
+from pathlib import Path
+import re
+import subprocess
+import sys
+import sysconfig
+import tomllib
+from types import ModuleType
+from urllib.parse import urlsplit, urlunsplit
+
+SPECS = {
+    "quant_platform_kit": ("quant-platform-kit", "QuantPlatformKit"),
+    "us_equity_strategies": ("us-equity-strategies", "UsEquityStrategies"),
+}
+REQUIRED = {
+    "quant_platform_kit", "quant_platform_kit.common.strategy_plugin_artifacts",
+    "us_equity_strategies", "us_equity_strategies.catalog",
+}
+
+
+class ProvenanceError(RuntimeError):
+    """Fixed reason codes only; never include a private path or metadata value."""
+
+
+def locked_sources(lockfile):
+    packages = tomllib.loads(Path(lockfile).read_text())["package"]
+    result = {}
+    for namespace, (name, repo) in SPECS.items():
+        matches = [p for p in packages if p["name"] == name]
+        if len(matches) != 1:
+            raise ProvenanceError("ambiguous_lock")
+        package = matches[0]
+        parsed = urlsplit(package.get("source", {}).get("git", ""))
+        url = urlunsplit((parsed.scheme, parsed.netloc, parsed.path, "", ""))
+        if (url != f"https://github.com/QuantStrategyLab/{repo}.git"
+                or not re.fullmatch(r"[0-9a-f]{40}", parsed.fragment)):
+            raise ProvenanceError("invalid_lock")
+        result[namespace] = dict(name=name, url=url, revision=parsed.fragment,
+                                 version=package["version"])
+    return result
+
+
+def git(root, *args):
+    result = subprocess.run(["git", "-C", str(root), *args], capture_output=True)
+    if result.returncode:
+        raise ProvenanceError("checkout_unproven")
+    return result.stdout
+
+
+class Audit:
+    def __init__(self, locked, mode, *, checkouts=None, distributions=metadata.distribution,
+                 sites=None, source_modes=None):
+        if mode not in {"installed-lock", "pinned-checkout"}:
+            raise ProvenanceError("explicit_mode_required")
+        if set(locked) != set(SPECS):
+            raise ProvenanceError("incomplete_lock")
+        self.locked = locked
+        self.mode = mode
+        self.source_modes = dict.fromkeys(SPECS, mode) if source_modes is None else source_modes
+        if (set(self.source_modes) != set(SPECS)
+                or not set(self.source_modes.values()) <= {"installed-lock", "pinned-checkout"}
+                or (mode == "installed-lock" and set(self.source_modes.values()) != {mode})):
+            raise ProvenanceError("explicit_sources_required")
+        self.checkouts = {k: Path(v).resolve() for k, v in (checkouts or {}).items()}
+        if set(self.checkouts) != {k for k, v in self.source_modes.items() if v == "pinned-checkout"}:
+            raise ProvenanceError("explicit_checkouts_required")
+        self.distributions = distributions
+        self.sites = tuple(Path(p).resolve() for p in (sites if sites is not None else {
+            sysconfig.get_path("purelib"), sysconfig.get_path("platlib")}))
+        self.doubles = {}
+        self.double_observations = 0
+
+    def _distribution(self, namespace):
+        expected = self.locked[namespace]
+        dist = self.distributions(expected["name"])
+        direct = json.loads(dist.read_text("direct_url.json") or "{}")
+        if (dist.version != expected["version"] or direct.get("dir_info", {}).get("editable")
+                or direct.get("url") != expected["url"]
+                or direct.get("vcs_info", {}).get("vcs") != "git"
+                or direct.get("vcs_info", {}).get("commit_id") != expected["revision"]):
+            raise ProvenanceError("installed_metadata_mismatch")
+        return dist
+
+    def _checkout(self, namespace):
+        root = self.checkouts[namespace]
+        if (Path(git(root, "rev-parse", "--show-toplevel").decode().strip()).resolve() != root
+                or git(root, "rev-parse", "HEAD").decode().strip()
+                != self.locked[namespace]["revision"]
+                or git(root, "status", "--porcelain", "--untracked-files=all")):
+            raise ProvenanceError("checkout_not_clean_locked_root")
+        return root
+
+    def _file(self, path, namespace, source):
+        resolved = path.resolve(strict=True)
+        if self.source_modes[namespace] == "installed-lock":
+            if not any(resolved.is_relative_to(root) for root in self.sites):
+                raise ProvenanceError("outside_distribution")
+            entries = [entry for entry in (source.files or ())
+                       if Path(source.locate_file(entry)).absolute() == path.absolute()]
+            if (len(entries) != 1 or entries[0].hash is None
+                    or entries[0].hash.mode != "sha256"):
+                raise ProvenanceError("missing_record_hash")
+            digest = base64.urlsafe_b64encode(hashlib.sha256(path.read_bytes()).digest())
+            if digest.decode().rstrip("=") != entries[0].hash.value:
+                raise ProvenanceError("record_hash_mismatch")
+            package_root = Path(source.locate_file(namespace)).resolve()
+        else:
+            package_root = source / "src" / namespace
+            if not resolved.is_relative_to(package_root):
+                raise ProvenanceError("outside_declared_checkout")
+            relative = path.absolute().relative_to(source).as_posix()
+            # The Git object supplies the expected bytes, including for ignored files.
+            if git(source, "show", f"HEAD:{relative}") != path.read_bytes():
+                raise ProvenanceError("checkout_content_mismatch")
+        if not resolved.is_relative_to(package_root):
+            raise ProvenanceError("symlink_escape")
+        return resolved
+
+    def require_fresh_process(self, modules=None):
+        # Files on disk cannot prove what an earlier module object executed.
+        modules = sys.modules if modules is None else modules
+        if any(name.split(".")[0] in SPECS for name in modules):
+            raise ProvenanceError("preloaded_shared_modules_unverified")
+
+    def audit(self, modules=None):
+        modules = dict(sys.modules if modules is None else modules)
+        sources = {namespace: (self._distribution(namespace) if self.source_modes[namespace] == "installed-lock"
+                              else self._checkout(namespace)) for namespace in SPECS}
+        observed = set()
+        for name, module in modules.items():
+            namespace = name.split(".")[0]
+            if namespace not in SPECS:
+                continue
+            if name in self.doubles and module is self.doubles[name]:
+                self.double_observations += 1
+                continue
+            spec = getattr(module, "__spec__", None)
+            filename = getattr(module, "__file__", None)
+            origin = getattr(spec, "origin", None)
+            if (not isinstance(module, ModuleType) or module.__name__ != name
+                    or getattr(spec, "name", None) != name or not filename or not origin):
+                raise ProvenanceError("unproven_module")
+            file = Path(filename)
+            if not file.is_absolute() or file.resolve() != Path(origin).resolve():
+                raise ProvenanceError("origin_disagreement")
+            checked = self._file(file, namespace, sources[namespace])
+            loader = getattr(spec, "loader", None)
+            if (type(loader) is not SourceFileLoader or file.suffix != ".py"
+                    or loader.name != name or Path(loader.path).resolve() != checked):
+                raise ProvenanceError("unproven_source_loader")
+            # A valid source hash does not validate a timestamp or unchecked-hash
+            # .pyc. Compare the code Python would load with a fresh source compile.
+            expected_code = compile(file.read_bytes(), str(file), "exec", dont_inherit=True)
+            if loader.get_code(name) != expected_code:
+                raise ProvenanceError("bytecode_source_mismatch")
+            if hasattr(module, "__path__"):
+                paths = tuple(Path(p).resolve() for p in module.__path__)
+                locations = tuple(Path(p).resolve() for p in (spec.submodule_search_locations or ()))
+                if paths != (checked.parent,) or locations != paths:
+                    raise ProvenanceError("package_search_path_mismatch")
+            observed.add(name)
+        return observed
+
+    @contextmanager
+    def scoped_double(self, name, double):
+        # Only an exact, already verified module can be temporarily replaced.
+        if name not in self.audit() or name in self.doubles:
+            raise ProvenanceError("double_boundary_unproven")
+        original = sys.modules[name]
+        self.doubles[name] = double
+        sys.modules[name] = double
+        try:
+            self.audit()
+            yield double
+        finally:
+            leaked = sys.modules.get(name) is not double
+            sys.modules[name] = original
+            del self.doubles[name]
+            self.audit()
+            if leaked:
+                raise ProvenanceError("double_boundary_changed")
+
+    def receipt(self):
+        observed = self.audit()
+        if self.doubles or not REQUIRED <= observed:
+            raise ProvenanceError("incomplete_real_inventory")
+        return {"mode": self.mode, "scope": "tests/test_shared_import_contract.py",
+                "origin_classes": self.source_modes, "real_modules": sorted(observed),
+                "revisions": {k: v["revision"] for k, v in self.locked.items()},
+                "scoped_double_observations": self.double_observations}

test_support/pytest_provenance.py

@@ -0,0 +1,108 @@
+"""Explicit strict lane; never auto-loaded by ordinary pytest runs."""
+import json
+from pathlib import Path
+
+import pytest
+
+from .provenance import Audit, ProvenanceError, locked_sources
+
+MODES = ("installed-lock",)
+
+EXPECTED = {"test_pinned_uri_parser", "test_pinned_profile_parser", "test_scoped_double_restores"}
+
+
+def pytest_addoption(parser):
+    group = parser.getgroup("shared import provenance")
+    group.addoption("--shared-import-mode", choices=MODES, required=True)
+    group.addoption("--qpk-checkout")
+    group.addoption("--ues-checkout")
+
+
+def pytest_configure(config):
+    roots = {}
+    for namespace, option in (("quant_platform_kit", "qpk_checkout"),
+                              ("us_equity_strategies", "ues_checkout")):
+        value = config.getoption(option)
+        if value:
+            roots[namespace] = value
+    config._shared_audit = Audit(locked_sources(config.rootpath / "uv.lock"),
+                                 config.getoption("shared_import_mode"), checkouts=roots)
+    config._shared_executed = set()
+    config._shared_collected = False
+    config._shared_failed = False
+
+
+def checked(config):
+    try:
+        config._shared_audit.audit()
+    except (ProvenanceError, OSError, ValueError, ImportError) as exc:
+        config._shared_failed = True
+        reason = str(exc) if isinstance(exc, ProvenanceError) else type(exc).__name__
+        raise pytest.UsageError("shared import provenance: " + reason) from None
+
+
+def pytest_sessionstart(session):
+    try:
+        session.config._shared_audit.require_fresh_process()
+    except ProvenanceError:
+        session.config._shared_failed = True
+        raise pytest.UsageError("shared import provenance: preloaded_shared_modules_unverified") from None
+    checked(session.config)
+
+
+def pytest_collection_finish(session):
+    config = session.config
+    expected_path = (config.rootpath / "tests/test_shared_import_contract.py").resolve()
+    if ({item.name for item in session.items} != EXPECTED
+            or len(session.items) != len(EXPECTED)
+            or any(Path(item.path).resolve() != expected_path for item in session.items)):
+        config._shared_failed = True
+        raise pytest.UsageError("shared import provenance: incomplete_audited_selection")
+    checked(config)  # Same process, after test modules/conftests have been imported.
+    config._shared_collected = True
+
+
+def boundary(item):
+    checked(item.config)
+    outcome = yield
+    checked(item.config)
+    if outcome.excinfo:
+        item.config._shared_failed = True
+
+
+@pytest.hookimpl(hookwrapper=True, tryfirst=True)
+def pytest_runtest_setup(item):
+    yield from boundary(item)
+
+
+@pytest.hookimpl(hookwrapper=True, tryfirst=True)
+def pytest_runtest_call(item):
+    yield from boundary(item)
+
+
+@pytest.hookimpl(hookwrapper=True, tryfirst=True)
+def pytest_runtest_teardown(item):
+    yield from boundary(item)
+
+
+@pytest.hookimpl(hookwrapper=True)
+def pytest_runtest_makereport(item, call):
+    report = (yield).get_result()
+    if report.failed or report.skipped:
+        item.config._shared_failed = True
+    if report.when == "call" and report.passed:
+        item.config._shared_executed.add(item.name)
+
+
+def pytest_sessionfinish(session, exitstatus):
+    config = session.config
+    if (exitstatus or config._shared_failed or not config._shared_collected
+            or config._shared_executed != EXPECTED):
+        return
+    try:
+        receipt = config._shared_audit.receipt()
+    except (ProvenanceError, OSError, ValueError, ImportError):
+        session.exitstatus = 1
+        return
+    reporter = config.pluginmanager.getplugin("terminalreporter")
+    reporter.write_line("SHARED_IMPORT_PROVENANCE_COMPLETE " + json.dumps(receipt, sort_keys=True))