From 4cae021359f056ff70e9b0c390f83fabc4349351 Mon Sep 17 00:00:00 2001 From: Yernat Yestekov Date: Thu, 13 Aug 2026 11:22:03 -0700 Subject: [PATCH 1/4] feat(cli): verify offline evidence bundle integrity --- noxfile.py | 46 +++ pyproject.toml | 2 +- src/raes_adapters/bundle_verifier.py | 408 +++++++++++++++++++++++++++ src/raes_adapters/entrypoint.py | 20 ++ tests/test_bundle_verifier.py | 368 ++++++++++++++++++++++++ 5 files changed, 843 insertions(+), 1 deletion(-) create mode 100644 src/raes_adapters/bundle_verifier.py create mode 100644 src/raes_adapters/entrypoint.py create mode 100644 tests/test_bundle_verifier.py diff --git a/noxfile.py b/noxfile.py index dad1857..0369acf 100644 --- a/noxfile.py +++ b/noxfile.py @@ -27,6 +27,8 @@ from __future__ import annotations +import hashlib +import json from pathlib import Path import shutil import sys @@ -572,6 +574,50 @@ def _distributions(session: nox.Session) -> None: IMPORT_PACKAGE, env={"PYTHONPATH": "", "PYTHONSAFEPATH": "1"}, ) + verifier_bundle = workdir / "verifier-bundle" + if verifier_bundle.exists(): + shutil.rmtree(verifier_bundle) + verifier_bundle.mkdir() + verifier_payload = b"installed verifier probe\n" + (verifier_bundle / "evidence.bin").write_bytes(verifier_payload) + (verifier_bundle / "inventory.json").write_text( + json.dumps( + { + "artifacts": [ + { + "media_type": "application/octet-stream", + "path": "evidence.bin", + "sha256": hashlib.sha256(verifier_payload).hexdigest(), + "size_bytes": len(verifier_payload), + } + ] + }, + indent=2, + sort_keys=True, + ) + + "\n", + encoding="utf-8", + ) + _run( + session, + str(venv / "bin" / "raes-adapters"), + "verify-bundle", + "--bundle", + str(verifier_bundle), + "--format", + "json", + env={"PYTHONPATH": "", "PYTHONSAFEPATH": "1"}, + ) + (verifier_bundle / "evidence.bin").write_bytes(b"tampered\n") + _run( + session, + str(venv / "bin" / "raes-adapters"), + "verify-bundle", + "--bundle", + str(verifier_bundle), + success_codes=[3], + env={"PYTHONPATH": "", "PYTHONSAFEPATH": "1"}, + ) session.log("clean install: cyberbattlesim extra conformance") conformance_venv = workdir / "venv-cyberbattlesim" diff --git a/pyproject.toml b/pyproject.toml index 089e9e4..3fbec84 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -67,7 +67,7 @@ cyborg = ["raes-env-packs==3.6.2"] nasim = ["nasim==0.12.0", "gymnasium==0.26.3", "numpy==1.26.4", "raes-env-packs==3.6.2"] [project.scripts] -raes-adapters = "raes_adapters.cli:main" +raes-adapters = "raes_adapters.entrypoint:main" [project.urls] Homepage = "https://github.com/OpenRAE/adapters" diff --git a/src/raes_adapters/bundle_verifier.py b/src/raes_adapters/bundle_verifier.py new file mode 100644 index 0000000..62f8801 --- /dev/null +++ b/src/raes_adapters/bundle_verifier.py @@ -0,0 +1,408 @@ +"""Offline, integrity-only verification for inventory-sealed bundles.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import stat +import sys +from collections import deque +from dataclasses import dataclass +from pathlib import Path, PurePosixPath +from typing import NoReturn + +EXIT_VERIFIED = 0 +EXIT_USAGE = 2 +EXIT_INVALID = 3 +EXIT_INTERNAL = 70 + +MAX_FILES = 100_000 +MAX_INVENTORIES = 100_000 +MAX_ENTRIES = 200_000 +MAX_UNIQUE_BYTES = 4 * 1024**3 +MAX_ARTIFACT_BYTES = 500 * 1024 +MAX_INVENTORY_BYTES = 16 * 1024**2 +MAX_DEPTH = 32 +MAX_PATH_BYTES = 1_024 + +_INVENTORY_NAME = "inventory.json" +_ENTRY_KEYS = {"media_type", "path", "sha256", "size_bytes"} + + +class BundleInvalid(ValueError): + """A stable integrity failure safe to render to a user.""" + + def __init__(self, code: str) -> None: + super().__init__(code) + self.code = code + + +class _UsageFailure(Exception): + pass + + +class _Parser(argparse.ArgumentParser): + def error(self, message: str) -> NoReturn: + del message + raise _UsageFailure + + +@dataclass(frozen=True) +class VerificationCard: + status: str + code: str + files: int = 0 + inventories: int = 0 + entries: int = 0 + unique_bytes: int = 0 + + def payload(self) -> dict[str, object]: + return { + "claim": "integrity-only", + "code": self.code, + "counts": { + "entries": self.entries, + "files": self.files, + "inventories": self.inventories, + "unique_bytes": self.unique_bytes, + }, + "disclaimers": [ + "semantic fidelity not assessed", + "capture completeness not assessed", + ], + "status": self.status, + } + + +@dataclass(frozen=True) +class _FileRecord: + relative: str + content: bytes + size: int + sha256: str + identity: tuple[int, int, int, int, int, int] + + +def _invalid(code: str) -> NoReturn: + raise BundleInvalid(code) + + +def _json_object(pairs: list[tuple[str, object]]) -> dict[str, object]: + result: dict[str, object] = {} + for key, value in pairs: + if key in result: + _invalid("bundle.inventory.duplicate-key") + result[key] = value + return result + + +def _relative_path(value: object) -> str: + if not isinstance(value, str) or not value or "\x00" in value: + _invalid("bundle.inventory.path-invalid") + try: + encoded = value.encode("utf-8") + except UnicodeEncodeError: + _invalid("bundle.inventory.path-invalid") + if len(encoded) > MAX_PATH_BYTES: + _invalid("bundle.inventory.path-invalid") + path = PurePosixPath(value) + if path.is_absolute() or ".." in path.parts or "." in path.parts or "\\" in value: + _invalid("bundle.inventory.path-invalid") + if len(path.parts) > MAX_DEPTH: + _invalid("bundle.limit.depth") + normalized = path.as_posix() + if normalized != value: + _invalid("bundle.inventory.path-invalid") + return normalized + + +def _scan(root: Path) -> dict[str, Path]: + files: dict[str, Path] = {} + pending: list[tuple[Path, int]] = [(root, 0)] + scanned_entries = 0 + while pending: + directory, depth = pending.pop() + if depth > MAX_DEPTH: + _invalid("bundle.limit.depth") + try: + entries = os.scandir(directory) + except OSError: + _invalid("bundle.filesystem.unreadable") + try: + with entries: + for entry in entries: + scanned_entries += 1 + if scanned_entries > MAX_FILES: + _invalid("bundle.limit.files") + relative = Path(entry.path).relative_to(root).as_posix() + _relative_path(relative) + try: + mode = entry.stat(follow_symlinks=False).st_mode + except OSError: + _invalid("bundle.filesystem.unreadable") + if stat.S_ISLNK(mode): + _invalid("bundle.filesystem.symlink") + if stat.S_ISDIR(mode): + pending.append((Path(entry.path), depth + 1)) + continue + if not stat.S_ISREG(mode): + _invalid("bundle.filesystem.special-file") + files[relative] = Path(entry.path) + except OSError: + _invalid("bundle.filesystem.unreadable") + return files + + +def _identity(info: os.stat_result) -> tuple[int, int, int, int, int, int]: + return ( + info.st_dev, + info.st_ino, + info.st_mode, + info.st_size, + info.st_mtime_ns, + info.st_ctime_ns, + ) + + +def _read_stable(root: Path, relative: str, *, inventory: bool) -> _FileRecord: + candidate = root / relative + flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0) + try: + before_path = candidate.lstat() + descriptor = os.open(candidate, flags) + try: + before = os.fstat(descriptor) + if not stat.S_ISREG(before.st_mode) or _identity(before_path) != _identity(before): + _invalid("bundle.filesystem.mutated") + limit = MAX_INVENTORY_BYTES if inventory else MAX_ARTIFACT_BYTES + chunks: list[bytes] = [] + total = 0 + while True: + chunk = os.read(descriptor, min(1024 * 1024, limit + 1 - total)) + if not chunk: + break + chunks.append(chunk) + total += len(chunk) + if total > limit: + code = ( + "bundle.limit.inventory-bytes" + if inventory + else "bundle.limit.artifact-bytes" + ) + _invalid(code) + after = os.fstat(descriptor) + finally: + os.close(descriptor) + after_path = candidate.lstat() + except BundleInvalid: + raise + except OSError: + _invalid("bundle.filesystem.mutated") + if _identity(before) != _identity(after) or _identity(after) != _identity(after_path): + _invalid("bundle.filesystem.mutated") + content = b"".join(chunks) + return _FileRecord( + relative=relative, + content=content, + size=len(content), + sha256=hashlib.sha256(content).hexdigest(), + identity=_identity(after), + ) + + +def _inventory_payload(record: _FileRecord) -> list[object]: + try: + payload = json.loads(record.content, object_pairs_hook=_json_object) + except BundleInvalid: + raise + except (UnicodeDecodeError, json.JSONDecodeError): + _invalid("bundle.inventory.malformed") + if not isinstance(payload, dict) or set(payload) != {"artifacts"}: + _invalid("bundle.inventory.malformed") + artifacts = payload["artifacts"] + if not isinstance(artifacts, list): + _invalid("bundle.inventory.malformed") + return artifacts + + +def _entry(item: object) -> tuple[str, int, str]: + if not isinstance(item, dict) or set(item) != _ENTRY_KEYS: + _invalid("bundle.inventory.entry-malformed") + relative = _relative_path(item["path"]) + size = item["size_bytes"] + digest = item["sha256"] + media_type = item["media_type"] + if isinstance(size, bool) or not isinstance(size, int) or size < 0: + _invalid("bundle.inventory.entry-malformed") + if ( + not isinstance(digest, str) + or len(digest) != 64 + or any(character not in "0123456789abcdef" for character in digest) + or not isinstance(media_type, str) + or not media_type + ): + _invalid("bundle.inventory.entry-malformed") + return relative, size, digest + + +def verify_bundle(bundle: Path) -> VerificationCard: + """Verify exact flat or transitive inventory closure without side effects.""" + + try: + root_stat = bundle.lstat() + except OSError: + _invalid("bundle.root.invalid") + if stat.S_ISLNK(root_stat.st_mode) or not stat.S_ISDIR(root_stat.st_mode): + _invalid("bundle.root.invalid") + root_identity = _identity(root_stat) + root = bundle.resolve() + files = _scan(root) + if _INVENTORY_NAME not in files: + _invalid("bundle.inventory.missing") + + records: dict[str, _FileRecord] = {} + referenced: set[str] = set() + inventories_seen: set[str] = set() + inventory_queue = deque([_INVENTORY_NAME]) + entries_count = 0 + unique_bytes = 0 + + def record(relative: str, *, inventory: bool) -> _FileRecord: + nonlocal unique_bytes + existing = records.get(relative) + if existing is not None: + return existing + if relative not in files: + _invalid("bundle.inventory.member-missing") + loaded = _read_stable(root, relative, inventory=inventory) + records[relative] = loaded + unique_bytes += loaded.size + if unique_bytes > MAX_UNIQUE_BYTES: + _invalid("bundle.limit.unique-bytes") + return loaded + + while inventory_queue: + inventory_path = inventory_queue.popleft() + if inventory_path in inventories_seen: + _invalid("bundle.inventory.duplicate") + inventories_seen.add(inventory_path) + if len(inventories_seen) > MAX_INVENTORIES: + _invalid("bundle.limit.inventories") + inventory_record = record(inventory_path, inventory=True) + base = PurePosixPath(inventory_path).parent + for item in _inventory_payload(inventory_record): + entries_count += 1 + if entries_count > MAX_ENTRIES: + _invalid("bundle.limit.entries") + child, expected_size, expected_digest = _entry(item) + joined = (base / child).as_posix() + joined = _relative_path(joined) + if joined == _INVENTORY_NAME or joined in referenced: + _invalid("bundle.inventory.duplicate-path") + referenced.add(joined) + is_inventory = PurePosixPath(joined).name == _INVENTORY_NAME + actual = record(joined, inventory=is_inventory) + if actual.size != expected_size or actual.sha256 != expected_digest: + _invalid("bundle.inventory.member-mismatch") + if is_inventory: + inventory_queue.append(joined) + + expected = set(files) - {_INVENTORY_NAME} + if referenced != expected: + _invalid("bundle.inventory.membership-mismatch") + final_files = _scan(root) + if set(final_files) != set(files): + _invalid("bundle.filesystem.mutated") + try: + if _identity(bundle.lstat()) != root_identity: + _invalid("bundle.filesystem.mutated") + for relative, loaded in records.items(): + current = (root / relative).lstat() + if not stat.S_ISREG(current.st_mode) or _identity(current) != loaded.identity: + _invalid("bundle.filesystem.mutated") + except BundleInvalid: + raise + except OSError: + _invalid("bundle.filesystem.mutated") + return VerificationCard( + status="verified", + code="bundle.integrity.verified", + files=len(files), + inventories=len(inventories_seen), + entries=entries_count, + unique_bytes=unique_bytes, + ) + + +def render_card(card: VerificationCard, output_format: str) -> str: + payload = card.payload() + if output_format == "json": + return json.dumps(payload, sort_keys=True, separators=(",", ":")) + counts = payload["counts"] + assert isinstance(counts, dict) + if output_format == "markdown": + return "\n".join( + ( + "# Bundle integrity card", + "", + f"- Status: `{card.status}`", + f"- Code: `{card.code}`", + "- Claim: integrity-only", + f"- Files: {counts['files']}", + f"- Inventories: {counts['inventories']}", + f"- Entries: {counts['entries']}", + f"- Unique bytes: {counts['unique_bytes']}", + "- Semantic fidelity: not assessed", + "- Capture completeness: not assessed", + ) + ) + return "\n".join( + ( + f"status: {card.status}", + f"code: {card.code}", + "claim: integrity-only", + f"files: {counts['files']}", + f"inventories: {counts['inventories']}", + f"entries: {counts['entries']}", + f"unique-bytes: {counts['unique_bytes']}", + "semantic-fidelity: not-assessed", + "capture-completeness: not-assessed", + ) + ) + + +def _parser() -> _Parser: + parser = _Parser(prog="raes-adapters verify-bundle") + parser.add_argument("--bundle", type=Path, required=True) + parser.add_argument("--format", choices=("json", "terminal", "markdown"), default="terminal") + return parser + + +def main(argv: list[str] | None = None) -> int: + try: + args = _parser().parse_args(argv) + try: + card = verify_bundle(args.bundle) + result = EXIT_VERIFIED + except BundleInvalid as error: + card = VerificationCard(status="invalid", code=error.code) + result = EXIT_INVALID + print(render_card(card, args.format)) + return result + except _UsageFailure: + print("verify-bundle.usage.invalid: invalid command line", file=sys.stderr) + return EXIT_USAGE + except Exception: + print("verify-bundle.internal.failure: internal verification failure", file=sys.stderr) + return EXIT_INTERNAL + + +__all__ = [ + "BundleInvalid", + "VerificationCard", + "main", + "render_card", + "verify_bundle", +] diff --git a/src/raes_adapters/entrypoint.py b/src/raes_adapters/entrypoint.py new file mode 100644 index 0000000..ef2a154 --- /dev/null +++ b/src/raes_adapters/entrypoint.py @@ -0,0 +1,20 @@ +"""Lazy command dispatcher preserving offline verifier isolation.""" + +from __future__ import annotations + +import sys +from collections.abc import Sequence + + +def main(argv: Sequence[str] | None = None) -> int: + arguments = list(sys.argv[1:] if argv is None else argv) + if arguments[:1] == ["verify-bundle"]: + from raes_adapters.bundle_verifier import main as verify_main + + return verify_main(arguments[1:]) + from raes_adapters.cli import main as researcher_main + + return researcher_main(arguments) + + +__all__ = ["main"] diff --git a/tests/test_bundle_verifier.py b/tests/test_bundle_verifier.py new file mode 100644 index 0000000..a4091d5 --- /dev/null +++ b/tests/test_bundle_verifier.py @@ -0,0 +1,368 @@ +"""Adversarial checks for the installed offline bundle verifier.""" + +from __future__ import annotations + +import hashlib +import json +import os +from pathlib import Path + +import pytest + +from raes_adapters import bundle_verifier +from raes_adapters.entrypoint import main + + +def _entry( + path: str, content: bytes, media_type: str = "application/octet-stream" +) -> dict[str, object]: + return { + "media_type": media_type, + "path": path, + "sha256": hashlib.sha256(content).hexdigest(), + "size_bytes": len(content), + } + + +def _write_inventory(root: Path, entries: list[dict[str, object]]) -> None: + (root / "inventory.json").write_text( + json.dumps({"artifacts": entries}, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + + +def test_flat_bundle_and_all_renderers_are_deterministic(tmp_path: Path) -> None: + payload = b"bounded evidence\n" + (tmp_path / "evidence.bin").write_bytes(payload) + _write_inventory(tmp_path, [_entry("evidence.bin", payload)]) + + card = bundle_verifier.verify_bundle(tmp_path) + + assert card.status == "verified" + assert bundle_verifier.render_card(card, "json") == bundle_verifier.render_card(card, "json") + assert "integrity-only" in bundle_verifier.render_card(card, "terminal") + assert "Semantic fidelity: not assessed" in bundle_verifier.render_card(card, "markdown") + + +def test_transitive_inventory_closure_is_verified(tmp_path: Path) -> None: + child = tmp_path / "run" + child.mkdir() + evidence = b"{}\n" + (child / "evidence.json").write_bytes(evidence) + _write_inventory(child, [_entry("evidence.json", evidence, "application/json")]) + child_inventory = (child / "inventory.json").read_bytes() + _write_inventory( + tmp_path, + [_entry("run/inventory.json", child_inventory, "application/json")], + ) + + card = bundle_verifier.verify_bundle(tmp_path) + + assert card.inventories == 2 + assert card.entries == 2 + + +@pytest.mark.parametrize( + ("mutation", "code"), + [ + ("tamper", "bundle.inventory.member-mismatch"), + ("extra", "bundle.inventory.membership-mismatch"), + ("missing", "bundle.inventory.member-missing"), + ("traversal", "bundle.inventory.path-invalid"), + ("duplicate", "bundle.inventory.duplicate-path"), + ], +) +def test_invalid_membership_and_tampering_are_rejected( + tmp_path: Path, mutation: str, code: str +) -> None: + payload = b"evidence" + (tmp_path / "item").write_bytes(payload) + entries = [_entry("item", payload)] + if mutation == "tamper": + entries[0]["sha256"] = "0" * 64 + elif mutation == "extra": + (tmp_path / "extra").write_bytes(b"x") + elif mutation == "missing": + entries[0]["path"] = "absent" + elif mutation == "traversal": + entries[0]["path"] = "../item" + elif mutation == "duplicate": + entries.append(dict(entries[0])) + _write_inventory(tmp_path, entries) + + with pytest.raises(bundle_verifier.BundleInvalid, match=code): + bundle_verifier.verify_bundle(tmp_path) + + +def test_symlink_is_rejected_even_when_not_in_inventory(tmp_path: Path) -> None: + outside = tmp_path.parent / "outside" + outside.write_bytes(b"outside") + (tmp_path / "link").symlink_to(outside) + _write_inventory(tmp_path, []) + + with pytest.raises(bundle_verifier.BundleInvalid, match="bundle.filesystem.symlink"): + bundle_verifier.verify_bundle(tmp_path) + + +def test_entrypoint_uses_documented_exit_codes( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + _write_inventory(tmp_path, []) + assert main(["verify-bundle", "--bundle", str(tmp_path), "--format", "json"]) == 0 + assert json.loads(capsys.readouterr().out)["status"] == "verified" + + (tmp_path / "extra").write_bytes(b"x") + assert main(["verify-bundle", "--bundle", str(tmp_path)]) == 3 + assert "status: invalid" in capsys.readouterr().out + assert main(["verify-bundle"]) == 2 + assert "usage.invalid" in capsys.readouterr().err + + +def test_verify_dispatch_does_not_import_simulator_modules( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _write_inventory(tmp_path, []) + imported: list[str] = [] + original = __import__ + + def guarded(name: str, *args: object, **kwargs: object) -> object: + if name.startswith(("CybORG", "cyberbattle", "nasim")): + imported.append(name) + return original(name, *args, **kwargs) + + monkeypatch.setattr("builtins.__import__", guarded) + assert main(["verify-bundle", "--bundle", str(tmp_path)]) == 0 + assert imported == [] + + +def test_fixed_admission_limits_match_the_public_contract() -> None: + assert bundle_verifier.MAX_FILES == 100_000 + assert bundle_verifier.MAX_INVENTORIES == 100_000 + assert bundle_verifier.MAX_ENTRIES == 200_000 + assert bundle_verifier.MAX_UNIQUE_BYTES == 4 * 1024**3 + assert bundle_verifier.MAX_ARTIFACT_BYTES == 500 * 1024 + assert bundle_verifier.MAX_INVENTORY_BYTES == 16 * 1024**2 + assert bundle_verifier.MAX_DEPTH == 32 + assert bundle_verifier.MAX_PATH_BYTES == 1_024 + + +@pytest.mark.parametrize( + ("limit", "value", "code"), + [ + ("MAX_FILES", 0, "bundle.limit.files"), + ("MAX_INVENTORIES", 0, "bundle.limit.inventories"), + ("MAX_ENTRIES", 0, "bundle.limit.entries"), + ("MAX_UNIQUE_BYTES", 0, "bundle.limit.unique-bytes"), + ("MAX_ARTIFACT_BYTES", 0, "bundle.limit.artifact-bytes"), + ("MAX_INVENTORY_BYTES", 0, "bundle.limit.inventory-bytes"), + ("MAX_DEPTH", 0, "bundle.limit.depth"), + ("MAX_PATH_BYTES", 1, "bundle.inventory.path-invalid"), + ], +) +def test_every_admission_limit_fails_closed( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + limit: str, + value: int, + code: str, +) -> None: + payload = b"x" + (tmp_path / "item").write_bytes(payload) + _write_inventory(tmp_path, [_entry("item", payload)]) + monkeypatch.setattr(bundle_verifier, limit, value) + + with pytest.raises(bundle_verifier.BundleInvalid, match=code): + bundle_verifier.verify_bundle(tmp_path) + + +@pytest.mark.parametrize( + "payload", + [ + b"not json", + b"[]", + b'{"artifacts":{},"extra":1}', + b'{"artifacts":{}}', + b'{"artifacts":[],"artifacts":[]}', + ], +) +def test_malformed_inventories_are_rejected(tmp_path: Path, payload: bytes) -> None: + (tmp_path / "inventory.json").write_bytes(payload) + + with pytest.raises(bundle_verifier.BundleInvalid, match="bundle.inventory"): + bundle_verifier.verify_bundle(tmp_path) + + +@pytest.mark.parametrize( + ("field", "value"), + [ + ("media_type", ""), + ("size_bytes", True), + ("size_bytes", -1), + ("sha256", "A" * 64), + ("sha256", "0" * 63), + ("path", "/item"), + ("path", "../item"), + ("path", "./item"), + ("path", "item\\other"), + ("path", "item//other"), + ("path", "item\x00suffix"), + ], +) +def test_malformed_inventory_entries_are_rejected( + tmp_path: Path, field: str, value: object +) -> None: + payload = b"x" + (tmp_path / "item").write_bytes(payload) + entry = _entry("item", payload) + entry[field] = value + _write_inventory(tmp_path, [entry]) + + with pytest.raises(bundle_verifier.BundleInvalid, match="bundle.inventory"): + bundle_verifier.verify_bundle(tmp_path) + + +def test_inventory_entry_requires_exact_keys(tmp_path: Path) -> None: + payload = b"x" + (tmp_path / "item").write_bytes(payload) + entry = _entry("item", payload) + del entry["media_type"] + _write_inventory(tmp_path, [entry]) + + with pytest.raises(bundle_verifier.BundleInvalid, match="bundle.inventory.entry-malformed"): + bundle_verifier.verify_bundle(tmp_path) + + +def test_special_files_are_rejected(tmp_path: Path) -> None: + if not hasattr(os, "mkfifo"): + pytest.skip("FIFO creation is unavailable") + os.mkfifo(tmp_path / "pipe") + _write_inventory(tmp_path, []) + + with pytest.raises(bundle_verifier.BundleInvalid, match="bundle.filesystem.special-file"): + bundle_verifier.verify_bundle(tmp_path) + + +def test_root_must_be_a_real_directory_with_an_inventory(tmp_path: Path) -> None: + missing = tmp_path / "missing" + with pytest.raises(bundle_verifier.BundleInvalid, match="bundle.root.invalid"): + bundle_verifier.verify_bundle(missing) + + regular = tmp_path / "regular" + regular.write_bytes(b"x") + with pytest.raises(bundle_verifier.BundleInvalid, match="bundle.root.invalid"): + bundle_verifier.verify_bundle(regular) + + empty = tmp_path / "empty" + empty.mkdir() + with pytest.raises(bundle_verifier.BundleInvalid, match="bundle.inventory.missing"): + bundle_verifier.verify_bundle(empty) + + target = tmp_path / "target" + target.mkdir() + _write_inventory(target, []) + root_link = tmp_path / "root-link" + root_link.symlink_to(target, target_is_directory=True) + with pytest.raises(bundle_verifier.BundleInvalid, match="bundle.root.invalid"): + bundle_verifier.verify_bundle(root_link) + + +def test_mutation_during_hashing_is_rejected( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _write_inventory(tmp_path, []) + original = bundle_verifier._identity + calls = 0 + + def changing_identity(info: os.stat_result) -> tuple[int, int, int, int, int, int]: + nonlocal calls + calls += 1 + identity = original(info) + if calls == 5: + return (*identity[:-1], identity[-1] + 1) + return identity + + monkeypatch.setattr(bundle_verifier, "_identity", changing_identity) + with pytest.raises(bundle_verifier.BundleInvalid, match="bundle.filesystem.mutated"): + bundle_verifier.verify_bundle(tmp_path) + + +def test_files_added_while_hashing_are_rejected( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _write_inventory(tmp_path, []) + original = bundle_verifier._inventory_payload + + def mutate(record: bundle_verifier._FileRecord) -> list[object]: + result = original(record) + (tmp_path / "late-file").write_bytes(b"late") + return result + + monkeypatch.setattr(bundle_verifier, "_inventory_payload", mutate) + with pytest.raises(bundle_verifier.BundleInvalid, match="bundle.filesystem.mutated"): + bundle_verifier.verify_bundle(tmp_path) + + +def test_verification_is_read_only(tmp_path: Path) -> None: + payload = b"evidence" + (tmp_path / "item").write_bytes(payload) + _write_inventory(tmp_path, [_entry("item", payload)]) + before = { + path.name: (path.read_bytes(), path.stat().st_mtime_ns) for path in tmp_path.iterdir() + } + + bundle_verifier.verify_bundle(tmp_path) + + after = {path.name: (path.read_bytes(), path.stat().st_mtime_ns) for path in tmp_path.iterdir()} + assert after == before + + +def test_cards_are_stable_and_disclose_only_integrity(tmp_path: Path) -> None: + _write_inventory(tmp_path, []) + card = bundle_verifier.verify_bundle(tmp_path) + counts = card.payload()["counts"] + assert isinstance(counts, dict) + assert bundle_verifier.render_card(card, "json") == json.dumps( + card.payload(), sort_keys=True, separators=(",", ":") + ) + assert bundle_verifier.render_card(card, "terminal") == "\n".join( + ( + "status: verified", + "code: bundle.integrity.verified", + "claim: integrity-only", + f"files: {counts['files']}", + f"inventories: {counts['inventories']}", + f"entries: {counts['entries']}", + f"unique-bytes: {counts['unique_bytes']}", + "semantic-fidelity: not-assessed", + "capture-completeness: not-assessed", + ) + ) + markdown = bundle_verifier.render_card(card, "markdown") + assert markdown.startswith("# Bundle integrity card\n") + assert "Semantic fidelity: not assessed" in markdown + assert "Capture completeness: not assessed" in markdown + + +def test_invalid_output_does_not_echo_sensitive_paths( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + secret = "customer-token-should-not-leak" + _write_inventory(tmp_path, [_entry(secret, b"absent")]) + + assert main(["verify-bundle", "--bundle", str(tmp_path)]) == 3 + captured = capsys.readouterr() + assert secret not in captured.out + assert secret not in captured.err + + +def test_unexpected_failures_use_exit_70( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + def fail(_bundle: Path) -> bundle_verifier.VerificationCard: + raise RuntimeError("sensitive internal detail") + + monkeypatch.setattr(bundle_verifier, "verify_bundle", fail) + assert main(["verify-bundle", "--bundle", str(tmp_path)]) == 70 + captured = capsys.readouterr() + assert "internal.failure" in captured.err + assert "sensitive internal detail" not in captured.err From df7a9d91fb415c3ebee4076b2ec286165e4b5b91 Mon Sep 17 00:00:00 2001 From: Yernat Yestekov Date: Thu, 13 Aug 2026 11:48:43 -0700 Subject: [PATCH 2/4] fix(cli): satisfy bundle verifier quality gate --- pyproject.toml | 1 + src/raes_adapters/bundle_verifier.py | 463 ++++++++++++++++++--------- src/raes_adapters/entrypoint.py | 2 + tests/test_bundle_verifier.py | 4 +- 4 files changed, 311 insertions(+), 159 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 3fbec84..6be0d0b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -124,6 +124,7 @@ select = ["E", "F", "I", "UP", "B", "SIM", "C4"] "src/raes_adapters/_gym_backend/provisioner.py" = ["UP004"] "src/raes_adapters/_gym_backend/evaluator.py" = ["UP004"] "src/raes_adapters/_gym_backend/participant_runtime.py" = ["UP004"] +"src/raes_adapters/bundle_verifier.py" = ["UP004"] "src/raes_adapters/cyberbattlesim/backend/driver.py" = ["UP004"] "src/raes_adapters/cyberbattlesim/backend/evaluator.py" = ["UP004"] "src/raes_adapters/cyberbattlesim/backend/orchestrator.py" = ["UP004"] diff --git a/src/raes_adapters/bundle_verifier.py b/src/raes_adapters/bundle_verifier.py index 62f8801..6f0ea3e 100644 --- a/src/raes_adapters/bundle_verifier.py +++ b/src/raes_adapters/bundle_verifier.py @@ -29,6 +29,11 @@ _INVENTORY_NAME = "inventory.json" _ENTRY_KEYS = {"media_type", "path", "sha256", "size_bytes"} +_CODE_ENTRY_MALFORMED = "bundle.inventory.entry-malformed" +_CODE_FILESYSTEM_MUTATED = "bundle.filesystem.mutated" +_CODE_FILESYSTEM_UNREADABLE = "bundle.filesystem.unreadable" +_CODE_INVENTORY_MALFORMED = "bundle.inventory.malformed" +_CODE_PATH_INVALID = "bundle.inventory.path-invalid" class BundleInvalid(ValueError): @@ -40,17 +45,23 @@ def __init__(self, code: str) -> None: class _UsageFailure(Exception): - pass + """An intentionally detail-free command-line usage failure.""" class _Parser(argparse.ArgumentParser): + """Argument parser that maps usage errors to the documented exit code.""" + def error(self, message: str) -> NoReturn: + """Raise a private usage exception without echoing input paths.""" + del message raise _UsageFailure @dataclass(frozen=True) -class VerificationCard: +class VerificationCard(object): + """Deterministic integrity-only result safe for public rendering.""" + status: str code: str files: int = 0 @@ -59,6 +70,8 @@ class VerificationCard: unique_bytes: int = 0 def payload(self) -> dict[str, object]: + """Return the stable machine-readable card payload.""" + return { "claim": "integrity-only", "code": self.code, @@ -77,7 +90,9 @@ def payload(self) -> dict[str, object]: @dataclass(frozen=True) -class _FileRecord: +class _FileRecord(object): + """One content digest bound to a stable filesystem identity.""" + relative: str content: bytes size: int @@ -86,10 +101,14 @@ class _FileRecord: def _invalid(code: str) -> NoReturn: + """Raise a redaction-safe bundle integrity failure.""" + raise BundleInvalid(code) def _json_object(pairs: list[tuple[str, object]]) -> dict[str, object]: + """Build a JSON object while rejecting duplicate member names.""" + result: dict[str, object] = {} for key, value in pairs: if key in result: @@ -98,27 +117,87 @@ def _json_object(pairs: list[tuple[str, object]]) -> dict[str, object]: return result -def _relative_path(value: object) -> str: +def _path_text(value: object) -> str: + """Require a non-empty, NUL-free text path.""" + if not isinstance(value, str) or not value or "\x00" in value: - _invalid("bundle.inventory.path-invalid") + _invalid(_CODE_PATH_INVALID) + return value + + +def _check_path_bytes(value: str) -> None: + """Enforce the UTF-8 path byte limit.""" + try: encoded = value.encode("utf-8") except UnicodeEncodeError: - _invalid("bundle.inventory.path-invalid") + _invalid(_CODE_PATH_INVALID) if len(encoded) > MAX_PATH_BYTES: - _invalid("bundle.inventory.path-invalid") - path = PurePosixPath(value) + _invalid(_CODE_PATH_INVALID) + + +def _check_path_shape(value: str, path: PurePosixPath) -> None: + """Reject non-normal, escaping, or over-deep POSIX paths.""" + if path.is_absolute() or ".." in path.parts or "." in path.parts or "\\" in value: - _invalid("bundle.inventory.path-invalid") + _invalid(_CODE_PATH_INVALID) if len(path.parts) > MAX_DEPTH: _invalid("bundle.limit.depth") - normalized = path.as_posix() - if normalized != value: - _invalid("bundle.inventory.path-invalid") - return normalized + if path.as_posix() != value: + _invalid(_CODE_PATH_INVALID) + + +def _relative_path(value: object) -> str: + """Validate and normalize a contained inventory-relative path.""" + + text = _path_text(value) + _check_path_bytes(text) + path = PurePosixPath(text) + _check_path_shape(text, path) + return path.as_posix() + + +def _directory_entries(directory: Path, remaining: int) -> list[os.DirEntry[str]]: + """Read at most the remaining bounded number of directory entries.""" + + collected: list[os.DirEntry[str]] = [] + try: + with os.scandir(directory) as entries: + for entry in entries: + if len(collected) >= remaining: + _invalid("bundle.limit.files") + collected.append(entry) + except BundleInvalid: + raise + except OSError: + _invalid(_CODE_FILESYSTEM_UNREADABLE) + return collected + + +def _entry_mode(entry: os.DirEntry[str]) -> int: + """Read an entry mode without following symbolic links.""" + + try: + return entry.stat(follow_symlinks=False).st_mode + except OSError: + _invalid(_CODE_FILESYSTEM_UNREADABLE) + + +def _entry_kind(mode: int) -> str: + """Classify an admitted regular file or directory.""" + + if stat.S_ISLNK(mode): + _invalid("bundle.filesystem.symlink") + if stat.S_ISDIR(mode): + return "directory" + if not stat.S_ISREG(mode): + _invalid("bundle.filesystem.special-file") + return "file" def _scan(root: Path) -> dict[str, Path]: + """Scan a bounded regular-file tree without following links.""" + files: dict[str, Path] = {} pending: list[tuple[Path, int]] = [(root, 0)] scanned_entries = 0 @@ -126,36 +205,22 @@ def _scan(root: Path) -> dict[str, Path]: directory, depth = pending.pop() if depth > MAX_DEPTH: _invalid("bundle.limit.depth") - try: - entries = os.scandir(directory) - except OSError: - _invalid("bundle.filesystem.unreadable") - try: - with entries: - for entry in entries: - scanned_entries += 1 - if scanned_entries > MAX_FILES: - _invalid("bundle.limit.files") - relative = Path(entry.path).relative_to(root).as_posix() - _relative_path(relative) - try: - mode = entry.stat(follow_symlinks=False).st_mode - except OSError: - _invalid("bundle.filesystem.unreadable") - if stat.S_ISLNK(mode): - _invalid("bundle.filesystem.symlink") - if stat.S_ISDIR(mode): - pending.append((Path(entry.path), depth + 1)) - continue - if not stat.S_ISREG(mode): - _invalid("bundle.filesystem.special-file") - files[relative] = Path(entry.path) - except OSError: - _invalid("bundle.filesystem.unreadable") + entries = _directory_entries(directory, MAX_FILES - scanned_entries) + scanned_entries += len(entries) + for entry in entries: + relative = Path(entry.path).relative_to(root).as_posix() + _relative_path(relative) + kind = _entry_kind(_entry_mode(entry)) + if kind == "directory": + pending.append((Path(entry.path), depth + 1)) + else: + files[relative] = Path(entry.path) return files def _identity(info: os.stat_result) -> tuple[int, int, int, int, int, int]: + """Return the filesystem fields used to detect concurrent mutation.""" + return ( info.st_dev, info.st_ino, @@ -166,7 +231,27 @@ def _identity(info: os.stat_result) -> tuple[int, int, int, int, int, int]: ) +def _read_content(descriptor: int, *, inventory: bool) -> bytes: + """Read one descriptor while enforcing its class-specific byte limit.""" + + limit = MAX_INVENTORY_BYTES if inventory else MAX_ARTIFACT_BYTES + chunks: list[bytes] = [] + total = 0 + while True: + chunk = os.read(descriptor, min(1024 * 1024, limit + 1 - total)) + if not chunk: + break + chunks.append(chunk) + total += len(chunk) + if total > limit: + code = "bundle.limit.inventory-bytes" if inventory else "bundle.limit.artifact-bytes" + _invalid(code) + return b"".join(chunks) + + def _read_stable(root: Path, relative: str, *, inventory: bool) -> _FileRecord: + """Hash a contained regular file and reject identity changes during I/O.""" + candidate = root / relative flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0) try: @@ -175,23 +260,8 @@ def _read_stable(root: Path, relative: str, *, inventory: bool) -> _FileRecord: try: before = os.fstat(descriptor) if not stat.S_ISREG(before.st_mode) or _identity(before_path) != _identity(before): - _invalid("bundle.filesystem.mutated") - limit = MAX_INVENTORY_BYTES if inventory else MAX_ARTIFACT_BYTES - chunks: list[bytes] = [] - total = 0 - while True: - chunk = os.read(descriptor, min(1024 * 1024, limit + 1 - total)) - if not chunk: - break - chunks.append(chunk) - total += len(chunk) - if total > limit: - code = ( - "bundle.limit.inventory-bytes" - if inventory - else "bundle.limit.artifact-bytes" - ) - _invalid(code) + _invalid(_CODE_FILESYSTEM_MUTATED) + content = _read_content(descriptor, inventory=inventory) after = os.fstat(descriptor) finally: os.close(descriptor) @@ -199,10 +269,9 @@ def _read_stable(root: Path, relative: str, *, inventory: bool) -> _FileRecord: except BundleInvalid: raise except OSError: - _invalid("bundle.filesystem.mutated") + _invalid(_CODE_FILESYSTEM_MUTATED) if _identity(before) != _identity(after) or _identity(after) != _identity(after_path): - _invalid("bundle.filesystem.mutated") - content = b"".join(chunks) + _invalid(_CODE_FILESYSTEM_MUTATED) return _FileRecord( relative=relative, content=content, @@ -213,42 +282,61 @@ def _read_stable(root: Path, relative: str, *, inventory: bool) -> _FileRecord: def _inventory_payload(record: _FileRecord) -> list[object]: + """Decode one exact-shape inventory object.""" + try: payload = json.loads(record.content, object_pairs_hook=_json_object) except BundleInvalid: raise except (UnicodeDecodeError, json.JSONDecodeError): - _invalid("bundle.inventory.malformed") + _invalid(_CODE_INVENTORY_MALFORMED) if not isinstance(payload, dict) or set(payload) != {"artifacts"}: - _invalid("bundle.inventory.malformed") + _invalid(_CODE_INVENTORY_MALFORMED) artifacts = payload["artifacts"] if not isinstance(artifacts, list): - _invalid("bundle.inventory.malformed") + _invalid(_CODE_INVENTORY_MALFORMED) return artifacts +def _entry_size(value: object) -> int: + """Validate one non-negative, non-boolean byte count.""" + + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + _invalid(_CODE_ENTRY_MALFORMED) + return value + + +def _entry_digest(value: object) -> str: + """Validate one lower-case SHA-256 hex digest.""" + + if not isinstance(value, str) or len(value) != 64: + _invalid(_CODE_ENTRY_MALFORMED) + if any(character not in "0123456789abcdef" for character in value): + _invalid(_CODE_ENTRY_MALFORMED) + return value + + +def _check_media_type(value: object) -> None: + """Require a non-empty inventory media type label.""" + + if not isinstance(value, str) or not value: + _invalid(_CODE_ENTRY_MALFORMED) + + def _entry(item: object) -> tuple[str, int, str]: + """Validate and project one exact-shape inventory entry.""" + if not isinstance(item, dict) or set(item) != _ENTRY_KEYS: - _invalid("bundle.inventory.entry-malformed") + _invalid(_CODE_ENTRY_MALFORMED) relative = _relative_path(item["path"]) - size = item["size_bytes"] - digest = item["sha256"] - media_type = item["media_type"] - if isinstance(size, bool) or not isinstance(size, int) or size < 0: - _invalid("bundle.inventory.entry-malformed") - if ( - not isinstance(digest, str) - or len(digest) != 64 - or any(character not in "0123456789abcdef" for character in digest) - or not isinstance(media_type, str) - or not media_type - ): - _invalid("bundle.inventory.entry-malformed") + size = _entry_size(item["size_bytes"]) + digest = _entry_digest(item["sha256"]) + _check_media_type(item["media_type"]) return relative, size, digest -def verify_bundle(bundle: Path) -> VerificationCard: - """Verify exact flat or transitive inventory closure without side effects.""" +def _root_details(bundle: Path) -> tuple[Path, tuple[int, int, int, int, int, int]]: + """Admit one real directory root and capture its identity.""" try: root_stat = bundle.lstat() @@ -256,92 +344,147 @@ def verify_bundle(bundle: Path) -> VerificationCard: _invalid("bundle.root.invalid") if stat.S_ISLNK(root_stat.st_mode) or not stat.S_ISDIR(root_stat.st_mode): _invalid("bundle.root.invalid") - root_identity = _identity(root_stat) - root = bundle.resolve() - files = _scan(root) - if _INVENTORY_NAME not in files: - _invalid("bundle.inventory.missing") - - records: dict[str, _FileRecord] = {} - referenced: set[str] = set() - inventories_seen: set[str] = set() - inventory_queue = deque([_INVENTORY_NAME]) - entries_count = 0 - unique_bytes = 0 - - def record(relative: str, *, inventory: bool) -> _FileRecord: - nonlocal unique_bytes - existing = records.get(relative) - if existing is not None: - return existing - if relative not in files: - _invalid("bundle.inventory.member-missing") - loaded = _read_stable(root, relative, inventory=inventory) - records[relative] = loaded - unique_bytes += loaded.size - if unique_bytes > MAX_UNIQUE_BYTES: - _invalid("bundle.limit.unique-bytes") - return loaded - - while inventory_queue: - inventory_path = inventory_queue.popleft() - if inventory_path in inventories_seen: - _invalid("bundle.inventory.duplicate") - inventories_seen.add(inventory_path) - if len(inventories_seen) > MAX_INVENTORIES: - _invalid("bundle.limit.inventories") - inventory_record = record(inventory_path, inventory=True) - base = PurePosixPath(inventory_path).parent - for item in _inventory_payload(inventory_record): - entries_count += 1 - if entries_count > MAX_ENTRIES: - _invalid("bundle.limit.entries") - child, expected_size, expected_digest = _entry(item) - joined = (base / child).as_posix() - joined = _relative_path(joined) - if joined == _INVENTORY_NAME or joined in referenced: - _invalid("bundle.inventory.duplicate-path") - referenced.add(joined) - is_inventory = PurePosixPath(joined).name == _INVENTORY_NAME - actual = record(joined, inventory=is_inventory) - if actual.size != expected_size or actual.sha256 != expected_digest: - _invalid("bundle.inventory.member-mismatch") - if is_inventory: - inventory_queue.append(joined) - - expected = set(files) - {_INVENTORY_NAME} - if referenced != expected: - _invalid("bundle.inventory.membership-mismatch") - final_files = _scan(root) - if set(final_files) != set(files): - _invalid("bundle.filesystem.mutated") + return bundle.resolve(), _identity(root_stat) + + +def _assert_current_identity( + path: Path, + expected: tuple[int, int, int, int, int, int], + *, + regular: bool, +) -> None: + """Require a path to retain its admitted filesystem identity.""" + try: - if _identity(bundle.lstat()) != root_identity: - _invalid("bundle.filesystem.mutated") - for relative, loaded in records.items(): - current = (root / relative).lstat() - if not stat.S_ISREG(current.st_mode) or _identity(current) != loaded.identity: - _invalid("bundle.filesystem.mutated") - except BundleInvalid: - raise + current = path.lstat() except OSError: - _invalid("bundle.filesystem.mutated") - return VerificationCard( - status="verified", - code="bundle.integrity.verified", - files=len(files), - inventories=len(inventories_seen), - entries=entries_count, - unique_bytes=unique_bytes, - ) + _invalid(_CODE_FILESYSTEM_MUTATED) + if regular and not stat.S_ISREG(current.st_mode): + _invalid(_CODE_FILESYSTEM_MUTATED) + if _identity(current) != expected: + _invalid(_CODE_FILESYSTEM_MUTATED) + + +class _BundleVerifier(object): + """Stateful bounded walk over one bundle's transitive inventory closure.""" + + def __init__(self, bundle: Path) -> None: + """Capture the root and initialize bounded verification state.""" + + self.bundle = bundle + self.root, self.root_identity = _root_details(bundle) + self.files = _scan(self.root) + if _INVENTORY_NAME not in self.files: + _invalid("bundle.inventory.missing") + self.records: dict[str, _FileRecord] = {} + self.referenced: set[str] = set() + self.inventories_seen: set[str] = set() + self.inventory_queue = deque([_INVENTORY_NAME]) + self.entries_count = 0 + self.unique_bytes = 0 + + def _record(self, relative: str, *, inventory: bool) -> _FileRecord: + """Read each unique path once and charge it to the byte budget.""" + + existing = self.records.get(relative) + if existing is None: + if relative not in self.files: + _invalid("bundle.inventory.member-missing") + existing = _read_stable(self.root, relative, inventory=inventory) + self.records[relative] = existing + self.unique_bytes += existing.size + if self.unique_bytes > MAX_UNIQUE_BYTES: + _invalid("bundle.limit.unique-bytes") + return existing + + def _start_inventory(self, inventory_path: str) -> tuple[PurePosixPath, list[object]]: + """Admit one not-yet-seen inventory and return its base and entries.""" + + if inventory_path in self.inventories_seen: + _invalid("bundle.inventory.duplicate") + self.inventories_seen.add(inventory_path) + if len(self.inventories_seen) > MAX_INVENTORIES: + _invalid("bundle.limit.inventories") + record = self._record(inventory_path, inventory=True) + return PurePosixPath(inventory_path).parent, _inventory_payload(record) + + def _verify_item(self, base: PurePosixPath, item: object) -> None: + """Verify one inventory member and enqueue nested inventories.""" + + self.entries_count += 1 + if self.entries_count > MAX_ENTRIES: + _invalid("bundle.limit.entries") + child, expected_size, expected_digest = _entry(item) + joined = _relative_path((base / child).as_posix()) + if joined == _INVENTORY_NAME or joined in self.referenced: + _invalid("bundle.inventory.duplicate-path") + self.referenced.add(joined) + is_inventory = PurePosixPath(joined).name == _INVENTORY_NAME + actual = self._record(joined, inventory=is_inventory) + if actual.size != expected_size or actual.sha256 != expected_digest: + _invalid("bundle.inventory.member-mismatch") + if is_inventory: + self.inventory_queue.append(joined) + + def _verify_closure(self) -> None: + """Walk all flat or transitive inventory entries breadth-first.""" + + while self.inventory_queue: + inventory_path = self.inventory_queue.popleft() + base, items = self._start_inventory(inventory_path) + for item in items: + self._verify_item(base, item) + + def _verify_membership(self) -> None: + """Require exact membership beyond the root inventory itself.""" + + expected = set(self.files) - {_INVENTORY_NAME} + if self.referenced != expected: + _invalid("bundle.inventory.membership-mismatch") + + def _verify_unchanged(self) -> None: + """Re-scan and re-stat all read paths to reject concurrent mutation.""" + + final_files = _scan(self.root) + if set(final_files) != set(self.files): + _invalid(_CODE_FILESYSTEM_MUTATED) + _assert_current_identity(self.bundle, self.root_identity, regular=False) + for relative, loaded in self.records.items(): + _assert_current_identity(self.root / relative, loaded.identity, regular=True) + + def verify(self) -> VerificationCard: + """Run closure, membership, and final identity verification.""" + + self._verify_closure() + self._verify_membership() + self._verify_unchanged() + return VerificationCard( + status="verified", + code="bundle.integrity.verified", + files=len(self.files), + inventories=len(self.inventories_seen), + entries=self.entries_count, + unique_bytes=self.unique_bytes, + ) + + +def verify_bundle(bundle: Path) -> VerificationCard: + """Verify exact flat or transitive inventory closure without side effects.""" + + return _BundleVerifier(bundle).verify() def render_card(card: VerificationCard, output_format: str) -> str: - payload = card.payload() + """Render one deterministic JSON, terminal, or Markdown integrity card.""" + if output_format == "json": - return json.dumps(payload, sort_keys=True, separators=(",", ":")) - counts = payload["counts"] - assert isinstance(counts, dict) + return json.dumps(card.payload(), sort_keys=True, separators=(",", ":")) + counts = { + "entries": card.entries, + "files": card.files, + "inventories": card.inventories, + "unique_bytes": card.unique_bytes, + } if output_format == "markdown": return "\n".join( ( @@ -374,6 +517,8 @@ def render_card(card: VerificationCard, output_format: str) -> str: def _parser() -> _Parser: + """Build the isolated verifier command-line parser.""" + parser = _Parser(prog="raes-adapters verify-bundle") parser.add_argument("--bundle", type=Path, required=True) parser.add_argument("--format", choices=("json", "terminal", "markdown"), default="terminal") @@ -381,6 +526,8 @@ def _parser() -> _Parser: def main(argv: list[str] | None = None) -> int: + """Run the verifier with its stable 0, 2, 3, and 70 exit contract.""" + try: args = _parser().parse_args(argv) try: diff --git a/src/raes_adapters/entrypoint.py b/src/raes_adapters/entrypoint.py index ef2a154..093c4ba 100644 --- a/src/raes_adapters/entrypoint.py +++ b/src/raes_adapters/entrypoint.py @@ -7,6 +7,8 @@ def main(argv: Sequence[str] | None = None) -> int: + """Dispatch the offline verifier without importing simulator adapters.""" + arguments = list(sys.argv[1:] if argv is None else argv) if arguments[:1] == ["verify-bundle"]: from raes_adapters.bundle_verifier import main as verify_main diff --git a/tests/test_bundle_verifier.py b/tests/test_bundle_verifier.py index a4091d5..00b095a 100644 --- a/tests/test_bundle_verifier.py +++ b/tests/test_bundle_verifier.py @@ -39,7 +39,9 @@ def test_flat_bundle_and_all_renderers_are_deterministic(tmp_path: Path) -> None card = bundle_verifier.verify_bundle(tmp_path) assert card.status == "verified" - assert bundle_verifier.render_card(card, "json") == bundle_verifier.render_card(card, "json") + first_json = bundle_verifier.render_card(card, "json") + second_json = bundle_verifier.render_card(card, "json") + assert first_json == second_json assert "integrity-only" in bundle_verifier.render_card(card, "terminal") assert "Semantic fidelity: not assessed" in bundle_verifier.render_card(card, "markdown") From 4bea2174c08b4448b5a1596071112a23d6d56f76 Mon Sep 17 00:00:00 2001 From: Brad Edwards Date: Fri, 14 Aug 2026 18:45:17 +0200 Subject: [PATCH 3/4] fix(cli): harden offline bundle verification --- docs/bundle-verifier.md | 92 ++ .../offline-bundle-verifier-guardrails.md | 183 ++++ docs/index.md | 2 + mkdocs.yml | 2 + src/raes_adapters/_bundle_command.py | 105 +++ src/raes_adapters/bundle_verifier.py | 890 +++++++++++++----- src/raes_adapters/entrypoint.py | 2 +- tests/test_bundle_verifier.py | 274 +++++- 8 files changed, 1322 insertions(+), 228 deletions(-) create mode 100644 docs/bundle-verifier.md create mode 100644 docs/decisions/offline-bundle-verifier-guardrails.md create mode 100644 src/raes_adapters/_bundle_command.py diff --git a/docs/bundle-verifier.md b/docs/bundle-verifier.md new file mode 100644 index 0000000..27ab36b --- /dev/null +++ b/docs/bundle-verifier.md @@ -0,0 +1,92 @@ +# Offline bundle verifier + +The base `raes-adapters` installation includes a simulator-independent command +for checking the byte integrity of an evidence bundle: + +```shell +raes-adapters verify-bundle --bundle exported-evidence +``` + +The command is read-only and offline. It does not import a simulator, repair an +inventory, write a report, recompute an aggregate, or contact a network +service. The selected path is necessarily visible in the process argument +list, so do not put secrets in bundle directory names; the command itself never +echoes that path. + +## What is verified + +The selected root must be a real directory containing `inventory.json`. Each +inventory has exactly one `artifacts` array, and each entry has exactly these +fields: + +```json +{ + "media_type": "application/json", + "path": "run.json", + "sha256": "3a6eb0790f39ac87c94f3856b2dd2c5d110e6811602261a9a923d3bb23adc8b7", + "size_bytes": 4 +} +``` + +Paths are normalized relative POSIX paths. A referenced file named +`inventory.json` is parsed as a nested inventory relative to its own directory. +Verification requires exact global closure: every declared file must exist and +match its size and lowercase SHA-256 digest, and every regular file other than +the root inventory must be declared exactly once. + +The verifier rejects links, special files, mount escapes, unreadable members, +extra or missing files, malformed or duplicate JSON fields, traversal paths, +and mutation during enumeration or hashing. Containment is rooted in pinned +filesystem descriptors. On a platform without equivalent handle-relative, +no-follow, directory, and non-blocking primitives, verification fails closed +with `bundle.filesystem.unsupported`. + +## Output formats + +Terminal output is the default. JSON and Markdown are deterministic alternatives: + +```shell +raes-adapters verify-bundle --bundle exported-evidence --format json +raes-adapters verify-bundle --bundle exported-evidence --format markdown +``` + +Every format reports only a stable status/code, bounded counts, the +`integrity-only` claim, and explicit disclaimers that semantic fidelity and +capture completeness were not assessed. Paths, member names, hashes, media +types, rejected values, file bytes, and exception details are never rendered. +The card is command presentation, not a RAES evidence model or a durable +attestation. + +## Fixed limits + +Limits are security constants and cannot be changed by flags or environment: + +| Limit | Value | +| --- | ---: | +| Regular files, including inventories | 100,000 | +| Inventories | 100,000 | +| Inventory entries | 200,000 | +| Bytes hashed across unique paths | 4 GiB | +| Bytes in one non-inventory artifact | 500 KiB | +| Bytes in one inventory | 16 MiB | +| Relative path components | 32 | +| UTF-8 bytes in one relative path | 1,024 | + +Directory traversal has a separate fixed work bound, so empty directories do +not consume the regular-file budget and cannot create unbounded work. Artifact +hashes are streamed in bounded chunks; the 4 GiB I/O ceiling is not a memory +allocation budget. + +## Exit status + +| Exit | Meaning | +| --- | --- | +| `0` | The exact byte-inventory closure was verified. | +| `2` | The command line was invalid. | +| `3` | The bundle was invalid, tampered, unreadable, unsupported, or over a fixed limit. | +| `70` | An unexpected internal defect occurred; no exception detail is exposed. | + +A successful result is a point-in-time byte-integrity judgment. It does not +establish semantic validity, evidence completeness, privacy, conformance, +reproducibility, aggregate agreement, or scientific support. Backend-specific +commands remain responsible for those separately governed checks. diff --git a/docs/decisions/offline-bundle-verifier-guardrails.md b/docs/decisions/offline-bundle-verifier-guardrails.md new file mode 100644 index 0000000..1e943f7 --- /dev/null +++ b/docs/decisions/offline-bundle-verifier-guardrails.md @@ -0,0 +1,183 @@ +# Offline evidence-bundle verifier guardrails + +GitHub issue #92 is the authority for this requirement-free deliverable. This +note fixes the architecture and security boundaries the implementation must +respect. It defines no RAES contract, evidence semantics, or implementation +plan. + +## Keep byte integrity separate from evidence meaning + +`raes-adapters verify-bundle --bundle PATH` is an always-installed, offline, +read-only verifier for the repository's existing `inventory.json` seal shape. +It proves only that one regular-file tree is closed over exact relative paths, +declared sizes, and SHA-256 digests, including nested inventories. It does not +prove that capture was complete, redaction was adequate, a RAES model is valid, +an aggregate is correct, a backend is conformant, or a scientific conclusion +is supported. + +The integrity command must remain independent of simulator extras and backend +modules. In particular, it does not replace +`raes_adapters.cyborg.reproduction.verify_bundle()`: that backend-local +function validates the frozen CAGE-2 protocol, RAES artifact joins, evidence +closure, aggregates, tiers, and report text. Those semantic checks may run only +after the generic byte boundary succeeds and remain separately named and +reported. + +The inventory object is the exact producer shape already emitted by +`raes_adapters.cli._seal_inventory()` and +`raes_adapters.cyborg.reproduction._seal_inventory()`: + +```json +{"artifacts":[{"media_type":"application/json","path":"run.json","sha256":"...","size_bytes":1}]} +``` + +This repository must not promote that local seal into a second RAES evidence +schema, add semantic fields, or place it in `raes_adapters.base`. The verifier +may use private in-process records and a deterministic presentation card, but +neither is a portable evidence DTO, diagnostic schema, or persistence format. + +## Canonical incumbents + +| Concern | Incumbent and required boundary | +| --- | --- | +| Installed command | `[project.scripts]` in `pyproject.toml`, the `raes-adapters` distribution identity, and the existing researcher `cli.main()` exit/error conventions. Use one lazy top-level dispatcher so `verify-bundle` imports neither the researcher shell nor simulator modules. Do not add another console script or command framework. | +| Inventory production | `cli._seal_inventory()`, `cyborg.reproduction._seal_inventory()`, `raes_operations.run_artifacts.atomic_write_json_artifact()`, relative POSIX names, SHA-256, and inventory-last sealing. Verification consumes those final bytes; it does not call a writer or change producer semantics. | +| Strict input handling | The duplicate-key rejection and exact-shape behavior already used by `cli._strict_json()` and `cyborg.reproduction.load_strict_json()`. The verifier needs an isolated bounded byte parser because importing either semantic command path would violate offline isolation; matching tests, not a new schema registry, keep the shapes aligned. | +| Failure hygiene | Stable researcher exit mappings, fixed input-free messages, and the default-deny principles in `base.redaction`. Verifier output contains only allowlisted codes, dispositions, fixed disclaimers, and bounded counts, so it must never render a rejected path, JSON value, exception, argv, environment value, or traceback. | +| Semantic validation | Published RAES models and validators, environment-pack validation, task/run joins, conformance runners, and backend-local reproduction verification remain downstream owners. A digest match never bypasses them. | +| Persistence | None. Artifact producers retain exclusive roots, atomic writes, and inventory-last sealing. The verifier writes no report, cache, temporary extraction, repaired inventory, access log, or lock file; stdout/stderr are its only output surfaces. | +| Packaging and workflow | The single `pyproject.toml`, single `uv.lock`, `_verification_envs()`, existing `tests` and `distributions` sessions, clean-wheel probe, strict docs build, `.github/workflows/ci.yml`, and `PR Gate`. Add no verifier-only workflow or simulator dependency. Native simulator qualification remains the existing Ubuntu-authoritative lane and is not part of integrity verification. | + +## Fixed admission limits + +The limits are issue-owned security constants, not defaults or tuning knobs: + +| Limit | Fixed value | Meaning | +| --- | ---: | --- | +| Regular files | 100,000 | Every discovered regular file, including inventories, is charged once. Directory traversal work must also remain bounded. | +| Inventories | 100,000 | Every distinct `inventory.json` admitted through transitive closure. | +| Inventory entries | 200,000 | Sum of entries across all admitted inventories. | +| Unique bytes | 4 GiB | Sum of bytes hashed per unique admitted path; hard-linked paths do not evade the budget. | +| Artifact bytes | 500 KiB | Maximum for each non-inventory regular file. | +| Inventory bytes | 16 MiB | Maximum raw size of each inventory before parsing. | +| Path depth | 32 | Maximum normalized POSIX path-component count. | +| Path bytes | 1,024 | Maximum UTF-8 byte length of each inventory-relative path. | + +Limits are checked before unbounded allocation or work and failures use the +stable invalid/tampered exit (`3`). The 4 GiB ceiling is an I/O admission +budget, not permission to retain 4 GiB in memory: artifacts are hashed in +bounded chunks, only the current bounded inventory payload is parsed, and +records retain metadata rather than file content. Limit counters have one +definition shared by flat and nested bundles; nested inventories do not reset +budgets. + +## Filesystem and parser security boundary + +The selected root is untrusted and may mutate concurrently. Containment must +therefore be established by filesystem handles, not by pathname checks alone: + +- open and pin the real root directory without following a symlink or reparse + point, then enumerate and open every descendant relative to already admitted + directory handles; +- apply no-follow semantics to every path component, require intermediate + components to remain directories and terminal members to be regular files, + and use non-blocking admission where a raced FIFO/device could otherwise + hang before its type is checked; +- pin root, directory, and file identities around enumeration and hashing, + then recheck membership and identities before reporting success; +- reject links, sockets, devices, FIFOs, mount/reparse escapes, unreadable + members, replacements, additions, removals, and metadata/content mutation; + no byte outside the selected root may be read even transiently; and +- never silently drop no-follow or handle-relative guarantees on a platform + lacking the required primitives. Supply an equivalent safe primitive or fail + closed; portability cannot weaken containment. + +`Path.resolve()`, `rglob()`, `is_file()`, a pre-open `lstat()`, or +`O_NOFOLLOW` on only the final component cannot establish this boundary under +concurrent directory replacement. Inventory paths are logical POSIX paths and +must be opened component-by-component; do not reinterpret a complete inventory +string using host drive, UNC, or separator rules. + +The root inventory is mandatory. Its object has exactly the `artifacts` key; +each entry has exactly `media_type`, `path`, `sha256`, and `size_bytes`. +Parsing rejects invalid UTF-8, duplicate JSON keys, non-object roots, +non-array artifacts, unknown/missing entry fields, booleans as sizes, +negative sizes, non-lowercase/non-64-character SHA-256 text, empty media types, +non-finite constants, excessive parser nesting, and every malformed path. +Paths are non-empty, normalized, relative POSIX text with no NUL, absolute +root, `.`/`..`, repeated separator, backslash, or over-limit encoding. +Parser depth/resource errors caused by input are invalid input (`3`), not an +internal failure (`70`). + +Closure is global and exact. Each declared path appears once, every declared +member exists and matches, every discovered regular file other than the root +inventory is reached by exactly one inventory entry, and every referenced file +named `inventory.json` is parsed transitively. An inventory cannot list itself, +appear twice, reset the base to the bundle root, or leave extra files hidden in +a nested directory. + +## Cross-cutting path + +| Layer the command passes | Required treatment | +| --- | --- | +| Authentication/authorization | None: this is a local process acting with the invoking user's filesystem authority. It adds no HTTP route, daemon, remote identity, participant authority, or policy decision. A future network surface must use RAES strict-default control-plane security rather than exposing this function directly. | +| Secrets and environment bindings | No credential, token, secret resolver, `.env`, ambient configuration, or environment-selected policy is read. `--bundle` is the only path input; users must not encode secrets in path names because argv is OS-visible. The command never echoes that path. | +| CLI/config shape | Closed `argparse` options: `verify-bundle`, required `--bundle`, and fixed `--format` choices. Limits, inventory names, algorithms, parser modules, import paths, network locations, and output destinations are not configurable by flags or environment. | +| Filesystem parser/policy gate | Descriptor-rooted containment, regular-file admission, strict POSIX path validation, exact JSON shape, duplicate rejection, fixed limits, digest/size comparison, global inventory closure, and mutation checks all pass before success is rendered. | +| OS/process exposure | No shell, subprocess, archive extraction, plugin discovery, home-directory scan, temporary output, or runtime download. The bundle path necessarily appears in process argv but never in output, logs, artifacts, or child processes. File descriptors are close-on-exec even though no child is launched. | +| Error envelope | `0` is verified, `2` closed usage failure, `3` invalid/tampered/unreadable/over-limit input, and `70` an unexpected internal defect. Expected hostile input must not reach `70`. JSON, terminal, and Markdown renderers emit deterministic allowlisted content only; internal failures go to stderr without details or traceback. | +| Logging/observability | No library logging. Success may expose fixed claim text and bounded counts; failure exposes only a stable code/status. Paths, member names, digests, media types, rejected values, file bytes, exception details, and timing are not observability fields. | +| Persistence/network/imports | No writes or network access. Lazy dispatch imports only stdlib-backed verifier code for this command and must work from the base clean-installed wheel without simulator extras. | + +## Extension seam + +The backend extension seam is the existing inventory producer shape: a new +adapter that emits the same seal needs no verifier registration or backend-name +branch. The presentation seam is the pure `--format` renderer over one +integrity-only result; adding an explicitly required renderer must not alter +verification or rerun filesystem access. Lazy command dispatch is the import +isolation seam. + +A future inventory version, digest algorithm, archive transport, remote object +store, or semantic claim requires a separately governed contract and threat +model. Do not prebuild a schema registry, algorithm plugin, filesystem service, +or configurable policy in anticipation of it. + +## Gotchas and anti-patterns + +- Do not reuse the CybORG reproduction verifier as the generic command or move + its protocol, aggregate, tier, leakage-scan, or scientific checks into shared + code. +- Do not call producer-side sealers, atomic writers, RAES contract loaders, or + simulator imports while verifying bytes. +- Do not use path resolution plus prefix comparison as the containment proof, + follow an intermediate link after an earlier check, or trust a directory + entry after reopening it by pathname. +- Do not buffer all artifacts or all inventories, enumerate an unbounded tree, + parse before checking byte limits, or let a special-file race block. +- Do not accept permissive JSON, extra fields, duplicate keys/paths, platform- + native separators, digest aliases, algorithm negotiation, or caller-raised + limits. +- Do not expose a `--repair`, `--write-report`, `--force`, `--follow-links`, + `--fetch`, `--schema`, or `--simulator` path. +- Do not render the selected root, member paths, hashes, rejected content, + exception text, traceback, environment, or argv in any format or log. +- Do not describe a green card as semantic validity, completeness, privacy, + conformance, reproducibility, aggregate agreement, or scientific support. + +## Non-goals and implementation boundaries + +- No simulator installation, import, execution, qualification, or native + conformance is performed. +- No RAES/environment-pack model, schema, validator, diagnostic envelope, + exception hierarchy, capability, profile, fixture corpus, evidence type, or + policy gate is added or replaced. +- No semantic artifact parsing, aggregate recomputation, leakage/privacy scan, + source admission, provenance validation, task/run join, or conclusion is + performed. +- No archive extraction, remote URI, network service, authentication system, + database, cache, repaired bundle, persisted report, or background monitor is + introduced. +- Verification is a point-in-time byte-integrity judgment over one selected + regular-file tree; it is not a durable attestation and cannot prevent later + mutation. diff --git a/docs/index.md b/docs/index.md index 2661ba6..f3352eb 100644 --- a/docs/index.md +++ b/docs/index.md @@ -19,6 +19,7 @@ semantic and protocol authority. - [Repository overview](https://github.com/OpenRAE/adapters#readme) - [Installed researcher command](researcher-command.md) +- [Offline bundle verifier](bundle-verifier.md) - [NASim researcher command](nasim-researcher-command.md) - [CyberBattleSim researcher command](cyberbattlesim-researcher-command.md) - [Contribution guide](https://github.com/OpenRAE/adapters/blob/dev/CONTRIBUTING.md) @@ -30,6 +31,7 @@ semantic and protocol authority. - [CybORG conformance-composition guardrails](decisions/cyborg-conformance-guardrails.md) - [CybORG researcher run-and-evidence command guardrails](decisions/cyborg-researcher-command-guardrails.md) - [CybORG/CAGE-2 protocol-reproduction guardrails](decisions/cyborg-cage2-protocol-reproduction-guardrails.md) +- [Offline evidence-bundle verifier guardrails](decisions/offline-bundle-verifier-guardrails.md) - [CybORG/CAGE-2 downstream environment-pack guardrails](decisions/cyborg-cage2-example-pack-guardrails.md) - [NASim researcher run-and-evidence command guardrails](decisions/nasim-researcher-command-guardrails.md) - [CyberBattleSim qualification guardrails](decisions/cyberbattlesim-qualification-guardrails.md) diff --git a/mkdocs.yml b/mkdocs.yml index 055ba2e..f537afe 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -31,6 +31,7 @@ markdown_extensions: nav: - Home: index.md - Researcher command: researcher-command.md + - Offline bundle verifier: bundle-verifier.md - NASim researcher command: nasim-researcher-command.md - CyberBattleSim researcher command: cyberbattlesim-researcher-command.md - Maintainers: @@ -51,6 +52,7 @@ nav: - CybORG conformance-composition guardrails: decisions/cyborg-conformance-guardrails.md - CybORG researcher run-and-evidence command guardrails: decisions/cyborg-researcher-command-guardrails.md - CybORG/CAGE-2 protocol-reproduction guardrails: decisions/cyborg-cage2-protocol-reproduction-guardrails.md + - Offline evidence-bundle verifier guardrails: decisions/offline-bundle-verifier-guardrails.md - CybORG/CAGE-2 downstream environment-pack guardrails: decisions/cyborg-cage2-example-pack-guardrails.md - NASim researcher run-and-evidence command guardrails: decisions/nasim-researcher-command-guardrails.md - CyberBattleSim qualification guardrails: decisions/cyberbattlesim-qualification-guardrails.md diff --git a/src/raes_adapters/_bundle_command.py b/src/raes_adapters/_bundle_command.py new file mode 100644 index 0000000..a363e1e --- /dev/null +++ b/src/raes_adapters/_bundle_command.py @@ -0,0 +1,105 @@ +"""Deterministic presentation shell for offline bundle verification.""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path +from typing import NoReturn + +from raes_adapters import bundle_verifier + +EXIT_VERIFIED = 0 +EXIT_USAGE = 2 +EXIT_INVALID = 3 +EXIT_INTERNAL = 70 + + +class _UsageFailure(Exception): + """An intentionally detail-free command-line usage failure.""" + + +class _Parser(argparse.ArgumentParser): + """Argument parser that maps usage errors to the documented exit code.""" + + def error(self, message: str) -> NoReturn: + """Raise a private usage exception without echoing input paths.""" + + del message + raise _UsageFailure + + +def render_card(card: bundle_verifier.VerificationCard, output_format: str) -> str: + """Render one deterministic JSON, terminal, or Markdown integrity card.""" + + if output_format == "json": + return json.dumps(card.payload(), sort_keys=True, separators=(",", ":")) + counts = { + "entries": card.entries, + "files": card.files, + "inventories": card.inventories, + "unique_bytes": card.unique_bytes, + } + if output_format == "markdown": + return "\n".join( + ( + "# Bundle integrity card", + "", + f"- Status: `{card.status}`", + f"- Code: `{card.code}`", + "- Claim: integrity-only", + f"- Files: {counts['files']}", + f"- Inventories: {counts['inventories']}", + f"- Entries: {counts['entries']}", + f"- Unique bytes: {counts['unique_bytes']}", + "- Semantic fidelity: not assessed", + "- Capture completeness: not assessed", + ) + ) + return "\n".join( + ( + f"status: {card.status}", + f"code: {card.code}", + "claim: integrity-only", + f"files: {counts['files']}", + f"inventories: {counts['inventories']}", + f"entries: {counts['entries']}", + f"unique-bytes: {counts['unique_bytes']}", + "semantic-fidelity: not-assessed", + "capture-completeness: not-assessed", + ) + ) + + +def _parser() -> _Parser: + """Build the isolated verifier command-line parser.""" + + parser = _Parser(prog="raes-adapters verify-bundle") + parser.add_argument("--bundle", type=Path, required=True) + parser.add_argument("--format", choices=("json", "terminal", "markdown"), default="terminal") + return parser + + +def main(argv: list[str] | None = None) -> int: + """Run the verifier with its stable 0, 2, 3, and 70 exit contract.""" + + try: + args = _parser().parse_args(argv) + try: + card = bundle_verifier.verify_bundle(args.bundle) + result = EXIT_VERIFIED + except bundle_verifier.BundleInvalid as error: + card = bundle_verifier.VerificationCard(status="invalid", code=error.code) + result = EXIT_INVALID + print(render_card(card, args.format)) + return result + except _UsageFailure: + print("verify-bundle.usage.invalid: invalid command line", file=sys.stderr) + return EXIT_USAGE + except Exception: + print("verify-bundle.internal.failure: internal verification failure", file=sys.stderr) + return EXIT_INTERNAL + + +__all__ = ["main"] diff --git a/src/raes_adapters/bundle_verifier.py b/src/raes_adapters/bundle_verifier.py index 6f0ea3e..c484d7d 100644 --- a/src/raes_adapters/bundle_verifier.py +++ b/src/raes_adapters/bundle_verifier.py @@ -2,22 +2,15 @@ from __future__ import annotations -import argparse import hashlib import json import os import stat -import sys from collections import deque from dataclasses import dataclass from pathlib import Path, PurePosixPath from typing import NoReturn -EXIT_VERIFIED = 0 -EXIT_USAGE = 2 -EXIT_INVALID = 3 -EXIT_INTERNAL = 70 - MAX_FILES = 100_000 MAX_INVENTORIES = 100_000 MAX_ENTRIES = 200_000 @@ -27,6 +20,11 @@ MAX_DEPTH = 32 MAX_PATH_BYTES = 1_024 +_MAX_DIRECTORIES = 100_000 +_MAX_JSON_DEPTH = 64 +_READ_CHUNK_BYTES = 1024 * 1024 +_FDINFO_MAX_BYTES = 4_096 + _INVENTORY_NAME = "inventory.json" _ENTRY_KEYS = {"media_type", "path", "sha256", "size_bytes"} _CODE_ENTRY_MALFORMED = "bundle.inventory.entry-malformed" @@ -35,6 +33,8 @@ _CODE_INVENTORY_MALFORMED = "bundle.inventory.malformed" _CODE_PATH_INVALID = "bundle.inventory.path-invalid" +_PATH_ADMISSION_ERRORS = (OSError, ValueError, UnicodeError) + class BundleInvalid(ValueError): """A stable integrity failure safe to render to a user.""" @@ -44,20 +44,6 @@ def __init__(self, code: str) -> None: self.code = code -class _UsageFailure(Exception): - """An intentionally detail-free command-line usage failure.""" - - -class _Parser(argparse.ArgumentParser): - """Argument parser that maps usage errors to the documented exit code.""" - - def error(self, message: str) -> NoReturn: - """Raise a private usage exception without echoing input paths.""" - - del message - raise _UsageFailure - - @dataclass(frozen=True) class VerificationCard(object): """Deterministic integrity-only result safe for public rendering.""" @@ -94,10 +80,31 @@ class _FileRecord(object): """One content digest bound to a stable filesystem identity.""" relative: str - content: bytes size: int sha256: str - identity: tuple[int, int, int, int, int, int] + identity: _Identity + content: bytes | None = None + + +_Identity = tuple[int, int, int, int, int, int] + + +@dataclass(frozen=True) +class _TreeSnapshot(object): + """One descriptor-rooted view of admitted files and directories.""" + + files: dict[str, _Identity] + directories: dict[str, _Identity] + + +@dataclass +class _ScanState(object): + """Bounded mutable counters for one descriptor-rooted traversal.""" + + files: dict[str, _Identity] + directories: dict[str, _Identity] + file_count: int = 0 + directory_count: int = 0 def _invalid(code: str) -> NoReturn: @@ -157,32 +164,6 @@ def _relative_path(value: object) -> str: return path.as_posix() -def _directory_entries(directory: Path, remaining: int) -> list[os.DirEntry[str]]: - """Read at most the remaining bounded number of directory entries.""" - - collected: list[os.DirEntry[str]] = [] - try: - with os.scandir(directory) as entries: - for entry in entries: - if len(collected) >= remaining: - _invalid("bundle.limit.files") - collected.append(entry) - except BundleInvalid: - raise - except OSError: - _invalid(_CODE_FILESYSTEM_UNREADABLE) - return collected - - -def _entry_mode(entry: os.DirEntry[str]) -> int: - """Read an entry mode without following symbolic links.""" - - try: - return entry.stat(follow_symlinks=False).st_mode - except OSError: - _invalid(_CODE_FILESYSTEM_UNREADABLE) - - def _entry_kind(mode: int) -> str: """Classify an admitted regular file or directory.""" @@ -195,30 +176,7 @@ def _entry_kind(mode: int) -> str: return "file" -def _scan(root: Path) -> dict[str, Path]: - """Scan a bounded regular-file tree without following links.""" - - files: dict[str, Path] = {} - pending: list[tuple[Path, int]] = [(root, 0)] - scanned_entries = 0 - while pending: - directory, depth = pending.pop() - if depth > MAX_DEPTH: - _invalid("bundle.limit.depth") - entries = _directory_entries(directory, MAX_FILES - scanned_entries) - scanned_entries += len(entries) - for entry in entries: - relative = Path(entry.path).relative_to(root).as_posix() - _relative_path(relative) - kind = _entry_kind(_entry_mode(entry)) - if kind == "directory": - pending.append((Path(entry.path), depth + 1)) - else: - files[relative] = Path(entry.path) - return files - - -def _identity(info: os.stat_result) -> tuple[int, int, int, int, int, int]: +def _identity(info: os.stat_result) -> _Identity: """Return the filesystem fields used to detect concurrent mutation.""" return ( @@ -231,65 +189,481 @@ def _identity(info: os.stat_result) -> tuple[int, int, int, int, int, int]: ) -def _read_content(descriptor: int, *, inventory: bool) -> bytes: - """Read one descriptor while enforcing its class-specific byte limit.""" +def _secure_descriptor_primitives_available() -> bool: + """Return whether this platform can enforce every containment guarantee.""" + + supports_dir_fd: set[object] = getattr(os, "supports_dir_fd", set()) + supports_fd: set[object] = getattr(os, "supports_fd", set()) + return all( + ( + bool(getattr(os, "O_NOFOLLOW", 0)), + bool(getattr(os, "O_DIRECTORY", 0)), + bool(getattr(os, "O_NONBLOCK", 0)), + os.open in supports_dir_fd, + os.scandir in supports_fd, + ) + ) + + +def _directory_flags() -> int: + """Return the fail-closed flags for opening one directory component.""" + + return os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0) + + +def _file_flags() -> int: + """Return non-blocking no-follow flags for a terminal regular file.""" + + return os.O_RDONLY | os.O_NONBLOCK | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0) + + +def _mount_id(descriptor: int) -> int | None: + """Read Linux's mount identity for one open descriptor, when available.""" + + path = f"/proc/self/fdinfo/{descriptor}" + flags = os.O_RDONLY | os.O_NONBLOCK | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0) + try: + fdinfo = os.open(path, flags) + except OSError: + return None + try: + content = os.read(fdinfo, _FDINFO_MAX_BYTES + 1) + except OSError: + return None + finally: + os.close(fdinfo) + if len(content) > _FDINFO_MAX_BYTES: + return None + for line in content.splitlines(): + if line.startswith(b"mnt_id:"): + try: + return int(line.partition(b":")[2].strip()) + except ValueError: + return None + return None + + +def _descriptor_stat(descriptor: int, code: str) -> os.stat_result: + """Read one open descriptor's identity through a bounded failure code.""" + + try: + return os.fstat(descriptor) + except OSError: + _invalid(code) + + +def _open_directory_at( + parent: int, + name: str, + expected: _Identity, + *, + root_device: int, + root_mount_id: int, +) -> int: + """Open and identity-pin one directory relative to an admitted parent.""" + + try: + descriptor = os.open(name, _directory_flags(), dir_fd=parent) + except OSError: + _invalid(_CODE_FILESYSTEM_MUTATED) + try: + current = _descriptor_stat(descriptor, _CODE_FILESYSTEM_MUTATED) + if ( + not stat.S_ISDIR(current.st_mode) + or current.st_dev != root_device + or _mount_id(descriptor) != root_mount_id + or _identity(current) != expected + ): + _invalid(_CODE_FILESYSTEM_MUTATED) + except BaseException: + os.close(descriptor) + raise + return descriptor + + +def _entry_stat(entry: os.DirEntry[str], code: str) -> os.stat_result: + """Read an enumerated entry without following a link.""" + + try: + return entry.stat(follow_symlinks=False) + except OSError: + _invalid(code) + + +def _scan_file(relative: str, info: os.stat_result, state: _ScanState) -> None: + """Charge and retain one regular-file identity.""" + + state.file_count += 1 + if state.file_count > MAX_FILES: + _invalid("bundle.limit.files") + state.files[relative] = _identity(info) + + +def _scan_child_directory( + descriptor: int, + entry: os.DirEntry[str], + relative: str, + info: os.stat_result, + state: _ScanState, + *, + depth: int, + root_device: int, + root_mount_id: int, + error_code: str, +) -> None: + """Open, scan, and close one identity-pinned descendant directory.""" + + state.directory_count += 1 + if state.directory_count > _MAX_DIRECTORIES: + _invalid("bundle.limit.directories") + expected = _identity(info) + state.directories[relative] = expected + child = _open_directory_at( + descriptor, + entry.name, + expected, + root_device=root_device, + root_mount_id=root_mount_id, + ) + try: + _scan_directory( + child, + relative, + state, + depth=depth + 1, + root_device=root_device, + root_mount_id=root_mount_id, + error_code=error_code, + ) + finally: + os.close(child) + + +def _scan_entry( + descriptor: int, + entry: os.DirEntry[str], + base: str, + state: _ScanState, + *, + depth: int, + root_device: int, + root_mount_id: int, + error_code: str, +) -> None: + """Classify and charge one descriptor-relative directory entry.""" + + relative = entry.name if not base else f"{base}/{entry.name}" + _relative_path(relative) + info = _entry_stat(entry, error_code) + kind = _entry_kind(info.st_mode) + if info.st_dev != root_device: + _invalid("bundle.filesystem.mount") + if kind == "directory": + _scan_child_directory( + descriptor, + entry, + relative, + info, + state, + depth=depth, + root_device=root_device, + root_mount_id=root_mount_id, + error_code=error_code, + ) + else: + _scan_file(relative, info, state) + + +def _scan_directory( + descriptor: int, + base: str, + state: _ScanState, + *, + depth: int, + root_device: int, + root_mount_id: int, + error_code: str, +) -> None: + """Enumerate one pinned directory while bounding depth and mutation.""" + + if depth > MAX_DEPTH: + _invalid("bundle.limit.depth") + before = _descriptor_stat(descriptor, error_code) + try: + with os.scandir(descriptor) as entries: + for entry in entries: + _scan_entry( + descriptor, + entry, + base, + state, + depth=depth, + root_device=root_device, + root_mount_id=root_mount_id, + error_code=error_code, + ) + except BundleInvalid: + raise + except OSError: + _invalid(error_code) + after = _descriptor_stat(descriptor, error_code) + if _identity(before) != _identity(after): + _invalid(_CODE_FILESYSTEM_MUTATED) + + +def _scan( + root_descriptor: int, + root_identity: _Identity, + root_mount_id: int, + *, + error_code: str = _CODE_FILESYSTEM_UNREADABLE, +) -> _TreeSnapshot: + """Scan one bounded tree only through descriptor-relative operations.""" + + scan_root = _open_directory_at( + root_descriptor, + ".", + root_identity, + root_device=root_identity[0], + root_mount_id=root_mount_id, + ) + state = _ScanState(files={}, directories={}) + try: + _scan_directory( + scan_root, + "", + state, + depth=0, + root_device=root_identity[0], + root_mount_id=root_mount_id, + error_code=error_code, + ) + finally: + os.close(scan_root) + return _TreeSnapshot(files=state.files, directories=state.directories) + + +def _open_member( + root_descriptor: int, + snapshot: _TreeSnapshot, + relative: str, + root_mount_id: int, +) -> tuple[int, os.stat_result]: + """Open one file through identity-pinned intermediate directories.""" + + current, owned_directory, expected_file = _open_member_parent( + root_descriptor, + snapshot, + relative, + root_mount_id, + ) + try: + return _open_regular_file( + current, PurePosixPath(relative).name, expected_file, root_mount_id + ) + finally: + if owned_directory is not None: + os.close(owned_directory) + + +def _open_member_parent( + root_descriptor: int, + snapshot: _TreeSnapshot, + relative: str, + root_mount_id: int, +) -> tuple[int, int | None, _Identity]: + """Traverse and pin every intermediate component of one member path.""" + + parts = PurePosixPath(relative).parts + expected_file = snapshot.files.get(relative) + if expected_file is None: + _invalid("bundle.inventory.member-missing") + current = root_descriptor + owned_directory: int | None = None + prefix: list[str] = [] + try: + for component in parts[:-1]: + prefix.append(component) + expected_directory = snapshot.directories.get("/".join(prefix)) + if expected_directory is None: + _invalid(_CODE_FILESYSTEM_MUTATED) + next_directory = _open_directory_at( + current, + component, + expected_directory, + root_device=expected_file[0], + root_mount_id=root_mount_id, + ) + if owned_directory is not None: + os.close(owned_directory) + owned_directory = next_directory + current = next_directory + return current, owned_directory, expected_file + except BaseException: + if owned_directory is not None: + os.close(owned_directory) + raise + + +def _open_regular_file( + parent: int, + name: str, + expected: _Identity, + root_mount_id: int, +) -> tuple[int, os.stat_result]: + """Open and identity-pin one non-blocking terminal regular file.""" + + try: + descriptor = os.open(name, _file_flags(), dir_fd=parent) + except OSError: + _invalid(_CODE_FILESYSTEM_MUTATED) + try: + before = _descriptor_stat(descriptor, _CODE_FILESYSTEM_MUTATED) + if ( + not stat.S_ISREG(before.st_mode) + or _mount_id(descriptor) != root_mount_id + or _identity(before) != expected + ): + _invalid(_CODE_FILESYSTEM_MUTATED) + except BaseException: + os.close(descriptor) + raise + return descriptor, before + - limit = MAX_INVENTORY_BYTES if inventory else MAX_ARTIFACT_BYTES - chunks: list[bytes] = [] +def _read_request_size(total: int, limit: int, unique_remaining: int | None) -> int: + """Bound the next read by the per-file and optional aggregate budgets.""" + + request = min(_READ_CHUNK_BYTES, limit + 1 - total) + if unique_remaining is not None: + request = min(request, max(1, unique_remaining + 1 - total)) + return request + + +def _hash_descriptor( + descriptor: int, + *, + inventory: bool, + retain_content: bool, + unique_remaining: int | None, +) -> tuple[int, str, bytes | None]: + """Stream one regular file under per-file and aggregate byte limits.""" + + limit, limit_code = _content_limit(inventory) + digest = hashlib.sha256() + chunks = _content_buffer(retain_content) total = 0 while True: - chunk = os.read(descriptor, min(1024 * 1024, limit + 1 - total)) + chunk = _read_checked_chunk( + descriptor, + total=total, + limit=limit, + limit_code=limit_code, + unique_remaining=unique_remaining, + ) if not chunk: break - chunks.append(chunk) total += len(chunk) - if total > limit: - code = "bundle.limit.inventory-bytes" if inventory else "bundle.limit.artifact-bytes" - _invalid(code) - return b"".join(chunks) + digest.update(chunk) + if chunks is not None: + chunks.append(chunk) + content = b"".join(chunks) if chunks is not None else None + return total, digest.hexdigest(), content + +def _content_limit(inventory: bool) -> tuple[int, str]: + """Return the byte limit and stable code for one file class.""" -def _read_stable(root: Path, relative: str, *, inventory: bool) -> _FileRecord: - """Hash a contained regular file and reject identity changes during I/O.""" + if inventory: + return MAX_INVENTORY_BYTES, "bundle.limit.inventory-bytes" + return MAX_ARTIFACT_BYTES, "bundle.limit.artifact-bytes" + + +def _content_buffer(retain_content: bool) -> list[bytes] | None: + """Allocate a buffer only for the currently parsed inventory.""" + + if retain_content: + return [] + return None + + +def _read_checked_chunk( + descriptor: int, + *, + total: int, + limit: int, + limit_code: str, + unique_remaining: int | None, +) -> bytes: + """Read one bounded chunk and reject limit crossings before retention.""" - candidate = root / relative - flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0) try: - before_path = candidate.lstat() - descriptor = os.open(candidate, flags) - try: - before = os.fstat(descriptor) - if not stat.S_ISREG(before.st_mode) or _identity(before_path) != _identity(before): - _invalid(_CODE_FILESYSTEM_MUTATED) - content = _read_content(descriptor, inventory=inventory) - after = os.fstat(descriptor) - finally: - os.close(descriptor) - after_path = candidate.lstat() - except BundleInvalid: - raise + chunk = os.read(descriptor, _read_request_size(total, limit, unique_remaining)) except OSError: _invalid(_CODE_FILESYSTEM_MUTATED) - if _identity(before) != _identity(after) or _identity(after) != _identity(after_path): + next_total = total + len(chunk) + if next_total > limit: + _invalid(limit_code) + if unique_remaining is not None and next_total > unique_remaining: + _invalid("bundle.limit.unique-bytes") + return chunk + + +def _read_stable( + root_descriptor: int, + snapshot: _TreeSnapshot, + relative: str, + root_mount_id: int, + *, + inventory: bool, + retain_content: bool, + unique_remaining: int | None, +) -> _FileRecord: + """Hash one contained regular file without retaining artifact bytes.""" + + descriptor, before = _open_member( + root_descriptor, + snapshot, + relative, + root_mount_id, + ) + try: + size, digest, content = _hash_descriptor( + descriptor, + inventory=inventory, + retain_content=retain_content, + unique_remaining=unique_remaining, + ) + after = _descriptor_stat(descriptor, _CODE_FILESYSTEM_MUTATED) + finally: + os.close(descriptor) + if _identity(before) != _identity(after): _invalid(_CODE_FILESYSTEM_MUTATED) return _FileRecord( relative=relative, - content=content, - size=len(content), - sha256=hashlib.sha256(content).hexdigest(), + size=size, + sha256=digest, identity=_identity(after), + content=content, ) def _inventory_payload(record: _FileRecord) -> list[object]: """Decode one exact-shape inventory object.""" + if record.content is None: + raise RuntimeError("inventory content was not retained") try: - payload = json.loads(record.content, object_pairs_hook=_json_object) + payload = json.loads( + record.content, + object_pairs_hook=_json_object, + parse_constant=_reject_json_constant, + ) except BundleInvalid: raise - except (UnicodeDecodeError, json.JSONDecodeError): + except (ValueError, RecursionError): _invalid(_CODE_INVENTORY_MALFORMED) + _check_json_depth(payload) if not isinstance(payload, dict) or set(payload) != {"artifacts"}: _invalid(_CODE_INVENTORY_MALFORMED) artifacts = payload["artifacts"] @@ -298,6 +672,27 @@ def _inventory_payload(record: _FileRecord) -> list[object]: return artifacts +def _reject_json_constant(value: str) -> NoReturn: + """Reject JSON's non-standard NaN and infinity spellings.""" + + del value + _invalid(_CODE_INVENTORY_MALFORMED) + + +def _check_json_depth(payload: object) -> None: + """Reject excessive parser nesting without recursive Python traversal.""" + + pending: list[tuple[object, int]] = [(payload, 1)] + while pending: + value, depth = pending.pop() + if depth > _MAX_JSON_DEPTH: + _invalid(_CODE_INVENTORY_MALFORMED) + if isinstance(value, dict): + pending.extend((child, depth + 1) for child in value.values()) + elif isinstance(value, list): + pending.extend((child, depth + 1) for child in value) + + def _entry_size(value: object) -> int: """Validate one non-negative, non-boolean byte count.""" @@ -335,46 +730,96 @@ def _entry(item: object) -> tuple[str, int, str]: return relative, size, digest -def _root_details(bundle: Path) -> tuple[Path, tuple[int, int, int, int, int, int]]: - """Admit one real directory root and capture its identity.""" +def _root_details(bundle: Path) -> tuple[int, _Identity, int]: + """Open and identity-pin one real directory root.""" + + if not _secure_descriptor_primitives_available(): + _invalid("bundle.filesystem.unsupported") + root_stat = _root_path_stat(bundle) + descriptor = _open_root_descriptor(bundle) + try: + opened, root_mount_id = _admit_open_root(descriptor, root_stat) + except BaseException: + os.close(descriptor) + raise + return descriptor, _identity(opened), root_mount_id + + +def _root_path_stat(bundle: Path) -> os.stat_result: + """Admit a lexical root that is a real directory rather than a link.""" try: root_stat = bundle.lstat() - except OSError: + except _PATH_ADMISSION_ERRORS: _invalid("bundle.root.invalid") if stat.S_ISLNK(root_stat.st_mode) or not stat.S_ISDIR(root_stat.st_mode): _invalid("bundle.root.invalid") - return bundle.resolve(), _identity(root_stat) + return root_stat + + +def _open_root_descriptor(bundle: Path) -> int: + """Open the selected root with directory and no-follow guarantees.""" + + try: + return os.open(bundle, _directory_flags()) + except _PATH_ADMISSION_ERRORS: + _invalid("bundle.root.invalid") + + +def _admit_open_root( + descriptor: int, + path_stat: os.stat_result, +) -> tuple[os.stat_result, int]: + """Bind a lexical root identity to its pinned descriptor and mount.""" + + opened = _descriptor_stat(descriptor, _CODE_FILESYSTEM_MUTATED) + if not stat.S_ISDIR(opened.st_mode) or _identity(opened) != _identity(path_stat): + _invalid(_CODE_FILESYSTEM_MUTATED) + root_mount_id = _mount_id(descriptor) + if root_mount_id is None: + _invalid("bundle.filesystem.unsupported") + return opened, root_mount_id def _assert_current_identity( - path: Path, - expected: tuple[int, int, int, int, int, int], - *, - regular: bool, + bundle: Path, + root_descriptor: int, + expected: _Identity, + root_mount_id: int, ) -> None: - """Require a path to retain its admitted filesystem identity.""" + """Require both the lexical root and pinned descriptor to stay identical.""" try: - current = path.lstat() - except OSError: + path_current = bundle.lstat() + except _PATH_ADMISSION_ERRORS: + _invalid(_CODE_FILESYSTEM_MUTATED) + descriptor_current = _descriptor_stat(root_descriptor, _CODE_FILESYSTEM_MUTATED) + if not stat.S_ISDIR(path_current.st_mode) or not stat.S_ISDIR(descriptor_current.st_mode): _invalid(_CODE_FILESYSTEM_MUTATED) - if regular and not stat.S_ISREG(current.st_mode): + if _identity(path_current) != expected or _identity(descriptor_current) != expected: _invalid(_CODE_FILESYSTEM_MUTATED) - if _identity(current) != expected: + if _mount_id(root_descriptor) != root_mount_id: _invalid(_CODE_FILESYSTEM_MUTATED) class _BundleVerifier(object): """Stateful bounded walk over one bundle's transitive inventory closure.""" - def __init__(self, bundle: Path) -> None: + def __init__( + self, + bundle: Path, + root_descriptor: int, + root_identity: _Identity, + root_mount_id: int, + ) -> None: """Capture the root and initialize bounded verification state.""" self.bundle = bundle - self.root, self.root_identity = _root_details(bundle) - self.files = _scan(self.root) - if _INVENTORY_NAME not in self.files: + self.root_descriptor = root_descriptor + self.root_identity = root_identity + self.root_mount_id = root_mount_id + self.snapshot = _scan(root_descriptor, root_identity, root_mount_id) + if _INVENTORY_NAME not in self.snapshot.files: _invalid("bundle.inventory.missing") self.records: dict[str, _FileRecord] = {} self.referenced: set[str] = set() @@ -383,19 +828,87 @@ def __init__(self, bundle: Path) -> None: self.entries_count = 0 self.unique_bytes = 0 - def _record(self, relative: str, *, inventory: bool) -> _FileRecord: + def _read_record( + self, + relative: str, + *, + inventory: bool, + retain_content: bool, + charge: bool, + ) -> _FileRecord: + """Read one record with optional aggregate-byte charging.""" + + remaining = MAX_UNIQUE_BYTES - self.unique_bytes if charge else None + return _read_stable( + self.root_descriptor, + self.snapshot, + relative, + self.root_mount_id, + inventory=inventory, + retain_content=retain_content, + unique_remaining=remaining, + ) + + @staticmethod + def _metadata_only(record: _FileRecord) -> _FileRecord: + """Drop buffered inventory bytes before retaining a record.""" + + return _FileRecord( + relative=record.relative, + size=record.size, + sha256=record.sha256, + identity=record.identity, + ) + + @staticmethod + def _same_record(first: _FileRecord, second: _FileRecord) -> bool: + """Compare stable file metadata without comparing buffered content.""" + + return ( + first.relative, + first.size, + first.sha256, + first.identity, + ) == ( + second.relative, + second.size, + second.sha256, + second.identity, + ) + + def _record( + self, + relative: str, + *, + inventory: bool, + retain_content: bool = False, + ) -> _FileRecord: """Read each unique path once and charge it to the byte budget.""" existing = self.records.get(relative) if existing is None: - if relative not in self.files: + if relative not in self.snapshot.files: _invalid("bundle.inventory.member-missing") - existing = _read_stable(self.root, relative, inventory=inventory) - self.records[relative] = existing - self.unique_bytes += existing.size - if self.unique_bytes > MAX_UNIQUE_BYTES: - _invalid("bundle.limit.unique-bytes") - return existing + loaded = self._read_record( + relative, + inventory=inventory, + retain_content=retain_content, + charge=True, + ) + self.records[relative] = self._metadata_only(loaded) + self.unique_bytes += loaded.size + return loaded + if not retain_content: + return existing + loaded = self._read_record( + relative, + inventory=inventory, + retain_content=True, + charge=False, + ) + if not self._same_record(existing, loaded): + _invalid(_CODE_FILESYSTEM_MUTATED) + return loaded def _start_inventory(self, inventory_path: str) -> tuple[PurePosixPath, list[object]]: """Admit one not-yet-seen inventory and return its base and entries.""" @@ -405,7 +918,7 @@ def _start_inventory(self, inventory_path: str) -> tuple[PurePosixPath, list[obj self.inventories_seen.add(inventory_path) if len(self.inventories_seen) > MAX_INVENTORIES: _invalid("bundle.limit.inventories") - record = self._record(inventory_path, inventory=True) + record = self._record(inventory_path, inventory=True, retain_content=True) return PurePosixPath(inventory_path).parent, _inventory_payload(record) def _verify_item(self, base: PurePosixPath, item: object) -> None: @@ -438,19 +951,27 @@ def _verify_closure(self) -> None: def _verify_membership(self) -> None: """Require exact membership beyond the root inventory itself.""" - expected = set(self.files) - {_INVENTORY_NAME} + expected = set(self.snapshot.files) - {_INVENTORY_NAME} if self.referenced != expected: _invalid("bundle.inventory.membership-mismatch") def _verify_unchanged(self) -> None: """Re-scan and re-stat all read paths to reject concurrent mutation.""" - final_files = _scan(self.root) - if set(final_files) != set(self.files): + final_snapshot = _scan( + self.root_descriptor, + self.root_identity, + self.root_mount_id, + error_code=_CODE_FILESYSTEM_MUTATED, + ) + if final_snapshot != self.snapshot: _invalid(_CODE_FILESYSTEM_MUTATED) - _assert_current_identity(self.bundle, self.root_identity, regular=False) - for relative, loaded in self.records.items(): - _assert_current_identity(self.root / relative, loaded.identity, regular=True) + _assert_current_identity( + self.bundle, + self.root_descriptor, + self.root_identity, + self.root_mount_id, + ) def verify(self) -> VerificationCard: """Run closure, membership, and final identity verification.""" @@ -461,7 +982,7 @@ def verify(self) -> VerificationCard: return VerificationCard( status="verified", code="bundle.integrity.verified", - files=len(self.files), + files=len(self.snapshot.files), inventories=len(self.inventories_seen), entries=self.entries_count, unique_bytes=self.unique_bytes, @@ -471,85 +992,16 @@ def verify(self) -> VerificationCard: def verify_bundle(bundle: Path) -> VerificationCard: """Verify exact flat or transitive inventory closure without side effects.""" - return _BundleVerifier(bundle).verify() - - -def render_card(card: VerificationCard, output_format: str) -> str: - """Render one deterministic JSON, terminal, or Markdown integrity card.""" - - if output_format == "json": - return json.dumps(card.payload(), sort_keys=True, separators=(",", ":")) - counts = { - "entries": card.entries, - "files": card.files, - "inventories": card.inventories, - "unique_bytes": card.unique_bytes, - } - if output_format == "markdown": - return "\n".join( - ( - "# Bundle integrity card", - "", - f"- Status: `{card.status}`", - f"- Code: `{card.code}`", - "- Claim: integrity-only", - f"- Files: {counts['files']}", - f"- Inventories: {counts['inventories']}", - f"- Entries: {counts['entries']}", - f"- Unique bytes: {counts['unique_bytes']}", - "- Semantic fidelity: not assessed", - "- Capture completeness: not assessed", - ) - ) - return "\n".join( - ( - f"status: {card.status}", - f"code: {card.code}", - "claim: integrity-only", - f"files: {counts['files']}", - f"inventories: {counts['inventories']}", - f"entries: {counts['entries']}", - f"unique-bytes: {counts['unique_bytes']}", - "semantic-fidelity: not-assessed", - "capture-completeness: not-assessed", - ) - ) - - -def _parser() -> _Parser: - """Build the isolated verifier command-line parser.""" - - parser = _Parser(prog="raes-adapters verify-bundle") - parser.add_argument("--bundle", type=Path, required=True) - parser.add_argument("--format", choices=("json", "terminal", "markdown"), default="terminal") - return parser - + root_descriptor, root_identity, root_mount_id = _root_details(bundle) + try: + return _BundleVerifier( + bundle, + root_descriptor, + root_identity, + root_mount_id, + ).verify() + finally: + os.close(root_descriptor) -def main(argv: list[str] | None = None) -> int: - """Run the verifier with its stable 0, 2, 3, and 70 exit contract.""" - try: - args = _parser().parse_args(argv) - try: - card = verify_bundle(args.bundle) - result = EXIT_VERIFIED - except BundleInvalid as error: - card = VerificationCard(status="invalid", code=error.code) - result = EXIT_INVALID - print(render_card(card, args.format)) - return result - except _UsageFailure: - print("verify-bundle.usage.invalid: invalid command line", file=sys.stderr) - return EXIT_USAGE - except Exception: - print("verify-bundle.internal.failure: internal verification failure", file=sys.stderr) - return EXIT_INTERNAL - - -__all__ = [ - "BundleInvalid", - "VerificationCard", - "main", - "render_card", - "verify_bundle", -] +__all__: list[str] = [] diff --git a/src/raes_adapters/entrypoint.py b/src/raes_adapters/entrypoint.py index 093c4ba..b70094b 100644 --- a/src/raes_adapters/entrypoint.py +++ b/src/raes_adapters/entrypoint.py @@ -11,7 +11,7 @@ def main(argv: Sequence[str] | None = None) -> int: arguments = list(sys.argv[1:] if argv is None else argv) if arguments[:1] == ["verify-bundle"]: - from raes_adapters.bundle_verifier import main as verify_main + from raes_adapters._bundle_command import main as verify_main return verify_main(arguments[1:]) from raes_adapters.cli import main as researcher_main diff --git a/tests/test_bundle_verifier.py b/tests/test_bundle_verifier.py index 00b095a..1ff6d8f 100644 --- a/tests/test_bundle_verifier.py +++ b/tests/test_bundle_verifier.py @@ -5,10 +5,13 @@ import hashlib import json import os +import sys +import tracemalloc from pathlib import Path import pytest +from raes_adapters import _bundle_command as bundle_command from raes_adapters import bundle_verifier from raes_adapters.entrypoint import main @@ -39,11 +42,11 @@ def test_flat_bundle_and_all_renderers_are_deterministic(tmp_path: Path) -> None card = bundle_verifier.verify_bundle(tmp_path) assert card.status == "verified" - first_json = bundle_verifier.render_card(card, "json") - second_json = bundle_verifier.render_card(card, "json") + first_json = bundle_command.render_card(card, "json") + second_json = bundle_command.render_card(card, "json") assert first_json == second_json - assert "integrity-only" in bundle_verifier.render_card(card, "terminal") - assert "Semantic fidelity: not assessed" in bundle_verifier.render_card(card, "markdown") + assert "integrity-only" in bundle_command.render_card(card, "terminal") + assert "Semantic fidelity: not assessed" in bundle_command.render_card(card, "markdown") def test_transitive_inventory_closure_is_verified(tmp_path: Path) -> None: @@ -97,7 +100,7 @@ def test_invalid_membership_and_tampering_are_rejected( def test_symlink_is_rejected_even_when_not_in_inventory(tmp_path: Path) -> None: - outside = tmp_path.parent / "outside" + outside = tmp_path.parent / f"{tmp_path.name}-outside" outside.write_bytes(b"outside") (tmp_path / "link").symlink_to(outside) _write_inventory(tmp_path, []) @@ -106,6 +109,95 @@ def test_symlink_is_rejected_even_when_not_in_inventory(tmp_path: Path) -> None: bundle_verifier.verify_bundle(tmp_path) +def test_intermediate_directory_swap_cannot_escape_bundle( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + container = tmp_path / "container" + container.mkdir() + local = container / "sub" + local.mkdir() + (local / "item.bin").write_bytes(b"local bytes") + outside = tmp_path.parent / f"{tmp_path.name}-outside" + outside.mkdir() + outside_payload = b"outside bytes" + (outside / "item.bin").write_bytes(outside_payload) + _write_inventory(tmp_path, [_entry("container/sub/item.bin", outside_payload)]) + parked = container / "sub.parked" + + def point_outside() -> None: + local.rename(parked) + local.symlink_to(outside, target_is_directory=True) + + def point_local() -> None: + local.unlink() + parked.rename(local) + + original_scan = bundle_verifier._scan + calls = 0 + + def raced_scan(*args: object, **kwargs: object) -> object: + nonlocal calls + calls += 1 + if calls == 2: + point_local() + result = original_scan(*args, **kwargs) + point_outside() + return result + + monkeypatch.setattr(bundle_verifier, "_scan", raced_scan) + try: + with pytest.raises(bundle_verifier.BundleInvalid, match="bundle.filesystem.mutated"): + bundle_verifier.verify_bundle(tmp_path) + finally: + if local.is_symlink(): + point_local() + + +def test_secure_descriptor_primitives_are_required( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _write_inventory(tmp_path, []) + monkeypatch.setattr(bundle_verifier.os, "O_NOFOLLOW", 0, raising=False) + + with pytest.raises(bundle_verifier.BundleInvalid, match="bundle.filesystem.unsupported"): + bundle_verifier.verify_bundle(tmp_path) + + +def test_mount_identity_primitive_is_required( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _write_inventory(tmp_path, []) + monkeypatch.setattr(bundle_verifier, "_mount_id", lambda _descriptor: None, raising=False) + + with pytest.raises(bundle_verifier.BundleInvalid, match="bundle.filesystem.unsupported"): + bundle_verifier.verify_bundle(tmp_path) + + +def test_raced_fifo_is_opened_nonblocking(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + if not hasattr(os, "mkfifo") or not hasattr(os, "O_NONBLOCK"): + pytest.skip("nonblocking FIFO admission is unavailable") + payload = b"regular" + item = tmp_path / "item" + item.write_bytes(payload) + _write_inventory(tmp_path, [_entry("item", payload)]) + original_flags = bundle_verifier._file_flags + raced = False + + def raced_flags() -> int: + nonlocal raced + flags = original_flags() + if not raced: + raced = True + assert flags & os.O_NONBLOCK + item.unlink() + os.mkfifo(item) + return flags + + monkeypatch.setattr(bundle_verifier, "_file_flags", raced_flags) + with pytest.raises(bundle_verifier.BundleInvalid, match="bundle.filesystem.mutated"): + bundle_verifier.verify_bundle(tmp_path) + + def test_entrypoint_uses_documented_exit_codes( tmp_path: Path, capsys: pytest.CaptureFixture[str] ) -> None: @@ -177,6 +269,51 @@ def test_every_admission_limit_fails_closed( bundle_verifier.verify_bundle(tmp_path) +def test_regular_file_limit_does_not_count_directories( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + (tmp_path / "one" / "two").mkdir(parents=True) + _write_inventory(tmp_path, []) + monkeypatch.setattr(bundle_verifier, "MAX_FILES", 1) + + card = bundle_verifier.verify_bundle(tmp_path) + + assert card.files == 1 + + +def test_directory_traversal_has_an_independent_limit( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + (tmp_path / "directory").mkdir() + _write_inventory(tmp_path, []) + monkeypatch.setattr(bundle_verifier, "_MAX_DIRECTORIES", 0, raising=False) + + with pytest.raises(bundle_verifier.BundleInvalid, match="bundle.limit.directories"): + bundle_verifier.verify_bundle(tmp_path) + + +def test_fixed_limits_accept_the_exact_boundary( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + payload = b"x" + (tmp_path / "item").write_bytes(payload) + _write_inventory(tmp_path, [_entry("item", payload)]) + inventory_size = (tmp_path / "inventory.json").stat().st_size + monkeypatch.setattr(bundle_verifier, "MAX_FILES", 2) + monkeypatch.setattr(bundle_verifier, "MAX_INVENTORIES", 1) + monkeypatch.setattr(bundle_verifier, "MAX_ENTRIES", 1) + monkeypatch.setattr(bundle_verifier, "MAX_UNIQUE_BYTES", inventory_size + len(payload)) + monkeypatch.setattr(bundle_verifier, "MAX_ARTIFACT_BYTES", len(payload)) + monkeypatch.setattr(bundle_verifier, "MAX_INVENTORY_BYTES", inventory_size) + + card = bundle_verifier.verify_bundle(tmp_path) + + assert card.files == 2 + assert card.inventories == 1 + assert card.entries == 1 + assert card.unique_bytes == inventory_size + len(payload) + + @pytest.mark.parametrize( "payload", [ @@ -194,6 +331,48 @@ def test_malformed_inventories_are_rejected(tmp_path: Path, payload: bytes) -> N bundle_verifier.verify_bundle(tmp_path) +@pytest.mark.parametrize("constant", ["NaN", "Infinity", "-Infinity"]) +def test_non_finite_json_constants_are_rejected(tmp_path: Path, constant: str) -> None: + (tmp_path / "inventory.json").write_text(f'{{"artifacts":{constant}}}', encoding="utf-8") + + with pytest.raises(bundle_verifier.BundleInvalid, match="bundle.inventory.malformed"): + bundle_verifier.verify_bundle(tmp_path) + + +def test_parser_recursion_is_invalid_input( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + nesting = 1_100 + payload = '{"artifacts":' + "[" * nesting + "[]" + "]" * nesting + "}" + (tmp_path / "inventory.json").write_text(payload, encoding="utf-8") + + assert main(["verify-bundle", "--bundle", str(tmp_path)]) == 3 + captured = capsys.readouterr() + assert "bundle.inventory.malformed" in captured.out + assert captured.err == "" + + +def test_oversized_json_integer_is_invalid_input( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + maximum_digits = sys.get_int_max_str_digits() + if maximum_digits == 0: + pytest.skip("Python integer conversion limit is disabled") + payload = ( + '{"artifacts":[{"media_type":"x","path":"x","sha256":"' + + "0" * 64 + + '","size_bytes":' + + "9" * (maximum_digits + 1) + + "}]}" + ) + (tmp_path / "inventory.json").write_text(payload, encoding="utf-8") + + assert main(["verify-bundle", "--bundle", str(tmp_path)]) == 3 + captured = capsys.readouterr() + assert "bundle.inventory.malformed" in captured.out + assert captured.err == "" + + @pytest.mark.parametrize( ("field", "value"), [ @@ -268,6 +447,35 @@ def test_root_must_be_a_real_directory_with_an_inventory(tmp_path: Path) -> None bundle_verifier.verify_bundle(root_link) +@pytest.mark.parametrize("suffix", ["\x00suffix", "\ud800"]) +def test_malformed_root_path_text_is_invalid_input( + tmp_path: Path, suffix: str, capsys: pytest.CaptureFixture[str] +) -> None: + root = f"{tmp_path}{suffix}" + + assert main(["verify-bundle", "--bundle", root]) == 3 + captured = capsys.readouterr() + assert "bundle.root.invalid" in captured.out + assert root not in captured.out + assert captured.err == "" + + +def test_undecodable_filesystem_name_is_invalid_input( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + if os.name != "posix": + pytest.skip("byte-oriented filesystem names are POSIX-specific") + name = os.fsencode(tmp_path) + b"/undecodable-\xff" + descriptor = os.open(name, os.O_WRONLY | os.O_CREAT, 0o600) + os.close(descriptor) + _write_inventory(tmp_path, []) + + assert main(["verify-bundle", "--bundle", str(tmp_path)]) == 3 + captured = capsys.readouterr() + assert "bundle.inventory.path-invalid" in captured.out + assert captured.err == "" + + def test_mutation_during_hashing_is_rejected( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: @@ -304,6 +512,31 @@ def mutate(record: bundle_verifier._FileRecord) -> list[object]: bundle_verifier.verify_bundle(tmp_path) +def test_root_replacement_while_hashing_is_rejected( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _write_inventory(tmp_path, []) + parked = tmp_path.parent / f"{tmp_path.name}-parked" + original = bundle_verifier._inventory_payload + + def replace_root(record: bundle_verifier._FileRecord) -> list[object]: + result = original(record) + tmp_path.rename(parked) + tmp_path.mkdir() + _write_inventory(tmp_path, []) + return result + + monkeypatch.setattr(bundle_verifier, "_inventory_payload", replace_root) + try: + with pytest.raises(bundle_verifier.BundleInvalid, match="bundle.filesystem.mutated"): + bundle_verifier.verify_bundle(tmp_path) + finally: + if parked.exists(): + (tmp_path / "inventory.json").unlink() + tmp_path.rmdir() + parked.rename(tmp_path) + + def test_verification_is_read_only(tmp_path: Path) -> None: payload = b"evidence" (tmp_path / "item").write_bytes(payload) @@ -318,15 +551,40 @@ def test_verification_is_read_only(tmp_path: Path) -> None: assert after == before +def test_artifact_hashing_does_not_retain_payloads(tmp_path: Path) -> None: + payload = b"x" * (128 * 1024) + source = tmp_path / "item-00" + source.write_bytes(payload) + entries = [_entry(source.name, payload)] + try: + for index in range(1, 64): + path = tmp_path / f"item-{index:02d}" + os.link(source, path) + entries.append(_entry(path.name, payload)) + except OSError: + pytest.skip("hard links are unavailable") + _write_inventory(tmp_path, entries) + + tracemalloc.start() + try: + card = bundle_verifier.verify_bundle(tmp_path) + _current, peak = tracemalloc.get_traced_memory() + finally: + tracemalloc.stop() + + assert card.files == 65 + assert peak < 4 * 1024**2 + + def test_cards_are_stable_and_disclose_only_integrity(tmp_path: Path) -> None: _write_inventory(tmp_path, []) card = bundle_verifier.verify_bundle(tmp_path) counts = card.payload()["counts"] assert isinstance(counts, dict) - assert bundle_verifier.render_card(card, "json") == json.dumps( + assert bundle_command.render_card(card, "json") == json.dumps( card.payload(), sort_keys=True, separators=(",", ":") ) - assert bundle_verifier.render_card(card, "terminal") == "\n".join( + assert bundle_command.render_card(card, "terminal") == "\n".join( ( "status: verified", "code: bundle.integrity.verified", @@ -339,7 +597,7 @@ def test_cards_are_stable_and_disclose_only_integrity(tmp_path: Path) -> None: "capture-completeness: not-assessed", ) ) - markdown = bundle_verifier.render_card(card, "markdown") + markdown = bundle_command.render_card(card, "markdown") assert markdown.startswith("# Bundle integrity card\n") assert "Semantic fidelity: not assessed" in markdown assert "Capture completeness: not assessed" in markdown From cb82f01d07e0bfcd76248c10e0f2cd038fa264be Mon Sep 17 00:00:00 2001 From: Brad Edwards Date: Fri, 14 Aug 2026 20:20:23 +0200 Subject: [PATCH 4/4] fix(cli): simplify bundle verifier internals --- src/raes_adapters/bundle_verifier.py | 105 +++++++++++++++------------ 1 file changed, 59 insertions(+), 46 deletions(-) diff --git a/src/raes_adapters/bundle_verifier.py b/src/raes_adapters/bundle_verifier.py index c484d7d..92b6606 100644 --- a/src/raes_adapters/bundle_verifier.py +++ b/src/raes_adapters/bundle_verifier.py @@ -7,6 +7,7 @@ import os import stat from collections import deque +from contextlib import suppress from dataclasses import dataclass from pathlib import Path, PurePosixPath from typing import NoReturn @@ -32,6 +33,7 @@ _CODE_FILESYSTEM_UNREADABLE = "bundle.filesystem.unreadable" _CODE_INVENTORY_MALFORMED = "bundle.inventory.malformed" _CODE_PATH_INVALID = "bundle.inventory.path-invalid" +_CODE_ROOT_INVALID = "bundle.root.invalid" _PATH_ADMISSION_ERRORS = (OSError, ValueError, UnicodeError) @@ -107,6 +109,15 @@ class _ScanState(object): directory_count: int = 0 +@dataclass(frozen=True) +class _ScanContext(object): + """Immutable filesystem boundary shared through one tree traversal.""" + + root_device: int + root_mount_id: int + error_code: str + + def _invalid(code: str) -> NoReturn: """Raise a redaction-safe bundle integrity failure.""" @@ -222,25 +233,33 @@ def _mount_id(descriptor: int) -> int | None: path = f"/proc/self/fdinfo/{descriptor}" flags = os.O_RDONLY | os.O_NONBLOCK | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0) + mount_id: int | None = None try: fdinfo = os.open(path, flags) except OSError: - return None - try: - content = os.read(fdinfo, _FDINFO_MAX_BYTES + 1) - except OSError: - return None - finally: - os.close(fdinfo) - if len(content) > _FDINFO_MAX_BYTES: - return None + fdinfo = None + if fdinfo is not None: + try: + content = os.read(fdinfo, _FDINFO_MAX_BYTES + 1) + except OSError: + content = None + finally: + os.close(fdinfo) + if content is not None and len(content) <= _FDINFO_MAX_BYTES: + mount_id = _parse_mount_id(content) + return mount_id + + +def _parse_mount_id(content: bytes) -> int | None: + """Parse one bounded fdinfo payload without exposing parser failures.""" + + mount_id: int | None = None for line in content.splitlines(): if line.startswith(b"mnt_id:"): - try: - return int(line.partition(b":")[2].strip()) - except ValueError: - return None - return None + with suppress(ValueError): + mount_id = int(line.partition(b":")[2].strip()) + break + return mount_id def _descriptor_stat(descriptor: int, code: str) -> os.stat_result: @@ -307,9 +326,7 @@ def _scan_child_directory( state: _ScanState, *, depth: int, - root_device: int, - root_mount_id: int, - error_code: str, + context: _ScanContext, ) -> None: """Open, scan, and close one identity-pinned descendant directory.""" @@ -322,8 +339,8 @@ def _scan_child_directory( descriptor, entry.name, expected, - root_device=root_device, - root_mount_id=root_mount_id, + root_device=context.root_device, + root_mount_id=context.root_mount_id, ) try: _scan_directory( @@ -331,9 +348,7 @@ def _scan_child_directory( relative, state, depth=depth + 1, - root_device=root_device, - root_mount_id=root_mount_id, - error_code=error_code, + context=context, ) finally: os.close(child) @@ -346,17 +361,15 @@ def _scan_entry( state: _ScanState, *, depth: int, - root_device: int, - root_mount_id: int, - error_code: str, + context: _ScanContext, ) -> None: """Classify and charge one descriptor-relative directory entry.""" relative = entry.name if not base else f"{base}/{entry.name}" _relative_path(relative) - info = _entry_stat(entry, error_code) + info = _entry_stat(entry, context.error_code) kind = _entry_kind(info.st_mode) - if info.st_dev != root_device: + if info.st_dev != context.root_device: _invalid("bundle.filesystem.mount") if kind == "directory": _scan_child_directory( @@ -366,9 +379,7 @@ def _scan_entry( info, state, depth=depth, - root_device=root_device, - root_mount_id=root_mount_id, - error_code=error_code, + context=context, ) else: _scan_file(relative, info, state) @@ -380,15 +391,13 @@ def _scan_directory( state: _ScanState, *, depth: int, - root_device: int, - root_mount_id: int, - error_code: str, + context: _ScanContext, ) -> None: """Enumerate one pinned directory while bounding depth and mutation.""" if depth > MAX_DEPTH: _invalid("bundle.limit.depth") - before = _descriptor_stat(descriptor, error_code) + before = _descriptor_stat(descriptor, context.error_code) try: with os.scandir(descriptor) as entries: for entry in entries: @@ -398,15 +407,13 @@ def _scan_directory( base, state, depth=depth, - root_device=root_device, - root_mount_id=root_mount_id, - error_code=error_code, + context=context, ) except BundleInvalid: raise except OSError: - _invalid(error_code) - after = _descriptor_stat(descriptor, error_code) + _invalid(context.error_code) + after = _descriptor_stat(descriptor, context.error_code) if _identity(before) != _identity(after): _invalid(_CODE_FILESYSTEM_MUTATED) @@ -428,15 +435,18 @@ def _scan( root_mount_id=root_mount_id, ) state = _ScanState(files={}, directories={}) + context = _ScanContext( + root_device=root_identity[0], + root_mount_id=root_mount_id, + error_code=error_code, + ) try: _scan_directory( scan_root, "", state, depth=0, - root_device=root_identity[0], - root_mount_id=root_mount_id, - error_code=error_code, + context=context, ) finally: os.close(scan_root) @@ -751,9 +761,9 @@ def _root_path_stat(bundle: Path) -> os.stat_result: try: root_stat = bundle.lstat() except _PATH_ADMISSION_ERRORS: - _invalid("bundle.root.invalid") + _invalid(_CODE_ROOT_INVALID) if stat.S_ISLNK(root_stat.st_mode) or not stat.S_ISDIR(root_stat.st_mode): - _invalid("bundle.root.invalid") + _invalid(_CODE_ROOT_INVALID) return root_stat @@ -761,9 +771,12 @@ def _open_root_descriptor(bundle: Path) -> int: """Open the selected root with directory and no-follow guarantees.""" try: - return os.open(bundle, _directory_flags()) + # The explicit local CLI path selects the trust root; descriptor-relative + # admission enforces containment beneath it. There is no ambient sandbox + # root against which this user-selected directory could be constrained. + return os.open(bundle, _directory_flags()) # NOSONAR except _PATH_ADMISSION_ERRORS: - _invalid("bundle.root.invalid") + _invalid(_CODE_ROOT_INVALID) def _admit_open_root(