diff --git a/docs/adr/0106-source-bound-connected-provider-loading.md b/docs/adr/0106-source-bound-connected-provider-loading.md new file mode 100644 index 000000000..43ed7f50f --- /dev/null +++ b/docs/adr/0106-source-bound-connected-provider-loading.md @@ -0,0 +1,78 @@ +--- +status: proposed +date: 2026-10-30 +related: 0078, 0099 +--- + +# 0106: Bind connected providers to source-only loading + +## Context and Problem Statement + +Connected-read reviews bind a selected provider and source, but ordinary discovery imports plugin entry points before +recording their origin. An in-place update can therefore make disk contents differ from code already executing. Python's +normal timestamp bytecode cache can also load older code after a same-size, same-timestamp source replacement. A plugin +cannot establish this boundary by checking its own source after its initializer has already run. + +## Considered Options + +- **Host-controlled source-only provider loading with a closed graph** (proposed). +- **Plugin self-reported file hashes**: the verifier itself may already be stale. +- **Assume installed files or version metadata identify loaded code**: neither proves which code was executed. +- **Rely on managed immutable deployments**: desirable, but not yet enforced for ordinary installed plugins. + +## Decision Outcome + +Chosen option: **host-controlled source-only provider loading with a closed graph**, because a trusted host must capture +and compile the bytes before provider code runs. Ordinary plugin discovery remains unchanged. A connected provider is +eligible only through a separately reviewed admission boundary that binds the exact owner-selected provider, source +graph, account and operation. Installed `heinzel.connected_sources` metadata may request capture only when it matches +the same distribution's ordinary module entry point; it does not assert trust, grant access, or select additional runtime +dependencies. The host checks the installed manifest and owns the trusted runtime baseline and the process-lifetime +capture catalog. Ordinary boot enables the guard; providers lacking a verifiable declaration remain available to ordinary +heads but cannot use the guarded connected-read path. Unknown dependencies, preloaded code, native extensions, unsupported loaders, graph +changes and ambiguous import ownership fail closed. The host rechecks the graph and grant before each operation and +fences later in-flight calls when changes are observed. A running request cannot be retroactively undone. + +Captured plugin code may to import only already-loaded host runtime modules and attributes sealed +after builtin discovery. It checks standard-library origins and loaders, the installed Pydantic manifest, and the +trusted running core's source root. A scoped import function returns the exact sealed objects or loads the plugin's +captured package; missing runtime dependencies are never loaded on the plugin's behalf. Sealed objects and specs are +rechecked before use. An ineligible plugin can still use ordinary discovery, but cannot use the source-guarded +connected-read path. An embedder constructing `Kernel` without a source catalog retains responsibility for provider +trust and does not obtain this guard. + +The public `captured_source_active()` predicate reports whether its exact calling module is executing under the host's +source loader with a verified snapshot and sealed runtime. It is false outside that initialization context and does +not grant permission or establish publisher trust. A plugin may use it to omit its captured-read service on older +cores or ordinary loading; current source binding and human approval remain separate requirements. + +The fixed host initialization imports `email.policy` for general MIME parsing before external plugin discovery. That +bootstrap and the Python interpreter's own imports are part of the trusted host runtime, not verified plugin code; +the source guard does not claim to prevent shadowing before that trusted initialization. No plugin declaration can +extend this bootstrap. A missing runtime module or attribute makes the plugin ineligible for guarded reads. + +A source digest establishes local continuity, not publisher authentication, immunity to same-user interference, or +provider account authority. Isolated provider processes or verified immutable deployments may be needed to preserve +ordinary module identity while applying this boundary. Do not reload or replace live plugin modules to obtain a digest. + +### Consequences + +This adds startup, compatibility and dependency-inventory work. Existing connected providers must add a matching +metadata declaration and use a verifiable non-editable installed distribution before guarded connected reads work; +unsupported plugins continue to work through ordinary heads but cannot receive this authority. Default boot enables +source capture for matching installed declarations; manually constructed kernels without the catalog remain an explicit +embedding trust boundary. The loader supports bounded pure-Python package trees and an already-loaded host runtime. +Native plugin extensions and unloaded runtime dependencies remain ineligible for guarded reads. + +This is a continuity mechanism for owner-selected plugin code, not a Python sandbox or publisher attestation. Static +import inspection and the scoped importer bind ordinary imports to captured bytes or sealed runtime objects. Python +reflection and dynamic imports remain capabilities of the already trusted installed code; the regression demonstrating +a dynamic helper import records that limit. Do not present the digest or capture predicate as proof that arbitrary +plugin code is safe, immutable against a same-user adversary, or authorized to use an account. Exact human-approved +account, operation and source grants remain necessary before a connected read. + +### Confirmation + +Focused tests exercise source-only compilation despite stale bytecode, refusal of already loaded packages and detection +of source changes. Independent review must establish closed dependency graphs, verified provider identity, current +binding checks and cancellation semantics before any real provider is admitted. diff --git a/docs/adr/README.md b/docs/adr/README.md index 517d8166f..f713cac7a 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -196,3 +196,4 @@ of inventing them. Two factual corrections came with it: ADR 0004 names the plug | [0098](0098-pages-showcases-the-product-and-discovers-public-plugin-metadata.md) | Pages showcases the product and discovers public plugin metadata | proposed | 2026-10-30 | | [0099](0099-reviewed-connected-reads-and-local-agent-notifications.md) | Review connected reads and local agent notifications separately | proposed | 2026-10-30 | | [0104](0104-reviewed-item-evidence-for-local-notifications.md) | Bind notification deduplication to observed source items | proposed | 2026-10-30 | +| [0106](0106-source-bound-connected-provider-loading.md) | Bind connected providers to source-only loading | proposed | 2026-10-30 | diff --git a/src/heinzel/__init__.py b/src/heinzel/__init__.py index 07c13b7fe..771272964 100644 --- a/src/heinzel/__init__.py +++ b/src/heinzel/__init__.py @@ -184,6 +184,7 @@ WidgetLet, Window, capability_read_schema, + captured_source_active, clean_text, draft_reporting_lines, free_busy_people, @@ -366,6 +367,7 @@ "Window", "__version__", "capability_read_schema", + "captured_source_active", "clean_text", "draft_reporting_lines", "free_busy_people", diff --git a/src/heinzel/kernel/__init__.py b/src/heinzel/kernel/__init__.py index a2f9f7327..7a640f15b 100644 --- a/src/heinzel/kernel/__init__.py +++ b/src/heinzel/kernel/__init__.py @@ -22,6 +22,7 @@ from heinzel.kernel._bounded_process import ProcessExecute, ProcessOutcome, run_bounded_process, signal_process_group from heinzel.kernel._capabilities import CAPABILITIES, MODES, NO_ADAPTER, Capability, Mode, Requirement from heinzel.kernel._capability_schema import capability_read_schema +from heinzel.kernel._captured_source import captured_source_active from heinzel.kernel._chat import ActionGate, ProposedCall from heinzel.kernel._chat_services import ( CheckInChatContext, @@ -241,6 +242,7 @@ StreamProvider, ThemeChoice, ) +from heinzel.kernel._source_loading import SourceGuardedConnectedService from heinzel.kernel._telemetry import ( HEINZEL_ID, VERBS, @@ -499,6 +501,7 @@ "SoundStream", "Soundscape", "SoundscapeLet", + "SourceGuardedConnectedService", "Span", "Status", "StatusItem", @@ -545,6 +548,7 @@ "Window", "boot", "capability_read_schema", + "captured_source_active", "check_repository_url", "check_target", "clean_text", diff --git a/src/heinzel/kernel/_boot.py b/src/heinzel/kernel/_boot.py index 5e607802d..35f0da951 100644 --- a/src/heinzel/kernel/_boot.py +++ b/src/heinzel/kernel/_boot.py @@ -1,14 +1,22 @@ """Boot: load the configuration, discover every module and build the kernel. What heads call on start-up.""" +import email.policy + from heinzel.kernel._config import KERNEL_SECTION, Config from heinzel.kernel._discovery import discover from heinzel.kernel._kernel import Kernel from heinzel.kernel._settings import Settings +from heinzel.kernel._source_loading import SourceDiscovery + +_installed_sources = SourceDiscovery(installed=True) +# Fixed host bootstrap for MIME parsing, independent of any plugin declaration. +_host_runtime = (email.policy,) -def boot(settings: Settings | None = None) -> Kernel: +def boot(settings: Settings | None = None, *, source_discovery: "SourceDiscovery | None" = None) -> Kernel: """Discover every module and return a ready kernel.""" + source_discovery = source_discovery or _installed_sources config = Config.load() settings = settings or config.settings(Settings, KERNEL_SECTION) - records, errors = discover(plugins=settings.plugins_enabled) - return Kernel(settings, records, errors, config) + records, errors = discover(plugins=settings.plugins_enabled, source_discovery=source_discovery) + return Kernel(settings, records, errors, config, source_discovery=source_discovery) diff --git a/src/heinzel/kernel/_captured_source.py b/src/heinzel/kernel/_captured_source.py new file mode 100644 index 000000000..9dcc99422 --- /dev/null +++ b/src/heinzel/kernel/_captured_source.py @@ -0,0 +1,36 @@ +"""Host execution marker for source-captured plugin initialization.""" + +import inspect +from collections.abc import Callable, Generator +from contextlib import contextmanager +from contextvars import ContextVar + +_active: ContextVar[tuple[dict[str, object], Callable[[], bool]] | None] = ContextVar("captured_source", default=None) + + +@contextmanager +def executing_captured_source(module_globals: dict[str, object], verify: Callable[[], bool]) -> Generator[None]: + """Set the host-owned marker only for one captured module execution.""" + token = _active.set((module_globals, verify)) + try: + yield + finally: + _active.reset(token) + + +def captured_source_active() -> bool: + """Whether the calling module is currently executing under the host's captured source loader. + + This reports source-loading context only. It does not grant access, establish publisher trust, or approve a read. + """ + marker = _active.get() + frame = inspect.currentframe() + try: + if marker is None or frame is None or frame.f_back is None or frame.f_back.f_globals is not marker[0]: + return False + try: + return marker[1]() + except Exception: # noqa: BLE001 - an unavailable source must not claim capture + return False + finally: + del frame diff --git a/src/heinzel/kernel/_discovery.py b/src/heinzel/kernel/_discovery.py index d918a611a..7dc1411dc 100644 --- a/src/heinzel/kernel/_discovery.py +++ b/src/heinzel/kernel/_discovery.py @@ -11,6 +11,7 @@ from collections.abc import Iterator from importlib.metadata import Distribution, PackageNotFoundError, distribution, entry_points from types import ModuleType +from typing import TYPE_CHECKING from heinzel import __version__ from heinzel.kernel._installation import CORE_DISTRIBUTION @@ -22,10 +23,15 @@ """Project-URL labels naming the source repository, normalized as in PEP 753, most specific first.""" ENTRY_POINT_GROUP = "heinzel.modules" +if TYPE_CHECKING: + from heinzel.kernel._source_loading import SourceDiscovery + logger = logging.getLogger(__name__) -def discover(*, plugins: bool = True) -> tuple[list[ModuleRecord], dict[str, str]]: +def discover( + *, plugins: bool = True, source_discovery: "SourceDiscovery | None" = None +) -> tuple[list[ModuleRecord], dict[str, str]]: """Find built-in modules and, unless disabled, plugin modules. A plugin that fails to import is reported in the returned errors and skipped, so one broken plugin never takes @@ -47,6 +53,15 @@ def discover(*, plugins: bool = True) -> tuple[list[ModuleRecord], dict[str, str builtin=False, url=repository_of(dist) if dist else None, ) + if source_discovery is not None: + try: + captured = source_discovery.try_capture(entry_point, origin) + except Exception as exc: # noqa: BLE001 - never reexecute a failed captured initializer + _skip(errors, entry_point.name, f"{type(exc).__name__}: {exc}") + continue + if captured is not None: + records.extend(captured.records) + continue try: target = entry_point.load() except Exception as exc: # noqa: BLE001 - isolate broken plugins diff --git a/src/heinzel/kernel/_kernel.py b/src/heinzel/kernel/_kernel.py index c070443ef..b75800b67 100644 --- a/src/heinzel/kernel/_kernel.py +++ b/src/heinzel/kernel/_kernel.py @@ -6,7 +6,7 @@ from dataclasses import dataclass, replace from functools import partial from pathlib import Path -from typing import Annotated, Any, cast, get_args, get_origin +from typing import TYPE_CHECKING, Annotated, Any, cast, get_args, get_origin import pluggy from pydantic import ConfigDict, ValidationError, validate_call @@ -51,6 +51,9 @@ from heinzel.kernel._settings import Settings from heinzel.kernel._spec import PROJECT_NAME, Spec +if TYPE_CHECKING: + from heinzel.kernel._source_loading import SourceDiscovery + logger = logging.getLogger(__name__) ADAPTER_ONLY_LETS = (ProviderLet, RequirementLet, SettingsLet, CredentialLet, LayerLet) @@ -106,6 +109,7 @@ def __init__( records: Iterable[ModuleRecord] = (), errors: Mapping[str, str] | None = None, config: Config | None = None, + source_discovery: "SourceDiscovery | None" = None, ) -> None: """Register modules. Disabled modules (and their submodules) are blocked, not registered. @@ -114,6 +118,7 @@ def __init__( records: Modules to register, typically from `heinzel.kernel.discover`. errors: Plugins that failed to load, by entry point name, for reporting. config: Configuration files and their content; none (defaults and environment only) if not given. + source_discovery: Optional host-owned source continuity catalog. A plugin module that collides with a registered one — same qualified name, a submodule named like one of its parent's operations, or a service type provided twice — is skipped and reported in `errors`, so one bad @@ -123,6 +128,7 @@ def __init__( HeinzelError: If built-in modules collide. """ self.config = config or Config.empty() + self.source_discovery = source_discovery self.settings = settings or self.config.settings(Settings, KERNEL_SECTION) self.errors: dict[str, str] = dict(errors or {}) self._manager = pluggy.PluginManager(PROJECT_NAME) diff --git a/src/heinzel/kernel/_source_loading.py b/src/heinzel/kernel/_source_loading.py new file mode 100644 index 000000000..cc0493d76 --- /dev/null +++ b/src/heinzel/kernel/_source_loading.py @@ -0,0 +1,887 @@ +"""Host source continuity for declared connected providers; separate from access and publisher trust.""" + +import ast +import builtins +import hashlib +import importlib.abc +import importlib.util +import io +import json +import os +import stat +import sys +import sysconfig +import zipfile +from collections.abc import Callable, Iterator, Mapping, Sequence +from dataclasses import dataclass +from importlib import machinery +from importlib.machinery import ModuleSpec +from importlib.metadata import EntryPoint, distribution +from importlib.resources.abc import Traversable, TraversableResources +from pathlib import Path +from types import MappingProxyType, ModuleType +from typing import cast + +from pydantic import JsonValue + +from heinzel.kernel._captured_source import executing_captured_source +from heinzel.kernel._connected_read import ( + ConnectedReadLease, + ConnectedReadMethod, + ConnectedReadReview, + ConnectedReadService, +) +from heinzel.kernel._discovery import walk +from heinzel.kernel._errors import HeinzelError +from heinzel.kernel._kernel import Kernel, ModuleRecord, Origin + +_MAX_FILES = 256 +_MAX_BYTES = 2_000_000 + + +def _read_source(path: Path) -> bytes: + """Read stable regular source bytes. + + Raises: + HeinzelError: If source is unsuitable or changed while read. + """ + fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK) + with os.fdopen(fd, "rb") as source: + before = os.fstat(source.fileno()) + if not stat.S_ISREG(before.st_mode) or before.st_size > _MAX_BYTES: + msg = "Source file is unsupported" + raise HeinzelError(msg) + data = source.read(_MAX_BYTES + 1) + after = os.fstat(source.fileno()) + if len(data) != before.st_size or ( + before.st_dev, + before.st_ino, + before.st_size, + before.st_mtime_ns, + before.st_ctime_ns, + ) != (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns, after.st_ctime_ns): + msg = "Source changed while inspected" + raise HeinzelError(msg) + return data + + +def _validate_path(path: Path) -> None: + """Refuse linked files and native code. + + Raises: + HeinzelError: If the path is unsupported. + """ + if path.is_symlink(): + msg = "Source package contains a linked file" + raise HeinzelError(msg) + if path.suffix in {".so", ".pyd", ".dll", ".dylib"}: + msg = "Native plugin code is unsupported" + raise HeinzelError(msg) + + +def _snapshot(root: Path) -> dict[str, bytes]: + """Capture a bounded Python-only package tree. + + Raises: + HeinzelError: If the tree is unavailable, unsupported or changed. + """ + if root.is_symlink() or not root.is_dir(): + msg = "Source package is unavailable" + raise HeinzelError(msg) + result: dict[str, bytes] = {} + for directory, names, files in os.walk(root, followlinks=False): + base = Path(directory) + for name in names: + if (base / name).is_symlink(): + msg = "Source package contains a linked directory" + raise HeinzelError(msg) + names[:] = [name for name in names if name != "__pycache__"] + for name in files: + path = base / name + _validate_path(path) + data = _read_source(path) + result[path.relative_to(root).as_posix()] = data + if len(result) > _MAX_FILES or sum(map(len, result.values())) > _MAX_BYTES: + msg = "Source package exceeds bounds" + raise HeinzelError(msg) + if "__init__.py" not in result: + msg = "Source package has no initializer" + raise HeinzelError(msg) + return result + + +@dataclass(frozen=True) +class SourceSnapshot: + """Immutable observed source bytes and a deterministic continuity digest.""" + + root: Path + files: tuple[tuple[str, bytes], ...] + + @classmethod + def capture(cls, root: Path) -> "SourceSnapshot": + return cls(root, tuple(sorted(_snapshot(root).items()))) + + def verify(self) -> None: + if tuple(sorted(_snapshot(self.root).items())) != self.files: + msg = "Plugin source changed; restart is required" + raise HeinzelError(msg) + + def digest(self) -> str: + digest = hashlib.sha256() + for name, data in self.files: + digest.update(len(name.encode()).to_bytes(4, "big")) + digest.update(name.encode()) + digest.update(len(data).to_bytes(8, "big")) + digest.update(data) + return digest.hexdigest() + + +def _check_dependencies(snapshot: SourceSnapshot, package: str, approved: frozenset[str]) -> None: + """Refuse syntactically declared imports outside the host-approved runtime set. + + Raises: + HeinzelError: If a declared external import is not in the approved set. + """ + allowed = set(sys.stdlib_module_names) | set(approved) | {package} + for name, data in snapshot.files: + if not name.endswith(".py"): + continue + tree = ast.parse(data, filename=name) + for node in ast.walk(tree): + if isinstance(node, ast.Import): + dependencies = [alias.name.split(".")[0] for alias in node.names] + elif isinstance(node, ast.ImportFrom) and node.level == 0 and node.module: + dependencies = [node.module.split(".")[0]] + else: + continue + if any(dependency not in allowed for dependency in dependencies): + msg = "Plugin declares an unapproved runtime dependency" + raise HeinzelError(msg) + + +class SealedRuntime: + """Already-loaded host runtime objects available to a captured plugin's static imports. + + This is not a sandbox: trusted runtime code and Python reflection remain outside the source graph. + """ + + def __init__(self, snapshot: SourceSnapshot, package: str, approved: frozenset[str]) -> None: + self.package = package + self.modules: dict[str, ModuleType] = {} + self.attributes: dict[tuple[str, str], object] = {} + self.specs: dict[str, tuple[object, object, object, object]] = {} + self.paths: dict[str, tuple[str, ...] | None] = {} + self.stdlib = Path(sysconfig.get_path("stdlib")).resolve() + self.site_roots = tuple(Path(sysconfig.get_path(key)).resolve() for key in ("purelib", "platlib")) + core = sys.modules["heinzel"] + core_file = vars(core).get("__file__") + if not isinstance(core_file, str): + msg = "The host core runtime has no verifiable origin" + raise HeinzelError(msg) + self.core_root = Path(core_file).resolve().parent + dist = distribution("pydantic") + if dist.files is None: + msg = "The host pydantic manifest is unavailable" + raise HeinzelError(msg) + locations = [dist.locate_file(item) for item in dist.files] + if any(not isinstance(location, Path) for location in locations): + msg = "The host pydantic manifest has unsupported locations" + raise HeinzelError(msg) + self.pydantic_files = {location.resolve() for location in locations if isinstance(location, Path)} + self.approved = approved + self._scan(snapshot) + + def _seal(self, name: str) -> ModuleType: + module = sys.modules.get(name) + if not isinstance(module, ModuleType): + msg = f"A declared host runtime module is not already loaded: {name}" + raise HeinzelError(msg) + spec = vars(module).get("__spec__") + if not isinstance(spec, ModuleSpec) or sys.modules.get(spec.name) is not module: + msg = f"A declared host runtime module has an invalid spec: {name}" + raise HeinzelError(msg) + origin = spec.origin + if origin in {"built-in", "frozen"}: + expected = machinery.BuiltinImporter if origin == "built-in" else machinery.FrozenImporter + canonical = expected.find_spec(spec.name) + if cast("object", spec.loader) is not expected or canonical is None or canonical.origin != origin: + msg = "A host runtime loader changed" + raise HeinzelError(msg) + else: + if not isinstance(origin, str) or vars(module).get("__file__") != origin: + msg = "A host runtime origin changed" + raise HeinzelError(msg) + path = Path(origin).resolve() + top = spec.name.split(".")[0] + valid = ( + (top == "heinzel" and path.is_relative_to(self.core_root)) + or (top == "pydantic" and path in self.pydantic_files) + or ( + top in sys.stdlib_module_names + and path.is_relative_to(self.stdlib) + and not any(path.is_relative_to(site) for site in self.site_roots) + and "site-packages" not in path.relative_to(self.stdlib).parts + and "dist-packages" not in path.relative_to(self.stdlib).parts + ) + ) + if not valid or not isinstance( + spec.loader, + ( + machinery.SourceFileLoader, + machinery.SourcelessFileLoader, + machinery.ExtensionFileLoader, + ), + ): + msg = f"A declared host runtime origin is unverified: {name}" + raise HeinzelError(msg) + self.modules[name] = module + self.specs[name] = (spec, spec.loader, spec.origin, vars(module).get("__file__")) + search = vars(module).get("__path__") + if search is not None: + if not isinstance(search, (list, tuple)): + msg = "A host runtime package path is unsupported" + raise HeinzelError(msg) + paths = tuple(cast("Sequence[object]", search)) + if ( + len(paths) != 1 + or not isinstance(paths[0], str) + or Path(paths[0]).resolve() != Path(origin).resolve().parent + ): + msg = "A host runtime package path changed" + raise HeinzelError(msg) + self.paths[name] = (paths[0],) + else: + self.paths[name] = None + return module + + def _scan(self, snapshot: SourceSnapshot) -> None: + for filename, data in snapshot.files: + if not filename.endswith(".py"): + continue + for node in ast.walk(ast.parse(data, filename=filename)): + if isinstance(node, ast.Import): + names = [alias.name for alias in node.names] + attrs: list[str] = [] + elif isinstance(node, ast.ImportFrom) and node.level == 0 and node.module: + names = [node.module] + attrs = [alias.name for alias in node.names] + else: + continue + for name in names: + self._seal_import(name, attrs) + + def _seal_import(self, name: str, attrs: list[str]) -> None: + top = name.split(".", maxsplit=1)[0] + if top == self.package: + return + if top not in self.approved and top not in sys.stdlib_module_names: + msg = "Plugin declares an unapproved runtime dependency" + raise HeinzelError(msg) + parts = name.split(".") + for end in range(1, len(parts) + 1): + self._seal(".".join(parts[:end])) + if end > 1: + parent_name = ".".join(parts[: end - 1]) + child = parts[end - 1] + if vars(self.modules[parent_name]).get(child) is not self.modules[".".join(parts[:end])]: + msg = "A host runtime submodule binding changed" + raise HeinzelError(msg) + self.attributes[(parent_name, child)] = self.modules[".".join(parts[:end])] + module = self.modules[name] + for attr in attrs: + if attr == "*" or attr not in vars(module): + msg = f"A declared host runtime attribute is not already loaded: {name}.{attr}" + raise HeinzelError(msg) + value = vars(module)[attr] + if isinstance(value, ModuleType): + child_spec = vars(value).get("__spec__") + if not isinstance(child_spec, ModuleSpec) or self._seal(child_spec.name) is not value: + msg = "A host runtime attribute module is unverified" + raise HeinzelError(msg) + self.attributes[(name, attr)] = value + + def verify(self) -> None: + """Refuse changes to the already-loaded modules and attributes used by the plugin. + + Raises: + HeinzelError: If a sealed object changed. + """ + for name, module in self.modules.items(): + spec, loader, origin, file = self.specs[name] + path = vars(module).get("__path__") + if ( + sys.modules.get(name) is not module + or vars(module).get("__spec__") is not spec + or not isinstance(spec, ModuleSpec) + or spec.loader is not loader + or spec.origin != origin + or vars(module).get("__file__") != file + or (tuple(cast("Sequence[object]", path)) if isinstance(path, (list, tuple)) else None) + != self.paths[name] + ): + msg = "The sealed host runtime module changed" + raise HeinzelError(msg) + for (name, attr), value in self.attributes.items(): + if vars(self.modules[name]).get(attr) is not value: + msg = "The sealed host runtime attribute changed" + raise HeinzelError(msg) + + def import_module( + self, + name: str, + caller_globals: dict[str, object] | None = None, + caller_locals: Mapping[str, object] | None = None, + fromlist: tuple[str, ...] | None = (), + level: int = 0, + ) -> object: + self.verify() + if level or name == self.package or name.startswith(self.package + "."): + return builtins.__import__(name, caller_globals, caller_locals, fromlist, level) + module = self.modules.get(name) + if module is None or sys.modules.get(name) is not module: + msg = "The host runtime import is not sealed" + raise HeinzelError(msg) + for attr in fromlist or (): + key = (name, attr) + if key not in self.attributes or vars(module).get(attr) is not self.attributes[key]: + msg = "The host runtime attribute changed" + raise HeinzelError(msg) + root = name.split(".", maxsplit=1)[0] + result = module if fromlist else self.modules[root] + if sys.modules.get(root) is not self.modules[root]: + msg = "The host runtime module changed" + raise HeinzelError(msg) + return result + + +class _CapturedResources(TraversableResources): + """Read package resources from captured bytes, never from the later disk state.""" + + def __init__(self, snapshot: SourceSnapshot, package_path: str) -> None: + self.snapshot = snapshot + self.prefix = package_path + "/" if package_path else "" + + def files(self) -> Traversable: + """Expose an immutable in-memory archive of the captured resource bytes.""" + self.snapshot.verify() + buffer = io.BytesIO() + with zipfile.ZipFile(buffer, "w", compression=zipfile.ZIP_STORED) as archive: + for name, data in self.snapshot.files: + archive.writestr(name, data) + return zipfile.Path(zipfile.ZipFile(buffer, "r"), at=self.prefix) + + def open_resource(self, resource: str) -> io.BufferedReader: + self.snapshot.verify() + if "/" in resource or resource in {"", ".", ".."}: + raise FileNotFoundError + data = dict(self.snapshot.files).get(self.prefix + resource) + if data is None: + raise FileNotFoundError + return io.BufferedReader(io.BytesIO(data)) + + def resource_path(self, resource: str) -> str: + raise FileNotFoundError(resource) + + def is_resource(self, path: str) -> bool: + self.snapshot.verify() + return "/" not in path and self.prefix + path in dict(self.snapshot.files) + + def contents(self) -> Iterator[str]: + self.snapshot.verify() + return iter( + sorted( + { + name[len(self.prefix) :].split("/", 1)[0] + for name, _ in self.snapshot.files + if name.startswith(self.prefix) + } + ) + ) + + +class SourceOnlyFinder(importlib.abc.MetaPathFinder, importlib.abc.Loader): + """Compile captured package bytes without relying on timestamp bytecode caches.""" + + def __init__(self, name: str, snapshot: SourceSnapshot, runtime: SealedRuntime | None = None) -> None: + if not name.isidentifier() or name in sys.modules or any(key.startswith(name + ".") for key in sys.modules): + msg = "Package was already loaded or has an invalid name" + raise HeinzelError(msg) + self.name = name + self.snapshot = snapshot + self.files = MappingProxyType(dict(snapshot.files)) + self.runtime = runtime + + def find_spec(self, fullname: str, path: object = None, target: ModuleType | None = None) -> ModuleSpec | None: + del path + if fullname != self.name and not fullname.startswith(self.name + "."): + return None + if target is not None: + msg = "Reloading a captured plugin is unsupported" + raise HeinzelError(msg) + self.snapshot.verify() + part = fullname[len(self.name) :].lstrip(".").replace(".", "/") + init = f"{part}/__init__.py" if part else "__init__.py" + module = f"{part}.py" + if init in self.files: + return importlib.util.spec_from_file_location( + fullname, + self.snapshot.root / init, + loader=self, + submodule_search_locations=[str((self.snapshot.root / init).parent)], + ) + if module in self.files: + return importlib.util.spec_from_file_location(fullname, self.snapshot.root / module, loader=self) + msg = "Module is outside the captured source graph" + raise ModuleNotFoundError(msg) + + def get_resource_reader(self, fullname: str) -> _CapturedResources | None: + if fullname != self.name and not fullname.startswith(self.name + "."): + return None + part = fullname[len(self.name) :].lstrip(".").replace(".", "/") + init = f"{part}/__init__.py" if part else "__init__.py" + return _CapturedResources(self.snapshot, part) if init in self.files else None + + def create_module(self, spec: ModuleSpec) -> None: + del spec + + def exec_module(self, module: ModuleType) -> None: + self.snapshot.verify() + spec = module.__spec__ + if spec is None or spec.origin is None: + msg = "The source-bound module has no origin" + raise HeinzelError(msg) + origin = spec.origin + path = Path(origin).relative_to(self.snapshot.root).as_posix() + code = compile(self.files[path], origin, "exec", dont_inherit=True) + if self.runtime is not None: + module.__dict__["__builtins__"] = dict(vars(builtins), __import__=self.runtime.import_module) + + def verify_context() -> bool: + active_spec = module.__spec__ + if ( + self.runtime is None + or sys.modules.get(module.__name__) is not module + or active_spec is None + or active_spec.loader is not self + ): + return False + self.snapshot.verify() + self.runtime.verify() + return True + + with executing_captured_source(module.__dict__, verify_context): + exec(code, module.__dict__) # noqa: S102 - explicitly owner-selected plugin source + self.snapshot.verify() + + +@dataclass(frozen=True) +class CapturedEntry: + """Host-owned association between source, entry-point records, and loader.""" + + finder: SourceOnlyFinder + records: tuple[ModuleRecord, ...] + approved_runtime: frozenset[str] + entry: EntryPoint + origin: Origin + require_declaration: bool + + def service[T](self, kernel: Kernel, record: ModuleRecord, contract: type[T]) -> T: + """Resolve only an exact recorded service whose class came from this source loader. + + Raises: + HeinzelError: If source, registration, or the live implementation changed. + """ + if not any(candidate is record for candidate in self.records) or kernel.module(record.qualname) is not record: + msg = "The source-bound module record changed" + raise HeinzelError(msg) + if self.entry.dist is not None: + _validate_distribution_source(self.entry, self.finder.snapshot.root, self.origin, self.require_declaration) + self.finder.snapshot.verify() + if self.finder.runtime is not None: + self.finder.runtime.verify() + instance = kernel.resolve_selected_service(contract, record) + cls = type(instance) + module = sys.modules.get(cls.__module__) + if ( + (not cls.__module__.startswith(self.finder.name + ".") and cls.__module__ != self.finder.name) + or module is None + or module.__spec__ is None + or module.__spec__.loader is not self.finder + ): + msg = "The source-bound service implementation changed" + raise HeinzelError(msg) + self.finder.snapshot.verify() + if self.finder.runtime is not None: + self.finder.runtime.verify() + return instance + + +def _check_entry_loaded(module: object, finder: SourceOnlyFinder) -> None: + if not isinstance(module, ModuleType) or module.__spec__ is None or module.__spec__.loader is not finder: + msg = "The entry point did not use the source loader" + raise HeinzelError(msg) + + +def capture_entry( + entry: EntryPoint, + root: Path, + origin: Origin, + approved_runtime: frozenset[str] = frozenset(), +) -> CapturedEntry: + """Load an exact selected entry point from captured source before its package is imported. + + The caller must derive the selected package and source root from trusted host discovery. + This creates no access authority; external imports use the already-loaded sealed runtime. + + Raises: + HeinzelError: If the entry point is unsuitable or returns an unbound package. + """ + if entry.module != entry.value or "." in entry.module: + msg = "The source-bound entry point must name one root package" + raise HeinzelError(msg) + snapshot = SourceSnapshot.capture(root) + _check_dependencies(snapshot, entry.module, approved_runtime) + if not root.is_absolute(): + msg = "The host source root must be absolute" + raise HeinzelError(msg) + runtime = SealedRuntime(snapshot, entry.module, approved_runtime) + finder = SourceOnlyFinder(entry.module, snapshot, runtime) + sys.meta_path.insert(0, finder) + try: + module = entry.load() + _check_entry_loaded(module, finder) + records = tuple(walk(module, origin)) + finder.snapshot.verify() + return CapturedEntry(finder, records, approved_runtime, entry, origin, _has_declaration(entry)) + except BaseException: + sys.meta_path.remove(finder) + raise + + +@dataclass(frozen=True) +class SourceBinding[T]: + """Exact host-selected provider binding for a source-guarded call.""" + + capture: CapturedEntry + kernel: Kernel + record: ModuleRecord + contract: type[T] + capability: str + + +class CaptureCatalog: + """Host-owned catalog; reuse only the same sealed package in this process.""" + + def __init__(self) -> None: + self._entries: dict[str, tuple[EntryPoint, Path, Origin, frozenset[str], bool, CapturedEntry]] = {} + + def capture(self, entry: EntryPoint, root: Path, origin: Origin, approved_runtime: frozenset[str]) -> CapturedEntry: + """Reuse the host's exact capture, or capture a fresh not-yet-imported entry point. + + Raises: + HeinzelError: If the host selection, source, or registration has changed. + """ + existing = self._entries.get(entry.module) + if existing: + old_entry, old_root, old_origin, old_runtime, old_declaration, captured = existing + require_declaration = _has_declaration(entry) + if (entry, root, origin, approved_runtime, require_declaration) != ( + old_entry, + old_root, + old_origin, + old_runtime, + old_declaration, + ): + msg = "The host source selection changed" + raise HeinzelError(msg) + captured.finder.snapshot.verify() + loaded = sys.modules.get(entry.module) + _check_entry_loaded(loaded, captured.finder) + return captured + captured = capture_entry(entry, root, origin, approved_runtime) + self._entries[entry.module] = (entry, root, origin, approved_runtime, captured.require_declaration, captured) + return captured + + def guarded_call[T](self, binding: SourceBinding[T], call: Callable[[T], object]) -> object: + """Check the same registered instance and source before and after a host-selected operation. + + Raises: + HeinzelError: If the result was produced after the source or instance changed. + """ + capture, kernel, record, contract, capability = ( + binding.capture, + binding.kernel, + binding.record, + binding.contract, + binding.capability, + ) + if not any(stored[-1] is capture for stored in self._entries.values()): + msg = "The source capture is not owned by this host catalog" + raise HeinzelError(msg) + if kernel.selected_capability_provider(capability)[0] is not record: + msg = "The selected capability changed" + raise HeinzelError(msg) + instance = capture.service(kernel, record, contract) + result = call(instance) + if kernel.selected_capability_provider(capability)[0] is not record: + msg = "The selected capability changed during the operation" + raise HeinzelError(msg) + if capture.service(kernel, record, contract) is not instance: + msg = "The source-bound service instance changed during the operation" + raise HeinzelError(msg) + return result + + +class SourceGuardedLease[T, L]: + """Source-bound lease fence; cleanup remains available after source changes.""" + + def __init__( + self, catalog: CaptureCatalog, binding: SourceBinding[T], lease: L, cleanup: Callable[[L], None] + ) -> None: + self._catalog, self._binding, self._lease, self._cleanup = catalog, binding, lease, cleanup + self._closed = False + + def call(self, operation: Callable[[L], object]) -> object: + """Run a host-selected lease operation with source and selected-provider checks. + + Raises: + HeinzelError: If the lease is closed or source authority changed. + """ + if self._closed: + msg = "The source-bound lease is closed" + raise HeinzelError(msg) + return self._catalog.guarded_call(self._binding, lambda _service: operation(self._lease)) + + def close(self) -> None: + """Release resources even if the source is stale; never perform a new read.""" + if not self._closed: + self._closed = True + self._cleanup(self._lease) + + +def open_source_lease[T, L]( + catalog: CaptureCatalog, + binding: SourceBinding[T], + opener: Callable[[T], L], + cleanup: Callable[[L], None], +) -> SourceGuardedLease[T, L]: + """Capture a lease only while its provider remains current; close if the post-check fails. + + Raises: + HeinzelError: If provider authority changed before or during opening. + """ + opened: list[L] = [] + + def open_once(service: T) -> L: + lease = opener(service) + opened.append(lease) + return lease + + try: + catalog.guarded_call(binding, open_once) + except BaseException: + for lease in opened: + cleanup(lease) + raise + return SourceGuardedLease(catalog, binding, opened[0], cleanup) + + +def _has_declaration(entry: EntryPoint) -> bool: + if entry.dist is None: + return False + declared = [ + item + for item in entry.dist.entry_points + if item.group == "heinzel.connected_sources" and item.name == entry.name + ] + return len(declared) == 1 and declared[0].value == entry.value + + +def _validate_distribution_source( + entry: EntryPoint, root: Path, origin: Origin, require_declaration: bool = False +) -> None: + """Bind the selected installed distribution's manifest to the exact source package. + + Raises: + HeinzelError: If the distribution, manifest or selected root cannot be established. + OSError: If installation metadata is unavailable. + ValueError: If metadata is malformed. + """ + dist = entry.dist + if dist is None or dist.files is None or entry not in dist.entry_points: + msg = "The installed distribution manifest is unavailable" + raise HeinzelError(msg) + if require_declaration and not _has_declaration(entry): + msg = "The installed connected source declaration changed" + raise HeinzelError(msg) + if origin.distribution != dist.name or origin.version != dist.version: + msg = "The selected distribution identity changed" + raise HeinzelError(msg) + direct_url = dist.read_text("direct_url.json") + if direct_url is not None: + metadata = cast("JsonValue", json.loads(direct_url)) + directory = metadata.get("dir_info", {}) if isinstance(metadata, dict) else None + if not isinstance(directory, dict) or directory.get("editable", False): + msg = "Editable source distributions cannot be source-bound" + raise HeinzelError(msg) + located = dist.locate_file(entry.module) + if ( + not root.is_absolute() + or root.is_symlink() + or not isinstance(located, Path) + or located.resolve() != root.resolve() + ): + msg = "The selected source root is not owned by the installed distribution" + raise HeinzelError(msg) + manifest: set[Path] = set() + for item in dist.files: + location = dist.locate_file(item) + if not isinstance(location, Path): + msg = "The installed distribution has an unsupported source location" + raise HeinzelError(msg) + manifest.add(location.resolve()) + snapshot = SourceSnapshot.capture(root) + if any((root / name).resolve() not in manifest for name, _ in snapshot.files): + msg = "The selected source graph is not covered by the distribution manifest" + raise HeinzelError(msg) + + +@dataclass(frozen=True) +class ReviewedSource: + """A host-supplied audit selection, not a plugin-provided attestation.""" + + package: str + root: Path + approved_runtime: frozenset[str] + + +class SourceDiscovery: + """Capture declared installed providers before loading their entry points.""" + + def __init__(self, selections: dict[str, ReviewedSource] | None = None, *, installed: bool = False) -> None: + self._selections = MappingProxyType(dict(selections or {})) + self._installed = installed + self._catalog = CaptureCatalog() + self._captured: dict[str, CapturedEntry] = {} + + @property + def catalog(self) -> CaptureCatalog: + """Host-owned capture catalog for source-guarded operations.""" + return self._catalog + + def try_capture(self, entry: EntryPoint, origin: Origin) -> CapturedEntry | None: + """Capture host-selected candidates before loading the entry point. + + Raises: + HeinzelError: If capture fails before import. + OSError: If source inspection fails. + SyntaxError: If source cannot be parsed. + ValueError: If source metadata is invalid. + """ + reviewed = self._selections.get(entry.name) + require_declaration = False + if reviewed is None and self._installed and entry.dist is not None: + declarations = [ + item + for item in entry.dist.entry_points + if item.group == "heinzel.connected_sources" and item.name == entry.name + ] + if len(declarations) == 1 and declarations[0].value == entry.value: + location = entry.dist.locate_file(entry.module) + if isinstance(location, Path): + reviewed = ReviewedSource(entry.module, location, frozenset({"heinzel", "pydantic"})) + require_declaration = True + if reviewed is None or entry.module != reviewed.package: + return None + try: + _validate_distribution_source(entry, reviewed.root, origin, require_declaration) + snapshot = SourceSnapshot.capture(reviewed.root) + _check_dependencies(snapshot, entry.module, reviewed.approved_runtime) + SealedRuntime(snapshot, entry.module, reviewed.approved_runtime) + except (HeinzelError, OSError, SyntaxError, ValueError): + return None + preloaded = entry.module in sys.modules + try: + capture = self._catalog.capture(entry, reviewed.root, origin, reviewed.approved_runtime) + except (HeinzelError, OSError, SyntaxError, ValueError): + if preloaded: + return None + raise + self._captured[entry.name] = capture + return capture + + def binding[T](self, kernel: Kernel, capability: str, contract: type[T]) -> SourceBinding[T]: + """Bind the host-selected record to the exact catalog-owned captured provider. + + Raises: + HeinzelError: If no captured source matches the current capability selection. + """ + record, provider = kernel.selected_capability_provider(capability) + matches = [capture for capture in self._captured.values() if any(item is record for item in capture.records)] + if len(matches) != 1: + msg = "The selected provider has no reviewed source capture" + raise HeinzelError(msg) + capture = matches[0] + factory_module = sys.modules.get(provider.factory.__module__) + if ( + factory_module is None + or factory_module.__spec__ is None + or factory_module.__spec__.loader is not capture.finder + ): + msg = "The selected capability implementation is outside the captured source" + raise HeinzelError(msg) + binding = SourceBinding(capture, kernel, record, contract, capability) + self._catalog.guarded_call(binding, lambda service: service) + return binding + + +class SourceGuardedConnectedLease(ConnectedReadLease): + """Host guard around a provider lease and its source revision.""" + + def __init__(self, lease: SourceGuardedLease[ConnectedReadService, ConnectedReadLease], source: str) -> None: + self._lease, self._source = lease, source + + def review(self) -> ConnectedReadReview: + observed = self._lease.call(lambda lease: lease.review()) + review = ConnectedReadReview.model_validate(observed) + digest = hashlib.sha256((self._source + "\0" + review.source_revision).encode()).hexdigest() + return review.model_copy(update={"source_revision": "host-source-v1:" + digest}) + + def read(self, arguments: Mapping[str, JsonValue]) -> JsonValue: + return cast("JsonValue", self._lease.call(lambda lease: lease.read(arguments))) + + def close(self) -> None: + self._lease.close() + + +class SourceGuardedConnectedService(ConnectedReadService): + """Host continuity wrapper; this is not approval or publisher verification.""" + + def __init__(self, discovery: SourceDiscovery, binding: SourceBinding[ConnectedReadService]) -> None: + self._discovery, self._binding = discovery, binding + + def supported_reads(self) -> tuple[ConnectedReadMethod, ...]: + value = self._discovery.catalog.guarded_call(self._binding, lambda service: service.supported_reads()) + if not isinstance(value, tuple) or any( + not isinstance(item, ConnectedReadMethod) for item in cast("tuple[object, ...]", value) + ): + msg = "The source-bound provider returned invalid methods" + raise HeinzelError(msg) + return cast("tuple[ConnectedReadMethod, ...]", value) + + def open_lease(self, session: str, operation: str, fixed_arguments: Mapping[str, JsonValue]) -> ConnectedReadLease: + lease = open_source_lease( + self._discovery.catalog, + self._binding, + lambda service: service.open_lease(session, operation, fixed_arguments), + lambda value: value.close(), + ) + capture = self._binding.capture + source = ( + str(capture.finder.snapshot.root) + + "\0" + + capture.finder.snapshot.digest() + + "\0" + + "\0".join(sorted(capture.approved_runtime)) + ) + return SourceGuardedConnectedLease(lease, source) diff --git a/src/heinzel/modules/heinzel/_connected_grants.py b/src/heinzel/modules/heinzel/_connected_grants.py index de0f5f7d3..79c048d03 100644 --- a/src/heinzel/modules/heinzel/_connected_grants.py +++ b/src/heinzel/modules/heinzel/_connected_grants.py @@ -24,7 +24,13 @@ SessionWriter, Settings, ) -from heinzel.kernel import CAPABILITIES, PRIVATE_WRITE, CapabilitySettings, capability_read_schema +from heinzel.kernel import ( + CAPABILITIES, + PRIVATE_WRITE, + CapabilitySettings, + SourceGuardedConnectedService, + capability_read_schema, +) from heinzel.kernel._approval import ( ApprovalConfirmation, ApprovalEvidence, @@ -257,7 +263,16 @@ def _human(self) -> None: def _selected(self, capability: str, method: str) -> tuple[ModuleRecord, ConnectedReadService]: try: - record, service = self.kernel.selected_capability_service(ConnectedReadService, capability) + record, _provider = self.kernel.selected_capability_provider(capability) + # Default boot binds source before resolving any provider service factory. + source_guard = self.kernel.source_discovery + service: ConnectedReadService + if source_guard is not None: + service = SourceGuardedConnectedService( + source_guard, source_guard.binding(self.kernel, capability, ConnectedReadService) + ) + else: + record, service = self.kernel.selected_capability_service(ConnectedReadService, capability) claims = service.supported_reads() if type(claims) is not tuple or len(claims) > _MAX_SUPPORTED_READS: raise ValueError # noqa: TRY301 diff --git a/tests/_source_distribution_fixture.py b/tests/_source_distribution_fixture.py new file mode 100644 index 000000000..b0dce549e --- /dev/null +++ b/tests/_source_distribution_fixture.py @@ -0,0 +1,14 @@ +"""Synthetic installed distribution metadata for source-bound discovery tests.""" + +from importlib.metadata import Distribution + + +def installed_entry(root, entry_name, package, *, declared=False): + info = root.parent / f"{entry_name}-1.0.dist-info" + info.mkdir(exist_ok=True) + (info / "METADATA").write_text(f"Metadata-Version: 2.1\nName: {entry_name}\nVersion: 1.0\n") + declaration = f"[heinzel.connected_sources]\n{entry_name} = {package}\n" if declared else "" + (info / "entry_points.txt").write_text(f"[heinzel.modules]\n{entry_name} = {package}\n" + declaration) + paths = sorted(path.relative_to(root.parent).as_posix() for path in root.rglob("*") if path.is_file()) + (info / "RECORD").write_text("".join(f"{path},,\n" for path in paths if "__pycache__" not in path)) + return next(iter(Distribution.at(info).entry_points)) diff --git a/tests/integration/source_discovery_prototype_test.py b/tests/integration/source_discovery_prototype_test.py new file mode 100644 index 000000000..ebbf56f17 --- /dev/null +++ b/tests/integration/source_discovery_prototype_test.py @@ -0,0 +1,209 @@ +"""Opt-in host source discovery with actual private review and fake provider reads.""" + +import sys +from uuid import uuid4 + +import pytest + +from heinzel import Config, CurrentUser, HeinzelError +from heinzel.kernel import ConnectedReadService +from heinzel.kernel._approval import arguments_hash +from heinzel.kernel._source_loading import SourceDiscovery +from heinzel.modules.heinzel._capability_review import describe_capability_read +from heinzel.modules.heinzel._connected_grants import ConnectedReadGrants +from tests._source_distribution_fixture import installed_entry +from tests.integration.chat_test import AUTH, web_client +from tests.integration.scheduled_connected_read_test import _approve_read + +pytestmark = pytest.mark.integration + + +@pytest.fixture +def bound_source(tmp_path, monkeypatch, settings): + from heinzel.kernel import _discovery + + package = "fixture_captured_provider_" + uuid4().hex + root = tmp_path / package + root.mkdir() + source = root / "__init__.py" + source.write_text("""from pathlib import Path +from heinzel import Module, ConnectedReadLease, ConnectedReadService +from heinzel import ConnectedReadMethod, ConnectedReadReview +module = Module("fictional", "Offline fixture") +owner = "" +schema = "" +reads = [] +constructed = [] +mutate_read = False +class Lease(ConnectedReadLease): + def __init__(self, session, operation, fixed): self.session, self.operation, self.fixed = session, operation, fixed + def review(self): + return ConnectedReadReview(owner=owner, session=self.session, operation=self.operation, schema_sha256=schema, + source_revision="fixture", account_identity="account-one", provider_scopes=("read",), + fixed_arguments=self.fixed) + def read(self, arguments): + reads.append(True) + if mutate_read: Path(__file__).write_text("# changed during read\\n") + return {"events": []} + def close(self): pass +class Provider(ConnectedReadService): + def __init__(self): constructed.append(True) + def supported_reads(self): return (ConnectedReadMethod(capability="calendar", method="events"),) + def open_lease(self, session, operation, fixed_arguments): return Lease(session, operation, dict(fixed_arguments)) +module.service(Provider) +class Calendar: + def events(self, window): return [] + def free_busy(self, people, window): return [] +module.provides("calendar", modes={"read"}, name="fictional", hosts=["fixture.invalid"])(Calendar) +""") + entry = installed_entry(root, "fictional", package, declared=True) + monkeypatch.setattr(_discovery, "entry_points", lambda **_kwargs: [entry]) + monkeypatch.syspath_prepend(str(tmp_path)) + config = Config(files=(), data={"capabilities": {"calendar": "fictional"}}) + monkeypatch.setattr(Config, "load", classmethod(lambda _cls, _cwd=None: config)) + from heinzel.kernel._boot import boot + + kernel = boot(settings) + sources = kernel.source_discovery + assert sources is not None + current = CurrentUser(tmp_path / "identity") + kernel.bind(CurrentUser, current) + client = web_client(kernel, tmp_path) + session = client.post("/api/heinzel/new-chat", headers=AUTH, json={"request_id": "new"}).json()["session_id"] + plugin = sys.modules[package] + with current.local("cli"): + plugin.owner = current.principal().id + plugin.schema = arguments_hash(describe_capability_read(kernel, "calendar", "events").schema) + yield client, kernel, current, session, plugin, source, sources + finder = plugin.__spec__.loader + if finder in sys.meta_path: + sys.meta_path.remove(finder) + sys.modules.pop(package, None) + + +def test_private_review_and_revocation_use_host_source_bound_provider(bound_source): + client, kernel, current, session, plugin, source, _ = bound_source + state = {"reads": 0} + proposal = _approve_read(client, kernel, current, session, state, rolling=False) + assert proposal.json()["review"]["source_revision"].startswith("host-source-v1:") + grants = kernel.resolve(ConnectedReadGrants) + with current.local("cli"), grants.crud.repository.main_writer(session) as writer: + assert grants.read(session, writer) == {"events": []} + assert plugin.reads == [True] + revoked = client.post( + "/api/heinzel/revoke-connected-read", headers=AUTH, json={"session_id": session, "request_id": "grant-one"} + ) + assert revoked.status_code == 200 + with current.local("cli"), grants.crud.repository.main_writer(session) as writer, pytest.raises(HeinzelError): + grants.read(session, writer) + assert plugin.reads == [True] + source.write_text("# changed after loading\n") + with pytest.raises(HeinzelError): + kernel.source_discovery.binding(kernel, "calendar", ConnectedReadService) + + +def test_unreviewed_source_cannot_open_a_connected_lease(bound_source): + client, kernel, _current, session, plugin, _source, _ = bound_source + kernel.source_discovery = SourceDiscovery({}) + proposed = client.post( + "/api/heinzel/propose-connected-read", + headers=AUTH, + json={ + "session_id": session, + "request_id": "unreviewed", + "capability": "calendar", + "method": "events", + "fixed_arguments": {"window": {"since": "2026-01-01T00:00:00Z", "until": "2026-01-02T00:00:00Z"}}, + }, + ) + assert proposed.status_code != 200 + assert plugin.reads == [] + assert plugin.constructed == [] + + +def test_stale_source_is_rejected_before_service_constructor(bound_source): + client, _kernel, _current, session, plugin, source, _ = bound_source + source.write_text("# changed before provider construction\n") + response = client.post( + "/api/heinzel/propose-connected-read", + headers=AUTH, + json={ + "session_id": session, + "request_id": "stale", + "capability": "calendar", + "method": "events", + "fixed_arguments": {"window": {"since": "2026-01-01T00:00:00Z", "until": "2026-01-02T00:00:00Z"}}, + }, + ) + assert response.status_code != 200 + assert plugin.constructed == [] + + +def test_default_boot_undeclared_provider_is_rejected_before_construction( + bound_source, monkeypatch, settings, tmp_path +): + _client, _kernel, current, _session, plugin, source, _ = bound_source + from heinzel.kernel import _discovery + from heinzel.kernel._boot import boot + + entry = installed_entry(source.parent, "fictional", plugin.__name__, declared=False) + monkeypatch.setattr(_discovery, "entry_points", lambda **_kwargs: [entry]) + kernel = boot(settings) + kernel.bind(CurrentUser, current) + (tmp_path / "other").mkdir() + client = web_client(kernel, tmp_path / "other") + session = client.post("/api/heinzel/new-chat", headers=AUTH, json={"request_id": "new"}).json()["session_id"] + response = client.post( + "/api/heinzel/propose-connected-read", + headers=AUTH, + json={ + "session_id": session, + "request_id": "undeclared", + "capability": "calendar", + "method": "events", + "fixed_arguments": {"window": {"since": "2026-01-01T00:00:00Z", "until": "2026-01-02T00:00:00Z"}}, + }, + ) + assert response.status_code != 200 + assert plugin.constructed == [] + + +def test_approved_grant_fails_closed_after_source_change(bound_source): + client, kernel, current, session, plugin, source, _ = bound_source + _approve_read(client, kernel, current, session, {"reads": 0}, rolling=False) + source.write_text("# changed before the read\n") + grants = kernel.resolve(ConnectedReadGrants) + with current.local("cli"), grants.crud.repository.main_writer(session) as writer, pytest.raises(HeinzelError): + grants.read(session, writer) + assert plugin.reads == [] + + +def test_in_flight_source_change_discards_provider_result(bound_source): + client, kernel, current, session, plugin, _source, _ = bound_source + _approve_read(client, kernel, current, session, {"reads": 0}, rolling=False) + plugin.mutate_read = True + grants = kernel.resolve(ConnectedReadGrants) + with current.local("cli"), grants.crud.repository.main_writer(session) as writer, pytest.raises(HeinzelError): + grants.read(session, writer) + assert plugin.reads == [True] + + +def test_distribution_manifest_change_after_approval_denies_read(bound_source): + client, kernel, current, session, plugin, source, _ = bound_source + _approve_read(client, kernel, current, session, {"reads": 0}, rolling=False) + (source.parent.parent / "fictional-1.0.dist-info" / "RECORD").write_text("") + grants = kernel.resolve(ConnectedReadGrants) + with current.local("cli"), grants.crud.repository.main_writer(session) as writer, pytest.raises(HeinzelError): + grants.read(session, writer) + assert plugin.reads == [] + + +def test_declared_source_removal_revokes_bound_read(bound_source): + client, kernel, current, session, plugin, source, _ = bound_source + _approve_read(client, kernel, current, session, {"reads": 0}, rolling=False) + metadata = source.parent.parent / "fictional-1.0.dist-info" / "entry_points.txt" + metadata.write_text(f"[heinzel.modules]\nfictional = {source.parent.name}\n") + grants = kernel.resolve(ConnectedReadGrants) + with current.local("cli"), grants.crud.repository.main_writer(session) as writer, pytest.raises(HeinzelError): + grants.read(session, writer) + assert plugin.reads == [] diff --git a/tests/unit/public_api_test.py b/tests/unit/public_api_test.py index 7293901df..c5c3b68d2 100644 --- a/tests/unit/public_api_test.py +++ b/tests/unit/public_api_test.py @@ -189,6 +189,7 @@ "Window", "__version__", "capability_read_schema", + "captured_source_active", "clean_text", "draft_reporting_lines", "free_busy_people", @@ -419,6 +420,7 @@ "SoundStream", "Soundscape", "SoundscapeLet", + "SourceGuardedConnectedService", "Span", "Status", "StatusItem", @@ -468,6 +470,7 @@ "Window", "boot", "capability_read_schema", + "captured_source_active", "check_repository_url", "check_target", "clean_text", diff --git a/tests/unit/verified_source_prototype_test.py b/tests/unit/verified_source_prototype_test.py new file mode 100644 index 000000000..dced96230 --- /dev/null +++ b/tests/unit/verified_source_prototype_test.py @@ -0,0 +1,526 @@ +"""Source-only loader experiment does not rely on Python timestamp caches.""" + +import builtins +import importlib +import os +import py_compile +import sys +from importlib.metadata import EntryPoint +from importlib.util import spec_from_file_location +from types import ModuleType + +import pytest + +from heinzel.kernel import Config, ConnectedReadService, Kernel, Origin, Settings, captured_source_active +from heinzel.kernel._errors import HeinzelError +from heinzel.kernel._source_loading import ( + CaptureCatalog, + SealedRuntime, + SourceBinding, + SourceOnlyFinder, + SourceSnapshot, + open_source_lease, +) +from tests._source_distribution_fixture import installed_entry + +pytestmark = pytest.mark.unit + + +def test_public_capture_marker_is_limited_to_the_exact_current_module(tmp_path, monkeypatch): + root = tmp_path / "scoped_marker_fixture" + root.mkdir() + (root / "__init__.py").write_text( + "import builtins\nfrom heinzel import Module, captured_source_active\n" + "root_before = captured_source_active()\nforeign = builtins.source_marker_foreign()\n" + "from . import child\nroot_after = captured_source_active()\n" + "def late(): return captured_source_active()\nmodule = Module('scope', 'fixture')\n" + ) + (root / "child.py").write_text( + "from heinzel import captured_source_active\nchild_active = captured_source_active()\n" + ) + monkeypatch.setattr(builtins, "source_marker_foreign", captured_source_active, raising=False) + + # The foreign function must call the predicate from its own globals, not as the plugin itself. + def foreign(): + return captured_source_active() + + monkeypatch.setattr(builtins, "source_marker_foreign", foreign) + monkeypatch.syspath_prepend(str(tmp_path)) + assert not captured_source_active() + entry = EntryPoint(name="scope", value=root.name, group="heinzel.modules") + catalog = CaptureCatalog() + captured = catalog.capture( + entry, root, Origin(distribution="fixture", version="1", builtin=False), frozenset({"heinzel"}) + ) + try: + plugin = sys.modules[root.name] + assert plugin.root_before is True + assert plugin.root_after is True + assert plugin.child.child_active is True + assert plugin.foreign is False + assert plugin.late() is False + assert not captured_source_active() + finally: + sys.meta_path.remove(captured.finder) + sys.modules.pop(root.name + ".child", None) + sys.modules.pop(root.name, None) + + +def test_public_capture_marker_is_false_for_ordinary_and_unsealed_loaders(tmp_path, monkeypatch): + root = tmp_path / "ordinary_marker_fixture" + root.mkdir() + (root / "__init__.py").write_text("from heinzel import captured_source_active\nactive = captured_source_active()\n") + monkeypatch.syspath_prepend(str(tmp_path)) + ordinary = importlib.import_module(root.name) + assert ordinary.active is False + sys.modules.pop(root.name) + finder = SourceOnlyFinder(root.name, SourceSnapshot.capture(root)) + sys.meta_path.insert(0, finder) + try: + unsealed = importlib.import_module(root.name) + assert unsealed.active is False + finally: + sys.meta_path.remove(finder) + sys.modules.pop(root.name, None) + + +def test_unloaded_shadowed_stdlib_is_refused_before_plugin_initializer(tmp_path, monkeypatch): + root = tmp_path / "shadow_source_fixture" + root.mkdir() + marker = tmp_path / "executed" + (root / "__init__.py").write_text(f"import mailbox\nopen({str(marker)!r}, 'w').close()\n") + (tmp_path / "mailbox.py").write_text(f"open({str(marker)!r}, 'w').close()\n") + monkeypatch.syspath_prepend(str(tmp_path)) + monkeypatch.delitem(sys.modules, "mailbox", raising=False) + entry = EntryPoint(name="shadow", value=root.name, group="heinzel.modules") + with pytest.raises(HeinzelError, match="not already loaded"): + CaptureCatalog().capture(entry, root, Origin(distribution="fixture", version="1", builtin=False), frozenset()) + assert not marker.exists() + + +def test_shadowed_external_runtime_is_refused_before_plugin_initializer(tmp_path, monkeypatch): + root = tmp_path / "external_shadow_fixture" + root.mkdir() + marker = tmp_path / "executed" + (root / "__init__.py").write_text(f"import pydantic\nopen({str(marker)!r}, 'w').close()\n") + fake = tmp_path / "pydantic.py" + fake.write_text("pass\n") + module = ModuleType("pydantic") + module.__spec__ = spec_from_file_location("pydantic", fake) + module.__file__ = str(fake) + monkeypatch.setitem(sys.modules, "pydantic", module) + entry = EntryPoint(name="shadow", value=root.name, group="heinzel.modules") + with pytest.raises(HeinzelError, match="origin is unverified"): + CaptureCatalog().capture( + entry, root, Origin(distribution="fixture", version="1", builtin=False), frozenset({"pydantic"}) + ) + assert not marker.exists() + + +def test_preloaded_shadowed_stdlib_is_refused_before_plugin_initializer(tmp_path, monkeypatch): + root = tmp_path / "cached_shadow_fixture" + root.mkdir() + marker = tmp_path / "executed" + (root / "__init__.py").write_text(f"import mailbox\nopen({str(marker)!r}, 'w').close()\n") + fake = tmp_path / "mailbox.py" + fake.write_text("pass\n") + module = ModuleType("mailbox") + module.__spec__ = spec_from_file_location("mailbox", fake) + module.__file__ = str(fake) + monkeypatch.setitem(sys.modules, "mailbox", module) + entry = EntryPoint(name="shadow", value=root.name, group="heinzel.modules") + with pytest.raises(HeinzelError, match="origin is unverified"): + CaptureCatalog().capture(entry, root, Origin(distribution="fixture", version="1", builtin=False), frozenset()) + assert not marker.exists() + + +def test_sealed_runtime_refuses_changed_cached_modules_and_attributes(tmp_path, monkeypatch): + import pathlib + + root = tmp_path / "sealed_fixture" + root.mkdir() + (root / "__init__.py").write_text("from pathlib import Path\n") + runtime = SealedRuntime(SourceSnapshot.capture(root), root.name, frozenset()) + assert runtime.import_module("pathlib", fromlist=("Path",)) is pathlib + with monkeypatch.context() as patch: + patch.setattr(pathlib, "Path", object()) + with pytest.raises(HeinzelError, match="attribute changed"): + runtime.import_module("pathlib", fromlist=("Path",)) + with monkeypatch.context() as patch: + patch.setitem(sys.modules, "pathlib", ModuleType("pathlib")) + with pytest.raises(HeinzelError, match="module changed"): + runtime.import_module("pathlib", fromlist=("Path",)) + + +def test_sealed_runtime_rejects_replaced_submodule(tmp_path, monkeypatch): + + root = tmp_path / "sealed_submodule_fixture" + root.mkdir() + (root / "__init__.py").write_text("from collections.abc import Mapping\n") + runtime = SealedRuntime(SourceSnapshot.capture(root), root.name, frozenset()) + monkeypatch.setitem(sys.modules, "collections.abc", ModuleType("collections.abc")) + with pytest.raises(HeinzelError, match="module changed"): + runtime.import_module("collections.abc", fromlist=("Mapping",)) + + +def test_sealed_runtime_rejects_changed_package_search_path(tmp_path, monkeypatch): + import collections + + root = tmp_path / "sealed_path_fixture" + root.mkdir() + (root / "__init__.py").write_text("from collections.abc import Mapping\n") + runtime = SealedRuntime(SourceSnapshot.capture(root), root.name, frozenset()) + monkeypatch.setattr(collections, "__path__", [str(tmp_path)]) + with pytest.raises(HeinzelError, match="module changed"): + runtime.import_module("collections.abc", fromlist=("Mapping",)) + + +def test_source_loader_ignores_stale_bytecode_and_detects_in_place_change(tmp_path): + root = tmp_path / "plugin" + root.mkdir() + entry = root / "__init__.py" + entry.write_text("value = 'old'\n") + py_compile.compile(str(entry), doraise=True) + before = entry.stat() + entry.write_text("value = 'new'\n") + os.utime(entry, ns=(before.st_atime_ns, before.st_mtime_ns)) + snapshot = SourceSnapshot.capture(root) + finder = SourceOnlyFinder("sample_verified_plugin", snapshot) + sys.meta_path.insert(0, finder) + try: + loaded = importlib.import_module(finder.name) + assert loaded.value == "new" + entry.write_text("value = 'two'\n") + with pytest.raises(HeinzelError, match="changed"): + snapshot.verify() + finally: + sys.meta_path.remove(finder) + sys.modules.pop(finder.name, None) + + +def test_preloaded_and_symlinked_packages_are_refused(tmp_path): + root = tmp_path / "plugin" + root.mkdir() + (root / "__init__.py").write_text("pass\n") + snapshot = SourceSnapshot.capture(root) + with pytest.raises(HeinzelError, match="already loaded"): + SourceOnlyFinder("sys", snapshot) + (root / "link.py").symlink_to(root / "__init__.py") + with pytest.raises(HeinzelError, match="linked"): + SourceSnapshot.capture(root) + + +def test_changed_package_refuses_later_captured_submodule_import(tmp_path): + root = tmp_path / "plugin" + root.mkdir() + (root / "__init__.py").write_text("pass\n") + module = root / "reader.py" + module.write_text("value = 'one'\n") + snapshot = SourceSnapshot.capture(root) + finder = SourceOnlyFinder("sample_lazy_plugin", snapshot) + sys.meta_path.insert(0, finder) + try: + importlib.import_module(finder.name) + module.write_text("value = 'two'\n") + with pytest.raises(HeinzelError, match="changed"): + importlib.import_module(finder.name + ".reader") + finally: + sys.meta_path.remove(finder) + sys.modules.pop(finder.name + ".reader", None) + sys.modules.pop(finder.name, None) + + +def test_external_native_extensions_and_missing_modules_refuse(tmp_path): + root = tmp_path / "plugin" + root.mkdir() + (root / "__init__.py").write_text("pass\n") + (root / "native.so").write_bytes(b"not a native library") + with pytest.raises(HeinzelError, match="Native"): + SourceSnapshot.capture(root) + (root / "native.so").unlink() + finder = SourceOnlyFinder("sample_missing_plugin", SourceSnapshot.capture(root)) + sys.meta_path.insert(0, finder) + try: + importlib.import_module(finder.name) + with pytest.raises(ModuleNotFoundError, match="outside"): + importlib.import_module(finder.name + ".unknown") + finally: + sys.meta_path.remove(finder) + sys.modules.pop(finder.name, None) + + +def _exercise_wrong_selection(catalog, captured, kernel, record): + foreign = SourceBinding(captured, kernel, record, ConnectedReadService, "mail") + with pytest.raises(HeinzelError): + catalog.guarded_call(foreign, lambda value: value.read()) + + +def _exercise_guarded_lease(catalog, binding, entry, old, new): + _exercise_wrong_selection(catalog, binding.capture, binding.kernel, binding.record) + closed = [] + lease = open_source_lease(catalog, binding, lambda _service: {"read": "private"}, closed.append) + assert lease.call(lambda value: value["read"]) == "private" + with pytest.raises(HeinzelError, match="changed"): + lease.call(lambda _value: entry.write_text(old)) + lease.close() + assert closed == [{"read": "private"}] + entry.write_text(new) + + def open_after_change(_service): + entry.write_text(old) + return {"read": "opened"} + + with pytest.raises(HeinzelError, match="changed"): + open_source_lease(catalog, binding, open_after_change, closed.append) + assert closed[-1] == {"read": "opened"} + + +def _exercise_captured_resources(captured, ep, entry, old): + from importlib.resources import files + + resource = files(ep.module).joinpath("metadata.txt") + assert resource.read_text() == "captured resource" + with pytest.raises(ValueError, match="requires mode"): + files(ep.module).joinpath("new.txt").open("w") + reader = captured.finder.get_resource_reader(ep.module) + assert reader is not None + with reader.open_resource("metadata.txt") as stream: + assert stream.read() == b"captured resource" + with pytest.raises(FileNotFoundError): + reader.open_resource("../metadata.txt") + with pytest.raises(FileNotFoundError): + reader.resource_path("metadata.txt") + entry.write_text(old) + with pytest.raises(HeinzelError, match="changed"): + files(ep.module) + with pytest.raises(HeinzelError, match="changed"): + reader.open_resource("metadata.txt") + + +def test_real_entry_point_records_and_registered_provider_use_captured_source(tmp_path): + root = tmp_path / "fake_bound_plugin" + root.mkdir() + entry = root / "__init__.py" + old = """from pathlib import Path +from heinzel.kernel import ConnectedReadService, Module +assert Path(__file__).is_file() +module = Module("fixture", "fixture") +class Reader(ConnectedReadService): + label = "old" + def read(self): return self.label +module.service(Reader) +class Calendar: + def events(self, window): return [] + def free_busy(self, people, window): return [] +module.provides("calendar", modes={"read"}, name="fixture", hosts=["fixture.invalid"])(Calendar) +""" + new = old.replace('"old"', '"new"') + entry.write_text(old) + py_compile.compile(str(entry), doraise=True) + before = entry.stat() + entry.write_text(new) + os.utime(entry, ns=(before.st_atime_ns, before.st_mtime_ns)) + (root / "metadata.txt").write_text("captured resource") + package = root / "child" + package.mkdir() + (package / "__init__.py").write_text('from heinzel.kernel import Module\nmodule = Module("child", "child")\n') + ep = EntryPoint(name="fake", value="fake_bound_plugin", group="heinzel.modules") + catalog = CaptureCatalog() + origin = Origin(distribution="fixture", version="1", builtin=False) + captured = catalog.capture(ep, root, origin, frozenset({"heinzel"})) + try: + assert [r.qualname for r in captured.records] == ["fixture", "fixture.child"] + assert catalog.capture(ep, root, origin, frozenset({"heinzel"})) is captured + kernel = Kernel( + Settings(), captured.records, config=Config(files=(), data={"capabilities": {"calendar": "fixture"}}) + ) + record, actual = kernel.selected_capability_service(ConnectedReadService, "calendar") + service = captured.service(kernel, record, ConnectedReadService) + second_kernel = Kernel( + Settings(), captured.records, config=Config(files=(), data={"capabilities": {"calendar": "fixture"}}) + ) + second_record, second_actual = second_kernel.selected_capability_service(ConnectedReadService, "calendar") + assert captured.service(second_kernel, second_record, ConnectedReadService) is second_actual + assert second_record is record + assert service is actual + assert service.label == "new" + binding = SourceBinding(captured, kernel, record, ConnectedReadService, "calendar") + assert catalog.guarded_call(binding, lambda value: value.read()) == "new" + _exercise_guarded_lease(catalog, binding, entry, old, new) + entry.write_text(new) + + def change_source(value): + entry.write_text(old) + return value.read() + + with pytest.raises(HeinzelError, match="changed"): + catalog.guarded_call(binding, change_source) + entry.write_text(new) + assert sys.modules[ep.module].__file__ == str(entry) + assert list(sys.modules[ep.module].__path__) == [str(root)] + _exercise_captured_resources(captured, ep, entry, old) + with pytest.raises(HeinzelError, match="changed"): + captured.service(kernel, record, ConnectedReadService) + finally: + sys.meta_path.remove(captured.finder) + for name in [ep.module + ".child", ep.module]: + sys.modules.pop(name, None) + + +def test_unknown_declared_dependency_refused_before_initializer(tmp_path): + from importlib.metadata import EntryPoint + + from heinzel.kernel import Origin + from heinzel.kernel._source_loading import capture_entry + + root = tmp_path / "unknown_dependency_plugin" + root.mkdir() + (root / "__init__.py").write_text("import unapproved_third_party\n") + ep = EntryPoint(name="fake", value="unknown_dependency_plugin", group="heinzel.modules") + with pytest.raises(HeinzelError, match="unapproved"): + capture_entry(ep, root, Origin(distribution="fixture", version="1", builtin=False)) + assert ep.module not in sys.modules + + +def test_static_dependency_check_is_not_a_dynamic_import_security_boundary(tmp_path, monkeypatch): + from importlib.metadata import EntryPoint + + from heinzel.kernel import Origin + from heinzel.kernel._source_loading import CaptureCatalog + + root = tmp_path / "dynamic_import_fixture" + root.mkdir() + (root / "__init__.py").write_text( + "import importlib\nfrom heinzel.kernel import Module\n" + 'dependency = importlib.import_module("unapproved_helper_fixture")\nmodule = Module("demo", "demo")\n' + ) + (tmp_path / "unapproved_helper_fixture.py").write_text("value = 42\n") + monkeypatch.syspath_prepend(str(tmp_path)) + ep = EntryPoint(name="demo", value="dynamic_import_fixture", group="heinzel.modules") + catalog = CaptureCatalog() + origin = Origin(distribution="fixture", version="1", builtin=False) + captured = catalog.capture(ep, root, origin, frozenset({"heinzel"})) + assert catalog.capture(ep, root, origin, frozenset({"heinzel"})) is captured + try: + assert sys.modules[ep.module].dependency.value == 42 + finally: + sys.meta_path.remove(captured.finder) + sys.modules.pop(ep.module, None) + sys.modules.pop("unapproved_helper_fixture", None) + + +def test_opt_in_real_discovery_preserves_ordinary_compatibility_and_reuses_capture(tmp_path, monkeypatch): + from heinzel.kernel import _discovery + from heinzel.kernel._source_loading import ReviewedSource, SourceDiscovery + + package = "fake_discovery_source" + root = tmp_path / package + root.mkdir() + entry_file = root / "__init__.py" + entry_file.write_text( + "from heinzel.kernel import ConnectedReadService, Module\n" + 'module = Module("fixture", "fixture")\n' + "class Reader(ConnectedReadService): pass\n" + "module.service(Reader)\n" + "class Calendar:\n" + " def events(self, window): return []\n" + " def free_busy(self, people, window): return []\n" + 'module.provides("calendar", modes={"read"}, name="fixture", hosts=["fixture.invalid"])(Calendar)\n' + ) + ep = installed_entry(root, "fixture", package) + monkeypatch.setattr(_discovery, "entry_points", lambda **_kwargs: [ep]) + monkeypatch.syspath_prepend(str(tmp_path)) + sources = SourceDiscovery({ep.name: ReviewedSource(package, root, frozenset({"heinzel"}))}) + try: + records, errors = _discovery.discover(source_discovery=sources) + assert errors == {} + again, errors = _discovery.discover(source_discovery=sources) + assert errors == {} + record = next(item for item in records if item.qualname == "fixture") + assert next(item for item in again if item.qualname == "fixture") is record + kernel = Kernel(Settings(), records, config=Config(files=(), data={"capabilities": {"calendar": "fixture"}})) + binding = sources.binding(kernel, "calendar", ConnectedReadService) + assert binding.record is record + entry_file.write_text("# changed after discovery\n") + with pytest.raises(HeinzelError, match="changed"): + sources.binding(kernel, "calendar", ConnectedReadService) + ordinary, _ = _discovery.discover() + assert any(item.qualname == "fixture" for item in ordinary) + finally: + loaded = sys.modules.pop(package, None) + if loaded is not None: + finder = loaded.__spec__.loader + if finder in sys.meta_path: + sys.meta_path.remove(finder) + + +def test_mismatched_distribution_root_and_editable_install_never_capture(tmp_path): + from heinzel.kernel._source_loading import ReviewedSource, SourceDiscovery + + installed = tmp_path / "installed_demo" + installed.mkdir() + (installed / "__init__.py").write_text("raise RuntimeError('initializer must not run')\n") + entry = installed_entry(installed, "demo", "installed_demo") + foreign = tmp_path / "unrelated" + foreign.mkdir() + (foreign / "__init__.py").write_text("raise RuntimeError('initializer must not run')\n") + origin = Origin(distribution=entry.dist.name, version=entry.dist.version, builtin=False) + mismatch = SourceDiscovery({entry.name: ReviewedSource(entry.module, foreign, frozenset())}) + assert mismatch.try_capture(entry, origin) is None + assert entry.module not in sys.modules + (installed / "unlisted.py").write_text("pass\n") + missing_manifest = SourceDiscovery({entry.name: ReviewedSource(entry.module, installed, frozenset())}) + assert missing_manifest.try_capture(entry, origin) is None + assert entry.module not in sys.modules + (installed / "unlisted.py").unlink() + (tmp_path / "demo-1.0.dist-info" / "direct_url.json").write_text('{"dir_info":{"editable":true}}') + editable = SourceDiscovery({entry.name: ReviewedSource(entry.module, installed, frozenset())}) + assert editable.try_capture(entry, origin) is None + assert entry.module not in sys.modules + + +def test_default_boot_denies_undeclared_preloaded_provider_but_keeps_ordinary_module(tmp_path, monkeypatch): + from heinzel.kernel import _boot, _discovery + from heinzel.kernel._source_loading import SourceDiscovery + + package = "undeclared_boot_provider" + root = tmp_path / package + root.mkdir() + (root / "__init__.py").write_text('from heinzel.kernel import Module\nmodule = Module("fixture", "fixture")\n') + entry = installed_entry(root, "fixture", package) + monkeypatch.setattr(_discovery, "entry_points", lambda **_kwargs: [entry]) + monkeypatch.setattr(_boot, "_installed_sources", SourceDiscovery(installed=True)) + monkeypatch.syspath_prepend(str(tmp_path)) + try: + loaded = importlib.import_module(package) + kernel = _boot.boot(Settings()) + assert kernel.module("fixture").module is loaded.module + assert kernel.source_discovery is not None + assert not kernel.source_discovery._captured + finally: + sys.modules.pop(package, None) + + +def test_unknown_runtime_dependency_keeps_ordinary_discovery_without_verified_read(tmp_path, monkeypatch): + from heinzel.kernel import _boot, _discovery + from heinzel.kernel._source_loading import SourceDiscovery + + package = "extra_dependency_fixture" + root = tmp_path / package + root.mkdir() + (root / "__init__.py").write_text( + 'import external_dependency_fixture\nfrom heinzel.kernel import Module\nmodule = Module("fixture", "fixture")\n' + ) + (tmp_path / "external_dependency_fixture.py").write_text("value = 1\n") + entry = installed_entry(root, "fixture", package, declared=True) + monkeypatch.setattr(_discovery, "entry_points", lambda **_kwargs: [entry]) + monkeypatch.setattr(_boot, "_installed_sources", SourceDiscovery(installed=True)) + monkeypatch.syspath_prepend(str(tmp_path)) + try: + kernel = _boot.boot(Settings()) + assert kernel.module("fixture") + assert kernel.source_discovery is not None + assert not kernel.source_discovery._captured + finally: + sys.modules.pop(package, None) + sys.modules.pop("external_dependency_fixture", None)