diff --git a/docs/repo-sentinel-baseline-review.md b/docs/repo-sentinel-baseline-review.md index 3f39e2d..b4a1b93 100644 --- a/docs/repo-sentinel-baseline-review.md +++ b/docs/repo-sentinel-baseline-review.md @@ -382,6 +382,88 @@ add `pull_request_target`, secrets, caches, Check API writes, permissions or repository enforcement. Rollback removes the acquisition helper/tests/docs; the merged reader and materializer remain independently usable. +### Authoritative Worker Core + +`scripts/repo_sentinel_authoritative.py` adds the data-only worker core for a +future authoritative gate. It is deliberately separate from GitHub event +parsing and from the dedicated-App publisher. The caller supplies validated +repository identity, pull request number, exact base/head object IDs, a trusted +base object database, a scratch root and an evidence root. The worker neither +derives identity from pull request text nor publishes a status. + +The worker reads the trusted base with `read_snapshot()`, acquires the exact +head with `acquire_pull_snapshot()`, and computes D1 directly from immutable +snapshot records. A path is changed when it is absent from base or its mode or +blob object ID differs; a path is deleted when it is absent from head. There is +no rename inference and no GitHub changed-files API input. Indexed maps plus +deterministic sorting keep the operation O(n log n). + +Before materialization or scanner execution, changed and deleted paths are +checked against the protected control plane: + +- `.reposentinel.toml` and `.reposentinel-baseline.json`; +- `.github/workflows/**` and `.github/actions/**`; +- the existing gate, acquisition, reader, materializer, authoritative worker + and integration-test scripts. + +Protected-path matching uses the materializer's portable-v1 ASCII case-alias +model: `A-Z` map to `a-z`, while every other code point remains unchanged. +This closes host aliases such as `.RepoSentinel.toml` without introducing +general Unicode case folding or weakening exact path and subtree boundaries. + +The exact `repo-sentinel-lite==0.8.1` wheel was also inspected for target-owned +suppression mechanisms. Its inline pattern is the concatenation of +`r"repo-sentinel:\s*"` and +`r"allow(?:\s+(?P[A-Za-z0-9_.\-, ]+))?"`, matched case-insensitively. +With no rule list it allows all findings on the finding line; with a +comma-separated kind/rule-ID list it allows matching findings. The scanner +checks the finding line and the immediately preceding line. It attempts +`utf-8`, `utf-8-sig`, `utf-16` and `cp1252` text decoding. + +The worker uses a conservative source-control policy: if any changed or +deleted file contains that directive in either the base or head snapshot, the +change is classified as protected rather than scanned. This can block an +ordinary edit to a file that already carries a legitimate suppression, but it +prevents pull-request-owned source annotations from weakening the +authoritative result without introducing a bypass channel. + +Scanner configuration resolution in `0.8.1` reads only +`/.reposentinel.toml`; it does not search parent directories, the +home directory or environment-selected alternate paths. A changed or deleted +root config is protected. The default baseline is always disabled. When the +base snapshot contains `.reposentinel-baseline.json`, its exact bytes are +written exclusively to a verifier-owned temporary file and passed with an +explicit `--baseline` argument. Head-owned default-baseline discovery never +becomes authoritative. + +Only the admitted head snapshot is passed to `materialized_snapshot()`. The +worker additionally requires the acquired object database and materialized +target to stay below the caller-owned scratch root, while the target remains +disjoint from both object databases, the trusted execution directory and the +evidence root. The scanner runs as `python -I -m repo_sentinel` from a separate +trusted temporary control directory with `shell=False`, bounded stdout/stderr, +an explicit timeout and a report-size cap. Pull-request files named +`repo_sentinel.py`, `sitecustomize.py`, `usercustomize.py` or `*.pth` remain +materialized data rather than import sources. + +Raw scanner output, report text and target paths are not printed. The result +contains a fixed verdict, exact base/head identities, changed/deleted counts, +report size and SHA-256, and the verified scanner version. Scanner errors, +malformed or missing reports, timeouts, size violations and cleanup failures +become fixed-code infrastructure refusals. Error findings block, warnings keep +the scanner's existing non-blocking contract, and ordinary deletion paths are +accounted separately rather than passed as changed-file arguments. A +deletion-only delta still materializes the head and runs repository-level +checks with an empty changed-path tuple, preserving evidence such as a missing +required-file warning. Only an identical base/head snapshot with no changed or +deleted paths may take the deterministic no-scan PASS path. + +This draft core does not add workflow YAML, credentials, GitHub App key or token +handling, Commit Status or Checks API calls, repository settings, or the +authoritative context name. Activation remains a later boundary after worker +and signer review. Rollback removes this module, its isolated tests and this +section without changing the existing informational workflow or baseline. + ## Relationship To Issue #5 This record closed [issue #5](https://github.com/stacknil/sec-writeups-public/issues/5) diff --git a/scripts/repo_sentinel_authoritative.py b/scripts/repo_sentinel_authoritative.py new file mode 100644 index 0000000..b9e335c --- /dev/null +++ b/scripts/repo_sentinel_authoritative.py @@ -0,0 +1,932 @@ +"""Run the data-only core of the future authoritative Repo Sentinel gate.""" + +from __future__ import annotations + +import hashlib +import json +import math +import os +import re +import stat +import subprocess +import sys +import threading +from collections.abc import Callable, Iterator +from contextlib import contextmanager +from dataclasses import dataclass, field, replace +from enum import Enum +from pathlib import Path +from tempfile import TemporaryDirectory + +from repo_sentinel_acquire import ( + AcquiredSnapshot, + AcquisitionLimits, + AcquisitionRefused, + acquire_pull_snapshot, +) +from repo_sentinel_materialize import ( + MaterializationRefused, + MaterializedSnapshot, + materialized_snapshot, +) +from repo_sentinel_reader import ( + ReaderLimits, + ReaderRefused, + Snapshot, + SnapshotFile, + read_snapshot, +) + +SCANNER_DISTRIBUTION = "repo-sentinel-lite" +SCANNER_VERSION = "0.8.1" +SCANNER_VERSION_LINE = f"repo-sentinel {SCANNER_VERSION}" + +_OID_PATTERN = re.compile(r"[0-9a-f]{40}|[0-9a-f]{64}") +_REPOSITORY_PATTERN = re.compile(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+") +_INLINE_SUPPRESSION_PATTERN = re.compile( + r"repo-sentinel:\s*allow(?:\s+(?P[A-Za-z0-9_.\-, ]+))?", + re.IGNORECASE, +) +_SCANNER_TEXT_ENCODINGS = ("utf-8", "utf-8-sig", "utf-16", "cp1252") +_PORTABLE_V1_ASCII_CASE_ALIAS = { + codepoint: codepoint + (ord("a") - ord("A")) + for codepoint in range(ord("A"), ord("Z") + 1) +} + + +def _portable_v1_alias(value: str) -> str: + return value.translate(_PORTABLE_V1_ASCII_CASE_ALIAS) + + +_PROTECTED_EXACT_PATHS = frozenset( + { + ".reposentinel.toml", + ".reposentinel-baseline.json", + "scripts/repo_sentinel_gate.py", + "scripts/repo_sentinel_authoritative.py", + "scripts/repo_sentinel_acquire.py", + "scripts/repo_sentinel_reader.py", + "scripts/repo_sentinel_materialize.py", + "scripts/test_repo_sentinel_integration.py", + } +) +_PROTECTED_PATH_PREFIXES = ( + ".github/workflows/", + ".github/actions/", +) +_PROTECTED_EXACT_ALIASES = frozenset( + _portable_v1_alias(path) for path in _PROTECTED_EXACT_PATHS +) +_PROTECTED_PATH_PREFIX_ALIASES = tuple( + _portable_v1_alias(prefix) for prefix in _PROTECTED_PATH_PREFIXES +) + +_REFUSAL_CODES = frozenset( + { + "base_reader_refused", + "cleanup_failed", + "filesystem_io_failed", + "head_acquisition_refused", + "head_materialization_refused", + "head_reader_refused", + "head_snapshot_mismatch", + "invalid_request", + "invalid_snapshot", + "protected_control_change", + "report_collision", + "report_io_failed", + "report_missing", + "report_oversize", + "scanner_failed", + "scanner_launch_failed", + "scanner_output_limit", + "scanner_result_invalid", + "scanner_timeout", + "scanner_version_mismatch", + "source_suppression_change", + "unexpected_failure", + "unsafe_root_layout", + } +) + + +class GateVerdict(str, Enum): + """Stable outcomes that a later authority publisher may sign.""" + + PASS = "PASS" + SCANNER_FINDING = "SCANNER_FINDING" + INFRASTRUCTURE_REFUSAL = "INFRASTRUCTURE_REFUSAL" + PROTECTED_CONTROL_CHANGE = "PROTECTED_CONTROL_CHANGE" + + +class WorkerRefused(RuntimeError): + """Internal fixed-code refusal without paths or hostile output.""" + + def __init__(self, code: str) -> None: + safe_code = code if code in _REFUSAL_CODES else "unexpected_failure" + self.code = safe_code + super().__init__(safe_code) + + +@dataclass(frozen=True, slots=True) +class AuthoritativeGateRequest: + """Validated event identity and caller-owned trust roots.""" + + repository_identity: str = field(repr=False) + pull_number: int + base_oid: str + head_oid: str + remote: str | Path = field(repr=False) + trusted_repository: Path = field(repr=False) + scratch_root: Path = field(repr=False) + evidence_root: Path = field(repr=False) + + +@dataclass(frozen=True, slots=True) +class AuthoritativeLimits: + """Bounds owned by the worker rather than target repository policy.""" + + scanner_timeout_seconds: float = 60.0 + max_capture_bytes: int = 64 * 1024 + max_report_bytes: int = 8 * 1024 * 1024 + + def __post_init__(self) -> None: + if ( + type(self.scanner_timeout_seconds) not in (int, float) + or not math.isfinite(self.scanner_timeout_seconds) + or self.scanner_timeout_seconds <= 0 + or type(self.max_capture_bytes) is not int + or self.max_capture_bytes <= 0 + or type(self.max_report_bytes) is not int + or self.max_report_bytes <= 0 + ): + raise WorkerRefused("invalid_request") + + +@dataclass(frozen=True, slots=True) +class AuthoritativeGateResult: + """Bounded evidence suitable for a separately reviewed signer.""" + + verdict: GateVerdict + repository_identity: str = field(repr=False) + pull_number: int + base_oid: str + head_oid: str + changed_count: int + deleted_count: int + report_sha256: str | None + report_size: int + scanner_version: str | None + refusal_code: str | None + report_path: Path | None = field(default=None, repr=False, compare=False) + + +@dataclass(frozen=True, slots=True) +class SnapshotDelta: + """Exact D1 path differences without rename inference.""" + + changed_paths: tuple[str, ...] = field(repr=False) + deleted_paths: tuple[str, ...] = field(repr=False) + + +@dataclass(frozen=True, slots=True) +class ScannerInvocation: + """Trusted scanner inputs; target-controlled names stay out of repr.""" + + target_root: Path = field(repr=False) + changed_paths: tuple[str, ...] = field(repr=False) + baseline_path: Path | None = field(repr=False) + report_path: Path = field(repr=False) + execution_directory: Path = field(repr=False) + timeout_seconds: float + max_capture_bytes: int + + +@dataclass(frozen=True, slots=True) +class ScannerExecution: + """Bounded child result; captured bytes are never rendered by the worker.""" + + returncode: int + scanner_version: str + stdout: bytes = field(repr=False) + stderr: bytes = field(repr=False) + + +@dataclass(frozen=True, slots=True) +class _CommandResult: + returncode: int + stdout: bytes = field(repr=False) + stderr: bytes = field(repr=False) + + +SnapshotReader = Callable[..., Snapshot] +PullAcquirer = Callable[..., Iterator[AcquiredSnapshot]] +SnapshotMaterializer = Callable[..., Iterator[MaterializedSnapshot]] +ScannerRunner = Callable[[ScannerInvocation], ScannerExecution] + + +def _regular_directory(path: Path) -> Path: + try: + info = path.lstat() + reparse = getattr(info, "st_file_attributes", 0) & getattr( + stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0 + ) + if reparse or not stat.S_ISDIR(info.st_mode): + raise WorkerRefused("unsafe_root_layout") + return path.resolve(strict=True) + except WorkerRefused: + raise + except OSError: + raise WorkerRefused("unsafe_root_layout") from None + + +def _overlaps(left: Path, right: Path) -> bool: + return left == right or left.is_relative_to(right) or right.is_relative_to(left) + + +def _request_identity_is_valid(request: AuthoritativeGateRequest) -> bool: + return ( + type(request.repository_identity) is str + and len(request.repository_identity) <= 200 + and _REPOSITORY_PATTERN.fullmatch(request.repository_identity) is not None + and type(request.pull_number) is int + and 0 < request.pull_number <= 2_147_483_647 + and type(request.base_oid) is str + and type(request.head_oid) is str + and _OID_PATTERN.fullmatch(request.base_oid) is not None + and _OID_PATTERN.fullmatch(request.head_oid) is not None + and len(request.base_oid) == len(request.head_oid) + ) + + +def _validate_request( + request: AuthoritativeGateRequest, +) -> tuple[Path, Path, Path]: + if not _request_identity_is_valid(request): + raise WorkerRefused("invalid_request") + + trusted_repository = _regular_directory(request.trusted_repository) + scratch_root = _regular_directory(request.scratch_root) + evidence_root = _regular_directory(request.evidence_root) + roots = (trusted_repository, scratch_root, evidence_root) + if any( + _overlaps(left, right) + for index, left in enumerate(roots) + for right in roots[index + 1 :] + ): + raise WorkerRefused("unsafe_root_layout") + return trusted_repository, scratch_root, evidence_root + + +def _snapshot_files(snapshot: Snapshot) -> dict[str, SnapshotFile]: + files: dict[str, SnapshotFile] = {} + for item in snapshot.files: + if item.path in files: + raise WorkerRefused("invalid_snapshot") + files[item.path] = item + return files + + +def diff_snapshots(base: Snapshot, head: Snapshot) -> SnapshotDelta: + """Return the exact D1 added/modified/mode/deleted path sets.""" + + base_files = _snapshot_files(base) + head_files = _snapshot_files(head) + changed = tuple( + sorted( + path + for path, item in head_files.items() + if path not in base_files + or (item.mode, item.oid) != (base_files[path].mode, base_files[path].oid) + ) + ) + deleted = tuple(sorted(path for path in base_files if path not in head_files)) + return SnapshotDelta(changed, deleted) + + +def _is_protected_path(path: str) -> bool: + alias = _portable_v1_alias(path) + return alias in _PROTECTED_EXACT_ALIASES or alias.startswith( + _PROTECTED_PATH_PREFIX_ALIASES + ) + + +def _contains_inline_suppression(data: bytes) -> bool: + for encoding in _SCANNER_TEXT_ENCODINGS: + try: + text = data.decode(encoding) + except (UnicodeDecodeError, UnicodeError): + continue + if _INLINE_SUPPRESSION_PATTERN.search(text) is not None: + return True + return False + + +def _control_refusal( + base: Snapshot, + head: Snapshot, + delta: SnapshotDelta, +) -> str | None: + touched_paths = (*delta.changed_paths, *delta.deleted_paths) + if any(_is_protected_path(path) for path in touched_paths): + return "protected_control_change" + + base_files = _snapshot_files(base) + head_files = _snapshot_files(head) + for path in touched_paths: + base_item = base_files.get(path) + head_item = head_files.get(path) + if (base_item is not None and _contains_inline_suppression(base_item.data)) or ( + head_item is not None and _contains_inline_suppression(head_item.data) + ): + return "source_suppression_change" + return None + + +def _scanner_environment() -> dict[str, str]: + environment = { + key: value + for key, value in os.environ.items() + if not key.upper().startswith("PYTHON") + and not key.upper().startswith("REPO_SENTINEL") + } + environment["PYTHONNOUSERSITE"] = "1" + return environment + + +def _consume_stream( + stream: object, + limit: int, + process: subprocess.Popen[bytes], + overflow: threading.Event, + failed: threading.Event, + output: list[bytes], +) -> None: + collected = bytearray() + reader = stream.read + try: + while True: + chunk = reader(8192) + if not chunk: + break + remaining = max(0, limit + 1 - len(collected)) + if remaining: + collected.extend(chunk[:remaining]) + if len(collected) > limit and not overflow.is_set(): + overflow.set() + try: + process.kill() + except OSError: + pass + except (OSError, ValueError): + failed.set() + output.append(bytes(collected[: limit + 1])) + + +def _run_command_bounded( + command: list[str], + *, + cwd: Path, + timeout_seconds: float, + capture_limit: int, +) -> _CommandResult: + try: + process = subprocess.Popen( + command, + cwd=cwd, + env=_scanner_environment(), + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + shell=False, + close_fds=True, + ) + except OSError: + raise WorkerRefused("scanner_launch_failed") from None + + assert process.stdout is not None + assert process.stderr is not None + overflow = threading.Event() + capture_failed = threading.Event() + stdout_parts: list[bytes] = [] + stderr_parts: list[bytes] = [] + threads = ( + threading.Thread( + target=_consume_stream, + args=( + process.stdout, + capture_limit, + process, + overflow, + capture_failed, + stdout_parts, + ), + daemon=True, + ), + threading.Thread( + target=_consume_stream, + args=( + process.stderr, + capture_limit, + process, + overflow, + capture_failed, + stderr_parts, + ), + daemon=True, + ), + ) + for thread in threads: + thread.start() + + timed_out = False + try: + process.wait(timeout=timeout_seconds) + except subprocess.TimeoutExpired: + timed_out = True + try: + process.kill() + except OSError: + pass + process.wait() + finally: + for thread in threads: + thread.join(timeout=5.0) + process.stdout.close() + process.stderr.close() + + if timed_out: + raise WorkerRefused("scanner_timeout") + if overflow.is_set(): + raise WorkerRefused("scanner_output_limit") + if capture_failed.is_set() or any(thread.is_alive() for thread in threads): + raise WorkerRefused("scanner_failed") + return _CommandResult( + process.returncode, + stdout_parts[0] if stdout_parts else b"", + stderr_parts[0] if stderr_parts else b"", + ) + + +def _run_trusted_scanner(invocation: ScannerInvocation) -> ScannerExecution: + version = _run_command_bounded( + [sys.executable, "-I", "-m", "repo_sentinel", "--version"], + cwd=invocation.execution_directory, + timeout_seconds=invocation.timeout_seconds, + capture_limit=invocation.max_capture_bytes, + ) + try: + version_line = version.stdout.decode("utf-8", errors="strict").strip() + except UnicodeDecodeError: + raise WorkerRefused("scanner_version_mismatch") from None + if ( + version.returncode != 0 + or version.stderr + or version_line != SCANNER_VERSION_LINE + ): + raise WorkerRefused("scanner_version_mismatch") + + baseline_arguments = ["--no-default-baseline"] + if invocation.baseline_path is not None: + baseline_arguments.extend(["--baseline", str(invocation.baseline_path)]) + command = [ + sys.executable, + "-I", + "-m", + "repo_sentinel", + "scan", + *baseline_arguments, + "--changed-files", + "--fail-on-severity", + "error", + "--format", + "json", + "--output", + str(invocation.report_path), + str(invocation.target_root), + "--", + *invocation.changed_paths, + ] + result = _run_command_bounded( + command, + cwd=invocation.execution_directory, + timeout_seconds=invocation.timeout_seconds, + capture_limit=invocation.max_capture_bytes, + ) + return ScannerExecution( + result.returncode, + SCANNER_VERSION, + result.stdout, + result.stderr, + ) + + +def _regular_file( + path: Path, + *, + missing_code: str, + invalid_code: str, +) -> os.stat_result: + try: + info = path.lstat() + except FileNotFoundError: + raise WorkerRefused(missing_code) from None + except OSError: + raise WorkerRefused(invalid_code) from None + reparse = getattr(info, "st_file_attributes", 0) & getattr( + stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0 + ) + if reparse or not stat.S_ISREG(info.st_mode): + raise WorkerRefused(invalid_code) + return info + + +def _validated_report( + report_path: Path, + execution: ScannerExecution, + limits: AuthoritativeLimits, +) -> tuple[bytes, GateVerdict]: + if ( + execution.scanner_version != SCANNER_VERSION + or len(execution.stdout) > limits.max_capture_bytes + or len(execution.stderr) > limits.max_capture_bytes + ): + raise WorkerRefused("scanner_result_invalid") + if execution.returncode not in (0, 1): + raise WorkerRefused("scanner_failed") + + info = _regular_file( + report_path, + missing_code="report_missing", + invalid_code="report_io_failed", + ) + if info.st_size > limits.max_report_bytes: + raise WorkerRefused("report_oversize") + try: + report = report_path.read_bytes() + except OSError: + raise WorkerRefused("report_io_failed") from None + if len(report) != info.st_size: + raise WorkerRefused("report_io_failed") + try: + decoded = report.decode("utf-8", errors="strict") + parsed = json.loads(decoded) + except (UnicodeDecodeError, json.JSONDecodeError): + raise WorkerRefused("scanner_result_invalid") from None + if not isinstance(parsed, dict): + raise WorkerRefused("scanner_result_invalid") + findings = parsed.get("findings") + missing_files = parsed.get("missing_files") + suspicious_files = parsed.get("suspicious_files") + if ( + not isinstance(findings, list) + or not isinstance(missing_files, dict) + or not isinstance(suspicious_files, list) + ): + raise WorkerRefused("scanner_result_invalid") + + has_error = False + for finding in findings: + if not isinstance(finding, dict): + raise WorkerRefused("scanner_result_invalid") + severity = finding.get("severity") + if severity not in ("warning", "error"): + raise WorkerRefused("scanner_result_invalid") + has_error = has_error or severity == "error" + if (execution.returncode == 1) != has_error: + raise WorkerRefused("scanner_result_invalid") + verdict = GateVerdict.SCANNER_FINDING if has_error else GateVerdict.PASS + return report, verdict + + +def _write_private(path: Path, data: bytes, refusal: str) -> None: + descriptor: int | None = None + created = False + try: + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_BINARY", 0) + descriptor = os.open(path, flags, 0o600) + created = True + with os.fdopen(descriptor, "wb") as output: + descriptor = None + if output.write(data) != len(data): + raise OSError("short write") + output.flush() + os.fsync(output.fileno()) + info = _regular_file( + path, + missing_code=refusal, + invalid_code=refusal, + ) + if info.st_size != len(data) or path.read_bytes() != data: + raise OSError("readback mismatch") + except FileExistsError: + raise WorkerRefused( + "report_collision" if refusal == "report_io_failed" else refusal + ) from None + except WorkerRefused: + if descriptor is not None: + os.close(descriptor) + if created: + try: + path.unlink(missing_ok=True) + except OSError: + raise WorkerRefused("cleanup_failed") from None + raise + except OSError: + if descriptor is not None: + os.close(descriptor) + if created: + try: + path.unlink(missing_ok=True) + except OSError: + raise WorkerRefused("cleanup_failed") from None + raise WorkerRefused(refusal) from None + + +@contextmanager +def _private_temporary_directory(root: Path, prefix: str) -> Iterator[Path]: + temporary: TemporaryDirectory[str] | None = None + try: + try: + temporary = TemporaryDirectory(prefix=prefix, dir=root) + directory = Path(temporary.name) + _regular_directory(directory) + except OSError: + raise WorkerRefused("filesystem_io_failed") from None + yield directory + finally: + if temporary is not None: + try: + temporary.cleanup() + except OSError: + raise WorkerRefused("cleanup_failed") from None + + +def _base_baseline(snapshot: Snapshot) -> bytes | None: + for item in snapshot.files: + if item.path == ".reposentinel-baseline.json": + return item.data + return None + + +def _result( + request: AuthoritativeGateRequest, + verdict: GateVerdict, + *, + changed_count: int = 0, + deleted_count: int = 0, + report: bytes | None = None, + scanner_version: str | None = None, + refusal_code: str | None = None, +) -> AuthoritativeGateResult: + if _request_identity_is_valid(request): + repository_identity = request.repository_identity + pull_number = request.pull_number + base_oid = request.base_oid + head_oid = request.head_oid + else: + repository_identity = "" + pull_number = 0 + base_oid = "" + head_oid = "" + return AuthoritativeGateResult( + verdict=verdict, + repository_identity=repository_identity, + pull_number=pull_number, + base_oid=base_oid, + head_oid=head_oid, + changed_count=changed_count, + deleted_count=deleted_count, + report_sha256=( + hashlib.sha256(report).hexdigest() if report is not None else None + ), + report_size=len(report) if report is not None else 0, + scanner_version=scanner_version, + refusal_code=refusal_code, + ) + + +def _execute( + request: AuthoritativeGateRequest, + *, + limits: AuthoritativeLimits, + acquisition_limits: AcquisitionLimits, + reader_limits: ReaderLimits, + snapshot_reader: SnapshotReader, + pull_acquirer: PullAcquirer, + snapshot_materializer: SnapshotMaterializer, + scanner_runner: ScannerRunner, +) -> tuple[AuthoritativeGateResult, bytes | None, Path]: + trusted_repository, scratch_root, evidence_root = _validate_request(request) + try: + base_snapshot = snapshot_reader( + trusted_repository, + request.base_oid, + limits=reader_limits, + ) + except ReaderRefused: + raise WorkerRefused("base_reader_refused") from None + if base_snapshot.commit_oid != request.base_oid: + raise WorkerRefused("invalid_snapshot") + + try: + with pull_acquirer( + request.remote, + request.pull_number, + request.head_oid, + scratch_root, + acquisition_limits=acquisition_limits, + reader_limits=reader_limits, + ) as acquired: + if acquired.snapshot.commit_oid != request.head_oid: + raise WorkerRefused("head_snapshot_mismatch") + acquired_repository = _regular_directory(acquired.repository) + if ( + acquired_repository == scratch_root + or not acquired_repository.is_relative_to(scratch_root) + ): + raise WorkerRefused("unsafe_root_layout") + delta = diff_snapshots(base_snapshot, acquired.snapshot) + control_refusal = _control_refusal( + base_snapshot, + acquired.snapshot, + delta, + ) + if control_refusal is not None: + return ( + _result( + request, + GateVerdict.PROTECTED_CONTROL_CHANGE, + changed_count=len(delta.changed_paths), + deleted_count=len(delta.deleted_paths), + refusal_code=control_refusal, + ), + None, + evidence_root, + ) + if not delta.changed_paths and not delta.deleted_paths: + return ( + _result( + request, + GateVerdict.PASS, + changed_count=0, + deleted_count=len(delta.deleted_paths), + ), + None, + evidence_root, + ) + + with _private_temporary_directory( + scratch_root, "repo-sentinel-control-" + ) as control_root: + baseline_path: Path | None = None + baseline = _base_baseline(base_snapshot) + if baseline is not None: + baseline_path = control_root / "baseline.json" + _write_private(baseline_path, baseline, "filesystem_io_failed") + + with _private_temporary_directory( + evidence_root, "repo-sentinel-report-" + ) as report_root: + report_path = report_root / "report.json" + with snapshot_materializer( + acquired.repository, + request.head_oid, + scratch_root, + limits=reader_limits, + ) as materialized: + if materialized.snapshot != acquired.snapshot: + raise WorkerRefused("head_snapshot_mismatch") + materialized_root = _regular_directory(materialized.root) + if ( + materialized_root == scratch_root + or not materialized_root.is_relative_to(scratch_root) + ): + raise WorkerRefused("unsafe_root_layout") + if any( + _overlaps(materialized_root, protected_root) + for protected_root in ( + trusted_repository, + acquired_repository, + control_root, + evidence_root, + ) + ): + raise WorkerRefused("unsafe_root_layout") + invocation = ScannerInvocation( + target_root=materialized_root, + changed_paths=delta.changed_paths, + baseline_path=baseline_path, + report_path=report_path, + execution_directory=control_root, + timeout_seconds=limits.scanner_timeout_seconds, + max_capture_bytes=limits.max_capture_bytes, + ) + execution = scanner_runner(invocation) + report, verdict = _validated_report( + report_path, + execution, + limits, + ) + return ( + _result( + request, + verdict, + changed_count=len(delta.changed_paths), + deleted_count=len(delta.deleted_paths), + report=report, + scanner_version=execution.scanner_version, + ), + report, + evidence_root, + ) + except ReaderRefused: + raise WorkerRefused("head_reader_refused") from None + except AcquisitionRefused as error: + code = ( + "cleanup_failed" + if str(error) == "cleanup_failed" + else "head_acquisition_refused" + ) + raise WorkerRefused(code) from None + except MaterializationRefused as error: + code = ( + "cleanup_failed" + if str(error) == "cleanup_failed" + else "head_materialization_refused" + ) + raise WorkerRefused(code) from None + + +def _persist_report( + result: AuthoritativeGateResult, + report: bytes, + evidence_root: Path, +) -> AuthoritativeGateResult: + report_path = evidence_root / f"authoritative-report-{result.head_oid}.json" + _write_private(report_path, report, "report_io_failed") + return replace(result, report_path=report_path) + + +def run_authoritative_gate( + request: AuthoritativeGateRequest, + *, + limits: AuthoritativeLimits | None = None, + acquisition_limits: AcquisitionLimits | None = None, + reader_limits: ReaderLimits | None = None, + snapshot_reader: SnapshotReader = read_snapshot, + pull_acquirer: PullAcquirer = acquire_pull_snapshot, + snapshot_materializer: SnapshotMaterializer = materialized_snapshot, + scanner_runner: ScannerRunner = _run_trusted_scanner, +) -> AuthoritativeGateResult: + """Run the trusted worker core without publishing or printing a result.""" + + if limits is None: + limits = AuthoritativeLimits() + if acquisition_limits is None: + acquisition_limits = AcquisitionLimits() + if reader_limits is None: + reader_limits = ReaderLimits() + + try: + result, report, evidence_root = _execute( + request, + limits=limits, + acquisition_limits=acquisition_limits, + reader_limits=reader_limits, + snapshot_reader=snapshot_reader, + pull_acquirer=pull_acquirer, + snapshot_materializer=snapshot_materializer, + scanner_runner=scanner_runner, + ) + if report is not None: + result = _persist_report(result, report, evidence_root) + return result + except WorkerRefused as error: + return _result( + request, + GateVerdict.INFRASTRUCTURE_REFUSAL, + refusal_code=error.code, + ) + # A signer must receive a fixed fail-closed result, never an exception string. + except Exception: # noqa: BLE001 + return _result( + request, + GateVerdict.INFRASTRUCTURE_REFUSAL, + refusal_code="unexpected_failure", + ) + + +__all__ = [ + "SCANNER_DISTRIBUTION", + "SCANNER_VERSION", + "AuthoritativeGateRequest", + "AuthoritativeGateResult", + "AuthoritativeLimits", + "GateVerdict", + "ScannerExecution", + "ScannerInvocation", + "SnapshotDelta", + "diff_snapshots", + "run_authoritative_gate", +] diff --git a/tests/test_repo_sentinel_authoritative.py b/tests/test_repo_sentinel_authoritative.py new file mode 100644 index 0000000..bd71729 --- /dev/null +++ b/tests/test_repo_sentinel_authoritative.py @@ -0,0 +1,1204 @@ +from __future__ import annotations + +import hashlib +import importlib.metadata +import io +import json +import shutil +import subprocess +import sys +import unittest +from contextlib import contextmanager, redirect_stderr, redirect_stdout +from pathlib import Path +from tempfile import TemporaryDirectory +from unittest.mock import patch + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "scripts")) + +import repo_sentinel_authoritative as authoritative +from repo_sentinel_acquire import AcquiredSnapshot, AcquisitionRefused +from repo_sentinel_materialize import MaterializedSnapshot +from repo_sentinel_reader import Snapshot, SnapshotFile + +BASE_OID = "1" * 40 +HEAD_OID = "2" * 40 +BASELINE = b'{"schema_version":1,"generated_at":"2026-01-01T00:00:00Z","findings":[]}\n' +EMPTY_REPORT = { + "findings": [], + "missing_files": {}, + "suspicious_files": [], +} + + +def exact_scanner_available() -> bool: + try: + return importlib.metadata.version("repo-sentinel-lite") == "0.8.1" + except importlib.metadata.PackageNotFoundError: + return False + + +def snapshot_file( + path: str, + data: bytes = b"fixture\n", + *, + mode: str = "100644", + oid: str | None = None, +) -> SnapshotFile: + digest = oid or hashlib.sha1(data).hexdigest() + return SnapshotFile(path, mode, digest, data) + + +def snapshot( + commit_oid: str, + *files: SnapshotFile, +) -> Snapshot: + return Snapshot( + commit_oid, "3" * 40, tuple(sorted(files, key=lambda item: item.path)) + ) + + +def scanner_execution( + returncode: int = 0, + *, + stdout: bytes = b"", + stderr: bytes = b"", +) -> authoritative.ScannerExecution: + return authoritative.ScannerExecution( + returncode, + authoritative.SCANNER_VERSION, + stdout, + stderr, + ) + + +def write_report( + invocation: authoritative.ScannerInvocation, + report: object = EMPTY_REPORT, +) -> None: + invocation.report_path.write_text( + json.dumps(report, sort_keys=True) + "\n", + encoding="utf-8", + ) + + +class WorkerHarness: + def __init__(self, test: unittest.TestCase, base: Snapshot, head: Snapshot) -> None: + self.base = base + self.head = head + self.directories: list[TemporaryDirectory[str]] = [] + self.trusted_repository = self._directory(test, "authoritative-base-") + self.scratch_root = self._directory(test, "authoritative-scratch-") + self.evidence_root = self._directory(test, "authoritative-evidence-") + self.head_repository = self.scratch_root / "head.git" + self.head_repository.mkdir() + self.target_root = self.scratch_root / "target" + self.acquirer_exited = False + self.materializer_exited = False + + def _directory(self, test: unittest.TestCase, prefix: str) -> Path: + temporary = TemporaryDirectory(prefix=prefix) + self.directories.append(temporary) + test.addCleanup(temporary.cleanup) + return Path(temporary.name) + + @property + def request(self) -> authoritative.AuthoritativeGateRequest: + return authoritative.AuthoritativeGateRequest( + repository_identity="stacknil/sec-writeups-public", + pull_number=7, + base_oid=self.base.commit_oid, + head_oid=self.head.commit_oid, + remote="https://example.com/repository.git", + trusted_repository=self.trusted_repository, + scratch_root=self.scratch_root, + evidence_root=self.evidence_root, + ) + + def reader(self, _repository: Path, oid: str, **_kwargs: object) -> Snapshot: + if oid != self.base.commit_oid: + raise AssertionError("worker read an unexpected base object") + return self.base + + @contextmanager + def acquirer(self, *_args: object, **_kwargs: object): + try: + yield AcquiredSnapshot( + self.head, + f"refs/pull/{self.request.pull_number}/head", + self.head_repository, + ) + finally: + self.acquirer_exited = True + + @contextmanager + def materializer(self, *_args: object, **_kwargs: object): + self.target_root.mkdir() + for item in self.head.files: + target = self.target_root.joinpath(*item.path.split("/")) + target.parent.mkdir(parents=True, exist_ok=True) + target.write_bytes(item.data) + try: + yield MaterializedSnapshot(self.head, self.target_root) + finally: + self.materializer_exited = True + shutil.rmtree(self.target_root) + + def run( + self, + scanner: authoritative.ScannerRunner, + *, + limits: authoritative.AuthoritativeLimits | None = None, + ) -> authoritative.AuthoritativeGateResult: + return authoritative.run_authoritative_gate( + self.request, + limits=limits or authoritative.AuthoritativeLimits(), + snapshot_reader=self.reader, + pull_acquirer=self.acquirer, + snapshot_materializer=self.materializer, + scanner_runner=scanner, + ) + + +class SnapshotDiffTests(unittest.TestCase): + def test_d1_added_modified_mode_changed_deleted_and_identical(self) -> None: + original = snapshot_file("same.txt", b"same") + base = snapshot( + BASE_OID, + original, + snapshot_file("blob.txt", b"old"), + snapshot_file("mode.txt", b"mode", mode="100644"), + snapshot_file("deleted.txt", b"gone"), + ) + head = snapshot( + HEAD_OID, + original, + snapshot_file("blob.txt", b"new"), + snapshot_file("mode.txt", b"mode", mode="100755"), + snapshot_file("added.txt", b"added"), + ) + + delta = authoritative.diff_snapshots(base, head) + + self.assertEqual( + delta.changed_paths, + ("added.txt", "blob.txt", "mode.txt"), + ) + self.assertEqual(delta.deleted_paths, ("deleted.txt",)) + self.assertEqual( + authoritative.diff_snapshots(base, snapshot(HEAD_OID, *base.files)), + authoritative.SnapshotDelta((), ()), + ) + + def test_d1_preserves_exact_unicode_identity(self) -> None: + nfc = "notes/caf\N{LATIN SMALL LETTER E WITH ACUTE}.md" + nfd = "notes/cafe\N{COMBINING ACUTE ACCENT}.md" + base = snapshot(BASE_OID, snapshot_file(nfc)) + head = snapshot(HEAD_OID, snapshot_file(nfd)) + + delta = authoritative.diff_snapshots(base, head) + + self.assertEqual(delta.changed_paths, (nfd,)) + self.assertEqual(delta.deleted_paths, (nfc,)) + + def test_duplicate_snapshot_path_fails_closed(self) -> None: + duplicate = snapshot_file("same.txt") + invalid = Snapshot(BASE_OID, "3" * 40, (duplicate, duplicate)) + + with self.assertRaises(authoritative.WorkerRefused) as raised: + authoritative.diff_snapshots(invalid, snapshot(HEAD_OID)) + + self.assertEqual(raised.exception.code, "invalid_snapshot") + + +class AuthoritativeWorkerTests(unittest.TestCase): + def harness( + self, + *, + base_files: tuple[SnapshotFile, ...] = (), + head_files: tuple[SnapshotFile, ...] = (), + ) -> WorkerHarness: + return WorkerHarness( + self, + snapshot(BASE_OID, *base_files), + snapshot(HEAD_OID, *head_files), + ) + + def passing_scanner( + self, + calls: list[authoritative.ScannerInvocation] | None = None, + ) -> authoritative.ScannerRunner: + def scanner( + invocation: authoritative.ScannerInvocation, + ) -> authoritative.ScannerExecution: + if calls is not None: + calls.append(invocation) + write_report(invocation) + return scanner_execution() + + return scanner + + def test_pass_uses_base_baseline_and_trusted_execution_directory(self) -> None: + base_files = ( + snapshot_file("README.md"), + snapshot_file(".reposentinel-baseline.json", BASELINE), + ) + head_files = (*base_files, snapshot_file("notes/new.md", b"new\n")) + harness = self.harness(base_files=base_files, head_files=head_files) + calls: list[authoritative.ScannerInvocation] = [] + observed_baseline: list[bytes] = [] + + def scanner( + invocation: authoritative.ScannerInvocation, + ) -> authoritative.ScannerExecution: + calls.append(invocation) + assert invocation.baseline_path is not None + observed_baseline.append(invocation.baseline_path.read_bytes()) + write_report(invocation) + return scanner_execution() + + result = harness.run(scanner) + + self.assertEqual(result.verdict, authoritative.GateVerdict.PASS) + self.assertEqual(result.changed_count, 1) + self.assertEqual(result.deleted_count, 0) + self.assertEqual(result.scanner_version, "0.8.1") + self.assertIsNotNone(result.report_path) + assert result.report_path is not None + self.assertEqual( + result.report_sha256, + hashlib.sha256(result.report_path.read_bytes()).hexdigest(), + ) + invocation = calls[0] + self.assertEqual(invocation.changed_paths, ("notes/new.md",)) + self.assertEqual(observed_baseline, [BASELINE]) + self.assertNotEqual(invocation.execution_directory, invocation.target_root) + self.assertFalse( + invocation.execution_directory.is_relative_to(invocation.target_root) + ) + self.assertFalse(invocation.report_path.is_relative_to(invocation.target_root)) + self.assertTrue(harness.acquirer_exited) + self.assertTrue(harness.materializer_exited) + + def test_baseline_bytes_are_selected_from_base_snapshot(self) -> None: + baseline_oid = "4" * 40 + base_baseline = snapshot_file( + ".reposentinel-baseline.json", + BASELINE, + oid=baseline_oid, + ) + head_baseline = snapshot_file( + ".reposentinel-baseline.json", + b'{"head_owned":true}\n', + oid=baseline_oid, + ) + harness = self.harness( + base_files=(base_baseline,), + head_files=(head_baseline, snapshot_file("notes/new.md")), + ) + observed_baseline: list[bytes] = [] + + def scanner( + invocation: authoritative.ScannerInvocation, + ) -> authoritative.ScannerExecution: + assert invocation.baseline_path is not None + observed_baseline.append(invocation.baseline_path.read_bytes()) + write_report(invocation) + return scanner_execution() + + result = harness.run(scanner) + + self.assertEqual(result.verdict, authoritative.GateVerdict.PASS) + self.assertEqual(observed_baseline, [BASELINE]) + + def test_absent_base_baseline_disables_explicit_baseline(self) -> None: + harness = self.harness(head_files=(snapshot_file("new.md"),)) + calls: list[authoritative.ScannerInvocation] = [] + + result = harness.run(self.passing_scanner(calls)) + + self.assertEqual(result.verdict, authoritative.GateVerdict.PASS) + self.assertIsNone(calls[0].baseline_path) + + def test_scanner_error_finding_is_a_bounded_result(self) -> None: + harness = self.harness(head_files=(snapshot_file("new.md"),)) + + def scanner( + invocation: authoritative.ScannerInvocation, + ) -> authoritative.ScannerExecution: + report = { + "findings": [{"severity": "error"}], + "missing_files": {}, + "suspicious_files": [], + } + write_report(invocation, report) + return scanner_execution(1) + + result = harness.run(scanner) + + self.assertEqual(result.verdict, authoritative.GateVerdict.SCANNER_FINDING) + self.assertIsNone(result.refusal_code) + self.assertGreater(result.report_size, 0) + + def test_warning_remains_non_blocking(self) -> None: + harness = self.harness(head_files=(snapshot_file("new.md"),)) + + def scanner( + invocation: authoritative.ScannerInvocation, + ) -> authoritative.ScannerExecution: + report = { + "findings": [{"severity": "warning"}], + "missing_files": {}, + "suspicious_files": [], + } + write_report(invocation, report) + return scanner_execution() + + result = harness.run(scanner) + + self.assertEqual(result.verdict, authoritative.GateVerdict.PASS) + + def test_identical_snapshot_does_not_launch_scanner(self) -> None: + existing = snapshot_file("ordinary.txt") + harness = self.harness( + base_files=(existing,), + head_files=(existing,), + ) + + result = harness.run(lambda _invocation: self.fail("scanner must not run")) + + self.assertEqual(result.verdict, authoritative.GateVerdict.PASS) + self.assertEqual(result.changed_count, 0) + self.assertEqual(result.deleted_count, 0) + self.assertIsNone(result.scanner_version) + self.assertIsNone(result.report_path) + + def test_deletion_only_snapshot_runs_scanner_with_empty_changed_paths(self) -> None: + existing = snapshot_file("ordinary.txt") + harness = self.harness(base_files=(existing,)) + calls: list[authoritative.ScannerInvocation] = [] + + result = harness.run(self.passing_scanner(calls)) + + self.assertEqual(result.verdict, authoritative.GateVerdict.PASS) + self.assertEqual(result.changed_count, 0) + self.assertEqual(result.deleted_count, 1) + self.assertEqual(result.scanner_version, "0.8.1") + self.assertIsNotNone(result.report_path) + self.assertEqual(len(calls), 1) + self.assertEqual(calls[0].changed_paths, ()) + + def test_deletion_only_scanner_failure_fails_closed(self) -> None: + harness = self.harness(base_files=(snapshot_file("ordinary.txt"),)) + + def failing_scanner( + _invocation: authoritative.ScannerInvocation, + ) -> authoritative.ScannerExecution: + raise authoritative.WorkerRefused("scanner_launch_failed") + + result = harness.run(failing_scanner) + + self.assertEqual( + result.verdict, + authoritative.GateVerdict.INFRASTRUCTURE_REFUSAL, + ) + self.assertEqual(result.refusal_code, "scanner_launch_failed") + + def test_protected_exact_and_subtree_changes_block_before_scanner(self) -> None: + cases = ( + ".reposentinel.toml", + ".reposentinel-baseline.json", + "scripts/repo_sentinel_authoritative.py", + "scripts/repo_sentinel_acquire.py", + "scripts/repo_sentinel_reader.py", + "scripts/repo_sentinel_materialize.py", + "scripts/repo_sentinel_gate.py", + "scripts/test_repo_sentinel_integration.py", + ".github/workflows/attack.yml", + ".github/actions/local/action.yml", + ) + for path in cases: + with self.subTest(path=path): + harness = self.harness(head_files=(snapshot_file(path),)) + result = harness.run( + lambda _invocation: self.fail("scanner must not run") + ) + self.assertEqual( + result.verdict, + authoritative.GateVerdict.PROTECTED_CONTROL_CHANGE, + ) + self.assertEqual(result.refusal_code, "protected_control_change") + + def test_deleting_protected_control_blocks(self) -> None: + protected = snapshot_file(".github/workflows/gate.yml") + harness = self.harness(base_files=(protected,)) + + result = harness.run(lambda _invocation: self.fail("scanner must not run")) + + self.assertEqual( + result.verdict, + authoritative.GateVerdict.PROTECTED_CONTROL_CHANGE, + ) + + def test_protected_ascii_case_aliases_block_before_scanner(self) -> None: + cases = ( + ".RepoSentinel.toml", + ".REPOSENTINEL.TOML", + ".reposentinel.TOML", + ".GitHub/workflows/test.yml", + ".github/Workflows/test.yml", + ".GITHUB/" + "ACTIONS/" + "example/" + "action.yml", + "Scripts/repo_sentinel_gate.py", + "scripts/Repo_Sentinel_Authoritative.py", + ) + for path in cases: + for change in ("changed", "deleted"): + with self.subTest(path=path, change=change): + item = snapshot_file(path) + harness = self.harness( + base_files=(item,) if change == "deleted" else (), + head_files=(item,) if change == "changed" else (), + ) + + result = harness.run( + lambda _invocation: self.fail("scanner must not run") + ) + + self.assertEqual( + result.verdict, + authoritative.GateVerdict.PROTECTED_CONTROL_CHANGE, + ) + self.assertEqual(result.refusal_code, "protected_control_change") + + def test_nearby_names_do_not_match_protected_paths(self) -> None: + cases = ( + ".reposentinel.toml.example", + ".reposentinel.tomlx", + ".github/workflows2/test.yml", + ".github/workflow/test.yml", + ".github/actions2/action.yml", + ".github/action/action.yml", + "scripts/repo_sentinel_authoritative.pyx", + "scripts/repo_sentinel_gate.py.backup", + ) + for path in cases: + with self.subTest(path=path): + harness = self.harness(head_files=(snapshot_file(path),)) + calls: list[authoritative.ScannerInvocation] = [] + + result = harness.run(self.passing_scanner(calls)) + + self.assertEqual(result.verdict, authoritative.GateVerdict.PASS) + self.assertEqual(len(calls), 1) + self.assertEqual(calls[0].changed_paths, (path,)) + + @unittest.skipUnless( + exact_scanner_available(), + "requires the exact production scanner", + ) + def test_mixed_case_config_exploit_blocks_before_real_scanner(self) -> None: + harness = self.harness( + head_files=( + snapshot_file( + ".RepoSentinel.toml", + b'ignore_globs = [".env"]\n', + ), + snapshot_file( + ".env", + b"TOKEN=" + + b"ghp_" + + bytes(range(ord("A"), ord("Z") + 1)) + + bytes(range(ord("a"), ord("j") + 1)) + + b"\n", + ), + ) + ) + calls: list[authoritative.ScannerInvocation] = [] + + def scanner( + invocation: authoritative.ScannerInvocation, + ) -> authoritative.ScannerExecution: + calls.append(invocation) + return authoritative._run_trusted_scanner(invocation) + + result = harness.run(scanner) + + self.assertEqual( + result.verdict, + authoritative.GateVerdict.PROTECTED_CONTROL_CHANGE, + ) + self.assertEqual(result.refusal_code, "protected_control_change") + self.assertEqual(calls, []) + + @unittest.skipUnless( + exact_scanner_available(), + "requires the exact production scanner", + ) + def test_real_scanner_preserves_license_deletion_warning(self) -> None: + readme = snapshot_file("README.md") + license_file = snapshot_file("LICENSE") + gitignore = snapshot_file(".gitignore") + harness = self.harness( + base_files=(readme, license_file, gitignore), + head_files=(readme, gitignore), + ) + calls: list[authoritative.ScannerInvocation] = [] + + def scanner( + invocation: authoritative.ScannerInvocation, + ) -> authoritative.ScannerExecution: + calls.append(invocation) + return authoritative._run_trusted_scanner(invocation) + + result = harness.run(scanner) + + self.assertEqual(result.verdict, authoritative.GateVerdict.PASS) + self.assertEqual(result.changed_count, 0) + self.assertEqual(result.deleted_count, 1) + self.assertEqual(result.scanner_version, "0.8.1") + self.assertIsNotNone(result.report_path) + self.assertGreater(result.report_size, 0) + self.assertIsNotNone(result.report_sha256) + self.assertEqual(len(calls), 1) + self.assertEqual(calls[0].changed_paths, ()) + assert result.report_path is not None + report = json.loads(result.report_path.read_text(encoding="utf-8")) + self.assertEqual( + [ + (finding["rule_id"], finding["path"], finding["severity"]) + for finding in report["findings"] + ], + [("repo.required_file_missing", "LICENSE", "warning")], + ) + + def test_inline_suppression_change_blocks_exact_scanner_syntax(self) -> None: + controls = ( + b"# repo-sentinel: allow\nvalue\n", + b"# RePo-SeNtInEl: allow secret.high_entropy, github_token\nvalue\n", + ) + for data in controls: + with self.subTest(data=data): + harness = self.harness( + head_files=(snapshot_file("notes/controlled.md", data),) + ) + result = harness.run( + lambda _invocation: self.fail("scanner must not run") + ) + self.assertEqual( + result.verdict, + authoritative.GateVerdict.PROTECTED_CONTROL_CHANGE, + ) + self.assertEqual(result.refusal_code, "source_suppression_change") + + def test_deleting_file_with_inline_suppression_blocks(self) -> None: + controlled = snapshot_file( + "notes/controlled.md", + b"# repo-sentinel: allow secret.high_entropy\nfixture\n", + ) + harness = self.harness(base_files=(controlled,)) + + result = harness.run(lambda _invocation: self.fail("scanner must not run")) + + self.assertEqual(result.refusal_code, "source_suppression_change") + + def test_similar_non_directive_text_does_not_block(self) -> None: + harness = self.harness( + head_files=( + snapshot_file( + "notes/ordinary.md", + b"The repo-sentinel policy allows reviewed fixtures.\n", + ), + ) + ) + + result = harness.run(self.passing_scanner()) + + self.assertEqual(result.verdict, authoritative.GateVerdict.PASS) + + def test_hostile_report_content_is_never_printed_or_interpreted(self) -> None: + harness = self.harness(head_files=(snapshot_file("new.md"),)) + payloads = ( + "::warning::payload", + "::error::payload", + "::add-mask::payload", + "::stop-commands::payload", + "\u001b[31mcontrol-like\u001b[0m", + ) + + def scanner( + invocation: authoritative.ScannerInvocation, + ) -> authoritative.ScannerExecution: + report = {**EMPTY_REPORT, "hostile_data": payloads} + write_report(invocation, report) + return scanner_execution( + stdout=b"::warning::stdout\n", + stderr=b"::error::stderr\n", + ) + + stdout = io.StringIO() + stderr = io.StringIO() + with redirect_stdout(stdout), redirect_stderr(stderr): + result = harness.run(scanner) + + self.assertEqual(result.verdict, authoritative.GateVerdict.PASS) + self.assertEqual(stdout.getvalue(), "") + self.assertEqual(stderr.getvalue(), "") + assert result.report_path is not None + persisted = result.report_path.read_text(encoding="utf-8") + for payload in payloads: + self.assertIn(json.dumps(payload)[1:-1], persisted) + + def test_malicious_filename_is_hidden_and_materializer_failure_is_closed( + self, + ) -> None: + path = "::error::not-a-workflow-command" + harness = self.harness(head_files=(snapshot_file(path),)) + + @contextmanager + def refusing_materializer(*_args: object, **_kwargs: object): + raise authoritative.MaterializationRefused("unsupported_host_path") + yield # pragma: no cover + + result = authoritative.run_authoritative_gate( + harness.request, + snapshot_reader=harness.reader, + pull_acquirer=harness.acquirer, + snapshot_materializer=refusing_materializer, + scanner_runner=lambda _invocation: self.fail("scanner must not run"), + ) + + self.assertEqual( + result.verdict, authoritative.GateVerdict.INFRASTRUCTURE_REFUSAL + ) + self.assertEqual(result.refusal_code, "head_materialization_refused") + self.assertNotIn(path, repr(result)) + + def test_missing_oversized_malformed_and_inconsistent_reports_fail_closed( + self, + ) -> None: + cases: tuple[tuple[str, authoritative.ScannerRunner, str], ...] = ( + ( + "missing", + lambda _invocation: scanner_execution(), + "report_missing", + ), + ( + "oversized", + self._raw_report_scanner(b"x" * 65), + "report_oversize", + ), + ( + "malformed", + self._raw_report_scanner(b"not-json\n"), + "scanner_result_invalid", + ), + ( + "exit-mismatch", + self._raw_report_scanner( + (json.dumps(EMPTY_REPORT) + "\n").encode(), + returncode=1, + ), + "scanner_result_invalid", + ), + ) + for name, scanner, refusal in cases: + with self.subTest(name=name): + harness = self.harness(head_files=(snapshot_file("new.md"),)) + result = harness.run( + scanner, + limits=authoritative.AuthoritativeLimits(max_report_bytes=64), + ) + self.assertEqual( + result.verdict, + authoritative.GateVerdict.INFRASTRUCTURE_REFUSAL, + ) + self.assertEqual(result.refusal_code, refusal) + self.assertEqual(list(harness.evidence_root.iterdir()), []) + + @staticmethod + def _raw_report_scanner( + data: bytes, + *, + returncode: int = 0, + ) -> authoritative.ScannerRunner: + def scanner( + invocation: authoritative.ScannerInvocation, + ) -> authoritative.ScannerExecution: + invocation.report_path.write_bytes(data) + return scanner_execution(returncode) + + return scanner + + def test_scanner_exception_cleans_every_temporary_boundary(self) -> None: + harness = self.harness(head_files=(snapshot_file("new.md"),)) + transient_paths: list[Path] = [] + + def scanner( + invocation: authoritative.ScannerInvocation, + ) -> authoritative.ScannerExecution: + transient_paths.extend( + [ + invocation.execution_directory, + invocation.report_path.parent, + invocation.target_root, + ] + ) + raise RuntimeError("hostile raw detail") + + result = harness.run(scanner) + + self.assertEqual( + result.verdict, authoritative.GateVerdict.INFRASTRUCTURE_REFUSAL + ) + self.assertEqual(result.refusal_code, "unexpected_failure") + self.assertTrue(harness.acquirer_exited) + self.assertTrue(harness.materializer_exited) + self.assertTrue(all(not path.exists() for path in transient_paths)) + self.assertNotIn("hostile raw detail", repr(result)) + + def test_boundary_failures_return_fixed_infrastructure_codes(self) -> None: + cases: list[ + tuple[ + str, + authoritative.SnapshotReader, + authoritative.PullAcquirer, + authoritative.SnapshotMaterializer, + authoritative.ScannerRunner, + str, + ] + ] = [] + for name, worker_code in ( + ("scanner-launch", "scanner_launch_failed"), + ("scanner-timeout", "scanner_timeout"), + ("scanner-output", "scanner_output_limit"), + ): + harness = self.harness(head_files=(snapshot_file("new.md"),)) + + def refused_scanner( + _invocation: authoritative.ScannerInvocation, + code: str = worker_code, + ) -> authoritative.ScannerExecution: + raise authoritative.WorkerRefused(code) + + cases.append( + ( + name, + harness.reader, + harness.acquirer, + harness.materializer, + refused_scanner, + worker_code, + ) + ) + + for name, reader, acquirer, materializer, scanner, code in cases: + with self.subTest(name=name): + request = reader.__self__.request + result = authoritative.run_authoritative_gate( + request, + snapshot_reader=reader, + pull_acquirer=acquirer, + snapshot_materializer=materializer, + scanner_runner=scanner, + ) + self.assertEqual( + result.verdict, + authoritative.GateVerdict.INFRASTRUCTURE_REFUSAL, + ) + self.assertEqual(result.refusal_code, code) + + def test_base_and_head_read_failures_are_distinguished(self) -> None: + base_harness = self.harness() + + def refused_base(*_args: object, **_kwargs: object) -> Snapshot: + raise authoritative.ReaderRefused("object_unavailable") + + base_result = authoritative.run_authoritative_gate( + base_harness.request, + snapshot_reader=refused_base, + ) + self.assertEqual(base_result.refusal_code, "base_reader_refused") + + head_harness = self.harness() + + @contextmanager + def refused_head(*_args: object, **_kwargs: object): + raise authoritative.ReaderRefused("object_unavailable") + yield # pragma: no cover + + head_result = authoritative.run_authoritative_gate( + head_harness.request, + snapshot_reader=head_harness.reader, + pull_acquirer=refused_head, + ) + self.assertEqual(head_result.refusal_code, "head_reader_refused") + + def test_acquisition_refusal_is_fixed_and_cleans(self) -> None: + harness = self.harness() + + @contextmanager + def refused_acquisition(*_args: object, **_kwargs: object): + raise AcquisitionRefused("fetch_failed") + yield # pragma: no cover + + result = authoritative.run_authoritative_gate( + harness.request, + snapshot_reader=harness.reader, + pull_acquirer=refused_acquisition, + ) + + self.assertEqual(result.refusal_code, "head_acquisition_refused") + + def test_final_report_is_exclusive_and_never_overwrites_evidence(self) -> None: + harness = self.harness(head_files=(snapshot_file("new.md"),)) + final_path = harness.evidence_root / f"authoritative-report-{HEAD_OID}.json" + original = b"reviewed evidence\n" + final_path.write_bytes(original) + + result = harness.run(self.passing_scanner()) + + self.assertEqual( + result.verdict, authoritative.GateVerdict.INFRASTRUCTURE_REFUSAL + ) + self.assertEqual(result.refusal_code, "report_collision") + self.assertEqual(final_path.read_bytes(), original) + + def test_acquisition_cleanup_failure_is_infrastructure(self) -> None: + harness = self.harness() + + @contextmanager + def cleanup_failure(*_args: object, **_kwargs: object): + yield AcquiredSnapshot( + harness.head, "refs/pull/7/head", harness.head_repository + ) + raise AcquisitionRefused("cleanup_failed") + + result = authoritative.run_authoritative_gate( + harness.request, + snapshot_reader=harness.reader, + pull_acquirer=cleanup_failure, + snapshot_materializer=harness.materializer, + scanner_runner=self.passing_scanner(), + ) + + self.assertEqual( + result.verdict, authoritative.GateVerdict.INFRASTRUCTURE_REFUSAL + ) + self.assertEqual(result.refusal_code, "cleanup_failed") + + def test_unsafe_overlapping_roots_fail_before_reader(self) -> None: + harness = self.harness() + request = authoritative.AuthoritativeGateRequest( + repository_identity="stacknil/sec-writeups-public", + pull_number=7, + base_oid=BASE_OID, + head_oid=HEAD_OID, + remote="https://example.com/repository.git", + trusted_repository=harness.trusted_repository, + scratch_root=harness.trusted_repository, + evidence_root=harness.evidence_root, + ) + + result = authoritative.run_authoritative_gate( + request, + snapshot_reader=lambda *_args, **_kwargs: self.fail("reader must not run"), + ) + + self.assertEqual(result.refusal_code, "unsafe_root_layout") + + def test_materialized_target_cannot_overlap_head_object_database(self) -> None: + harness = self.harness(head_files=(snapshot_file("new.md"),)) + materializer_exited = False + + @contextmanager + def overlapping_materializer(*_args: object, **_kwargs: object): + nonlocal materializer_exited + try: + yield MaterializedSnapshot(harness.head, harness.head_repository) + finally: + materializer_exited = True + + result = authoritative.run_authoritative_gate( + harness.request, + snapshot_reader=harness.reader, + pull_acquirer=harness.acquirer, + snapshot_materializer=overlapping_materializer, + scanner_runner=lambda _invocation: self.fail("scanner must not run"), + ) + + self.assertEqual( + result.verdict, authoritative.GateVerdict.INFRASTRUCTURE_REFUSAL + ) + self.assertEqual(result.refusal_code, "unsafe_root_layout") + self.assertTrue(materializer_exited) + self.assertTrue(harness.acquirer_exited) + + def test_default_repr_hides_paths_and_untrusted_strings(self) -> None: + secret_path = Path("D:/private/control") + request = authoritative.AuthoritativeGateRequest( + "owner/repository", + 1, + BASE_OID, + HEAD_OID, + "https://example.invalid/private.git", + secret_path, + secret_path / "scratch", + secret_path / "evidence", + ) + result = authoritative.AuthoritativeGateResult( + authoritative.GateVerdict.PASS, + "owner/repository", + 1, + BASE_OID, + HEAD_OID, + 1, + 0, + "a" * 64, + 1, + "0.8.1", + None, + secret_path / "raw-report.json", + ) + + for rendered in (repr(request), repr(result)): + self.assertNotIn("private", rendered) + self.assertNotIn("owner/repository", rendered) + self.assertNotIn("example.invalid", rendered) + + def test_invalid_identity_is_not_reflected_in_refusal_result(self) -> None: + harness = self.harness() + hostile_oid = "::error::must-remain-data" + request = authoritative.AuthoritativeGateRequest( + repository_identity="stacknil/sec-writeups-public", + pull_number=7, + base_oid=hostile_oid, + head_oid=HEAD_OID, + remote="https://example.com/repository.git", + trusted_repository=harness.trusted_repository, + scratch_root=harness.scratch_root, + evidence_root=harness.evidence_root, + ) + + result = authoritative.run_authoritative_gate( + request, + snapshot_reader=lambda *_args, **_kwargs: self.fail("reader must not run"), + ) + + self.assertEqual(result.refusal_code, "invalid_request") + self.assertEqual(result.pull_number, 0) + self.assertEqual(result.base_oid, "") + self.assertEqual(result.head_oid, "") + self.assertNotIn(hostile_oid, repr(result)) + + +class ScannerRunnerTests(unittest.TestCase): + def directory(self, prefix: str) -> Path: + temporary = TemporaryDirectory(prefix=prefix) + self.addCleanup(temporary.cleanup) + return Path(temporary.name) + + def test_command_is_isolated_array_with_explicit_policy_inputs(self) -> None: + control = self.directory("scanner-control-") + target = self.directory("scanner-target-") + report_root = self.directory("scanner-report-") + baseline = control / "baseline.json" + baseline.write_bytes(BASELINE) + report_path = report_root / "report.json" + marker = target / "executed" + for name in ( + "repo_sentinel.py", + "sitecustomize.py", + "usercustomize.py", + "fixture.pth", + ): + (target / name).write_text( + f'from pathlib import Path\nPath(r"{marker}").touch()\n', + encoding="utf-8", + ) + calls: list[tuple[list[str], Path]] = [] + + def command_runner( + command: list[str], + *, + cwd: Path, + timeout_seconds: float, + capture_limit: int, + ) -> authoritative._CommandResult: + del timeout_seconds, capture_limit + calls.append((command, cwd)) + if "--version" in command: + return authoritative._CommandResult(0, b"repo-sentinel 0.8.1\n", b"") + report_path.write_text(json.dumps(EMPTY_REPORT) + "\n", encoding="utf-8") + return authoritative._CommandResult(0, b"", b"") + + invocation = authoritative.ScannerInvocation( + target, + ("repo_sentinel.py", "sitecustomize.py", "fixture.pth"), + baseline, + report_path, + control, + 10.0, + 1024, + ) + with patch.object( + authoritative, "_run_command_bounded", side_effect=command_runner + ): + execution = authoritative._run_trusted_scanner(invocation) + + self.assertEqual(execution.returncode, 0) + self.assertFalse(marker.exists()) + self.assertEqual(len(calls), 2) + command, cwd = calls[1] + self.assertEqual(command[:4], [sys.executable, "-I", "-m", "repo_sentinel"]) + self.assertEqual(cwd, control) + self.assertNotEqual(cwd, target) + self.assertIn("--no-default-baseline", command) + self.assertEqual(command[command.index("--baseline") + 1], str(baseline)) + self.assertEqual(command[command.index("--output") + 1], str(report_path)) + self.assertEqual( + command[command.index("--") + 1 :], list(invocation.changed_paths) + ) + + def test_real_bounded_command_rejects_timeout_and_output_overflow(self) -> None: + cwd = self.directory("bounded-command-") + cases = ( + ( + [sys.executable, "-c", "import time; time.sleep(2)"], + 0.05, + 1024, + "scanner_timeout", + ), + ( + [sys.executable, "-c", "import sys; sys.stdout.write('x' * 4096)"], + 5.0, + 64, + "scanner_output_limit", + ), + ) + for command, timeout, limit, code in cases: + with self.subTest(code=code): + with self.assertRaises(authoritative.WorkerRefused) as raised: + authoritative._run_command_bounded( + command, + cwd=cwd, + timeout_seconds=timeout, + capture_limit=limit, + ) + self.assertEqual(raised.exception.code, code) + + @unittest.skipUnless( + exact_scanner_available(), + "requires the exact production scanner", + ) + def test_real_pinned_scanner_keeps_target_python_inert(self) -> None: + control = self.directory("scanner-real-control-") + target = self.directory("scanner-real-target-") + report_root = self.directory("scanner-real-report-") + marker = target / "target-python-executed" + marker_code = f'import pathlib; pathlib.Path(r"{marker}").touch()\n' + for name in ( + "repo_sentinel.py", + "sitecustomize.py", + "usercustomize.py", + "fixture.pth", + ): + (target / name).write_text(marker_code, encoding="utf-8") + for name in ("README.md", "LICENSE", ".gitignore"): + (target / name).write_text("fixture\n", encoding="utf-8") + invocation = authoritative.ScannerInvocation( + target, + ( + "repo_sentinel.py", + "sitecustomize.py", + "usercustomize.py", + "fixture.pth", + ), + None, + report_root / "report.json", + control, + 10.0, + 64 * 1024, + ) + + execution = authoritative._run_trusted_scanner(invocation) + + self.assertEqual(execution.returncode, 0) + self.assertEqual(execution.scanner_version, "0.8.1") + self.assertFalse(marker.exists()) + report = json.loads(invocation.report_path.read_text(encoding="utf-8")) + self.assertEqual(report["findings"], []) + + +def run_git(repository: Path, *arguments: str) -> str: + return subprocess.run( + ["git", *arguments], + cwd=repository, + check=True, + capture_output=True, + text=True, + encoding="utf-8", + ).stdout.strip() + + +class CrossLayerWorkerTests(unittest.TestCase): + def test_real_local_acquisition_reader_and_materializer(self) -> None: + roots = [ + TemporaryDirectory(prefix=f"authoritative-real-{name}-") + for name in ("repo", "scratch", "evidence") + ] + for temporary in roots: + self.addCleanup(temporary.cleanup) + repository, scratch, evidence = (Path(item.name) for item in roots) + run_git(repository, "init", "--quiet") + run_git(repository, "config", "user.name", "Contract Test") + run_git(repository, "config", "user.email", "contract@example.com") + run_git(repository, "config", "commit.gpgsign", "false") + (repository / "README.md").write_text("Base.\n", encoding="utf-8") + (repository / "LICENSE").write_text("CC BY 4.0\n", encoding="utf-8") + (repository / ".gitignore").write_text("\n", encoding="utf-8") + run_git(repository, "add", "--all") + run_git(repository, "commit", "--quiet", "-m", "test: base") + base_oid = run_git(repository, "rev-parse", "HEAD") + notes = repository / "notes" + notes.mkdir() + exact_data = b"exact acquired bytes\x00remain data\n" + (notes / "safe & exact!.md").write_bytes(exact_data) + run_git(repository, "add", "--all") + run_git(repository, "commit", "--quiet", "-m", "test: head") + head_oid = run_git(repository, "rev-parse", "HEAD") + run_git(repository, "update-ref", "refs/pull/1/head", head_oid) + observed: dict[str, object] = {} + + def scanner( + invocation: authoritative.ScannerInvocation, + ) -> authoritative.ScannerExecution: + observed["cwd"] = invocation.execution_directory + observed["target"] = invocation.target_root + observed["bytes"] = ( + invocation.target_root / "notes" / "safe & exact!.md" + ).read_bytes() + write_report(invocation) + return scanner_execution() + + request = authoritative.AuthoritativeGateRequest( + "stacknil/sec-writeups-public", + 1, + base_oid, + head_oid, + repository, + repository, + scratch, + evidence, + ) + result = authoritative.run_authoritative_gate(request, scanner_runner=scanner) + + self.assertEqual(result.verdict, authoritative.GateVerdict.PASS) + self.assertEqual(result.changed_count, 1) + self.assertEqual(observed["bytes"], exact_data) + self.assertNotEqual(observed["cwd"], observed["target"]) + self.assertFalse(any(scratch.iterdir())) + self.assertEqual(len(list(evidence.iterdir())), 1) + + +if __name__ == "__main__": + unittest.main()