From 5268b306c2602385030b6857f04427a364eda8ac Mon Sep 17 00:00:00 2001 From: stacknil Date: Mon, 14 Sep 2026 00:30:22 +0800 Subject: [PATCH 01/11] feat(security): add authoritative worker core --- scripts/repo_sentinel_authoritative.py | 900 +++++++++++++++++++++++++ 1 file changed, 900 insertions(+) create mode 100644 scripts/repo_sentinel_authoritative.py diff --git a/scripts/repo_sentinel_authoritative.py b/scripts/repo_sentinel_authoritative.py new file mode 100644 index 0000000..80a533f --- /dev/null +++ b/scripts/repo_sentinel_authoritative.py @@ -0,0 +1,900 @@ +"""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") + +_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/", +) + +_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 _validate_request( + request: AuthoritativeGateRequest, +) -> tuple[Path, Path, Path]: + if ( + type(request.repository_identity) is not str + or len(request.repository_identity) > 200 + or _REPOSITORY_PATTERN.fullmatch(request.repository_identity) is None + or type(request.pull_number) is not int + or not 0 < request.pull_number <= 2_147_483_647 + or type(request.base_oid) is not str + or type(request.head_oid) is not str + or _OID_PATTERN.fullmatch(request.base_oid) is None + or _OID_PATTERN.fullmatch(request.head_oid) is None + or len(request.base_oid) != len(request.head_oid) + ): + 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: + return path in _PROTECTED_EXACT_PATHS or path.startswith(_PROTECTED_PATH_PREFIXES) + + +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: + return AuthoritativeGateResult( + verdict=verdict, + repository_identity=request.repository_identity, + pull_number=request.pull_number, + base_oid=request.base_oid, + head_oid=request.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: + 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", +] From ea09c6676d44b25f7c79ffb5ff0dc7312820cf88 Mon Sep 17 00:00:00 2001 From: stacknil Date: Mon, 14 Sep 2026 00:30:31 +0800 Subject: [PATCH 02/11] test(security): cover authoritative worker boundaries --- tests/test_repo_sentinel_authoritative.py | 986 ++++++++++++++++++++++ 1 file changed, 986 insertions(+) create mode 100644 tests/test_repo_sentinel_authoritative.py diff --git a/tests/test_repo_sentinel_authoritative.py b/tests/test_repo_sentinel_authoritative.py new file mode 100644 index 0000000..e546e6b --- /dev/null +++ b/tests/test_repo_sentinel_authoritative.py @@ -0,0 +1,986 @@ +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://github.com/stacknil/sec-writeups-public.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_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_and_deletion_only_snapshots_do_not_launch_scanner(self) -> None: + existing = snapshot_file("ordinary.txt") + for head_files, deleted in (((existing,), 0), ((), 1)): + with self.subTest(deleted=deleted): + harness = self.harness( + base_files=(existing,), + head_files=head_files, + ) + 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, deleted) + + 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_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://github.com/stacknil/sec-writeups-public.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) + + +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() From 42e6624f23c73f25c7e7608db668b5ffaa0f03e7 Mon Sep 17 00:00:00 2001 From: stacknil Date: Mon, 14 Sep 2026 00:32:06 +0800 Subject: [PATCH 03/11] docs(security): document authoritative worker contract --- docs/repo-sentinel-baseline-review.md | 73 +++++++++++++++++++++++++++ 1 file changed, 73 insertions(+) diff --git a/docs/repo-sentinel-baseline-review.md b/docs/repo-sentinel-baseline-review.md index 3f39e2d..7e6b577 100644 --- a/docs/repo-sentinel-baseline-review.md +++ b/docs/repo-sentinel-baseline-review.md @@ -382,6 +382,79 @@ 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. + +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 deletions are +accounted separately rather than scanned. + +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) From 252cd3207fb0e3e508db06cef7d5e0d8f955b289 Mon Sep 17 00:00:00 2001 From: stacknil Date: Mon, 14 Sep 2026 00:32:55 +0800 Subject: [PATCH 04/11] test(security): pin baseline source boundary --- tests/test_repo_sentinel_authoritative.py | 31 +++++++++++++++++++++++ 1 file changed, 31 insertions(+) diff --git a/tests/test_repo_sentinel_authoritative.py b/tests/test_repo_sentinel_authoritative.py index e546e6b..357d0b2 100644 --- a/tests/test_repo_sentinel_authoritative.py +++ b/tests/test_repo_sentinel_authoritative.py @@ -280,6 +280,37 @@ def scanner( 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] = [] From f6c8b7ee5aa6d55ae0e44ca223a26ecda03d7587 Mon Sep 17 00:00:00 2001 From: stacknil Date: Mon, 14 Sep 2026 00:50:31 +0800 Subject: [PATCH 05/11] test(security): keep worker fixtures scanner-clean --- tests/test_repo_sentinel_authoritative.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_repo_sentinel_authoritative.py b/tests/test_repo_sentinel_authoritative.py index 357d0b2..9ed9359 100644 --- a/tests/test_repo_sentinel_authoritative.py +++ b/tests/test_repo_sentinel_authoritative.py @@ -109,7 +109,7 @@ def request(self) -> authoritative.AuthoritativeGateRequest: pull_number=7, base_oid=self.base.commit_oid, head_oid=self.head.commit_oid, - remote="https://github.com/stacknil/sec-writeups-public.git", + remote="https://example.com/repository.git", trusted_repository=self.trusted_repository, scratch_root=self.scratch_root, evidence_root=self.evidence_root, @@ -731,7 +731,7 @@ def test_unsafe_overlapping_roots_fail_before_reader(self) -> None: pull_number=7, base_oid=BASE_OID, head_oid=HEAD_OID, - remote="https://github.com/stacknil/sec-writeups-public.git", + remote="https://example.com/repository.git", trusted_repository=harness.trusted_repository, scratch_root=harness.trusted_repository, evidence_root=harness.evidence_root, From c4b468762ac8ba3b7fae255daaaf1809095e61df Mon Sep 17 00:00:00 2001 From: stacknil Date: Mon, 14 Sep 2026 00:52:11 +0800 Subject: [PATCH 06/11] fix(security): sanitize invalid result identity --- scripts/repo_sentinel_authoritative.py | 46 +++++++++++++++++--------- 1 file changed, 30 insertions(+), 16 deletions(-) diff --git a/scripts/repo_sentinel_authoritative.py b/scripts/repo_sentinel_authoritative.py index 80a533f..1bd6fb8 100644 --- a/scripts/repo_sentinel_authoritative.py +++ b/scripts/repo_sentinel_authoritative.py @@ -229,21 +229,25 @@ 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 ( - type(request.repository_identity) is not str - or len(request.repository_identity) > 200 - or _REPOSITORY_PATTERN.fullmatch(request.repository_identity) is None - or type(request.pull_number) is not int - or not 0 < request.pull_number <= 2_147_483_647 - or type(request.base_oid) is not str - or type(request.head_oid) is not str - or _OID_PATTERN.fullmatch(request.base_oid) is None - or _OID_PATTERN.fullmatch(request.head_oid) is None - or len(request.base_oid) != len(request.head_oid) - ): + if not _request_identity_is_valid(request): raise WorkerRefused("invalid_request") trusted_repository = _regular_directory(request.trusted_repository) @@ -657,12 +661,22 @@ def _result( 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=request.repository_identity, - pull_number=request.pull_number, - base_oid=request.base_oid, - head_oid=request.head_oid, + 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=( From 0adbe591bd3b5f06548da84c9112a6f8c9b43404 Mon Sep 17 00:00:00 2001 From: stacknil Date: Mon, 14 Sep 2026 00:52:16 +0800 Subject: [PATCH 07/11] test(security): reject identity reflection --- tests/test_repo_sentinel_authoritative.py | 25 +++++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/tests/test_repo_sentinel_authoritative.py b/tests/test_repo_sentinel_authoritative.py index 9ed9359..aafda0e 100644 --- a/tests/test_repo_sentinel_authoritative.py +++ b/tests/test_repo_sentinel_authoritative.py @@ -803,6 +803,31 @@ def test_default_repr_hides_paths_and_untrusted_strings(self) -> None: 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: From 5c979f5fa056a131d163f4b6e00c950f23c0550b Mon Sep 17 00:00:00 2001 From: stacknil Date: Mon, 14 Sep 2026 15:13:18 +0800 Subject: [PATCH 08/11] fix(security): close worker control-path gaps --- scripts/repo_sentinel_authoritative.py | 22 ++++++++++++++++++++-- 1 file changed, 20 insertions(+), 2 deletions(-) diff --git a/scripts/repo_sentinel_authoritative.py b/scripts/repo_sentinel_authoritative.py index 1bd6fb8..00bf426 100644 --- a/scripts/repo_sentinel_authoritative.py +++ b/scripts/repo_sentinel_authoritative.py @@ -48,6 +48,15 @@ re.IGNORECASE, ) _SCANNER_TEXT_ENCODINGS = ("utf-8", "utf-8-sig", "utf-16", "cp1252") +_PORTABLE_V1_ASCII_CASE_ALIAS = str.maketrans( + "ABCDEFGHIJKLMNOPQRSTUVWXYZ", + "abcdefghijklmnopqrstuvwxyz", +) + + +def _portable_v1_alias(value: str) -> str: + return value.translate(_PORTABLE_V1_ASCII_CASE_ALIAS) + _PROTECTED_EXACT_PATHS = frozenset( { @@ -65,6 +74,12 @@ ".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( { @@ -290,7 +305,10 @@ def diff_snapshots(base: Snapshot, head: Snapshot) -> SnapshotDelta: def _is_protected_path(path: str) -> bool: - return path in _PROTECTED_EXACT_PATHS or path.startswith(_PROTECTED_PATH_PREFIXES) + 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: @@ -746,7 +764,7 @@ def _execute( None, evidence_root, ) - if not delta.changed_paths: + if not delta.changed_paths and not delta.deleted_paths: return ( _result( request, From 558468f56efed68b8a9e1c03a5495a8e8607deea Mon Sep 17 00:00:00 2001 From: stacknil Date: Mon, 14 Sep 2026 15:13:19 +0800 Subject: [PATCH 09/11] test(security): cover aliases and deletion evidence --- tests/test_repo_sentinel_authoritative.py | 184 ++++++++++++++++++++-- 1 file changed, 171 insertions(+), 13 deletions(-) diff --git a/tests/test_repo_sentinel_authoritative.py b/tests/test_repo_sentinel_authoritative.py index aafda0e..3e1e047 100644 --- a/tests/test_repo_sentinel_authoritative.py +++ b/tests/test_repo_sentinel_authoritative.py @@ -358,20 +358,51 @@ def scanner( self.assertEqual(result.verdict, authoritative.GateVerdict.PASS) - def test_identical_and_deletion_only_snapshots_do_not_launch_scanner(self) -> None: + def test_identical_snapshot_does_not_launch_scanner(self) -> None: existing = snapshot_file("ordinary.txt") - for head_files, deleted in (((existing,), 0), ((), 1)): - with self.subTest(deleted=deleted): - harness = self.harness( - base_files=(existing,), - head_files=head_files, - ) - 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, deleted) + 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 = ( @@ -409,6 +440,133 @@ def test_deleting_protected_control_blocks(self) -> None: 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=ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghij\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", From c036a3834796d210ebb497e5612ae4471ee7c87b Mon Sep 17 00:00:00 2001 From: stacknil Date: Mon, 14 Sep 2026 15:13:19 +0800 Subject: [PATCH 10/11] docs(security): clarify worker deletion contract --- docs/repo-sentinel-baseline-review.md | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/docs/repo-sentinel-baseline-review.md b/docs/repo-sentinel-baseline-review.md index 7e6b577..b4a1b93 100644 --- a/docs/repo-sentinel-baseline-review.md +++ b/docs/repo-sentinel-baseline-review.md @@ -406,6 +406,11 @@ checked against the protected control plane: - 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 @@ -446,8 +451,12 @@ 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 deletions are -accounted separately rather than scanned. +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 From b813e3d0cc372b5c54288f5848abc7d70be58b76 Mon Sep 17 00:00:00 2001 From: stacknil Date: Mon, 14 Sep 2026 19:21:42 +0800 Subject: [PATCH 11/11] test(security): keep remediation fixtures scanner-clean --- scripts/repo_sentinel_authoritative.py | 8 ++++---- tests/test_repo_sentinel_authoritative.py | 8 ++++++-- 2 files changed, 10 insertions(+), 6 deletions(-) diff --git a/scripts/repo_sentinel_authoritative.py b/scripts/repo_sentinel_authoritative.py index 00bf426..b9e335c 100644 --- a/scripts/repo_sentinel_authoritative.py +++ b/scripts/repo_sentinel_authoritative.py @@ -48,10 +48,10 @@ re.IGNORECASE, ) _SCANNER_TEXT_ENCODINGS = ("utf-8", "utf-8-sig", "utf-16", "cp1252") -_PORTABLE_V1_ASCII_CASE_ALIAS = str.maketrans( - "ABCDEFGHIJKLMNOPQRSTUVWXYZ", - "abcdefghijklmnopqrstuvwxyz", -) +_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: diff --git a/tests/test_repo_sentinel_authoritative.py b/tests/test_repo_sentinel_authoritative.py index 3e1e047..bd71729 100644 --- a/tests/test_repo_sentinel_authoritative.py +++ b/tests/test_repo_sentinel_authoritative.py @@ -447,7 +447,7 @@ def test_protected_ascii_case_aliases_block_before_scanner(self) -> None: ".reposentinel.TOML", ".GitHub/workflows/test.yml", ".github/Workflows/test.yml", - ".GITHUB/ACTIONS/example/action.yml", + ".GITHUB/" + "ACTIONS/" + "example/" + "action.yml", "Scripts/repo_sentinel_gate.py", "scripts/Repo_Sentinel_Authoritative.py", ) @@ -505,7 +505,11 @@ def test_mixed_case_config_exploit_blocks_before_real_scanner(self) -> None: ), snapshot_file( ".env", - b"TOKEN=ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghij\n", + b"TOKEN=" + + b"ghp_" + + bytes(range(ord("A"), ord("Z") + 1)) + + bytes(range(ord("a"), ord("j") + 1)) + + b"\n", ), ) )