From 4cae021359f056ff70e9b0c390f83fabc4349351 Mon Sep 17 00:00:00 2001 From: Yernat Yestekov Date: Thu, 13 Aug 2026 11:22:03 -0700 Subject: [PATCH 1/3] 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/3] 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 9fa14d105465e67af9704063a3539234f202c815 Mon Sep 17 00:00:00 2001 From: Yernat Yestekov Date: Thu, 13 Aug 2026 13:49:23 -0700 Subject: [PATCH 3/3] refactor(evidence): share inventory producer primitives --- src/raes_adapters/_inventory.py | 61 ++++++++++++++++++ src/raes_adapters/cli.py | 21 ++----- src/raes_adapters/cyborg/reproduction.py | 21 ++++--- tests/test_cyborg_reproduction.py | 22 ++++++- tests/test_inventory_primitives.py | 79 ++++++++++++++++++++++++ 5 files changed, 178 insertions(+), 26 deletions(-) create mode 100644 src/raes_adapters/_inventory.py create mode 100644 tests/test_inventory_primitives.py diff --git a/src/raes_adapters/_inventory.py b/src/raes_adapters/_inventory.py new file mode 100644 index 0000000..26a7f20 --- /dev/null +++ b/src/raes_adapters/_inventory.py @@ -0,0 +1,61 @@ +"""Private byte-level inventory primitives shared by adapter producers. + +This module describes files, not experiment meaning. It deliberately imports +neither simulators nor RAES contracts and defines no portable schema. +""" + +from __future__ import annotations + +import hashlib +from collections.abc import Callable, Iterable +from pathlib import Path + + +def media_type(path: Path) -> str: + """Return the established portable media type for a produced artifact.""" + + return "application/json" if path.suffix == ".json" else "application/octet-stream" + + +def inventory_entry( + root: Path, + path: Path, + *, + type_name: str | None = None, +) -> dict[str, object]: + """Describe one regular contained file with the established entry shape.""" + + resolved_root = root.resolve() + if path.is_symlink() or not path.is_file(): + raise ValueError("inventory member is not a regular file") + resolved_path = path.resolve() + if not resolved_path.is_relative_to(resolved_root): + raise ValueError("inventory member escapes its root") + try: + relative = path.relative_to(root).as_posix() + except ValueError as error: + raise ValueError("inventory member escapes its root") from error + content = path.read_bytes() + return { + "media_type": type_name if type_name is not None else media_type(path), + "path": relative, + "sha256": hashlib.sha256(content).hexdigest(), + "size_bytes": len(content), + } + + +def inventory_document( + root: Path, + members: Iterable[Path], + *, + type_for: Callable[[Path], str] = media_type, +) -> dict[str, object]: + """Build a deterministic inventory document without writing it.""" + + ordered = sorted(members, key=lambda item: item.relative_to(root).as_posix()) + return { + "artifacts": [inventory_entry(root, path, type_name=type_for(path)) for path in ordered] + } + + +__all__ = ["inventory_document", "inventory_entry", "media_type"] diff --git a/src/raes_adapters/cli.py b/src/raes_adapters/cli.py index cc55bda..a804fde 100644 --- a/src/raes_adapters/cli.py +++ b/src/raes_adapters/cli.py @@ -60,6 +60,7 @@ atomic_write_json_artifact, ) +from raes_adapters._inventory import inventory_document, media_type from raes_adapters.cyberbattlesim import ( load_qualification as load_cyberbattlesim_qualification, ) @@ -1315,26 +1316,16 @@ def _reserve_output(requested: Path) -> Path: def _media_type(path: Path) -> str: """Return the portable media type used in the artifact inventory.""" - return "application/json" if path.suffix == ".json" else "application/octet-stream" + return media_type(path) def _seal_inventory(output: Path) -> dict[str, object]: """Hash every portable artifact and atomically seal the inventory.""" - artifacts: list[dict[str, object]] = [] - for path in sorted(item for item in output.rglob("*") if item.is_file()): - if path.name == _INVENTORY_NAME: - continue - payload = path.read_bytes() - artifacts.append( - { - "media_type": _media_type(path), - "path": path.relative_to(output).as_posix(), - "sha256": hashlib.sha256(payload).hexdigest(), - "size_bytes": len(payload), - } - ) - inventory: dict[str, object] = {"artifacts": artifacts} + members = [ + path for path in output.rglob("*") if path.is_file() and path.name != _INVENTORY_NAME + ] + inventory = inventory_document(output, members, type_for=_media_type) atomic_write_json_artifact(output / _INVENTORY_NAME, inventory) return inventory diff --git a/src/raes_adapters/cyborg/reproduction.py b/src/raes_adapters/cyborg/reproduction.py index 773ad8c..bbbd412 100644 --- a/src/raes_adapters/cyborg/reproduction.py +++ b/src/raes_adapters/cyborg/reproduction.py @@ -42,6 +42,9 @@ atomic_write_json_artifact, ) +from raes_adapters._inventory import inventory_entry +from raes_adapters.bundle_verifier import verify_bundle as verify_integrity_bundle + from .driver import CyborgDriver, SourceInstalledCyborgDriver from .researcher import ( EpisodeEvidence, @@ -1393,17 +1396,14 @@ def _media_type(path: Path) -> str: def _inventory_entry(root: Path, path: Path) -> dict[str, object]: """Build one bounded relative inventory entry.""" - if path.is_symlink() or not path.is_file() or not path.resolve().is_relative_to(root.resolve()): - raise ValueError("inventory member is invalid") - content = path.read_bytes() - if len(content) >= _MAX_PUBLIC_FILE_BYTES: + try: + entry = inventory_entry(root, path, type_name=_media_type(path)) + except ValueError as error: + raise ValueError("inventory member is invalid") from error + size = entry["size_bytes"] + if not isinstance(size, int) or size >= _MAX_PUBLIC_FILE_BYTES: raise ValueError("public artifact exceeds the file-size limit") - return { - "path": path.relative_to(root).as_posix(), - "media_type": _media_type(path), - "sha256": hashlib.sha256(content).hexdigest(), - "size_bytes": len(content), - } + return entry def _seal_inventory(root: Path, members: Sequence[Path]) -> dict[str, object]: @@ -2836,6 +2836,7 @@ def verify_bundle( """Offline-verify every transitive inventory and recompute the frozen result.""" bundle = root.resolve() + verify_integrity_bundle(bundle) expected_root_directories = {"plans", "runs"} if ( bundle.is_symlink() diff --git a/tests/test_cyborg_reproduction.py b/tests/test_cyborg_reproduction.py index 5bb7029..52b5498 100644 --- a/tests/test_cyborg_reproduction.py +++ b/tests/test_cyborg_reproduction.py @@ -8,6 +8,7 @@ import pytest from raes_adapters import cli +from raes_adapters.bundle_verifier import BundleInvalid from raes_adapters.cyborg import reproduction from raes_adapters.cyborg.driver import ( _NativeEvaluationContext, @@ -20,6 +21,25 @@ PROJECT_ROOT = Path(__file__).parents[1] +def test_generic_integrity_precedes_backend_semantic_verification( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + bundle = tmp_path / "bundle" + bundle.mkdir() + (bundle / "inventory.json").write_text('{"artifacts":[]}\n', encoding="utf-8") + (bundle / "protocol.json").write_text("{}\n", encoding="utf-8") + + monkeypatch.setattr( + reproduction, + "load_strict_json", + lambda _path: pytest.fail("backend semantic checks ran before generic integrity"), + ) + + with pytest.raises(BundleInvalid, match="bundle.inventory.membership-mismatch"): + reproduction.verify_bundle(bundle) + + class StudyDriver: """Bounded native seam for the real scheduler and persistence path.""" @@ -560,5 +580,5 @@ def test_offline_verifier_requires_complete_transitive_inventories(tmp_path: Pat root_members = [bundle / item["path"] for item in root_inventory["artifacts"]] reproduction._seal_inventory(bundle, root_members) - with pytest.raises(ValueError, match="inventory membership is incomplete"): + with pytest.raises(BundleInvalid, match="bundle.inventory.membership-mismatch"): reproduction.verify_bundle(bundle, selection=selection) diff --git a/tests/test_inventory_primitives.py b/tests/test_inventory_primitives.py new file mode 100644 index 0000000..802d07a --- /dev/null +++ b/tests/test_inventory_primitives.py @@ -0,0 +1,79 @@ +"""Parity checks for shared private inventory producer plumbing.""" + +from __future__ import annotations + +import hashlib +import json +from pathlib import Path + +from raes_adapters import cli +from raes_adapters._inventory import inventory_document, inventory_entry +from raes_adapters.cyborg import reproduction + + +def _entry(path: str, content: bytes, media_type: str) -> dict[str, object]: + return { + "media_type": media_type, + "path": path, + "sha256": hashlib.sha256(content).hexdigest(), + "size_bytes": len(content), + } + + +def _canonical_bytes(payload: object) -> bytes: + return (json.dumps(payload, indent=2, sort_keys=True) + "\n").encode() + + +def test_generic_producer_wrapper_preserves_exact_inventory_bytes(tmp_path: Path) -> None: + nested = tmp_path / "nested" + nested.mkdir() + json_content = b'{"value":1}\n' + binary_content = b"\x00portable" + (nested / "record.json").write_bytes(json_content) + (tmp_path / "artifact.bin").write_bytes(binary_content) + expected = { + "artifacts": [ + _entry("artifact.bin", binary_content, "application/octet-stream"), + _entry("nested/record.json", json_content, "application/json"), + ] + } + + assert cli._seal_inventory(tmp_path) == expected + assert (tmp_path / "inventory.json").read_bytes() == _canonical_bytes(expected) + + +def test_cyborg_compatibility_wrappers_preserve_media_types_and_bytes(tmp_path: Path) -> None: + markdown_content = b"# Result\n" + gzip_content = b"compressed" + markdown = tmp_path / "report.md" + compressed = tmp_path / "attempt.json.gz" + markdown.write_bytes(markdown_content) + compressed.write_bytes(gzip_content) + expected = { + "artifacts": [ + _entry("attempt.json.gz", gzip_content, "application/gzip"), + _entry("report.md", markdown_content, "text/markdown"), + ] + } + + assert reproduction._inventory_entry(tmp_path, compressed) == expected["artifacts"][0] + assert reproduction._seal_inventory(tmp_path, [markdown, compressed]) == expected + assert (tmp_path / "inventory.json").read_bytes() == _canonical_bytes(expected) + + +def test_private_primitives_have_no_simulator_or_contract_authority(tmp_path: Path) -> None: + artifact = tmp_path / "record.json" + artifact.write_bytes(b"{}\n") + + assert inventory_entry(tmp_path, artifact)["path"] == "record.json" + assert inventory_document(tmp_path, [artifact]) == { + "artifacts": [inventory_entry(tmp_path, artifact)] + } + + source = (Path(__file__).parents[1] / "src/raes_adapters/_inventory.py").read_text( + encoding="utf-8" + ) + assert "raes_" not in source + assert "cyborg" not in source.casefold() + assert "nasim" not in source.casefold() + assert "cyberbattlesim" not in source.casefold()