From ac8630268faabf99951693e8c68d78036def6582 Mon Sep 17 00:00:00 2001 From: stacknil Date: Tue, 15 Sep 2026 19:39:16 +0800 Subject: [PATCH 1/8] feat(security): add trusted authority controller --- scripts/repo_sentinel_authority_bootstrap.sh | 40 + scripts/repo_sentinel_authority_controller.py | 738 ++++++++++++++++++ 2 files changed, 778 insertions(+) create mode 100755 scripts/repo_sentinel_authority_bootstrap.sh create mode 100644 scripts/repo_sentinel_authority_controller.py diff --git a/scripts/repo_sentinel_authority_bootstrap.sh b/scripts/repo_sentinel_authority_bootstrap.sh new file mode 100755 index 0000000..5adccf8 --- /dev/null +++ b/scripts/repo_sentinel_authority_bootstrap.sh @@ -0,0 +1,40 @@ +#!/bin/sh + +# The workflow launch envelope must supply both paths from reviewed configuration. +set -eu +umask 077 + +if [ "$#" -lt 2 ]; then + exit 2 +fi + +trusted_python=$1 +controller=$2 +shift 2 + +case "$trusted_python" in + /*) ;; + *) exit 2 ;; +esac +case "$controller" in + /*) ;; + *) exit 2 ;; +esac + +if [ ! -x "$trusted_python" ] || [ ! -f "$controller" ]; then + exit 2 +fi + +# These variables must not reach the dynamic loader for env or Python. They may +# already have affected this shell, which is why the parent launch is trusted. +unset LD_PRELOAD LD_LIBRARY_PATH +unset DYLD_INSERT_LIBRARIES DYLD_LIBRARY_PATH DYLD_FRAMEWORK_PATH +unset DYLD_FALLBACK_LIBRARY_PATH DYLD_FALLBACK_FRAMEWORK_PATH +unset BASH_ENV ENV CDPATH GLOBIGNORE + +exec /usr/bin/env -i \ + PATH=/usr/bin:/bin \ + LANG=C.UTF-8 \ + LC_ALL=C.UTF-8 \ + TZ=UTC \ + "$trusted_python" -I -S -B "$controller" "$@" diff --git a/scripts/repo_sentinel_authority_controller.py b/scripts/repo_sentinel_authority_controller.py new file mode 100644 index 0000000..b702d30 --- /dev/null +++ b/scripts/repo_sentinel_authority_controller.py @@ -0,0 +1,738 @@ +"""Orchestrate exact-commit authority from a sanitized trusted process.""" + +from __future__ import annotations + +import hashlib +import importlib +import json +import os +import platform +import re +import shutil +import stat +import subprocess +import sys +import tempfile +from collections.abc import Callable, Iterator, Mapping +from contextlib import contextmanager +from dataclasses import dataclass, field +from enum import Enum +from pathlib import Path +from types import ModuleType + +CONTROLLER_SCHEMA_VERSION = 1 +REPOSITORY_ID = 1_130_304_545 +OWNER_ID = 219_124_580 +REPOSITORY = "stacknil/sec-writeups-public" +REMOTE_URL = "https://github.com/stacknil/sec-writeups-public.git" +POLICY_EPOCH = "v1" +WORKER_POLICY_EPOCH = "repo-sentinel-authority-v1" +POLICY_BUNDLE_SHA256 = ( + "6f25ebb773ce1453e8de623bca5aaecc936f1f188288f8df20aedeadb3bf4612" +) +SCANNER_ARTIFACT_SHA256 = ( + "0a949a4d00c6e6ae37eba60a6cb74e4e15bc3ec5fce2f1d4c99aa0ef309b36e3" +) +EXPECTED_RUNTIME = ("cpython", "3.12.3", "Linux", "x86_64") +FIXED_ENVIRONMENT = { + "LANG": "C.UTF-8", + "LC_ALL": "C.UTF-8", + "PATH": "/usr/bin:/bin", + "TZ": "UTC", +} + +_OID = re.compile(r"(?:[0-9a-f]{40}|[0-9a-f]{64})\Z") +_DIGEST = re.compile(r"[0-9a-f]{64}\Z") +_DECIMAL = re.compile(r"[0-9]+\Z") +_SAFE_GIT_VERSION = re.compile( + r"git version [0-9]+(?:\.[0-9]+)+(?:\.[-+0-9A-Za-z.]+)?\n?\Z" +) +_WORKER_SEMANTIC_DOMAIN = b"repo-sentinel-commit-authority-result-v1\0" +_POLICY_REFUSALS = frozenset( + { + "coverage_policy_mismatch", + "policy_bundle_mirror_mismatch", + "protected_control_mismatch", + "suppression_manifest_mismatch", + } +) +_REFUSAL_CODES = frozenset( + { + "acquired_head_mismatch", + "acquisition_refused", + "cleanup_failed", + "environment_mismatch", + "git_identity_mismatch", + "invalid_head_oid", + "invalid_request", + "launch_not_isolated", + "policy_bundle_mismatch", + "repository_identity_mismatch", + "runtime_mismatch", + "unsafe_control_root", + "unsafe_root_layout", + "unexpected_failure", + "unknown_policy_epoch", + "worker_infrastructure_refusal", + "worker_result_invalid", + } +) +_REQUEST_OPTIONS = ( + "--repository-id", + "--owner-id", + "--repository", + "--remote-url", + "--pull-number", + "--head-oid", + "--policy-epoch", + "--policy-bundle-sha256", + "--scratch-root", + "--scanner-artifact", +) +_WORKER_RESULT_KEYS = frozenset( + { + "coverage_policy_sha256", + "files_policy_excluded", + "files_scanned", + "files_scanner_skipped", + "files_total", + "head_oid", + "policy_bundle_sha256", + "policy_epoch", + "policy_schema_version", + "protected_manifest_sha256", + "refusal_code", + "report_sha256", + "report_size", + "repository_id", + "scanner_artifact_sha256", + "scanner_distribution", + "scanner_version", + "semantic_sha256", + "suppression_manifest_sha256", + "verdict", + } +) +_TRUSTED_MODULE_PATHS = { + "repo_sentinel_acquire": "repo_sentinel_acquire.py", + "repo_sentinel_commit_authoritative": "repo_sentinel_commit_authoritative.py", + "repo_sentinel_materialize": "repo_sentinel_materialize.py", + "repo_sentinel_policy_bundle": "repo_sentinel_policy_bundle.py", + "repo_sentinel_reader": "repo_sentinel_reader.py", +} + + +class ControllerOutcome(str, Enum): + AUTHORITY_RESULT = "AUTHORITY_RESULT" + INFRASTRUCTURE_REFUSAL = "INFRASTRUCTURE_REFUSAL" + + +class ControllerRefused(RuntimeError): + def __init__(self, code: str) -> None: + self.code = code if code in _REFUSAL_CODES else "unexpected_failure" + super().__init__(self.code) + + +@dataclass(frozen=True, slots=True) +class ControllerRequest: + repository_id: int + owner_id: int + repository: str + remote_url: str = field(repr=False) + pull_number: int + head_oid: str + policy_epoch: str + expected_policy_bundle_sha256: str = field(repr=False) + scratch_root: Path = field(repr=False) + scanner_artifact: Path = field(repr=False) + + +@dataclass(frozen=True, slots=True) +class ControllerResult: + controller_schema_version: int + repository_id: int + head_oid: str | None + policy_epoch: str | None + policy_bundle_sha256: str + worker_result: dict[str, object] | None + worker_semantic_sha256: str | None + controller_outcome: ControllerOutcome + fixed_refusal_code: str | None + + +@dataclass(frozen=True, slots=True) +class RuntimeFacts: + implementation: str + python_version: str + os_family: str + architecture: str + + +@dataclass(frozen=True, slots=True) +class ControllerWorkspace: + root: Path = field(repr=False) + acquisition: Path = field(repr=False) + worker: Path = field(repr=False) + + +@dataclass(frozen=True, slots=True) +class TrustedStack: + acquire_pull_snapshot: Callable[..., Iterator[object]] + acquisition_refused: type[Exception] + worker_request: Callable[..., object] + run_worker: Callable[[object], object] + render_worker: Callable[[object], dict[str, object]] + + +RuntimeFactsProvider = Callable[[], RuntimeFacts] +GitProbe = Callable[[Path], None] +WorkspaceFactory = Callable[[Path], Iterator[ControllerWorkspace]] + + +def current_runtime_facts() -> RuntimeFacts: + return RuntimeFacts( + sys.implementation.name, + platform.python_version(), + platform.system(), + platform.machine(), + ) + + +def _has_reparse(info: os.stat_result) -> bool: + return bool( + getattr(info, "st_file_attributes", 0) + & getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0) + ) + + +def _without_aliases(path: Path, *, directory: bool) -> Path: + if not path.is_absolute(): + raise ControllerRefused("unsafe_root_layout") + current = path + while True: + try: + info = current.lstat() + except OSError: + raise ControllerRefused("unsafe_root_layout") from None + if stat.S_ISLNK(info.st_mode) or _has_reparse(info): + raise ControllerRefused("unsafe_root_layout") + if current.parent == current: + break + current = current.parent + try: + resolved = path.resolve(strict=True) + info = resolved.lstat() + except OSError: + raise ControllerRefused("unsafe_root_layout") from None + expected = stat.S_ISDIR if directory else stat.S_ISREG + if _has_reparse(info) or not expected(info.st_mode): + raise ControllerRefused("unsafe_root_layout") + return resolved + + +def _overlaps(left: Path, right: Path) -> bool: + return left == right or left.is_relative_to(right) or right.is_relative_to(left) + + +def _control_root() -> Path: + source = Path(__file__) + if not source.is_absolute(): + raise ControllerRefused("unsafe_control_root") + try: + controller = _without_aliases(source, directory=False) + root = _without_aliases(controller.parent.parent, directory=True) + except ControllerRefused: + raise ControllerRefused("unsafe_control_root") from None + if controller != root / "scripts" / Path(__file__).name: + raise ControllerRefused("unsafe_control_root") + return root + + +def _validate_launch(runtime_facts_provider: RuntimeFactsProvider) -> None: + flags = sys.flags + if not ( + flags.isolated + and flags.ignore_environment + and flags.no_user_site + and flags.no_site + and flags.safe_path + and flags.dont_write_bytecode + ): + raise ControllerRefused("launch_not_isolated") + if dict(os.environ) != FIXED_ENVIRONMENT: + raise ControllerRefused("environment_mismatch") + facts = runtime_facts_provider() + if ( + facts.implementation, + facts.python_version, + facts.os_family, + facts.architecture, + ) != EXPECTED_RUNTIME: + raise ControllerRefused("runtime_mismatch") + try: + _without_aliases(Path(sys.executable), directory=False) + except ControllerRefused: + raise ControllerRefused("runtime_mismatch") from None + + +def _trusted_git_probe(control_root: Path) -> None: + expected = Path("/usr/bin/git") + selected = shutil.which("git", path=FIXED_ENVIRONMENT["PATH"]) + if selected != str(expected): + raise ControllerRefused("git_identity_mismatch") + try: + if _without_aliases(expected, directory=False) != expected: + raise ControllerRefused("git_identity_mismatch") + completed = subprocess.run( + [str(expected), "--version"], + cwd=control_root, + env=FIXED_ENVIRONMENT, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + timeout=5, + check=False, + ) + except (ControllerRefused, OSError, subprocess.TimeoutExpired): + raise ControllerRefused("git_identity_mismatch") from None + if ( + completed.returncode != 0 + or len(completed.stdout) > 128 + or _SAFE_GIT_VERSION.fullmatch(completed.stdout.decode("ascii", "strict")) + is None + ): + raise ControllerRefused("git_identity_mismatch") + + +def _validated_control_file(root: Path, relative: str) -> Path: + expected = root.joinpath(*relative.split("/")) + try: + resolved = _without_aliases(expected, directory=False) + except ControllerRefused: + raise ControllerRefused("unsafe_control_root") from None + if resolved != expected: + raise ControllerRefused("unsafe_control_root") + return resolved + + +def _load_module(name: str, expected: Path) -> ModuleType: + existing = sys.modules.get(name) + module = existing if existing is not None else importlib.import_module(name) + module_file = getattr(module, "__file__", None) + if type(module_file) is not str: + raise ControllerRefused("unsafe_control_root") + try: + actual = _without_aliases(Path(module_file), directory=False) + except ControllerRefused: + raise ControllerRefused("unsafe_control_root") from None + if actual != expected: + raise ControllerRefused("unsafe_control_root") + return module + + +def load_trusted_stack(control_root: Path) -> TrustedStack: + scripts = _without_aliases(control_root / "scripts", directory=True) + expected = { + name: _validated_control_file(control_root, f"scripts/{relative}") + for name, relative in _TRUSTED_MODULE_PATHS.items() + } + scripts_text = str(scripts) + if scripts_text not in sys.path: + sys.path.insert(0, scripts_text) + acquisition = _load_module( + "repo_sentinel_acquire", expected["repo_sentinel_acquire"] + ) + worker = _load_module( + "repo_sentinel_commit_authoritative", + expected["repo_sentinel_commit_authoritative"], + ) + for name in ( + "repo_sentinel_materialize", + "repo_sentinel_policy_bundle", + "repo_sentinel_reader", + ): + _load_module(name, expected[name]) + return TrustedStack( + acquisition.acquire_pull_snapshot, + acquisition.AcquisitionRefused, + worker.CommitAuthoritativeRequest, + worker.run_commit_authoritative, + worker.result_dict, + ) + + +def _parse_decimal(value: str) -> int: + if _DECIMAL.fullmatch(value) is None: + raise ControllerRefused("invalid_request") + return int(value) + + +def parse_request(argv: list[str]) -> ControllerRequest: + if len(argv) != len(_REQUEST_OPTIONS) * 2: + raise ControllerRefused("invalid_request") + values: dict[str, str] = {} + for index in range(0, len(argv), 2): + option, value = argv[index : index + 2] + if option not in _REQUEST_OPTIONS or option in values or type(value) is not str: + raise ControllerRefused("invalid_request") + values[option] = value + if set(values) != set(_REQUEST_OPTIONS): + raise ControllerRefused("invalid_request") + return ControllerRequest( + repository_id=_parse_decimal(values["--repository-id"]), + owner_id=_parse_decimal(values["--owner-id"]), + repository=values["--repository"], + remote_url=values["--remote-url"], + pull_number=_parse_decimal(values["--pull-number"]), + head_oid=values["--head-oid"], + policy_epoch=values["--policy-epoch"], + expected_policy_bundle_sha256=values["--policy-bundle-sha256"], + scratch_root=Path(values["--scratch-root"]), + scanner_artifact=Path(values["--scanner-artifact"]), + ) + + +def _validate_request( + request: ControllerRequest, control_root: Path +) -> tuple[Path, Path]: + if ( + type(request.repository_id) is not int + or type(request.owner_id) is not int + or type(request.pull_number) is not int + or not 0 < request.pull_number <= 2_147_483_647 + or type(request.repository) is not str + or type(request.remote_url) is not str + or type(request.head_oid) is not str + or type(request.policy_epoch) is not str + or type(request.expected_policy_bundle_sha256) is not str + or not isinstance(request.scratch_root, Path) + or not isinstance(request.scanner_artifact, Path) + ): + raise ControllerRefused("invalid_request") + if ( + request.repository_id != REPOSITORY_ID + or request.owner_id != OWNER_ID + or request.repository != REPOSITORY + or request.remote_url != REMOTE_URL + ): + raise ControllerRefused("repository_identity_mismatch") + if request.policy_epoch != POLICY_EPOCH: + raise ControllerRefused("unknown_policy_epoch") + if request.expected_policy_bundle_sha256 != POLICY_BUNDLE_SHA256: + raise ControllerRefused("policy_bundle_mismatch") + if _OID.fullmatch(request.head_oid) is None: + raise ControllerRefused("invalid_head_oid") + scratch = _without_aliases(request.scratch_root, directory=True) + artifact = _without_aliases(request.scanner_artifact, directory=False) + if _overlaps(control_root, scratch): + raise ControllerRefused("unsafe_root_layout") + return scratch, artifact + + +@contextmanager +def controller_workspace(scratch_root: Path) -> Iterator[ControllerWorkspace]: + root: Path | None = None + try: + root = Path( + tempfile.mkdtemp(prefix="repo-sentinel-controller-", dir=scratch_root) + ) + root.chmod(0o700) + acquisition = root / "acquisition" + worker = root / "worker" + acquisition.mkdir(mode=0o700) + worker.mkdir(mode=0o700) + yield ControllerWorkspace(root, acquisition, worker) + finally: + if root is not None: + try: + shutil.rmtree(root) + except OSError: + raise ControllerRefused("cleanup_failed") from None + + +def _valid_digest(value: object, *, optional: bool = False) -> bool: + return (optional and value is None) or ( + type(value) is str and _DIGEST.fullmatch(value) is not None + ) + + +def _valid_count(value: object) -> bool: + return type(value) is int and value >= 0 + + +def _worker_semantic_digest(payload: Mapping[str, object]) -> str: + semantic = { + key: value for key, value in payload.items() if key != "semantic_sha256" + } + encoded = json.dumps( + semantic, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + return hashlib.sha256( + _WORKER_SEMANTIC_DOMAIN + len(encoded).to_bytes(8, "big") + encoded + ).hexdigest() + + +def _validate_worker_result( + value: object, + request: ControllerRequest, +) -> dict[str, object]: + if type(value) is not dict or set(value) != _WORKER_RESULT_KEYS: + raise ControllerRefused("worker_result_invalid") + payload = dict(value) + verdict = payload["verdict"] + if verdict == "INFRASTRUCTURE_REFUSAL": + raise ControllerRefused("worker_infrastructure_refusal") + if verdict not in {"PASS", "SCANNER_FINDING", "POLICY_ADMISSION_FAILURE"}: + raise ControllerRefused("worker_result_invalid") + counts = ( + payload["files_total"], + payload["files_scanned"], + payload["files_policy_excluded"], + payload["files_scanner_skipped"], + payload["report_size"], + ) + if not all(_valid_count(item) for item in counts): + raise ControllerRefused("worker_result_invalid") + if verdict in {"PASS", "SCANNER_FINDING"} and payload["files_total"] != sum( + counts[1:4] + ): + raise ControllerRefused("worker_result_invalid") + required_digests = ( + payload["policy_bundle_sha256"], + payload["protected_manifest_sha256"], + payload["suppression_manifest_sha256"], + payload["coverage_policy_sha256"], + payload["scanner_artifact_sha256"], + payload["semantic_sha256"], + ) + if not all(_valid_digest(item) for item in required_digests): + raise ControllerRefused("worker_result_invalid") + if not _valid_digest(payload["report_sha256"], optional=True): + raise ControllerRefused("worker_result_invalid") + refusal = payload["refusal_code"] + if verdict == "POLICY_ADMISSION_FAILURE": + if refusal not in _POLICY_REFUSALS: + raise ControllerRefused("worker_result_invalid") + elif refusal is not None: + raise ControllerRefused("worker_result_invalid") + if ( + payload["policy_schema_version"] != 1 + or payload["policy_epoch"] != WORKER_POLICY_EPOCH + or payload["policy_bundle_sha256"] != POLICY_BUNDLE_SHA256 + or payload["repository_id"] != REPOSITORY_ID + or payload["head_oid"] != request.head_oid + or payload["scanner_distribution"] != "repo-sentinel-lite" + or payload["scanner_version"] != "0.8.1" + or payload["scanner_artifact_sha256"] != SCANNER_ARTIFACT_SHA256 + or payload["semantic_sha256"] != _worker_semantic_digest(payload) + ): + raise ControllerRefused("worker_result_invalid") + if (payload["report_sha256"] is None) != (payload["report_size"] == 0): + raise ControllerRefused("worker_result_invalid") + return payload + + +def _refusal(request: ControllerRequest | None, code: str) -> ControllerResult: + safe_head = ( + request.head_oid + if request is not None and _OID.fullmatch(request.head_oid) is not None + else None + ) + safe_epoch = ( + request.policy_epoch + if request is not None and request.policy_epoch == POLICY_EPOCH + else None + ) + return ControllerResult( + CONTROLLER_SCHEMA_VERSION, + REPOSITORY_ID, + safe_head, + safe_epoch, + POLICY_BUNDLE_SHA256, + None, + None, + ControllerOutcome.INFRASTRUCTURE_REFUSAL, + code if code in _REFUSAL_CODES else "unexpected_failure", + ) + + +def _authority_result( + request: ControllerRequest, worker: dict[str, object] +) -> ControllerResult: + return ControllerResult( + CONTROLLER_SCHEMA_VERSION, + REPOSITORY_ID, + request.head_oid, + POLICY_EPOCH, + POLICY_BUNDLE_SHA256, + worker, + str(worker["semantic_sha256"]), + ControllerOutcome.AUTHORITY_RESULT, + None, + ) + + +def run_controller( + request: ControllerRequest, + *, + control_root: Path, + stack: TrustedStack, + runtime_facts_provider: RuntimeFactsProvider = current_runtime_facts, + git_probe: GitProbe = _trusted_git_probe, + workspace_factory: WorkspaceFactory = controller_workspace, +) -> ControllerResult: + """Run trusted orchestration without publishing an authority status.""" + + try: + facts = runtime_facts_provider() + if ( + facts.implementation, + facts.python_version, + facts.os_family, + facts.architecture, + ) != EXPECTED_RUNTIME: + raise ControllerRefused("runtime_mismatch") + control = _without_aliases(control_root, directory=True) + scratch, artifact = _validate_request(request, control) + policy = _without_aliases( + control / "policy" / "repo-sentinel-authority" / "v1", + directory=True, + ) + git_probe(control) + with workspace_factory(scratch) as workspace: + workspace_root = _without_aliases(workspace.root, directory=True) + acquisition_root = _without_aliases(workspace.acquisition, directory=True) + worker_root = _without_aliases(workspace.worker, directory=True) + if ( + not acquisition_root.is_relative_to(workspace_root) + or not worker_root.is_relative_to(workspace_root) + or _overlaps(acquisition_root, worker_root) + or _overlaps(control, workspace_root) + ): + raise ControllerRefused("unsafe_root_layout") + try: + with stack.acquire_pull_snapshot( + request.remote_url, + request.pull_number, + request.head_oid, + acquisition_root, + ) as acquired: + snapshot = getattr(acquired, "snapshot", None) + repository = getattr(acquired, "repository", None) + if getattr( + snapshot, "commit_oid", None + ) != request.head_oid or not isinstance(repository, Path): + raise ControllerRefused("acquired_head_mismatch") + acquired_root = _without_aliases(repository, directory=True) + if ( + acquired_root == acquisition_root + or not acquired_root.is_relative_to(acquisition_root) + or _overlaps(acquired_root, worker_root) + or _overlaps(acquired_root, control) + ): + raise ControllerRefused("unsafe_root_layout") + worker_request = stack.worker_request( + REPOSITORY_ID, + request.head_oid, + acquired_root, + policy, + POLICY_BUNDLE_SHA256, + artifact, + worker_root, + ) + try: + raw_worker = stack.render_worker( + stack.run_worker(worker_request) + ) + except ControllerRefused: + raise + except Exception: + raise ControllerRefused( + "worker_infrastructure_refusal" + ) from None + worker = _validate_worker_result(raw_worker, request) + except ControllerRefused: + raise + except stack.acquisition_refused as error: + code = ( + "cleanup_failed" + if str(error) == "cleanup_failed" + else "acquisition_refused" + ) + raise ControllerRefused(code) from None + except Exception: + raise ControllerRefused("acquisition_refused") from None + return _authority_result(request, worker) + except ControllerRefused as error: + return _refusal(request, error.code) + except Exception: + return _refusal(request, "unexpected_failure") + + +def result_dict(result: ControllerResult) -> dict[str, object]: + return { + "controller_outcome": result.controller_outcome.value, + "controller_schema_version": result.controller_schema_version, + "fixed_refusal_code": result.fixed_refusal_code, + "head_oid": result.head_oid, + "policy_bundle_sha256": result.policy_bundle_sha256, + "policy_epoch": result.policy_epoch, + "repository_id": result.repository_id, + "worker_result": result.worker_result, + "worker_semantic_sha256": result.worker_semantic_sha256, + } + + +def render_result(result: ControllerResult) -> str: + return ( + json.dumps( + result_dict(result), + ensure_ascii=True, + sort_keys=True, + separators=(",", ":"), + ) + + "\n" + ) + + +def main(argv: list[str] | None = None) -> int: + request: ControllerRequest | None = None + try: + _validate_launch(current_runtime_facts) + control = _control_root() + os.chdir(control) + stack = load_trusted_stack(control) + request = parse_request(list(sys.argv[1:] if argv is None else argv)) + result = run_controller(request, control_root=control, stack=stack) + except ControllerRefused as error: + result = _refusal(request, error.code) + except Exception: + result = _refusal(request, "unexpected_failure") + sys.stdout.write(render_result(result)) + if result.controller_outcome is ControllerOutcome.INFRASTRUCTURE_REFUSAL: + return 2 + assert result.worker_result is not None + return 0 if result.worker_result["verdict"] == "PASS" else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) + + +__all__ = [ + "ControllerOutcome", + "ControllerRequest", + "ControllerResult", + "RuntimeFacts", + "TrustedStack", + "current_runtime_facts", + "load_trusted_stack", + "main", + "parse_request", + "render_result", + "result_dict", + "run_controller", +] From e755aab17eed88761b3dce8043c8221e411997b5 Mon Sep 17 00:00:00 2001 From: stacknil Date: Tue, 15 Sep 2026 19:39:30 +0800 Subject: [PATCH 2/8] test(security): enforce controller isolation --- ...test_repo_sentinel_authority_controller.py | 622 ++++++++++++++++++ 1 file changed, 622 insertions(+) create mode 100644 tests/test_repo_sentinel_authority_controller.py diff --git a/tests/test_repo_sentinel_authority_controller.py b/tests/test_repo_sentinel_authority_controller.py new file mode 100644 index 0000000..e05154f --- /dev/null +++ b/tests/test_repo_sentinel_authority_controller.py @@ -0,0 +1,622 @@ +"""Contract tests for the trusted Repo Sentinel authority controller.""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +import tempfile +import unittest +from contextlib import contextmanager +from dataclasses import replace +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import patch + +ROOT = Path(__file__).resolve().parents[1] +SCRIPTS = ROOT / "scripts" +sys.path.insert(0, str(SCRIPTS)) + +import repo_sentinel_authority_controller as controller # noqa: E402 + +HEAD_OID = "b" * 40 +OTHER_OID = "c" * 40 +EXACT_RUNTIME = controller.RuntimeFacts("cpython", "3.12.3", "Linux", "x86_64") + + +class FakeAcquisitionRefused(ValueError): + pass + + +def worker_payload( + verdict: str = "PASS", + *, + head_oid: str = HEAD_OID, + refusal_code: str | None = None, +) -> dict[str, object]: + semantic = verdict != "INFRASTRUCTURE_REFUSAL" + report = semantic and verdict != "POLICY_ADMISSION_FAILURE" + payload: dict[str, object] = { + "coverage_policy_sha256": "a" * 64 if semantic else None, + "files_policy_excluded": 1 if report else 0, + "files_scanned": 2 if report else 0, + "files_scanner_skipped": 0, + "files_total": 3 if semantic else 0, + "head_oid": head_oid if semantic else "", + "policy_bundle_sha256": (controller.POLICY_BUNDLE_SHA256 if semantic else None), + "policy_epoch": controller.WORKER_POLICY_EPOCH if semantic else None, + "policy_schema_version": 1 if semantic else None, + "protected_manifest_sha256": "b" * 64 if semantic else None, + "refusal_code": refusal_code, + "report_sha256": "c" * 64 if report else None, + "report_size": 123 if report else 0, + "repository_id": controller.REPOSITORY_ID if semantic else 0, + "scanner_artifact_sha256": ( + controller.SCANNER_ARTIFACT_SHA256 if semantic else None + ), + "scanner_distribution": "repo-sentinel-lite" if semantic else None, + "scanner_version": "0.8.1" if semantic else None, + "semantic_sha256": None, + "suppression_manifest_sha256": "d" * 64 if semantic else None, + "verdict": verdict, + } + if semantic: + payload["semantic_sha256"] = controller._worker_semantic_digest(payload) + return payload + + +class Harness: + def __init__(self) -> None: + self.temporary = tempfile.TemporaryDirectory(prefix="controller-contract-") + self.root = Path(self.temporary.name) + self.control = self.root / "control" + self.scratch = self.root / "scratch" + self.artifact = self.root / "scanner.whl" + (self.control / "policy" / "repo-sentinel-authority" / "v1").mkdir(parents=True) + self.scratch.mkdir() + self.artifact.write_bytes(b"fixture") + self.request = controller.ControllerRequest( + controller.REPOSITORY_ID, + controller.OWNER_ID, + controller.REPOSITORY, + controller.REMOTE_URL, + 7, + HEAD_OID, + controller.POLICY_EPOCH, + controller.POLICY_BUNDLE_SHA256, + self.scratch, + self.artifact, + ) + self.acquired_head = HEAD_OID + self.acquisition_error: Exception | None = None + self.worker = worker_payload() + self.worker_arguments: tuple[object, ...] | None = None + self.git_calls = 0 + + def close(self) -> None: + self.temporary.cleanup() + + @contextmanager + def acquire( + self, + _remote: str, + _pull: int, + _head: str, + scratch: Path, + ): + if self.acquisition_error is not None: + raise self.acquisition_error + repository = scratch / "objects.git" + repository.mkdir() + snapshot = SimpleNamespace(commit_oid=self.acquired_head) + yield SimpleNamespace(snapshot=snapshot, repository=repository) + + def make_worker_request(self, *arguments: object) -> object: + self.worker_arguments = arguments + return arguments + + def git_probe(self, _control: Path) -> None: + self.git_calls += 1 + + def stack(self) -> controller.TrustedStack: + return controller.TrustedStack( + self.acquire, + FakeAcquisitionRefused, + self.make_worker_request, + lambda _request: self.worker, + lambda result: result, + ) + + def run( + self, + request: controller.ControllerRequest | None = None, + **kwargs: object, + ) -> controller.ControllerResult: + return controller.run_controller( + request or self.request, + control_root=self.control, + stack=self.stack(), + runtime_facts_provider=kwargs.pop( + "runtime_facts_provider", lambda: EXACT_RUNTIME + ), + git_probe=kwargs.pop("git_probe", self.git_probe), + **kwargs, + ) + + +class HarnessTestCase(unittest.TestCase): + def harness(self) -> Harness: + harness = Harness() + self.addCleanup(harness.close) + return harness + + +class RequestContractTests(HarnessTestCase): + def argv(self, harness: Harness) -> list[str]: + request = harness.request + return [ + "--repository-id", + str(request.repository_id), + "--owner-id", + str(request.owner_id), + "--repository", + request.repository, + "--remote-url", + request.remote_url, + "--pull-number", + str(request.pull_number), + "--head-oid", + request.head_oid, + "--policy-epoch", + request.policy_epoch, + "--policy-bundle-sha256", + request.expected_policy_bundle_sha256, + "--scratch-root", + str(request.scratch_root), + "--scanner-artifact", + str(request.scanner_artifact), + ] + + def test_strict_parser_accepts_only_complete_unique_option_pairs(self) -> None: + harness = self.harness() + argv = self.argv(harness) + self.assertEqual(controller.parse_request(argv), harness.request) + for candidate in ( + argv[:-2], + [*argv, "--extra", "value"], + ["--repository-id", "1", *argv[2:-2], "--repository-id", "2"], + ["--repository-id=1130304545", *argv[2:]], + ): + with self.subTest(candidate=candidate): + with self.assertRaisesRegex( + controller.ControllerRefused, "invalid_request" + ): + controller.parse_request(candidate) + + def test_repository_policy_and_head_inputs_fail_closed(self) -> None: + harness = self.harness() + cases = { + "repository-id": ( + replace(harness.request, repository_id=1), + "repository_identity_mismatch", + ), + "owner-id": ( + replace(harness.request, owner_id=1), + "repository_identity_mismatch", + ), + "repository": ( + replace(harness.request, repository="other/repository"), + "repository_identity_mismatch", + ), + "remote": ( + replace(harness.request, remote_url="https://example.com/repo.git"), + "repository_identity_mismatch", + ), + "epoch": ( + replace(harness.request, policy_epoch="v2"), + "unknown_policy_epoch", + ), + "digest": ( + replace(harness.request, expected_policy_bundle_sha256="0" * 64), + "policy_bundle_mismatch", + ), + "head": ( + replace(harness.request, head_oid="HEAD\n::error::secret"), + "invalid_head_oid", + ), + "pull": (replace(harness.request, pull_number=0), "invalid_request"), + } + for name, (request, code) in cases.items(): + with self.subTest(name=name): + result = harness.run(request) + self.assertEqual( + result.controller_outcome.value, "INFRASTRUCTURE_REFUSAL" + ) + self.assertEqual(result.fixed_refusal_code, code) + rendered = controller.render_result(result) + self.assertNotIn("secret", rendered) + self.assertNotIn("::error::", rendered) + + def test_scratch_must_be_absolute_non_aliasing_and_outside_control(self) -> None: + harness = self.harness() + for request in ( + replace(harness.request, scratch_root=Path("relative")), + replace(harness.request, scratch_root=harness.control), + ): + with self.subTest(path=request.scratch_root): + self.assertEqual( + harness.run(request).fixed_refusal_code, + "unsafe_root_layout", + ) + + +class OrchestrationTests(HarnessTestCase): + def test_pass_is_bounded_and_worker_handoff_is_commit_intrinsic(self) -> None: + harness = self.harness() + result = harness.run() + + self.assertEqual(result.controller_outcome.value, "AUTHORITY_RESULT") + self.assertIsNone(result.fixed_refusal_code) + self.assertEqual(result.worker_result, harness.worker) + self.assertEqual( + result.worker_semantic_sha256, + harness.worker["semantic_sha256"], + ) + self.assertEqual(harness.git_calls, 1) + self.assertIsNotNone(harness.worker_arguments) + arguments = harness.worker_arguments + assert arguments is not None + self.assertEqual(arguments[0], controller.REPOSITORY_ID) + self.assertEqual(arguments[1], HEAD_OID) + self.assertEqual( + arguments[3], + harness.control / "policy" / "repo-sentinel-authority" / "v1", + ) + self.assertEqual(arguments[4], controller.POLICY_BUNDLE_SHA256) + self.assertEqual(arguments[5], harness.artifact) + self.assertNotIn(harness.request.pull_number, arguments[0:2]) + self.assertEqual(list(harness.scratch.iterdir()), []) + + def test_semantic_failure_results_are_preserved(self) -> None: + harness = self.harness() + for verdict, refusal in ( + ("SCANNER_FINDING", None), + ("POLICY_ADMISSION_FAILURE", "protected_control_mismatch"), + ): + with self.subTest(verdict=verdict): + harness.worker = worker_payload(verdict, refusal_code=refusal) + result = harness.run() + self.assertEqual(result.controller_outcome.value, "AUTHORITY_RESULT") + self.assertEqual(result.worker_result, harness.worker) + + def test_wrong_acquired_head_and_acquisition_refusal_have_no_authority( + self, + ) -> None: + harness = self.harness() + harness.acquired_head = OTHER_OID + result = harness.run() + self.assertEqual(result.fixed_refusal_code, "acquired_head_mismatch") + self.assertIsNone(result.worker_result) + + harness = self.harness() + harness.acquisition_error = FakeAcquisitionRefused("fetch_failed\nsecret") + result = harness.run() + self.assertEqual(result.fixed_refusal_code, "acquisition_refused") + self.assertNotIn("secret", controller.render_result(result)) + + def test_worker_infrastructure_or_invalid_result_has_no_authority(self) -> None: + harness = self.harness() + harness.worker = worker_payload( + "INFRASTRUCTURE_REFUSAL", refusal_code="scanner_failed" + ) + self.assertEqual( + harness.run().fixed_refusal_code, + "worker_infrastructure_refusal", + ) + + harness = self.harness() + harness.worker = worker_payload() + harness.worker["head_oid"] = OTHER_OID + self.assertEqual(harness.run().fixed_refusal_code, "worker_result_invalid") + + def test_worker_semantic_digest_is_recomputed(self) -> None: + harness = self.harness() + harness.worker["files_scanned"] = 1 + harness.worker["files_scanner_skipped"] = 1 + self.assertEqual(harness.run().fixed_refusal_code, "worker_result_invalid") + + def test_wrong_runtime_stops_before_git_or_acquisition(self) -> None: + harness = self.harness() + result = harness.run( + runtime_facts_provider=lambda: controller.RuntimeFacts( + "cpython", "3.12.14", "Linux", "x86_64" + ) + ) + self.assertEqual(result.fixed_refusal_code, "runtime_mismatch") + self.assertEqual(harness.git_calls, 0) + + def test_cleanup_failure_overrides_semantic_success(self) -> None: + harness = self.harness() + + @contextmanager + def cleanup_failure(scratch: Path): + root = scratch / "workspace" + acquisition = root / "acquisition" + worker = root / "worker" + acquisition.mkdir(parents=True) + worker.mkdir() + yield controller.ControllerWorkspace(root, acquisition, worker) + raise controller.ControllerRefused("cleanup_failed") + + result = harness.run(workspace_factory=cleanup_failure) + self.assertEqual(result.fixed_refusal_code, "cleanup_failed") + self.assertIsNone(result.worker_result) + + def test_repeated_execution_is_independent_of_neutralized_parent_noise( + self, + ) -> None: + harness = self.harness() + first = controller.result_dict(harness.run()) + hostile = { + "HOME": "hostile", + "PYTHONPATH": "hostile", + "REPO_SENTINEL_CONFIG": "hostile", + "XDG_CONFIG_HOME": "hostile", + } + with patch.dict(os.environ, hostile, clear=False): + second = controller.result_dict(harness.run()) + self.assertEqual(first, second) + + def test_target_python_and_shell_files_remain_data(self) -> None: + harness = self.harness() + marker = harness.root / "target-executed" + original = harness.acquire + + @contextmanager + def malicious_target(*args: object, **kwargs: object): + with original(*args, **kwargs) as acquired: + repository = acquired.repository + payload = f"from pathlib import Path;Path({str(marker)!r}).touch()\n" + for name in ( + "sitecustomize.py", + "usercustomize.py", + "hashlib.py", + "repo_sentinel.py", + "setup.py", + "run.sh", + ): + (repository / name).write_text(payload, encoding="utf-8") + self.assertNotIn(str(repository), sys.path) + yield acquired + + stack = replace(harness.stack(), acquire_pull_snapshot=malicious_target) + result = controller.run_controller( + harness.request, + control_root=harness.control, + stack=stack, + runtime_facts_provider=lambda: EXACT_RUNTIME, + git_probe=harness.git_probe, + ) + self.assertEqual(result.controller_outcome.value, "AUTHORITY_RESULT") + self.assertFalse(marker.exists()) + + +def replace_flags(flags: SimpleNamespace, **changes: object) -> SimpleNamespace: + values = vars(flags).copy() + values.update(changes) + return SimpleNamespace(**values) + + +class LaunchAndImportContractTests(HarnessTestCase): + def isolated_flags(self) -> SimpleNamespace: + return SimpleNamespace( + isolated=1, + ignore_environment=1, + no_user_site=1, + no_site=1, + safe_path=True, + dont_write_bytecode=1, + ) + + def test_launch_requires_exact_fixed_environment_and_isolation_flags(self) -> None: + with ( + patch.object(controller.sys, "flags", self.isolated_flags()), + patch.dict(os.environ, controller.FIXED_ENVIRONMENT, clear=True), + ): + controller._validate_launch(lambda: EXACT_RUNTIME) + + hostile = dict(controller.FIXED_ENVIRONMENT, PYTHONPATH="marker") + with ( + patch.object(controller.sys, "flags", self.isolated_flags()), + patch.dict(os.environ, hostile, clear=True), + ): + with self.assertRaisesRegex( + controller.ControllerRefused, "environment_mismatch" + ): + controller._validate_launch(lambda: EXACT_RUNTIME) + + flags = replace_flags(self.isolated_flags(), isolated=0) + with ( + patch.object(controller.sys, "flags", flags), + patch.dict(os.environ, controller.FIXED_ENVIRONMENT, clear=True), + ): + with self.assertRaisesRegex( + controller.ControllerRefused, "launch_not_isolated" + ): + controller._validate_launch(lambda: EXACT_RUNTIME) + + def test_trusted_module_loader_binds_files_under_control_scripts(self) -> None: + stack = controller.load_trusted_stack(ROOT) + self.assertEqual( + stack.acquire_pull_snapshot.__module__, "repo_sentinel_acquire" + ) + self.assertEqual( + stack.run_worker.__module__, + "repo_sentinel_commit_authoritative", + ) + for name, relative in controller._TRUSTED_MODULE_PATHS.items(): + module = sys.modules[name] + self.assertEqual( + Path(module.__file__).resolve(), + (SCRIPTS / relative).resolve(), + ) + + def test_rendered_refusal_never_contains_raw_exception_or_workflow_commands( + self, + ) -> None: + harness = self.harness() + + @contextmanager + def injected(*_args: object, **_kwargs: object): + raise RuntimeError("::warning::\n\x1b[31msecret-path") + yield + + stack = replace(harness.stack(), acquire_pull_snapshot=injected) + result = controller.run_controller( + harness.request, + control_root=harness.control, + stack=stack, + runtime_facts_provider=lambda: EXACT_RUNTIME, + git_probe=harness.git_probe, + ) + rendered = controller.render_result(result) + self.assertNotIn("::warning::", rendered) + self.assertNotIn("secret-path", rendered) + self.assertNotIn("\x1b", rendered) + self.assertEqual(rendered.count("\n"), 1) + + +class BootstrapTests(unittest.TestCase): + def bootstrap_source(self) -> str: + return (SCRIPTS / "repo_sentinel_authority_bootstrap.sh").read_text( + encoding="utf-8" + ) + + def test_bootstrap_contract_is_static_and_mutation_sensitive(self) -> None: + source = self.bootstrap_source() + + def satisfies_contract(candidate: str) -> bool: + required = ( + "exec /usr/bin/env -i", + "PATH=/usr/bin:/bin", + '"$trusted_python" -I -S -B "$controller"', + "unset LD_PRELOAD LD_LIBRARY_PATH", + "unset BASH_ENV ENV CDPATH GLOBIGNORE", + ) + return all(item in candidate for item in required) + + self.assertTrue(satisfies_contract(source)) + mutations = ( + source.replace("env -i", "env"), + source.replace("PATH=/usr/bin:/bin", "PATH=$PATH"), + source.replace( + '"$trusted_python" -I -S -B "$controller"', + 'python -I -S -B "$controller"', + ), + source.replace(" -I -S -B ", " -S -B "), + ) + for mutation in mutations: + with self.subTest(): + self.assertFalse(satisfies_contract(mutation)) + self.assertNotIn("eval ", source) + self.assertNotIn("source ", source) + + @unittest.skipIf(os.name == "nt", "POSIX bootstrap executes on Linux CI") + def test_hostile_parent_cannot_shadow_python_stdlib_or_git(self) -> None: + with tempfile.TemporaryDirectory(prefix="bootstrap-hostile-") as temporary: + root = Path(temporary) + hostile = root / "hostile" + fake_bin = root / "bin" + home = root / "home" + hostile.mkdir() + fake_bin.mkdir() + home.mkdir() + marker = root / "marker" + output = root / "output.json" + startup = root / "startup.py" + probe = root / "probe.py" + marker_code = f"from pathlib import Path;Path({str(marker)!r}).touch()\n" + for name in ( + "hashlib.py", + "json.py", + "sitecustomize.py", + "usercustomize.py", + ): + (hostile / name).write_text(marker_code, encoding="utf-8") + (hostile / "marker.pth").write_text(marker_code, encoding="utf-8") + startup.write_text(marker_code, encoding="utf-8") + for name in ("git", "python"): + executable = fake_bin / name + executable.write_text( + f"#!/bin/sh\nprintf x >> {str(marker)!r}\nexit 99\n", + encoding="utf-8", + ) + executable.chmod(0o700) + probe.write_text( + "import json,os,shutil,subprocess,sys\n" + "from pathlib import Path\n" + "result=subprocess.run(['git','--version'],capture_output=True,check=True,text=True)\n" + "Path(sys.argv[1]).write_text(json.dumps({\n" + "'environment':dict(os.environ),\n" + "'executable':sys.executable,\n" + "'git':shutil.which('git'),\n" + "'git_version':result.stdout.strip(),\n" + "'isolated':sys.flags.isolated,\n" + "'no_site':sys.flags.no_site,\n" + "'dont_write_bytecode':sys.flags.dont_write_bytecode,\n" + "},sort_keys=True),encoding='utf-8')\n", + encoding="utf-8", + ) + environment = dict(os.environ) + environment.update( + { + "BASH_ENV": str(startup), + "ENV": str(startup), + "HOME": str(home), + "LD_LIBRARY_PATH": str(hostile), + "PATH": f"{fake_bin}{os.pathsep}{environment.get('PATH', '')}", + "PIP_CONFIG_FILE": str(root / "pip.ini"), + "PYTHONHOME": str(hostile), + "PYTHONINSPECT": "1", + "PYTHONPATH": str(hostile), + "PYTHONSTARTUP": str(startup), + "REPO_SENTINEL_CONFIG": "hostile", + "VIRTUAL_ENV": str(hostile), + "XDG_CONFIG_HOME": str(hostile), + } + ) + completed = subprocess.run( + [ + "/bin/sh", + str(SCRIPTS / "repo_sentinel_authority_bootstrap.sh"), + str(Path(sys.executable).resolve()), + str(probe), + str(output), + ], + cwd=root, + env=environment, + stdin=subprocess.DEVNULL, + capture_output=True, + timeout=20, + check=False, + ) + self.assertEqual(completed.returncode, 0, completed.stderr) + observed = json.loads(output.read_text(encoding="utf-8")) + self.assertEqual(observed["environment"], controller.FIXED_ENVIRONMENT) + self.assertEqual( + Path(observed["executable"]).resolve(), + Path(sys.executable).resolve(), + ) + self.assertEqual(observed["git"], "/usr/bin/git") + self.assertRegex(observed["git_version"], r"^git version [0-9]") + self.assertEqual(observed["isolated"], 1) + self.assertEqual(observed["no_site"], 1) + self.assertEqual(observed["dont_write_bytecode"], 1) + self.assertFalse(marker.exists()) + self.assertEqual(completed.stdout, b"") + self.assertEqual(completed.stderr, b"") + + +if __name__ == "__main__": + unittest.main() From d170a63213dc15a1bba111d8be3388f9d43cf374 Mon Sep 17 00:00:00 2001 From: stacknil Date: Tue, 15 Sep 2026 19:51:41 +0800 Subject: [PATCH 3/8] fix(security): tighten controller result boundary --- scripts/repo_sentinel_authority_controller.py | 25 +++++++++++++++-- ...test_repo_sentinel_authority_controller.py | 28 ++++++++++++++----- 2 files changed, 43 insertions(+), 10 deletions(-) diff --git a/scripts/repo_sentinel_authority_controller.py b/scripts/repo_sentinel_authority_controller.py index b702d30..fe5153f 100644 --- a/scripts/repo_sentinel_authority_controller.py +++ b/scripts/repo_sentinel_authority_controller.py @@ -295,11 +295,14 @@ def _trusted_git_probe(control_root: Path) -> None: ) except (ControllerRefused, OSError, subprocess.TimeoutExpired): raise ControllerRefused("git_identity_mismatch") from None + try: + version = completed.stdout.decode("ascii", "strict") + except UnicodeDecodeError: + raise ControllerRefused("git_identity_mismatch") from None if ( completed.returncode != 0 or len(completed.stdout) > 128 - or _SAFE_GIT_VERSION.fullmatch(completed.stdout.decode("ascii", "strict")) - is None + or _SAFE_GIT_VERSION.fullmatch(version) is None ): raise ControllerRefused("git_identity_mismatch") @@ -362,7 +365,7 @@ def load_trusted_stack(control_root: Path) -> TrustedStack: def _parse_decimal(value: str) -> int: - if _DECIMAL.fullmatch(value) is None: + if len(value) > 10 or _DECIMAL.fullmatch(value) is None: raise ControllerRefused("invalid_request") return int(value) @@ -422,6 +425,14 @@ def _validate_request( raise ControllerRefused("policy_bundle_mismatch") if _OID.fullmatch(request.head_oid) is None: raise ControllerRefused("invalid_head_oid") + for path in (request.scratch_root, request.scanner_artifact): + value = os.fspath(path) + if ( + not 0 < len(value) <= 4096 + or re.search(r"[\x00-\x1f\x7f]", value) + or ".." in path.parts + ): + raise ControllerRefused("invalid_request") scratch = _without_aliases(request.scratch_root, directory=True) artifact = _without_aliases(request.scanner_artifact, directory=False) if _overlaps(control_root, scratch): @@ -516,8 +527,16 @@ def _validate_worker_result( if verdict == "POLICY_ADMISSION_FAILURE": if refusal not in _POLICY_REFUSALS: raise ControllerRefused("worker_result_invalid") + if ( + any(counts[1:4]) + or payload["report_sha256"] is not None + or payload["report_size"] != 0 + ): + raise ControllerRefused("worker_result_invalid") elif refusal is not None: raise ControllerRefused("worker_result_invalid") + elif payload["report_sha256"] is None or payload["report_size"] == 0: + raise ControllerRefused("worker_result_invalid") if ( payload["policy_schema_version"] != 1 or payload["policy_epoch"] != WORKER_POLICY_EPOCH diff --git a/tests/test_repo_sentinel_authority_controller.py b/tests/test_repo_sentinel_authority_controller.py index e05154f..2ef42bb 100644 --- a/tests/test_repo_sentinel_authority_controller.py +++ b/tests/test_repo_sentinel_authority_controller.py @@ -187,6 +187,7 @@ def test_strict_parser_accepts_only_complete_unique_option_pairs(self) -> None: [*argv, "--extra", "value"], ["--repository-id", "1", *argv[2:-2], "--repository-id", "2"], ["--repository-id=1130304545", *argv[2:]], + ["--repository-id", "9" * 100, *argv[2:]], ): with self.subTest(candidate=candidate): with self.assertRaisesRegex( @@ -240,15 +241,22 @@ def test_repository_policy_and_head_inputs_fail_closed(self) -> None: def test_scratch_must_be_absolute_non_aliasing_and_outside_control(self) -> None: harness = self.harness() - for request in ( - replace(harness.request, scratch_root=Path("relative")), - replace(harness.request, scratch_root=harness.control), + for request, code in ( + ( + replace(harness.request, scratch_root=Path("relative")), + "unsafe_root_layout", + ), + ( + replace(harness.request, scratch_root=harness.control), + "unsafe_root_layout", + ), + ( + replace(harness.request, scratch_root=Path("bad\npath")), + "invalid_request", + ), ): with self.subTest(path=request.scratch_root): - self.assertEqual( - harness.run(request).fixed_refusal_code, - "unsafe_root_layout", - ) + self.assertEqual(harness.run(request).fixed_refusal_code, code) class OrchestrationTests(HarnessTestCase): @@ -577,10 +585,16 @@ def test_hostile_parent_cannot_shadow_python_stdlib_or_git(self) -> None: "LD_LIBRARY_PATH": str(hostile), "PATH": f"{fake_bin}{os.pathsep}{environment.get('PATH', '')}", "PIP_CONFIG_FILE": str(root / "pip.ini"), + "PIP_INDEX_URL": "https://example.invalid/simple", + "PYTHONBREAKPOINT": "marker.breakpoint", "PYTHONHOME": str(hostile), "PYTHONINSPECT": "1", "PYTHONPATH": str(hostile), + "PYTHONPYCACHEPREFIX": str(hostile), + "PYTHONSAFEPATH": "0", "PYTHONSTARTUP": str(startup), + "PYTHONUSERBASE": str(hostile), + "PYTHONWARNINGS": "error", "REPO_SENTINEL_CONFIG": "hostile", "VIRTUAL_ENV": str(hostile), "XDG_CONFIG_HOME": str(hostile), From dce1fee2059bcfbfeb50ad3b19ac546f5b9d313c Mon Sep 17 00:00:00 2001 From: stacknil Date: Tue, 15 Sep 2026 19:52:09 +0800 Subject: [PATCH 4/8] docs(security): record controller trust boundary --- .../repo-sentinel-authority-controller-v1.md | 369 ++++++++++++++++++ 1 file changed, 369 insertions(+) create mode 100644 docs/design-decisions/repo-sentinel-authority-controller-v1.md diff --git a/docs/design-decisions/repo-sentinel-authority-controller-v1.md b/docs/design-decisions/repo-sentinel-authority-controller-v1.md new file mode 100644 index 0000000..3cc1a95 --- /dev/null +++ b/docs/design-decisions/repo-sentinel-authority-controller-v1.md @@ -0,0 +1,369 @@ +# Repo Sentinel Authority Controller v1 + +Status: candidate controller boundary for independent review. It is not +production-active and publishes no GitHub status. + +## Problem + +The commit-intrinsic worker defines the frozen predicate `P_v(H)`, but it assumes +that a trusted Python process has already imported the reviewed worker and +policy code. That assumption leaves a missing boundary between future trusted +GitHub-hosted execution and semantic evaluation. + +The controller closes that gap for policy epoch `v1`: + +```text +trusted GitHub-hosted execution envelope + -> sanitized bootstrap + -> trusted controller Python process + -> exact H acquisition + -> commit-intrinsic P_v(H) + -> bounded canonical controller result +``` + +It does not authenticate a GitHub Actions invocation, publish authority, replace +the future signer, or certify a test-merge result. + +## Invariant + +Authority evaluation is reachable only after all of the following hold: + +```text +absolute trusted CPython 3.12.3 Linux x86_64 ++ isolated interpreter flags ++ exact fixed child environment ++ trusted control-root module identity ++ fixed production repository identity ++ exact refs/pull//head == H ++ fixed v1 policy bundle identity ++ distinct control, acquisition, and worker roots +``` + +The controller returns one of two outcomes: + +```text +AUTHORITY_RESULT +INFRASTRUCTURE_REFUSAL +``` + +Only worker `PASS`, `SCANNER_FINDING`, and `POLICY_ADMISSION_FAILURE` results +become `AUTHORITY_RESULT`. Acquisition, runtime, bootstrap, orchestration, +worker-infrastructure, and cleanup failures become `INFRASTRUCTURE_REFUSAL`. +An infrastructure refusal carries no worker result or semantic digest. + +## Design Decision + +### Separate bootstrap and controller + +`scripts/repo_sentinel_authority_bootstrap.sh` is a minimal POSIX bootstrap. It +does not discover Python from `PATH`, activate a virtual environment, source a +profile, import Python, or inspect repository content. The future trusted launch +configuration must supply: + +1. an absolute trusted Python interpreter path; +2. the absolute reviewed controller entrypoint; +3. the controller request arguments. + +The bootstrap clears known native-loader and shell-startup variables before it +executes `/usr/bin/env -i`. The Python child receives exactly: + +```text +PATH=/usr/bin:/bin +LANG=C.UTF-8 +LC_ALL=C.UTF-8 +TZ=UTC +``` + +It launches the absolute interpreter with: + +```text +-I -S -B +``` + +`-I` ignores Python environment and user-site influence, `-S` prevents site and +`.pth` startup processing, and `-B` prevents bytecode writes into the reviewed +control checkout. + +The shell cannot undo loader or startup effects that occurred before the shell +process began. The future workflow must therefore launch the bootstrap itself +without target-controlled `LD_PRELOAD`, `LD_LIBRARY_PATH`, `DYLD_*`, `BASH_ENV`, +or `ENV` state. On the approved Linux runtime, `/bin/sh` is the trusted, +non-interactive shell supplied by the runner image. + +### Exact runtime and Git identity + +The controller requires exactly: + +```text +implementation: cpython +python_version: 3.12.3 +os_family: Linux +architecture: x86_64 +``` + +`3.12`, a later `3.12.x`, `3.13`, and PyPy are not equivalent for epoch `v1`. +The workflow must pass the resolved absolute interpreter path rather than a +caller `PATH` lookup or a virtual-environment shim. + +The frozen acquisition module invokes `git`. The controller therefore fixes +`PATH` to `/usr/bin:/bin`, requires that lookup to select `/usr/bin/git`, rejects +symlink or reparse aliases, and runs a bounded `git --version` probe before +acquisition. A repository-local or caller-supplied `git` cannot participate. + +### Trusted control root + +The controller derives `CONTROL_ROOT` from its own absolute entrypoint. It does +not accept a caller-selected module root. Before imports, it rejects aliases and +requires the controller, acquisition, reader, materializer, policy-bundle, and +commit-worker modules to resolve to their exact reviewed files below: + +```text +CONTROL_ROOT/scripts/ +``` + +Only that scripts directory is explicitly added to `sys.path` after isolated +interpreter startup. `TARGET_ROOT`, the materialized commit, the caller's current +directory, `HOME`, and user site-packages are never added. + +### Request contract + +The strict request has exactly these fields: + +```text +repository_id +owner_id +repository +remote_url +pull_number +head_oid +policy_epoch +expected_policy_bundle_sha256 +scratch_root +scanner_artifact +``` + +Repository identity is fixed to: + +```text +repository_id = 1130304545 +owner_id = 219124580 +repository = stacknil/sec-writeups-public +remote_url = https://github.com/stacknil/sec-writeups-public.git +``` + +The parser accepts each named option exactly once. It does not accept shell +fragments, environment-variable names, module names, command names, or a policy +bundle path. All subprocess calls use argument arrays and `shell=False`. + +The operational pull number routes acquisition but is not passed into semantic +authority. No base SHA, merge base, current-main identity, test-merge SHA, PR +title, PR body, run ID, timestamp, or GitHub status field enters `P_v(H)`. + +### Static policy selection + +The only supported selector is: + +```text +v1 -> policy/repo-sentinel-authority/v1/ +``` + +The expected bundle digest is fixed in controller code: + +```text +6f25ebb773ce1453e8de623bca5aaecc936f1f188288f8df20aedeadb3bf4612 +``` + +The caller must repeat that digest, but cannot choose a different digest or +bundle path. Unknown epochs and digest disagreement fail before acquisition. +Future policy migration requires a separately reviewed controller change. + +### Root separation and cleanup + +The caller supplies an existing absolute `SCRATCH_ROOT`. The controller rejects +non-directories, symlinks, reparse points, unresolved paths, and overlap with +`CONTROL_ROOT`. + +Inside verifier-owned scratch it creates one private workspace with sibling +roots: + +```text +controller workspace/ + acquisition/ + worker/ +``` + +The fresh Git database must be a descendant of `acquisition/`; worker scratch is +the distinct sibling `worker/`. The target cannot replace controller modules or +policy files through layout. The acquisition context cleans its Git database, +the worker cleans materialization and evidence, and the controller removes the +outer workspace. Cleanup failure overrides any earlier semantic success. + +### Exact acquisition and worker handoff + +The controller reuses the frozen acquisition contract to fetch only: + +```text +refs/pull//head +``` + +It requires the acquired snapshot commit and fresh object database to bind exact +expected `H`. A mismatch is an infrastructure refusal. It then constructs the +frozen worker request from: + +```text +repository ID +exact H object database +trusted v1 policy directory +trusted expected bundle digest +digest-verified scanner wheel path +private worker scratch +``` + +Repository content remains data. The target does not supply imports, commands, +environment variables, bundle selection, or executable control files. + +### Bounded result envelope + +Stdout contains exactly one canonical JSON object and a final newline. The +controller envelope contains: + +```text +controller_schema_version +repository_id +head_oid +policy_epoch +policy_bundle_sha256 +worker_result +worker_semantic_sha256 +controller_outcome +fixed_refusal_code +``` + +Before returning authority, the controller validates the exact worker schema, +fixed repository and epoch identities, result counts, digest syntax, semantic +verdict/refusal relationship, report presence, and the worker's domain-separated +semantic digest. A malformed or infrastructure worker result is not reflected +as authority. + +No raw report, path, URL, environment value, subprocess output, exception text, +token, or credential appears in the envelope. There are no untrusted display +fields, so newline, escape-sequence, and GitHub workflow-command injection are +not possible through diagnostics. The controller emits no free-form stderr. + +## Threat And Failure Model + +The production trust root will be: + +```text +GitHub-hosted runner ++ reviewed immutable workflow SHA ++ reviewed bootstrap and controller files +``` + +The controller assumes that the runner kernel, dynamic loader, filesystem, +trusted workflow definition, absolute interpreter, `/usr/bin/env`, `/bin/sh`, +and `/usr/bin/git` are trusted. It does not defend against compromise of those +components. + +After that launch boundary, it defends against: + +- hostile `PYTHON*`, `HOME`, `XDG_*`, `PIP_*`, `REPO_SENTINEL_*`, virtualenv, + shell-startup, loader, locale, timezone, and caller `PATH` state reaching + Python or Git children; +- shadow `hashlib`, `json`, `sitecustomize`, `usercustomize`, `.pth`, fake + Python, and fake Git artifacts in ambient or target-controlled locations; +- caller-selected repository, owner, remote, bundle, epoch, module, command, or + control paths; +- ref movement or an acquired commit different from exact expected `H`; +- overlap or aliases between trusted controls and verifier scratch; +- target files being imported or executed; +- malformed, contradictory, or identity-reflecting worker results; +- raw Git, scanner, exception, environment, or filesystem diagnostics reaching + stdout or stderr; +- stale semantic success surviving required cleanup failure. + +Expected infrastructure failures include runtime mismatch, unavailable trusted +Git, network or acquisition refusal, filesystem errors, worker infrastructure +refusal, and cleanup failure. Future publication must publish nothing for these +outcomes. + +## Rejected Alternatives + +1. Inherit the parent environment and delete known variables. + Rejected because a denylist cannot establish a complete process contract. +2. Discover Python or Git from caller `PATH`. + Rejected because repository or user-local executables could shadow trusted + tools. +3. Use `/usr/bin/env python` or activate a virtual environment. + Rejected because interpreter identity would depend on ambient state. +4. Import controller modules before isolated interpreter entry. + Rejected because `PYTHONPATH`, site hooks, and user configuration may have + already affected imports. +5. Import worker code from acquired or materialized `H`. + Rejected because the evaluated commit must remain data rather than control. +6. Allow caller-selected policy paths or digests. + Rejected because the target could select its own authority policy. +7. Reuse base-relative or test-merge identity in `P_v(H)`. + Rejected because those values are not intrinsic properties of exact `H`. +8. Convert infrastructure refusal into a semantic failure status. + Rejected because future publication must distinguish no authority from an + authoritative negative result. +9. Include raw diagnostics for convenience. + Rejected because paths, URLs, credentials, hostile text, or workflow commands + could cross the public result boundary. + +## Compatibility + +This change adds a controller boundary without modifying: + +- workflow files, triggers, permissions, or branch/ruleset configuration; +- GitHub App, OIDC, signer, Commit Status, replay, or ticket protocols; +- pull-head acquisition refspec, protocol, redirect, depth, tag, timeout, + repository-size, credential-isolation, or cleanup behavior; +- reader, materializer, base-relative worker, commit-worker, scanner, policy + bundle, baseline, or report schemas. + +The controller currently supports only production repository identity and epoch +`v1`. That deliberate narrowness is a compatibility boundary, not a general +controller framework. + +## Validation + +Focused tests cover: + +- trusted bootstrap environment and flags; +- exact and wrong runtime identities; +- unknown epoch, wrong bundle digest, wrong repository identity, invalid and + mismatched `H`, and invalid pull numbers; +- acquisition refusal and worker infrastructure refusal; +- worker `PASS`, `SCANNER_FINDING`, and `POLICY_ADMISSION_FAILURE` preservation; +- hostile Python, home, XDG, pip, virtualenv, Repo Sentinel, loader, and path + environment state; +- fake Python, fake Git, shadow standard-library modules, startup hooks, user + customizations, and `.pth` files; +- target Python and executable-looking files remaining data; +- fixed stdout, no raw diagnostics, cleanup override, and repeated deterministic + results; +- mutation controls for environment inheritance, caller `PATH`, relative Python, + missing isolated mode, policy selection, exact-head binding, target imports, + raw exception output, and stale success after cleanup failure. + +The existing commit-worker, base-relative worker, acquisition, reader, +materializer, gate, and full repository suites remain required regressions. + +A production-style HTTPS probe must run on exact CPython 3.12.3 Linux x86_64 +through the bootstrap. Public evidence records only trusted Python and Git +identity, repository ID, `H`, epoch, bundle digest, worker verdict and semantic +digest, controller outcome, and cleanup status. It must not record local paths, +tokens, or raw subprocess output. + +## Rollback + +No workflow or status publication is activated. Before a future workflow adopts +this controller, rollback is deletion of the bootstrap, controller, tests, and +this decision record. + +After activation, rollback must point the immutable trusted workflow to a +previously reviewed controller/bootstrap SHA. It must not select policy or code +from target `H`, regenerate epoch `v1`, or weaken an infrastructure refusal into +a published semantic result. From cd4b181c82774870267f72097405def2ea3abac6 Mon Sep 17 00:00:00 2001 From: stacknil Date: Tue, 15 Sep 2026 19:57:20 +0800 Subject: [PATCH 5/8] test(security): isolate launch fixture identity --- tests/test_repo_sentinel_authority_controller.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/tests/test_repo_sentinel_authority_controller.py b/tests/test_repo_sentinel_authority_controller.py index 2ef42bb..e9ffbb8 100644 --- a/tests/test_repo_sentinel_authority_controller.py +++ b/tests/test_repo_sentinel_authority_controller.py @@ -428,8 +428,10 @@ def isolated_flags(self) -> SimpleNamespace: ) def test_launch_requires_exact_fixed_environment_and_isolation_flags(self) -> None: + executable = str(self.harness().artifact) with ( patch.object(controller.sys, "flags", self.isolated_flags()), + patch.object(controller.sys, "executable", executable), patch.dict(os.environ, controller.FIXED_ENVIRONMENT, clear=True), ): controller._validate_launch(lambda: EXACT_RUNTIME) @@ -437,6 +439,7 @@ def test_launch_requires_exact_fixed_environment_and_isolation_flags(self) -> No hostile = dict(controller.FIXED_ENVIRONMENT, PYTHONPATH="marker") with ( patch.object(controller.sys, "flags", self.isolated_flags()), + patch.object(controller.sys, "executable", executable), patch.dict(os.environ, hostile, clear=True), ): with self.assertRaisesRegex( @@ -447,6 +450,7 @@ def test_launch_requires_exact_fixed_environment_and_isolation_flags(self) -> No flags = replace_flags(self.isolated_flags(), isolated=0) with ( patch.object(controller.sys, "flags", flags), + patch.object(controller.sys, "executable", executable), patch.dict(os.environ, controller.FIXED_ENVIRONMENT, clear=True), ): with self.assertRaisesRegex( From 6cee46c37e3b4ac9e42dc1c499d16869fcc34e10 Mon Sep 17 00:00:00 2001 From: stacknil Date: Tue, 15 Sep 2026 20:25:54 +0800 Subject: [PATCH 6/8] test(security): assert silent refusal boundary --- ...test_repo_sentinel_authority_controller.py | 20 +++++++++++-------- 1 file changed, 12 insertions(+), 8 deletions(-) diff --git a/tests/test_repo_sentinel_authority_controller.py b/tests/test_repo_sentinel_authority_controller.py index e9ffbb8..c014682 100644 --- a/tests/test_repo_sentinel_authority_controller.py +++ b/tests/test_repo_sentinel_authority_controller.py @@ -2,13 +2,14 @@ from __future__ import annotations +import io import json import os import subprocess import sys import tempfile import unittest -from contextlib import contextmanager +from contextlib import contextmanager, redirect_stderr from dataclasses import replace from pathlib import Path from types import SimpleNamespace @@ -485,14 +486,17 @@ def injected(*_args: object, **_kwargs: object): yield stack = replace(harness.stack(), acquire_pull_snapshot=injected) - result = controller.run_controller( - harness.request, - control_root=harness.control, - stack=stack, - runtime_facts_provider=lambda: EXACT_RUNTIME, - git_probe=harness.git_probe, - ) + stderr = io.StringIO() + with redirect_stderr(stderr): + result = controller.run_controller( + harness.request, + control_root=harness.control, + stack=stack, + runtime_facts_provider=lambda: EXACT_RUNTIME, + git_probe=harness.git_probe, + ) rendered = controller.render_result(result) + self.assertEqual(stderr.getvalue(), "") self.assertNotIn("::warning::", rendered) self.assertNotIn("secret-path", rendered) self.assertNotIn("\x1b", rendered) From 5c40702ac003e0e236f75440848de457e26fee34 Mon Sep 17 00:00:00 2001 From: stacknil Date: Wed, 16 Sep 2026 10:35:53 +0800 Subject: [PATCH 7/8] fix(security): enforce exact worker identity types --- scripts/repo_sentinel_authority_controller.py | 17 +++++++- ...test_repo_sentinel_authority_controller.py | 41 +++++++++++++++++++ 2 files changed, 57 insertions(+), 1 deletion(-) diff --git a/scripts/repo_sentinel_authority_controller.py b/scripts/repo_sentinel_authority_controller.py index fe5153f..c86908a 100644 --- a/scripts/repo_sentinel_authority_controller.py +++ b/scripts/repo_sentinel_authority_controller.py @@ -494,10 +494,26 @@ def _validate_worker_result( raise ControllerRefused("worker_result_invalid") payload = dict(value) verdict = payload["verdict"] + if type(verdict) is not str: + raise ControllerRefused("worker_result_invalid") if verdict == "INFRASTRUCTURE_REFUSAL": raise ControllerRefused("worker_infrastructure_refusal") if verdict not in {"PASS", "SCANNER_FINDING", "POLICY_ADMISSION_FAILURE"}: raise ControllerRefused("worker_result_invalid") + identity_strings = ( + payload["policy_epoch"], + payload["head_oid"], + payload["scanner_distribution"], + payload["scanner_version"], + ) + refusal = payload["refusal_code"] + if ( + type(payload["policy_schema_version"]) is not int + or type(payload["repository_id"]) is not int + or not all(type(item) is str for item in identity_strings) + or (refusal is not None and type(refusal) is not str) + ): + raise ControllerRefused("worker_result_invalid") counts = ( payload["files_total"], payload["files_scanned"], @@ -523,7 +539,6 @@ def _validate_worker_result( raise ControllerRefused("worker_result_invalid") if not _valid_digest(payload["report_sha256"], optional=True): raise ControllerRefused("worker_result_invalid") - refusal = payload["refusal_code"] if verdict == "POLICY_ADMISSION_FAILURE": if refusal not in _POLICY_REFUSALS: raise ControllerRefused("worker_result_invalid") diff --git a/tests/test_repo_sentinel_authority_controller.py b/tests/test_repo_sentinel_authority_controller.py index c014682..4f5ec0b 100644 --- a/tests/test_repo_sentinel_authority_controller.py +++ b/tests/test_repo_sentinel_authority_controller.py @@ -329,6 +329,47 @@ def test_worker_infrastructure_or_invalid_result_has_no_authority(self) -> None: harness.worker["head_oid"] = OTHER_OID self.assertEqual(harness.run().fixed_refusal_code, "worker_result_invalid") + def test_worker_identity_fields_require_exact_types(self) -> None: + class StringAlias(str): + pass + + cases = ( + ("policy_schema_version", True), + ("policy_schema_version", 1.0), + ("repository_id", float(controller.REPOSITORY_ID)), + ("verdict", StringAlias("PASS")), + ("policy_epoch", StringAlias(controller.WORKER_POLICY_EPOCH)), + ("head_oid", StringAlias(HEAD_OID)), + ("scanner_distribution", StringAlias("repo-sentinel-lite")), + ("scanner_version", StringAlias("0.8.1")), + ) + for field, value in cases: + with self.subTest(field=field, value=value): + harness = self.harness() + harness.worker[field] = value + harness.worker["semantic_sha256"] = controller._worker_semantic_digest( + harness.worker + ) + result = harness.run() + self.assertEqual( + result.controller_outcome.value, + "INFRASTRUCTURE_REFUSAL", + ) + self.assertEqual(result.fixed_refusal_code, "worker_result_invalid") + self.assertIsNone(result.worker_result) + self.assertIsNone(result.worker_semantic_sha256) + + harness = self.harness() + harness.worker = worker_payload( + "POLICY_ADMISSION_FAILURE", + refusal_code=StringAlias("protected_control_mismatch"), + ) + result = harness.run() + self.assertEqual(result.controller_outcome.value, "INFRASTRUCTURE_REFUSAL") + self.assertEqual(result.fixed_refusal_code, "worker_result_invalid") + self.assertIsNone(result.worker_result) + self.assertIsNone(result.worker_semantic_sha256) + def test_worker_semantic_digest_is_recomputed(self) -> None: harness = self.harness() harness.worker["files_scanned"] = 1 From 33514ea6d03e374a2acd91fbd204fb1ead09433d Mon Sep 17 00:00:00 2001 From: stacknil Date: Wed, 16 Sep 2026 10:41:48 +0800 Subject: [PATCH 8/8] test(security): reuse semantic digest assertion --- tests/test_repo_sentinel_authority_controller.py | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/tests/test_repo_sentinel_authority_controller.py b/tests/test_repo_sentinel_authority_controller.py index 4f5ec0b..5596daf 100644 --- a/tests/test_repo_sentinel_authority_controller.py +++ b/tests/test_repo_sentinel_authority_controller.py @@ -152,6 +152,13 @@ def harness(self) -> Harness: self.addCleanup(harness.close) return harness + def assert_worker_semantic_digest( + self, + result: controller.ControllerResult, + expected: str | None, + ) -> None: + self.assertEqual(result.worker_semantic_sha256, expected) + class RequestContractTests(HarnessTestCase): def argv(self, harness: Harness) -> list[str]: @@ -268,8 +275,8 @@ def test_pass_is_bounded_and_worker_handoff_is_commit_intrinsic(self) -> None: self.assertEqual(result.controller_outcome.value, "AUTHORITY_RESULT") self.assertIsNone(result.fixed_refusal_code) self.assertEqual(result.worker_result, harness.worker) - self.assertEqual( - result.worker_semantic_sha256, + self.assert_worker_semantic_digest( + result, harness.worker["semantic_sha256"], ) self.assertEqual(harness.git_calls, 1) @@ -357,7 +364,7 @@ class StringAlias(str): ) self.assertEqual(result.fixed_refusal_code, "worker_result_invalid") self.assertIsNone(result.worker_result) - self.assertIsNone(result.worker_semantic_sha256) + self.assert_worker_semantic_digest(result, None) harness = self.harness() harness.worker = worker_payload( @@ -368,7 +375,7 @@ class StringAlias(str): self.assertEqual(result.controller_outcome.value, "INFRASTRUCTURE_REFUSAL") self.assertEqual(result.fixed_refusal_code, "worker_result_invalid") self.assertIsNone(result.worker_result) - self.assertIsNone(result.worker_semantic_sha256) + self.assert_worker_semantic_digest(result, None) def test_worker_semantic_digest_is_recomputed(self) -> None: harness = self.harness()