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..6be0d0b 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" @@ -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/_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/bundle_verifier.py b/src/raes_adapters/bundle_verifier.py new file mode 100644 index 0000000..6f0ea3e --- /dev/null +++ b/src/raes_adapters/bundle_verifier.py @@ -0,0 +1,555 @@ +"""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"} +_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): + """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): + """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.""" + + status: str + code: str + files: int = 0 + inventories: int = 0 + entries: int = 0 + unique_bytes: int = 0 + + def payload(self) -> dict[str, object]: + """Return the stable machine-readable card payload.""" + + 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(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] + + +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: + _invalid("bundle.inventory.duplicate-key") + result[key] = value + return result + + +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(_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(_CODE_PATH_INVALID) + if len(encoded) > MAX_PATH_BYTES: + _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(_CODE_PATH_INVALID) + if len(path.parts) > MAX_DEPTH: + _invalid("bundle.limit.depth") + 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 + 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]: + """Return the filesystem fields used to detect concurrent mutation.""" + + return ( + info.st_dev, + info.st_ino, + info.st_mode, + info.st_size, + info.st_mtime_ns, + info.st_ctime_ns, + ) + + +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: + 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 + except OSError: + _invalid(_CODE_FILESYSTEM_MUTATED) + if _identity(before) != _identity(after) or _identity(after) != _identity(after_path): + _invalid(_CODE_FILESYSTEM_MUTATED) + return _FileRecord( + relative=relative, + content=content, + size=len(content), + sha256=hashlib.sha256(content).hexdigest(), + identity=_identity(after), + ) + + +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(_CODE_INVENTORY_MALFORMED) + if not isinstance(payload, dict) or set(payload) != {"artifacts"}: + _invalid(_CODE_INVENTORY_MALFORMED) + artifacts = payload["artifacts"] + if not isinstance(artifacts, list): + _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(_CODE_ENTRY_MALFORMED) + relative = _relative_path(item["path"]) + size = _entry_size(item["size_bytes"]) + digest = _entry_digest(item["sha256"]) + _check_media_type(item["media_type"]) + 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.""" + + 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") + 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: + current = path.lstat() + except OSError: + _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: + """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 = 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/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/src/raes_adapters/entrypoint.py b/src/raes_adapters/entrypoint.py new file mode 100644 index 0000000..093c4ba --- /dev/null +++ b/src/raes_adapters/entrypoint.py @@ -0,0 +1,22 @@ +"""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: + """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 + + 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..00b095a --- /dev/null +++ b/tests/test_bundle_verifier.py @@ -0,0 +1,370 @@ +"""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" + 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") + + +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 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()