From 45e72bba21c1178f6ca6ee33e91e0d189a0afc50 Mon Sep 17 00:00:00 2001 From: Ramesh Padmanabhaiah <22363102+codeforester@users.noreply.github.com> Date: Mon, 28 Sep 2026 12:03:17 +0530 Subject: [PATCH 1/3] fix(python): break trust and projects import cycle --- cli/python/base_projects/test_preflight.py | 4 +- .../base_projects/workspace_onboarding.py | 8 +- cli/python/base_setup/manifest_trust.py | 322 +++++++++++++++ .../base_setup/manifest_trust_guidance.py | 158 ++++++++ cli/python/base_trust/engine.py | 19 +- cli/python/base_trust/guidance.py | 173 +------- .../tests/test_import_boundaries.py | 49 +++ cli/python/base_trust/trust_store.py | 368 +++--------------- 8 files changed, 615 insertions(+), 486 deletions(-) create mode 100644 cli/python/base_setup/manifest_trust.py create mode 100644 cli/python/base_setup/manifest_trust_guidance.py create mode 100644 cli/python/base_trust/tests/test_import_boundaries.py diff --git a/cli/python/base_projects/test_preflight.py b/cli/python/base_projects/test_preflight.py index 2c9160ac..0f29e456 100644 --- a/cli/python/base_projects/test_preflight.py +++ b/cli/python/base_projects/test_preflight.py @@ -8,10 +8,10 @@ from base_setup.manifest_model import BaseManifest from base_setup.test_requirements import check_test_requirements -from base_trust.guidance import print_blocked_command_text -from base_trust.trust_store import ( +from base_setup.manifest_trust import ( ManifestCommandTrustStore, compute_trust_identity, manifest_command_surfaces_from_manifest, ) +from base_setup.manifest_trust_guidance import print_blocked_command_text def project_test_preflight(ctx: base_cli.Context, manifest: BaseManifest) -> bool: diff --git a/cli/python/base_projects/workspace_onboarding.py b/cli/python/base_projects/workspace_onboarding.py index f176f205..10ecef7d 100644 --- a/cli/python/base_projects/workspace_onboarding.py +++ b/cli/python/base_projects/workspace_onboarding.py @@ -4,10 +4,6 @@ from dataclasses import dataclass from pathlib import Path -from base_trust.guidance import allow_command_text -from base_trust.trust_store import ManifestCommandTrustStore -from base_trust.trust_store import compute_trust_identity -from base_trust.trust_store import manifest_command_surfaces_from_manifest from base_projects.project_commands import test_command as manifest_test_command from base_projects.workspace_manifest import WorkspaceManifest from base_projects.workspace_repository_url import redact_repository_url @@ -15,6 +11,10 @@ from base_projects.workspace_statuses import workspace_manifest_project_statuses from base_setup.manifest import read_manifest from base_setup.manifest_loader import ManifestError +from base_setup.manifest_trust import ManifestCommandTrustStore +from base_setup.manifest_trust import compute_trust_identity +from base_setup.manifest_trust import manifest_command_surfaces_from_manifest +from base_setup.manifest_trust_guidance import allow_command_text from base_setup.runtime_inspection import runtime_verification_command diff --git a/cli/python/base_setup/manifest_trust.py b/cli/python/base_setup/manifest_trust.py new file mode 100644 index 00000000..dd73d5a9 --- /dev/null +++ b/cli/python/base_setup/manifest_trust.py @@ -0,0 +1,322 @@ +from __future__ import annotations + +import hashlib +import json +import os +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from base_cli_adapters.history import format_timestamp, utc_now +from base_cli_adapters.paths import base_state_root +from base_setup.git_commands import run_git +from base_setup.git_remote_parse import parse_origin_remote +from base_setup.manifest import BaseManifest +from base_setup.manifest import read_manifest +from base_setup.test_requirements import requirements_file_digest + +SCHEMA_VERSION = 1 +ALLOWED_COMMANDS = ["test", "run", "build", "demo", "activate"] +TRUST_RELATIVE_ROOT = Path("trust") / "manifest-commands" +TRUST_SCOPE_WARNING = ( + "Approval is bound to the base_manifest.yaml SHA-256 and declared local test " + "requirements SHA-256. Manifest or declared test requirements changes invalidate " + "approval; referenced scripts, direct executable files, Git HEAD, and uncommitted " + "working-tree changes are not independently verified or bound to approval. Re-review " + "those inputs before execution after repository changes." +) +TRUST_SCOPE = { + "approval_basis": "base_manifest.yaml_sha256+declared_test_requirements_sha256", + "manifest_changes_invalidate": True, + "declared_test_requirements_changes_invalidate": True, + "referenced_script_changes_invalidate": False, + "direct_executable_file_changes_invalidate": False, + "git_head_changes_invalidate": False, + "working_tree_changes_invalidate": False, + "warning": TRUST_SCOPE_WARNING, +} + + +def trust_scope_payload() -> dict[str, object]: + return dict(TRUST_SCOPE) + + +@dataclass(frozen=True) +class ManifestCommandTrustIdentity: + project_name: str + project_root: Path + manifest_path: Path + manifest_sha256: str + identity_key: str + test_requirements_sha256: str | None = None + git_root: Path | None = None + origin: str | None = None + head: str | None = None + + def project_payload(self) -> dict[str, str]: + payload = { + "name": self.project_name, + "root": str(self.project_root), + "manifest": str(self.manifest_path), + "manifest_sha256": self.manifest_sha256, + } + if self.test_requirements_sha256 is not None: + payload["test_requirements_sha256"] = self.test_requirements_sha256 + if self.git_root is not None: + payload["git_root"] = str(self.git_root) + if self.origin is not None: + payload["origin"] = self.origin + if self.head is not None: + payload["head"] = self.head + return payload + + +@dataclass(frozen=True) +class TrustStatus: + status: str + reason: str + identity: ManifestCommandTrustIdentity + record: dict[str, Any] | None = None + changed_record: dict[str, Any] | None = None + + @property + def is_allowed(self) -> bool: + return self.status == "allowed" + + +class ManifestCommandTrustStore: + def __init__(self, home: Path | None = None) -> None: + self.root = base_state_root(home) / TRUST_RELATIVE_ROOT + + def record_path(self, identity: ManifestCommandTrustIdentity) -> Path: + return self.root / f"{identity.identity_key}.json" + + def read_record(self, identity: ManifestCommandTrustIdentity) -> dict[str, Any] | None: + path = self.record_path(identity) + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError): + return None + if not isinstance(payload, dict) or payload.get("schema_version") != SCHEMA_VERSION: + return None + if identity_key_from_record(payload) != identity.identity_key: + return None + return payload + + def status(self, identity: ManifestCommandTrustIdentity) -> TrustStatus: + record = self.read_record(identity) + if record is not None: + return TrustStatus(status="allowed", reason="allowed", identity=identity, record=record) + + changed_record = self.find_changed_record(identity) + if changed_record is not None: + return TrustStatus( + status="blocked", + reason=trust_change_reason(identity, changed_record), + identity=identity, + changed_record=changed_record, + ) + + return TrustStatus(status="blocked", reason="not_allowed", identity=identity) + + def find_changed_record(self, identity: ManifestCommandTrustIdentity) -> dict[str, Any] | None: + if not self.root.is_dir(): + return None + for path in sorted(self.root.glob("*.json")): + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError): + continue + if not isinstance(payload, dict) or payload.get("schema_version") != SCHEMA_VERSION: + continue + project = payload.get("project") + if not isinstance(project, dict): + continue + if project.get("root") == str(identity.project_root) and project.get("manifest") == str( + identity.manifest_path + ): + return payload + return None + + def allow( + self, + identity: ManifestCommandTrustIdentity, + *, + base_version: str | None, + allowed_at: str | None = None, + ) -> Path: + path = self.record_path(identity) + payload = { + "schema_version": SCHEMA_VERSION, + "allowed_at": allowed_at or format_timestamp(utc_now()), + "allowed_by": "local-user", + "base_version": base_version, + "project": identity.project_payload(), + "allowed_commands": ALLOWED_COMMANDS, + } + write_json_atomic(path, payload) + return path + + def revoke(self, identity: ManifestCommandTrustIdentity) -> bool: + removed = False + paths = {self.record_path(identity)} + try: + records = tuple(self.root.iterdir()) + except FileNotFoundError: + return False + for path in records: + if path.suffix != ".json": + continue + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (FileNotFoundError, UnicodeError, json.JSONDecodeError): + continue + if not isinstance(payload, dict) or payload.get("schema_version") != SCHEMA_VERSION: + continue + project = payload.get("project") + if isinstance(project, dict) and project.get("root") == str(identity.project_root) and project.get( + "manifest" + ) == str(identity.manifest_path): + # Unlink the record in this store, never a path from its contents. + paths.add(path) + + for path in sorted(paths): + try: + path.unlink() + except FileNotFoundError: + continue + removed = True + return removed + + +def manifest_command_surfaces(manifest_path: Path) -> tuple[str, ...]: + manifest = read_manifest(manifest_path.expanduser().resolve()) + return manifest_command_surfaces_from_manifest(manifest) + + +def manifest_command_surfaces_from_manifest(manifest: BaseManifest) -> tuple[str, ...]: + surfaces = [] + if manifest.test is not None: + surfaces.append("test") + if manifest.commands: + surfaces.append("run") + if manifest.build is not None and manifest.build.targets: + surfaces.append("build") + if manifest.demo is not None: + surfaces.append("demo") + if manifest.activate.source: + surfaces.append("activate") + return tuple(surfaces) + + +def compute_trust_identity_for_manifest(manifest_path: Path) -> ManifestCommandTrustIdentity: + manifest = read_manifest(manifest_path.expanduser().resolve()) + return compute_trust_identity(manifest) + + +def compute_trust_identity(manifest: BaseManifest) -> ManifestCommandTrustIdentity: + canonical_manifest = manifest.path.resolve() + project_root = canonical_manifest.parent.resolve() + manifest_sha256 = sha256_file(canonical_manifest) + requirements_sha256 = requirements_file_digest(manifest) + git_root = git_repository_root(project_root) + origin = git_origin(project_root) + head = git_head(project_root) + identity_key = compute_identity_key(project_root, canonical_manifest, manifest_sha256, requirements_sha256) + return ManifestCommandTrustIdentity( + project_name=manifest.project_name, + project_root=project_root, + manifest_path=canonical_manifest, + manifest_sha256=manifest_sha256, + test_requirements_sha256=requirements_sha256, + identity_key=identity_key, + git_root=git_root, + origin=origin, + head=head, + ) + + +def sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def compute_identity_key( + project_root: Path, + manifest_path: Path, + manifest_sha256: str, + test_requirements_sha256: str | None = None, +) -> str: + payload = "\0".join([str(project_root), str(manifest_path), manifest_sha256, test_requirements_sha256 or ""]) + return hashlib.sha256(payload.encode("utf-8")).hexdigest() + + +def identity_key_from_record(record: dict[str, Any]) -> str | None: + project = record.get("project") + if not isinstance(project, dict): + return None + root = project.get("root") + manifest = project.get("manifest") + digest = project.get("manifest_sha256") + if not all(isinstance(value, str) and value for value in (root, manifest, digest)): + return None + requirements_digest = project.get("test_requirements_sha256") + if requirements_digest is not None and not isinstance(requirements_digest, str): + return None + return compute_identity_key(Path(root), Path(manifest), digest, requirements_digest) + + +def trust_change_reason(identity: ManifestCommandTrustIdentity, record: dict[str, Any]) -> str: + project = record.get("project") + if not isinstance(project, dict): + return "manifest_changed" + if project.get("manifest_sha256") != identity.manifest_sha256: + return "manifest_changed" + if project.get("test_requirements_sha256") != identity.test_requirements_sha256: + return "test_requirements_changed" + return "manifest_changed" + + +def git_repository_root(project_root: Path) -> Path | None: + result = run_git(project_root, ["rev-parse", "--show-toplevel"]) + if result.returncode != 0: + return None + value = result.stdout.strip() + return Path(value).resolve() if value else None + + +def git_origin(project_root: Path) -> str | None: + result = run_git(project_root, ["remote", "get-url", "origin"]) + if result.returncode != 0: + return None + remote_url = result.stdout.strip() + if not remote_url: + return None + remote_info = parse_origin_remote(remote_url, project_root) + return remote_info.sanitized_url if remote_info.valid and remote_info.sanitized_url else None + + +def git_head(project_root: Path) -> str | None: + result = run_git(project_root, ["rev-parse", "HEAD"]) + if result.returncode != 0: + return None + value = result.stdout.strip() + return value or None + + +def write_json_atomic(path: Path, payload: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temp_path = path.with_name(f".{path.name}.{os.getpid()}.tmp") + try: + temp_path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") + temp_path.chmod(0o600) + os.replace(temp_path, path) + path.chmod(0o600) + finally: + try: + temp_path.unlink() + except FileNotFoundError: + pass diff --git a/cli/python/base_setup/manifest_trust_guidance.py b/cli/python/base_setup/manifest_trust_guidance.py new file mode 100644 index 00000000..92084c3a --- /dev/null +++ b/cli/python/base_setup/manifest_trust_guidance.py @@ -0,0 +1,158 @@ +"""Trust review guidance shared by CLI commands and execution preflight.""" + +from __future__ import annotations + +import shlex +import sys +from typing import Any + +from .manifest_trust import TRUST_SCOPE_WARNING, ManifestCommandTrustIdentity, TrustStatus + + +def allow_command_text(identity: ManifestCommandTrustIdentity) -> str: + command = [ + "basectl", + "trust", + "allow", + identity.project_name, + "--manifest-sha256", + identity.manifest_sha256, + ] + if identity.test_requirements_sha256 is not None: + command.extend(("--test-requirements-sha256", identity.test_requirements_sha256)) + command.extend(("--workspace", str(identity.project_root.parent))) + return shlex.join(command) + + +def print_status_text(trust_status: TrustStatus, surfaces: tuple[str, ...]) -> None: + identity = trust_status.identity + if not surfaces: + print( + f"Manifest command trust is not required for project '{identity.project_name}': " + "the manifest declares no executable command surfaces." + ) + return + + if trust_status.is_allowed: + print(f"Manifest command trust is allowed for project '{identity.project_name}'.") + print_identity("Trusted identity", identity) + print() + print_trust_scope_warning() + return + + if trust_status.reason in {"manifest_changed", "test_requirements_changed"}: + print( + f"Manifest command trust is blocked for project '{identity.project_name}': " + "manifest or declared test requirements changed." + ) + if trust_status.reason == "test_requirements_changed": + print("Declared test requirements changed; review the requirements file before allowing commands.") + changed_project = (trust_status.changed_record or {}).get("project", {}) + if isinstance(changed_project, dict) and changed_project.get("manifest_sha256"): + print(f"Recorded Manifest SHA-256: {changed_project['manifest_sha256']}") + else: + print(f"Manifest command trust is blocked for project '{identity.project_name}'.") + print_identity("Current identity", identity) + print() + print_trust_scope_warning() + print() + print_review_guidance(identity, surfaces, stream=sys.stdout) + print() + print("Allow after review:") + print(f" {allow_command_text(identity)}") + + +def print_blocked_command_text( + trust_status: TrustStatus, + surfaces: tuple[str, ...], + *, + stream: Any, +) -> None: + identity = trust_status.identity + if trust_status.reason in {"manifest_changed", "test_requirements_changed"}: + print( + f"ERROR: Manifest command trust is blocked for project '{identity.project_name}': " + "manifest command or declared test requirements contract changed.", + file=stream, + ) + else: + print( + f"ERROR: Manifest-declared commands are not allowed for project " + f"'{identity.project_name}' on this machine.", + file=stream, + ) + print(f"Project root: {identity.project_root}", file=stream) + print(f"Manifest: {identity.manifest_path}", file=stream) + if trust_status.reason in {"manifest_changed", "test_requirements_changed"}: + changed_project = (trust_status.changed_record or {}).get("project", {}) + if isinstance(changed_project, dict) and changed_project.get("manifest_sha256"): + print(f"Recorded Manifest SHA-256: {changed_project['manifest_sha256']}", file=stream) + print(f"Manifest SHA-256: {identity.manifest_sha256}", file=stream) + if identity.test_requirements_sha256 is not None: + print(f"Test requirements SHA-256: {identity.test_requirements_sha256}", file=stream) + if identity.origin is not None: + print(f"Origin: {identity.origin}", file=stream) + print(file=stream) + print_trust_scope_warning(stream=stream) + print(file=stream) + print_review_guidance(identity, surfaces, stream=stream) + print(file=stream) + print("Allow after review:", file=stream) + print(f" {allow_command_text(identity)}", file=stream) + + +def print_review_guidance( + identity: ManifestCommandTrustIdentity, + surfaces: tuple[str, ...], + *, + stream: Any, +) -> None: + workspace = str(identity.project_root.parent) + print("Review first:", file=stream) + if "run" in surfaces: + print( + f" {shlex.join(['basectl', 'run', identity.project_name, '--list', '--workspace', workspace])}", + file=stream, + ) + if "build" in surfaces: + print( + f" {shlex.join(['basectl', 'build', identity.project_name, '--list', '--workspace', workspace])}", + file=stream, + ) + if "test" in surfaces: + print( + f" {shlex.join(['basectl', 'test', identity.project_name, '--dry-run', '--workspace', workspace])}", + file=stream, + ) + if "demo" in surfaces: + print( + f" {shlex.join(['basectl', 'demo', identity.project_name, '--dry-run', '--workspace', workspace])}", + file=stream, + ) + if "activate" in surfaces: + print( + f" Inspect activate.source entries in {identity.manifest_path} before running " + f"'basectl activate {identity.project_name}'.", + file=stream, + ) + + +def print_identity(title: str, identity: ManifestCommandTrustIdentity) -> None: + print(f"{title}:") + print(f" Project: {identity.project_name}") + print(f" Project root: {identity.project_root}") + print(f" Manifest: {identity.manifest_path}") + print(f" Manifest SHA-256: {identity.manifest_sha256}") + if identity.test_requirements_sha256 is not None: + print(f" Test requirements SHA-256: {identity.test_requirements_sha256}") + if identity.origin is not None: + print(f" Origin: {identity.origin}") + if identity.head is not None: + print(f" HEAD: {identity.head}") + + +def print_trust_scope_warning(*, stream: Any | None = None) -> None: + if stream is None: + stream = sys.stdout + print("Trust scope:", file=stream) + print(f" {TRUST_SCOPE_WARNING}", file=stream) diff --git a/cli/python/base_trust/engine.py b/cli/python/base_trust/engine.py index 21460c92..7168f2ce 100644 --- a/cli/python/base_trust/engine.py +++ b/cli/python/base_trust/engine.py @@ -3,17 +3,12 @@ import os import sys from pathlib import Path -from typing import Any +from typing import TYPE_CHECKING, Any import base_cli from base_cli_profile import base_cli_app from base_cli_adapters.history import base_version as read_base_version -from base_projects import engine as project_engine -from base_projects.project_discovery import Project -from base_projects.project_discovery import discover_projects_cached -from base_projects.project_discovery import read_project -from base_projects.workspace_context import resolve_workspace_root -from base_projects.workspace_scanner import ProjectDiscoveryError +from base_projects.workspace_errors import ProjectDiscoveryError from base_setup.manifest import read_manifest from base_setup.manifest_loader import ManifestError from .guidance import allow_command_text @@ -43,6 +38,9 @@ from .trust_store import trust_scope_payload from .trust_store import write_json_atomic # pylint: disable=unused-import +if TYPE_CHECKING: + from base_projects.project_discovery import Project + app = base_cli_app( name="base_trust", @@ -248,6 +246,9 @@ def workspace_status_command(ctx: base_cli.Context, workspace: str | None, outpu def workspace_status_projects(ctx: base_cli.Context, workspace: str | None) -> tuple[Project, ...]: + from base_projects.project_discovery import discover_projects_cached, read_project + from base_projects.workspace_context import resolve_workspace_root + workspace_root = resolve_workspace_root(ctx, workspace) projects_by_name = {project.name: project for project in discover_projects_cached(ctx, workspace_root)} @@ -266,6 +267,8 @@ def workspace_status_projects(ctx: base_cli.Context, workspace: str | None) -> t def workspace_status_active_project() -> Project | None: + from base_projects.project_discovery import read_project + if "BASE_TRUST_ACTIVE_PROJECT" in os.environ: active_name = os.environ.get("BASE_TRUST_ACTIVE_PROJECT") active_manifest = os.environ.get("BASE_TRUST_ACTIVE_PROJECT_MANIFEST") @@ -289,6 +292,8 @@ def resolve_trust_identity( project_name: str, workspace: str | None, ) -> ManifestCommandTrustIdentity: + from base_projects import engine as project_engine + project = project_engine.resolve_named_project(ctx, project_name, workspace) return compute_trust_identity_for_manifest(project.manifest_path) diff --git a/cli/python/base_trust/guidance.py b/cli/python/base_trust/guidance.py index 850ffeb7..199153d3 100644 --- a/cli/python/base_trust/guidance.py +++ b/cli/python/base_trust/guidance.py @@ -1,158 +1,21 @@ -"""Trust review guidance shared by CLI commands and execution preflight.""" +"""Compatibility exports for shared manifest trust guidance.""" from __future__ import annotations -import shlex -import sys -from typing import Any - -from .trust_store import TRUST_SCOPE_WARNING, ManifestCommandTrustIdentity, TrustStatus - - -def allow_command_text(identity: ManifestCommandTrustIdentity) -> str: - command = [ - "basectl", - "trust", - "allow", - identity.project_name, - "--manifest-sha256", - identity.manifest_sha256, - ] - if identity.test_requirements_sha256 is not None: - command.extend(("--test-requirements-sha256", identity.test_requirements_sha256)) - command.extend(("--workspace", str(identity.project_root.parent))) - return shlex.join(command) - - -def print_status_text(trust_status: TrustStatus, surfaces: tuple[str, ...]) -> None: - identity = trust_status.identity - if not surfaces: - print( - f"Manifest command trust is not required for project '{identity.project_name}': " - "the manifest declares no executable command surfaces." - ) - return - - if trust_status.is_allowed: - print(f"Manifest command trust is allowed for project '{identity.project_name}'.") - print_identity("Trusted identity", identity) - print() - print_trust_scope_warning() - return - - if trust_status.reason in {"manifest_changed", "test_requirements_changed"}: - print( - f"Manifest command trust is blocked for project '{identity.project_name}': " - "manifest or declared test requirements changed." - ) - if trust_status.reason == "test_requirements_changed": - print("Declared test requirements changed; review the requirements file before allowing commands.") - changed_project = (trust_status.changed_record or {}).get("project", {}) - if isinstance(changed_project, dict) and changed_project.get("manifest_sha256"): - print(f"Recorded Manifest SHA-256: {changed_project['manifest_sha256']}") - else: - print(f"Manifest command trust is blocked for project '{identity.project_name}'.") - print_identity("Current identity", identity) - print() - print_trust_scope_warning() - print() - print_review_guidance(identity, surfaces, stream=sys.stdout) - print() - print("Allow after review:") - print(f" {allow_command_text(identity)}") - - -def print_blocked_command_text( - trust_status: TrustStatus, - surfaces: tuple[str, ...], - *, - stream: Any, -) -> None: - identity = trust_status.identity - if trust_status.reason in {"manifest_changed", "test_requirements_changed"}: - print( - f"ERROR: Manifest command trust is blocked for project '{identity.project_name}': " - "manifest command or declared test requirements contract changed.", - file=stream, - ) - else: - print( - f"ERROR: Manifest-declared commands are not allowed for project " - f"'{identity.project_name}' on this machine.", - file=stream, - ) - print(f"Project root: {identity.project_root}", file=stream) - print(f"Manifest: {identity.manifest_path}", file=stream) - if trust_status.reason in {"manifest_changed", "test_requirements_changed"}: - changed_project = (trust_status.changed_record or {}).get("project", {}) - if isinstance(changed_project, dict) and changed_project.get("manifest_sha256"): - print(f"Recorded Manifest SHA-256: {changed_project['manifest_sha256']}", file=stream) - print(f"Manifest SHA-256: {identity.manifest_sha256}", file=stream) - if identity.test_requirements_sha256 is not None: - print(f"Test requirements SHA-256: {identity.test_requirements_sha256}", file=stream) - if identity.origin is not None: - print(f"Origin: {identity.origin}", file=stream) - print(file=stream) - print_trust_scope_warning(stream=stream) - print(file=stream) - print_review_guidance(identity, surfaces, stream=stream) - print(file=stream) - print("Allow after review:", file=stream) - print(f" {allow_command_text(identity)}", file=stream) - - -def print_review_guidance( - identity: ManifestCommandTrustIdentity, - surfaces: tuple[str, ...], - *, - stream: Any, -) -> None: - workspace = str(identity.project_root.parent) - print("Review first:", file=stream) - if "run" in surfaces: - print( - f" {shlex.join(['basectl', 'run', identity.project_name, '--list', '--workspace', workspace])}", - file=stream, - ) - if "build" in surfaces: - print( - f" {shlex.join(['basectl', 'build', identity.project_name, '--list', '--workspace', workspace])}", - file=stream, - ) - if "test" in surfaces: - print( - f" {shlex.join(['basectl', 'test', identity.project_name, '--dry-run', '--workspace', workspace])}", - file=stream, - ) - if "demo" in surfaces: - print( - f" {shlex.join(['basectl', 'demo', identity.project_name, '--dry-run', '--workspace', workspace])}", - file=stream, - ) - if "activate" in surfaces: - print( - f" Inspect activate.source entries in {identity.manifest_path} before running " - f"'basectl activate {identity.project_name}'.", - file=stream, - ) - - -def print_identity(title: str, identity: ManifestCommandTrustIdentity) -> None: - print(f"{title}:") - print(f" Project: {identity.project_name}") - print(f" Project root: {identity.project_root}") - print(f" Manifest: {identity.manifest_path}") - print(f" Manifest SHA-256: {identity.manifest_sha256}") - if identity.test_requirements_sha256 is not None: - print(f" Test requirements SHA-256: {identity.test_requirements_sha256}") - if identity.origin is not None: - print(f" Origin: {identity.origin}") - if identity.head is not None: - print(f" HEAD: {identity.head}") - - -def print_trust_scope_warning(*, stream: Any | None = None) -> None: - if stream is None: - stream = sys.stdout - print("Trust scope:", file=stream) - print(f" {TRUST_SCOPE_WARNING}", file=stream) +from base_setup.manifest_trust_guidance import ( + allow_command_text, + print_blocked_command_text, + print_identity, + print_review_guidance, + print_status_text, + print_trust_scope_warning, +) + +__all__ = [ + "allow_command_text", + "print_blocked_command_text", + "print_identity", + "print_review_guidance", + "print_status_text", + "print_trust_scope_warning", +] diff --git a/cli/python/base_trust/tests/test_import_boundaries.py b/cli/python/base_trust/tests/test_import_boundaries.py new file mode 100644 index 00000000..aa3ec43d --- /dev/null +++ b/cli/python/base_trust/tests/test_import_boundaries.py @@ -0,0 +1,49 @@ +from __future__ import annotations + +import os +import subprocess +import sys +import unittest +from pathlib import Path + + +REPO_ROOT = Path(__file__).resolve().parents[4] +CLI_PYTHON = REPO_ROOT / "cli" / "python" + + +def run_import_probe(source: str) -> subprocess.CompletedProcess[str]: + environment = dict(os.environ) + existing_pythonpath = environment.get("PYTHONPATH") + pythonpath = str(CLI_PYTHON) + if existing_pythonpath: + pythonpath = f"{pythonpath}{os.pathsep}{existing_pythonpath}" + environment["PYTHONPATH"] = pythonpath + return subprocess.run( + [sys.executable, "-c", source], + cwd=REPO_ROOT, + env=environment, + capture_output=True, + check=False, + text=True, + ) + + +class ImportBoundaryTests(unittest.TestCase): + def test_trust_engine_does_not_import_project_engine_or_onboarding(self) -> None: + result = run_import_probe( + "import sys; import base_trust.engine; " + "assert 'base_projects.engine' not in sys.modules; " + "assert 'base_projects.workspace_onboarding' not in sys.modules" + ) + self.assertEqual(result.returncode, 0, result.stderr) + + def test_project_engine_does_not_import_trust_engine(self) -> None: + result = run_import_probe( + "import sys; import base_projects.engine; " + "assert 'base_trust.engine' not in sys.modules" + ) + self.assertEqual(result.returncode, 0, result.stderr) + + +if __name__ == "__main__": + unittest.main() diff --git a/cli/python/base_trust/trust_store.py b/cli/python/base_trust/trust_store.py index dd73d5a9..312d9dea 100644 --- a/cli/python/base_trust/trust_store.py +++ b/cli/python/base_trust/trust_store.py @@ -1,322 +1,54 @@ -from __future__ import annotations - -import hashlib -import json -import os -from dataclasses import dataclass -from pathlib import Path -from typing import Any +"""Compatibility exports for the shared manifest trust contract.""" -from base_cli_adapters.history import format_timestamp, utc_now -from base_cli_adapters.paths import base_state_root -from base_setup.git_commands import run_git -from base_setup.git_remote_parse import parse_origin_remote -from base_setup.manifest import BaseManifest -from base_setup.manifest import read_manifest -from base_setup.test_requirements import requirements_file_digest +from __future__ import annotations -SCHEMA_VERSION = 1 -ALLOWED_COMMANDS = ["test", "run", "build", "demo", "activate"] -TRUST_RELATIVE_ROOT = Path("trust") / "manifest-commands" -TRUST_SCOPE_WARNING = ( - "Approval is bound to the base_manifest.yaml SHA-256 and declared local test " - "requirements SHA-256. Manifest or declared test requirements changes invalidate " - "approval; referenced scripts, direct executable files, Git HEAD, and uncommitted " - "working-tree changes are not independently verified or bound to approval. Re-review " - "those inputs before execution after repository changes." +from base_setup.git_commands import run_git # pylint: disable=unused-import +from base_setup.git_remote_parse import parse_origin_remote # pylint: disable=unused-import + +from base_setup.manifest_trust import ( + ALLOWED_COMMANDS, + SCHEMA_VERSION, + TRUST_RELATIVE_ROOT, + TRUST_SCOPE, + TRUST_SCOPE_WARNING, + ManifestCommandTrustIdentity, + ManifestCommandTrustStore, + TrustStatus, + compute_identity_key, + compute_trust_identity, + compute_trust_identity_for_manifest, + git_head, + git_origin, + git_repository_root, + identity_key_from_record, + manifest_command_surfaces, + manifest_command_surfaces_from_manifest, + sha256_file, + trust_change_reason, + trust_scope_payload, + write_json_atomic, ) -TRUST_SCOPE = { - "approval_basis": "base_manifest.yaml_sha256+declared_test_requirements_sha256", - "manifest_changes_invalidate": True, - "declared_test_requirements_changes_invalidate": True, - "referenced_script_changes_invalidate": False, - "direct_executable_file_changes_invalidate": False, - "git_head_changes_invalidate": False, - "working_tree_changes_invalidate": False, - "warning": TRUST_SCOPE_WARNING, -} - - -def trust_scope_payload() -> dict[str, object]: - return dict(TRUST_SCOPE) - - -@dataclass(frozen=True) -class ManifestCommandTrustIdentity: - project_name: str - project_root: Path - manifest_path: Path - manifest_sha256: str - identity_key: str - test_requirements_sha256: str | None = None - git_root: Path | None = None - origin: str | None = None - head: str | None = None - - def project_payload(self) -> dict[str, str]: - payload = { - "name": self.project_name, - "root": str(self.project_root), - "manifest": str(self.manifest_path), - "manifest_sha256": self.manifest_sha256, - } - if self.test_requirements_sha256 is not None: - payload["test_requirements_sha256"] = self.test_requirements_sha256 - if self.git_root is not None: - payload["git_root"] = str(self.git_root) - if self.origin is not None: - payload["origin"] = self.origin - if self.head is not None: - payload["head"] = self.head - return payload - - -@dataclass(frozen=True) -class TrustStatus: - status: str - reason: str - identity: ManifestCommandTrustIdentity - record: dict[str, Any] | None = None - changed_record: dict[str, Any] | None = None - - @property - def is_allowed(self) -> bool: - return self.status == "allowed" - - -class ManifestCommandTrustStore: - def __init__(self, home: Path | None = None) -> None: - self.root = base_state_root(home) / TRUST_RELATIVE_ROOT - - def record_path(self, identity: ManifestCommandTrustIdentity) -> Path: - return self.root / f"{identity.identity_key}.json" - - def read_record(self, identity: ManifestCommandTrustIdentity) -> dict[str, Any] | None: - path = self.record_path(identity) - try: - payload = json.loads(path.read_text(encoding="utf-8")) - except (OSError, UnicodeError, json.JSONDecodeError): - return None - if not isinstance(payload, dict) or payload.get("schema_version") != SCHEMA_VERSION: - return None - if identity_key_from_record(payload) != identity.identity_key: - return None - return payload - - def status(self, identity: ManifestCommandTrustIdentity) -> TrustStatus: - record = self.read_record(identity) - if record is not None: - return TrustStatus(status="allowed", reason="allowed", identity=identity, record=record) - - changed_record = self.find_changed_record(identity) - if changed_record is not None: - return TrustStatus( - status="blocked", - reason=trust_change_reason(identity, changed_record), - identity=identity, - changed_record=changed_record, - ) - - return TrustStatus(status="blocked", reason="not_allowed", identity=identity) - - def find_changed_record(self, identity: ManifestCommandTrustIdentity) -> dict[str, Any] | None: - if not self.root.is_dir(): - return None - for path in sorted(self.root.glob("*.json")): - try: - payload = json.loads(path.read_text(encoding="utf-8")) - except (OSError, UnicodeError, json.JSONDecodeError): - continue - if not isinstance(payload, dict) or payload.get("schema_version") != SCHEMA_VERSION: - continue - project = payload.get("project") - if not isinstance(project, dict): - continue - if project.get("root") == str(identity.project_root) and project.get("manifest") == str( - identity.manifest_path - ): - return payload - return None - - def allow( - self, - identity: ManifestCommandTrustIdentity, - *, - base_version: str | None, - allowed_at: str | None = None, - ) -> Path: - path = self.record_path(identity) - payload = { - "schema_version": SCHEMA_VERSION, - "allowed_at": allowed_at or format_timestamp(utc_now()), - "allowed_by": "local-user", - "base_version": base_version, - "project": identity.project_payload(), - "allowed_commands": ALLOWED_COMMANDS, - } - write_json_atomic(path, payload) - return path - - def revoke(self, identity: ManifestCommandTrustIdentity) -> bool: - removed = False - paths = {self.record_path(identity)} - try: - records = tuple(self.root.iterdir()) - except FileNotFoundError: - return False - for path in records: - if path.suffix != ".json": - continue - try: - payload = json.loads(path.read_text(encoding="utf-8")) - except (FileNotFoundError, UnicodeError, json.JSONDecodeError): - continue - if not isinstance(payload, dict) or payload.get("schema_version") != SCHEMA_VERSION: - continue - project = payload.get("project") - if isinstance(project, dict) and project.get("root") == str(identity.project_root) and project.get( - "manifest" - ) == str(identity.manifest_path): - # Unlink the record in this store, never a path from its contents. - paths.add(path) - - for path in sorted(paths): - try: - path.unlink() - except FileNotFoundError: - continue - removed = True - return removed - - -def manifest_command_surfaces(manifest_path: Path) -> tuple[str, ...]: - manifest = read_manifest(manifest_path.expanduser().resolve()) - return manifest_command_surfaces_from_manifest(manifest) - - -def manifest_command_surfaces_from_manifest(manifest: BaseManifest) -> tuple[str, ...]: - surfaces = [] - if manifest.test is not None: - surfaces.append("test") - if manifest.commands: - surfaces.append("run") - if manifest.build is not None and manifest.build.targets: - surfaces.append("build") - if manifest.demo is not None: - surfaces.append("demo") - if manifest.activate.source: - surfaces.append("activate") - return tuple(surfaces) - - -def compute_trust_identity_for_manifest(manifest_path: Path) -> ManifestCommandTrustIdentity: - manifest = read_manifest(manifest_path.expanduser().resolve()) - return compute_trust_identity(manifest) - - -def compute_trust_identity(manifest: BaseManifest) -> ManifestCommandTrustIdentity: - canonical_manifest = manifest.path.resolve() - project_root = canonical_manifest.parent.resolve() - manifest_sha256 = sha256_file(canonical_manifest) - requirements_sha256 = requirements_file_digest(manifest) - git_root = git_repository_root(project_root) - origin = git_origin(project_root) - head = git_head(project_root) - identity_key = compute_identity_key(project_root, canonical_manifest, manifest_sha256, requirements_sha256) - return ManifestCommandTrustIdentity( - project_name=manifest.project_name, - project_root=project_root, - manifest_path=canonical_manifest, - manifest_sha256=manifest_sha256, - test_requirements_sha256=requirements_sha256, - identity_key=identity_key, - git_root=git_root, - origin=origin, - head=head, - ) - - -def sha256_file(path: Path) -> str: - digest = hashlib.sha256() - with path.open("rb") as handle: - for chunk in iter(lambda: handle.read(1024 * 1024), b""): - digest.update(chunk) - return digest.hexdigest() - - -def compute_identity_key( - project_root: Path, - manifest_path: Path, - manifest_sha256: str, - test_requirements_sha256: str | None = None, -) -> str: - payload = "\0".join([str(project_root), str(manifest_path), manifest_sha256, test_requirements_sha256 or ""]) - return hashlib.sha256(payload.encode("utf-8")).hexdigest() - - -def identity_key_from_record(record: dict[str, Any]) -> str | None: - project = record.get("project") - if not isinstance(project, dict): - return None - root = project.get("root") - manifest = project.get("manifest") - digest = project.get("manifest_sha256") - if not all(isinstance(value, str) and value for value in (root, manifest, digest)): - return None - requirements_digest = project.get("test_requirements_sha256") - if requirements_digest is not None and not isinstance(requirements_digest, str): - return None - return compute_identity_key(Path(root), Path(manifest), digest, requirements_digest) - - -def trust_change_reason(identity: ManifestCommandTrustIdentity, record: dict[str, Any]) -> str: - project = record.get("project") - if not isinstance(project, dict): - return "manifest_changed" - if project.get("manifest_sha256") != identity.manifest_sha256: - return "manifest_changed" - if project.get("test_requirements_sha256") != identity.test_requirements_sha256: - return "test_requirements_changed" - return "manifest_changed" - - -def git_repository_root(project_root: Path) -> Path | None: - result = run_git(project_root, ["rev-parse", "--show-toplevel"]) - if result.returncode != 0: - return None - value = result.stdout.strip() - return Path(value).resolve() if value else None - - -def git_origin(project_root: Path) -> str | None: - result = run_git(project_root, ["remote", "get-url", "origin"]) - if result.returncode != 0: - return None - remote_url = result.stdout.strip() - if not remote_url: - return None - remote_info = parse_origin_remote(remote_url, project_root) - return remote_info.sanitized_url if remote_info.valid and remote_info.sanitized_url else None - - -def git_head(project_root: Path) -> str | None: - result = run_git(project_root, ["rev-parse", "HEAD"]) - if result.returncode != 0: - return None - value = result.stdout.strip() - return value or None - -def write_json_atomic(path: Path, payload: dict[str, Any]) -> None: - path.parent.mkdir(parents=True, exist_ok=True) - temp_path = path.with_name(f".{path.name}.{os.getpid()}.tmp") - try: - temp_path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") - temp_path.chmod(0o600) - os.replace(temp_path, path) - path.chmod(0o600) - finally: - try: - temp_path.unlink() - except FileNotFoundError: - pass +__all__ = [ + "ALLOWED_COMMANDS", + "SCHEMA_VERSION", + "TRUST_RELATIVE_ROOT", + "TRUST_SCOPE", + "TRUST_SCOPE_WARNING", + "ManifestCommandTrustIdentity", + "ManifestCommandTrustStore", + "TrustStatus", + "compute_identity_key", + "compute_trust_identity", + "compute_trust_identity_for_manifest", + "git_head", + "git_origin", + "git_repository_root", + "identity_key_from_record", + "manifest_command_surfaces", + "manifest_command_surfaces_from_manifest", + "sha256_file", + "trust_change_reason", + "trust_scope_payload", + "write_json_atomic", +] From 62bef9707cfc980fe410eb13cd15057e5ed14af7 Mon Sep 17 00:00:00 2001 From: Ramesh Padmanabhaiah <22363102+codeforester@users.noreply.github.com> Date: Mon, 28 Sep 2026 20:53:31 +0530 Subject: [PATCH 2/3] test(python): tighten trust import boundaries --- .../tests/test_compatibility_facades.py | 38 +++++++++++++++---- cli/python/base_trust/engine.py | 2 + .../tests/test_import_boundaries.py | 13 ++++++- cli/python/base_trust/trust_store.py | 3 -- 4 files changed, 43 insertions(+), 13 deletions(-) diff --git a/cli/python/base_setup/tests/test_compatibility_facades.py b/cli/python/base_setup/tests/test_compatibility_facades.py index bc0c2877..3daa09a7 100644 --- a/cli/python/base_setup/tests/test_compatibility_facades.py +++ b/cli/python/base_setup/tests/test_compatibility_facades.py @@ -1,7 +1,8 @@ from __future__ import annotations -from pathlib import Path import unittest +from pathlib import Path +from unittest import mock from base_devenv import report as devenv_report_impl from base_devcontainer import export as devcontainer_export_impl @@ -11,8 +12,8 @@ from base_setup import devcontainer_export from base_setup import engine as setup_engine from base_setup import manifest_checks +from base_setup import manifest_trust from base_setup import setup_reconcile -from base_trust import trust_store class CompatibilityFacadeTests(unittest.TestCase): @@ -84,12 +85,33 @@ def test_setup_engine_uses_focused_devenv_package(self) -> None: self.assertNotIn("def build_devenv_report", facade_source) self.assertNotIn("@dataclass", facade_source) - def test_trust_store_uses_focused_git_modules(self) -> None: - trust_store_source = Path(trust_store.__file__).read_text(encoding="utf-8") - - self.assertNotIn("from base_setup import git_remote", trust_store_source) - self.assertIn("from base_setup.git_commands import run_git", trust_store_source) - self.assertIn("from base_setup.git_remote_parse import parse_origin_remote", trust_store_source) + def test_manifest_trust_git_identity_helpers_use_focused_modules(self) -> None: + project_root = Path("/tmp/project") + with mock.patch.object(manifest_trust, "run_git") as run_git, mock.patch.object( + manifest_trust, "parse_origin_remote" + ) as parse_origin_remote: + resolved_project_root = project_root.resolve() + run_git.return_value = mock.Mock(returncode=0, stdout=f"{resolved_project_root}\n") + self.assertEqual(manifest_trust.git_repository_root(project_root), resolved_project_root) + + run_git.return_value = mock.Mock( + returncode=0, + stdout="https://github.com/basefoundry/base.git\n", + ) + parse_origin_remote.return_value = mock.Mock( + valid=True, + sanitized_url="https://github.com/basefoundry/base.git", + ) + self.assertEqual( + manifest_trust.git_origin(project_root), + "https://github.com/basefoundry/base.git", + ) + + run_git.assert_any_call(project_root, ["rev-parse", "--show-toplevel"]) + run_git.assert_any_call(project_root, ["remote", "get-url", "origin"]) + parse_origin_remote.assert_called_once_with( + "https://github.com/basefoundry/base.git", project_root + ) if __name__ == "__main__": diff --git a/cli/python/base_trust/engine.py b/cli/python/base_trust/engine.py index 7168f2ce..f4259f64 100644 --- a/cli/python/base_trust/engine.py +++ b/cli/python/base_trust/engine.py @@ -8,6 +8,8 @@ import base_cli from base_cli_profile import base_cli_app from base_cli_adapters.history import base_version as read_base_version +# This exception-only leaf has no project or trust dependencies, so it is safe +# at module scope while the behavioral project modules below remain lazy. from base_projects.workspace_errors import ProjectDiscoveryError from base_setup.manifest import read_manifest from base_setup.manifest_loader import ManifestError diff --git a/cli/python/base_trust/tests/test_import_boundaries.py b/cli/python/base_trust/tests/test_import_boundaries.py index aa3ec43d..3863ad31 100644 --- a/cli/python/base_trust/tests/test_import_boundaries.py +++ b/cli/python/base_trust/tests/test_import_boundaries.py @@ -32,8 +32,10 @@ class ImportBoundaryTests(unittest.TestCase): def test_trust_engine_does_not_import_project_engine_or_onboarding(self) -> None: result = run_import_probe( "import sys; import base_trust.engine; " - "assert 'base_projects.engine' not in sys.modules; " - "assert 'base_projects.workspace_onboarding' not in sys.modules" + "assert not any(module_name in sys.modules for module_name in (" + "'base_projects.engine', 'base_projects.project_discovery', " + "'base_projects.workspace_onboarding', 'base_projects.workspace_context', " + "'base_projects.workspace_scanner'))" ) self.assertEqual(result.returncode, 0, result.stderr) @@ -44,6 +46,13 @@ def test_project_engine_does_not_import_trust_engine(self) -> None: ) self.assertEqual(result.returncode, 0, result.stderr) + def test_shared_trust_modules_do_not_import_project_modules(self) -> None: + result = run_import_probe( + "import sys; import base_setup.manifest_trust; import base_setup.manifest_trust_guidance; " + "assert not any(name == 'base_projects' or name.startswith('base_projects.') for name in sys.modules)" + ) + self.assertEqual(result.returncode, 0, result.stderr) + if __name__ == "__main__": unittest.main() diff --git a/cli/python/base_trust/trust_store.py b/cli/python/base_trust/trust_store.py index 312d9dea..413c55c9 100644 --- a/cli/python/base_trust/trust_store.py +++ b/cli/python/base_trust/trust_store.py @@ -2,9 +2,6 @@ from __future__ import annotations -from base_setup.git_commands import run_git # pylint: disable=unused-import -from base_setup.git_remote_parse import parse_origin_remote # pylint: disable=unused-import - from base_setup.manifest_trust import ( ALLOWED_COMMANDS, SCHEMA_VERSION, From a5f22567078ecc29bca759e4310a9e38761003de Mon Sep 17 00:00:00 2001 From: Ramesh Padmanabhaiah <22363102+codeforester@users.noreply.github.com> Date: Mon, 28 Sep 2026 23:16:10 +0530 Subject: [PATCH 3/3] test(python): scan all production import boundaries --- .../tests/test_import_boundaries.py | 46 +++++++++++++++++++ 1 file changed, 46 insertions(+) diff --git a/cli/python/base_trust/tests/test_import_boundaries.py b/cli/python/base_trust/tests/test_import_boundaries.py index 3863ad31..06f51dec 100644 --- a/cli/python/base_trust/tests/test_import_boundaries.py +++ b/cli/python/base_trust/tests/test_import_boundaries.py @@ -1,5 +1,6 @@ from __future__ import annotations +import ast import os import subprocess import sys @@ -9,6 +10,17 @@ REPO_ROOT = Path(__file__).resolve().parents[4] CLI_PYTHON = REPO_ROOT / "cli" / "python" +PRODUCTION_PACKAGE_ROOTS = ( + REPO_ROOT / "cli" / "python" / "base_setup", + REPO_ROOT / "cli" / "python" / "base_trust", +) + +# `workspace_errors` contains exception definitions only and has no project or +# trust dependencies. Keep this one documented exception explicit while making +# every other production module-scope base_projects import fail the test. +ALLOWED_MODULE_SCOPE_PROJECT_IMPORTS = { + ("cli/python/base_trust/engine.py", "base_projects.workspace_errors"), +} def run_import_probe(source: str) -> subprocess.CompletedProcess[str]: @@ -28,7 +40,41 @@ def run_import_probe(source: str) -> subprocess.CompletedProcess[str]: ) +def production_python_files() -> list[Path]: + return sorted( + path + for package_root in PRODUCTION_PACKAGE_ROOTS + for path in package_root.rglob("*.py") + if "tests" not in path.parts + ) + + +def module_scope_project_imports(path: Path) -> set[str]: + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + imports: set[str] = set() + for node in tree.body: + if isinstance(node, ast.Import): + imports.update( + alias.name + for alias in node.names + if alias.name == "base_projects" or alias.name.startswith("base_projects.") + ) + elif isinstance(node, ast.ImportFrom): + module = node.module or "" + if module == "base_projects" or module.startswith("base_projects."): + imports.add(module) + return imports + + class ImportBoundaryTests(unittest.TestCase): + def test_production_module_scope_project_imports_are_explicit(self) -> None: + actual = { + (path.relative_to(REPO_ROOT).as_posix(), module) + for path in production_python_files() + for module in module_scope_project_imports(path) + } + self.assertEqual(actual, ALLOWED_MODULE_SCOPE_PROJECT_IMPORTS) + def test_trust_engine_does_not_import_project_engine_or_onboarding(self) -> None: result = run_import_probe( "import sys; import base_trust.engine; "