diff --git a/tests/test_execution_outputs.py b/tests/test_execution_outputs.py new file mode 100644 index 0000000..a44c0e0 --- /dev/null +++ b/tests/test_execution_outputs.py @@ -0,0 +1,371 @@ +"""Exact, bounded output locators carry no authority to access files.""" + +import hashlib +import unittest +from dataclasses import FrozenInstanceError, replace +from unittest.mock import patch + +from test_execution_contexts import load_module + + +class OutputContractTests(unittest.TestCase): + def setUp(self) -> None: + self.m = load_module("execution_outputs") + + def reference(self, **changes) -> object: + reference = self.m.OutputReference( + "7", 0, "output", "", "same.png", "posix", "a" * 64 + ) + return replace(reference, **changes) + + def test_exact_unicode_and_duplicate_locators_survive(self) -> None: + m = self.m + a = self.reference() + b = replace(a, image_ordinal=1) + bundle = m.OutputBundle((a, b), "complete", "reported") + self.assertIsNone(m.validate_output_bundle(bundle)) + self.assertEqual(m.output_payload_size(bundle), 552) + self.assertEqual( + m.output_payload_size(bundle), + 128 + + sum( + 128 + + sum( + load_module("execution_contexts").text_size(v, 4096) + for v in ( + r.output_node_id, + r.folder_type, + r.subfolder, + r.filename, + r.path_style, + r.root_key or "", + ) + ) + for r in bundle.references + ), + ) + variants = ("é.png", "e\u0301.png", "É.png", " é.png", "é.png ") + references = tuple( + self.reference(image_ordinal=i, filename=name, subfolder=" 雪 / Café ") + for i, name in enumerate(variants) + ) + exact = m.OutputBundle(references, "complete", "reported") + self.assertIsNone(m.validate_output_bundle(exact)) + self.assertEqual(tuple(r.filename for r in exact.references), variants) + self.assertEqual(exact.references[0].subfolder, " 雪 / Café ") + self.assertEqual( + m.output_payload_size(m.OutputBundle((), "complete", "none_reported")), 128 + ) + with self.assertRaises(FrozenInstanceError): + a.filename = "changed.png" + + def test_posix_backslash_is_literal(self) -> None: + m = self.m + reference = self.reference(subfolder=r"a\b/雪", filename=r"a\b:c.png") + self.assertIsNone(m.validate_output_reference(reference)) + self.assertEqual(reference.filename, r"a\b:c.png") + for subfolder in ( + "/absolute", + "../escape", + "a/../b", + "./a", + "a//b", + "a/", + ".", + "..", + ): + with self.subTest(subfolder=subfolder), self.assertRaises(ValueError): + m.validate_output_reference(replace(reference, subfolder=subfolder)) + for filename in ("", "/absolute", "a/b", ".", "..", "a\x00b", "\ud800"): + with self.subTest(filename=repr(filename)), self.assertRaises(ValueError): + m.validate_output_reference(replace(reference, filename=filename)) + + def test_windows_drive_unc_traversal_ads_rejected(self) -> None: + m = self.m + reference = self.reference( + path_style="windows", subfolder=r"a\雪/b", filename=" image.png " + ) + self.assertIsNone(m.validate_output_reference(reference)) + for subfolder in ( + r"C:\absolute", + "C:relative", + r"\\server\share", + r"\rooted", + "/rooted", + "../escape", + r"a\..\b", + r"a\.\b", + r"a\\b", + "a//b", + r"a:stream", + "a/", + "a\\", + ".", + "..", + ): + with self.subTest(subfolder=subfolder), self.assertRaises(ValueError): + m.validate_output_reference(replace(reference, subfolder=subfolder)) + for filename in (r"a\b.png", "a/b.png", "C:foo.png", "a.png:stream", ".", ".."): + with self.subTest(filename=filename), self.assertRaises(ValueError): + m.validate_output_reference(replace(reference, filename=filename)) + + def test_root_namespace_is_versioned_style_and_folder_specific(self) -> None: + m = self.m + roots = m.snapshot_output_roots("posix", "/a/../雪/root/", "/雪/root") + self.assertEqual(roots.path_style, "posix") + self.assertEqual( + roots.output_key, + hashlib.sha256('[1,"posix","output","/雪/root"]'.encode()).hexdigest(), + ) + self.assertEqual( + roots.temp_key, + hashlib.sha256('[1,"posix","temp","/雪/root"]'.encode()).hexdigest(), + ) + self.assertNotEqual(roots.output_key, roots.temp_key) + self.assertNotEqual( + roots.output_key, + m.snapshot_output_roots("posix", "/雪/Root", None).output_key, + ) + windows = m.snapshot_output_roots( + "windows", r"C:\a\..\雪\root", r"\\server\share\a\.." + ) + self.assertEqual( + windows.output_key, + hashlib.sha256( + r'[1,"windows","output","C:\\雪\\root"]'.encode() + ).hexdigest(), + ) + self.assertIsNotNone(windows.temp_key) + invalid = ( + None, + True, + b"/tmp", + "relative", + "", + "/a\x00b", + "/\ud800", + "/" + "x" * 8192, + "/" + "é" * 4096, + ) + for root in invalid: + with self.subTest(root_type=type(root).__name__): + snapshot = m.snapshot_output_roots("posix", root, "/tmp") + self.assertIsNone(snapshot.output_key) + self.assertIsNotNone(snapshot.temp_key) + self.assertIsNotNone( + m.snapshot_output_roots("posix", "/" + "x" * 8191, None).output_key + ) + self.assertIsNotNone( + m.snapshot_output_roots("posix", "/" + "é" * 4095 + "x", None).output_key + ) + for root in ( + r"\\server", + "\\\\server\\", + r"\\server\.", + r"1:\root", + r"C:relative", + r"\rooted", + "/rooted", + r"\\?\C:\root", + r"\\.\C:\root", + ): + with self.subTest(root=root): + self.assertIsNone( + m.snapshot_output_roots("windows", root, None).output_key + ) + for style in ("linux", "POSIX", None, True): + with self.assertRaises(ValueError): + m.snapshot_output_roots(style, "/tmp", "/tmp") + + def test_bounds_and_state_row_consistency(self) -> None: + m = self.m + a = self.reference() + valid = { + "complete": ("reported", "none_reported"), + "partial": ( + "invalid_evidence", + "unsupported", + "limit_reached", + "root_unavailable", + "budget_exhausted", + ), + "unavailable": ("unavailable", "invalid_evidence", "budget_exhausted"), + "conflicting": ("conflicting",), + } + reasons = ( + "reported", + "none_reported", + "invalid_evidence", + "unsupported", + "limit_reached", + "root_unavailable", + "budget_exhausted", + "unavailable", + "conflicting", + "unknown", + "", + ) + for state in (*valid, "not_captured", "unknown"): + for reason in reasons: + for references in ((), (a,)): + accepted = reason in valid.get(state, ()) + if state == "complete": + accepted = accepted and bool(references) == ( + reason == "reported" + ) + if state == "unavailable": + accepted = accepted and not references + if reason == "root_unavailable": + accepted = accepted and bool(references) + bundle = m.OutputBundle(references, state, reason) + with self.subTest(state=state, reason=reason, rows=len(references)): + if accepted: + self.assertIsNone(m.validate_output_bundle(bundle)) + else: + with self.assertRaises(ValueError): + m.validate_output_bundle(bundle) + null_root = replace(a, root_key=None) + for state, reason in ( + ("partial", "root_unavailable"), + ("conflicting", "conflicting"), + ): + m.validate_output_bundle(m.OutputBundle((null_root,), state, reason)) + with self.assertRaises(ValueError): + m.validate_output_bundle( + m.OutputBundle((null_root,), "complete", "reported") + ) + for ordinal in (0, 255): + m.validate_output_reference(replace(a, image_ordinal=ordinal)) + for changes in ( + {"image_ordinal": -1}, + {"image_ordinal": 256}, + {"image_ordinal": True}, + {"image_ordinal": 1.0}, + {"output_node_id": ""}, + {"output_node_id": "x" * 129}, + {"output_node_id": "\ud800"}, + {"output_node_id": "a\x00b"}, + {"filename": "é" * 513}, + {"subfolder": "雪" * 683}, + {"root_key": "A" * 64}, + {"root_key": "a" * 63}, + {"root_key": "z" * 64}, + {"folder_type": "input"}, + {"path_style": "native"}, + ): + with self.subTest(changes=changes), self.assertRaises(ValueError): + m.validate_output_reference(replace(a, **changes)) + ordered = ( + replace(a, output_node_id="10"), + replace(a, output_node_id="2"), + a, + replace(a, image_ordinal=1), + ) + m.validate_output_bundle(m.OutputBundle(ordered, "complete", "reported")) + for references in ( + (a, a), + tuple(reversed(ordered)), + tuple(replace(a, output_node_id=str(i).zfill(3)) for i in range(257)), + ): + with self.assertRaises(ValueError): + m.validate_output_bundle( + m.OutputBundle(references, "complete", "reported") + ) + maximum = replace( + a, + output_node_id="😀" * 128, + filename="x" * 1024, + subfolder="x" * 2048, + path_style="windows", + ) + self.assertIsNone(m.validate_output_reference(maximum)) + self.assertEqual( + m.output_payload_size(m.OutputBundle((maximum,), "complete", "reported")), + 3917, + ) + prefix = tuple(replace(maximum, image_ordinal=i) for i in range(34)) + tail = replace(maximum, image_ordinal=34, filename="x", subfolder="x" * 1400) + exact = m.OutputBundle((*prefix, tail), "complete", "reported") + self.assertEqual(m.output_payload_size(exact), 131072) + with self.assertRaises(ValueError): + m.output_payload_size( + replace( + exact, references=(*prefix, replace(tail, subfolder="x" * 1401)) + ) + ) + for changes in ( + {"format_version": True}, + {"format_version": 2}, + {"source": "host_cache"}, + {"references": [a]}, + ): + with self.subTest(changes=changes), self.assertRaises(ValueError): + m.validate_output_bundle( + replace(m.OutputBundle((a,), "complete", "reported"), **changes) + ) + + def test_no_coercion_or_filesystem_calls(self) -> None: + m = self.m + + class Hostile: + def __str__(self): + raise AssertionError("coerced arbitrary object") + + def __fspath__(self): + raise AssertionError("read arbitrary path protocol") + + def __iter__(self): + raise AssertionError("iterated arbitrary object") + + class Text(str): + pass + + class Sequence(tuple): + pass + + hostile = Hostile() + for field in ( + "output_node_id", + "folder_type", + "subfolder", + "filename", + "path_style", + "root_key", + ): + for value in (hostile, Text("output")): + with self.subTest(field=field), self.assertRaises(ValueError): + m.validate_output_reference(self.reference(**{field: value})) + for value in (hostile, Sequence((self.reference(),))): + with self.assertRaises(ValueError): + m.validate_output_bundle(m.OutputBundle(value, "complete", "reported")) + for field in ("state", "reason", "source"): + with self.subTest(field=field), self.assertRaises(ValueError): + m.validate_output_bundle( + replace( + m.OutputBundle((), "complete", "none_reported"), + **{field: hostile}, + ) + ) + for value in (hostile, Text("/tmp")): + self.assertIsNone(m.snapshot_output_roots("posix", value, None).output_key) + with ( + patch("builtins.open", side_effect=AssertionError("filesystem open")), + patch("os.stat", side_effect=AssertionError("filesystem stat")), + patch("os.lstat", side_effect=AssertionError("filesystem lstat")), + patch( + "os.path.realpath", side_effect=AssertionError("filesystem resolution") + ), + ): + roots = m.snapshot_output_roots( + "posix", "/nonexistent/output", "/nonexistent/temp" + ) + reference = self.reference(root_key=roots.output_key) + self.assertIsNone( + m.validate_output_bundle( + m.OutputBundle((reference,), "complete", "reported") + ) + ) + self.assertEqual( + m.output_payload_size(m.OutputBundle((), "unavailable", "unavailable")), + 128, + ) diff --git a/utils/execution_outputs.py b/utils/execution_outputs.py new file mode 100644 index 0000000..4899563 --- /dev/null +++ b/utils/execution_outputs.py @@ -0,0 +1,211 @@ +"""Bounded scalar output locators, without filesystem resolution or authority.""" + +import hashlib +import json +import ntpath +import posixpath +import re +from dataclasses import dataclass +from types import ModuleType + +from .execution_contexts import text_size + + +@dataclass(frozen=True) +class OutputReference: + output_node_id: str + image_ordinal: int + folder_type: str + subfolder: str + filename: str + path_style: str + root_key: str | None + + +@dataclass(frozen=True) +class OutputBundle: + references: tuple[OutputReference, ...] + state: str + reason: str + format_version: int = 1 + source: str = "host_final_history" + + +@dataclass(frozen=True) +class OutputRoots: + path_style: str + output_key: str | None + temp_key: str | None + + +MAX_REFERENCE_BYTES = 4096 +MAX_BUNDLE_BYTES = 128 * 1024 +MAX_IMAGE_POSITIONS = 256 +MAX_ROOT_BYTES = 8192 +REFERENCE_OVERHEAD = 128 +BUNDLE_OVERHEAD = 128 +STATE_REASONS = { + "complete": frozenset({"reported", "none_reported"}), + "partial": frozenset( + { + "invalid_evidence", + "unsupported", + "limit_reached", + "root_unavailable", + "budget_exhausted", + } + ), + "unavailable": frozenset({"unavailable", "invalid_evidence", "budget_exhausted"}), + "conflicting": frozenset({"conflicting"}), +} + + +def _path_module(path_style: str) -> ModuleType: + if type(path_style) is not str or path_style not in ("posix", "windows"): + raise ValueError("Invalid output path style") + return posixpath if path_style == "posix" else ntpath + + +def _scalar_size(value: str, limit: int) -> int: + size = text_size(value, limit) + if "\x00" in value: + raise ValueError("NUL in output scalar") + return size + + +def _root_key(path_style: str, folder_type: str, root: object) -> str | None: + try: + _scalar_size(root, MAX_ROOT_BYTES) + except ValueError: + return None + paths = _path_module(path_style) + if not paths.isabs(root): + return None + if path_style == "windows": + drive, _ = ntpath.splitdrive(root) + portable = root.replace("/", "\\") + if not drive or portable.startswith(("\\\\?\\", "\\\\.\\")): + return None + portable_drive = drive.replace("/", "\\") + if portable_drive.startswith("\\\\"): + parts = portable_drive[2:].split("\\") + if len(parts) != 2 or any(part in ("", ".", "..") for part in parts): + return None + elif re.fullmatch(r"[a-zA-Z]:", drive) is None: + return None + normalized = paths.normpath(root) + encoded = json.dumps( + [1, path_style, folder_type, normalized], + ensure_ascii=False, + allow_nan=False, + separators=(",", ":"), + ).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def snapshot_output_roots( + path_style: str, output_root: object, temp_root: object +) -> OutputRoots: + """Namespace bounded, lexically absolute configured roots without reading files.""" + _path_module(path_style) + return OutputRoots( + path_style, + _root_key(path_style, "output", output_root), + _root_key(path_style, "temp", temp_root), + ) + + +def _reference_size(reference: OutputReference) -> int: + if type(reference) is not OutputReference: + raise ValueError("Invalid output reference") + size = REFERENCE_OVERHEAD + if ( + type(reference.output_node_id) is not str + or not reference.output_node_id + or len(reference.output_node_id) > 128 + ): + raise ValueError("Invalid output node ID") + size += _scalar_size(reference.output_node_id, 512) + if type(reference.image_ordinal) is not int or not ( + 0 <= reference.image_ordinal < MAX_IMAGE_POSITIONS + ): + raise ValueError("Invalid output image ordinal") + size += _scalar_size(reference.folder_type, 6) + if reference.folder_type not in ("output", "temp"): + raise ValueError("Unsupported output folder type") + size += _scalar_size(reference.path_style, 7) + paths = _path_module(reference.path_style) + size += _scalar_size(reference.subfolder, 2048) + size += _scalar_size(reference.filename, 1024) + if not reference.filename or reference.filename in (".", ".."): + raise ValueError("Invalid output filename") + separators = ("/", "\\") if reference.path_style == "windows" else ("/",) + if any(separator in reference.filename for separator in separators): + raise ValueError("Output filename must be a basename") + if paths.isabs(reference.subfolder) or paths.splitdrive(reference.subfolder)[0]: + raise ValueError("Output subfolder must be relative") + if reference.path_style == "windows" and ( + ":" in reference.filename or ":" in reference.subfolder + ): + raise ValueError("Windows output colon or alternate stream") + if reference.subfolder: + parts = ( + re.split(r"[/\\]", reference.subfolder) + if reference.path_style == "windows" + else reference.subfolder.split("/") + ) + if any(part in ("", ".", "..") for part in parts): + raise ValueError("Unsafe or ambiguous output subfolder") + if reference.root_key is not None: + size += _scalar_size(reference.root_key, 64) + if re.fullmatch(r"[a-f0-9]{64}", reference.root_key) is None: + raise ValueError("Invalid output root key") + if size > MAX_REFERENCE_BYTES: + raise ValueError("Output reference payload limit reached") + return size + + +def validate_output_reference(reference: OutputReference) -> None: + _reference_size(reference) + + +def validate_output_bundle(bundle: OutputBundle) -> None: + output_payload_size(bundle) + + +def output_payload_size(bundle: OutputBundle) -> int: + """Validate the immutable capture and measure its charged scalar payload.""" + if ( + type(bundle) is not OutputBundle + or type(bundle.format_version) is not int + or bundle.format_version != 1 + or type(bundle.references) is not tuple + or len(bundle.references) > MAX_IMAGE_POSITIONS + ): + raise ValueError("Invalid output bundle") + for value in (bundle.state, bundle.reason, bundle.source): + _scalar_size(value, 128) + if bundle.source != "host_final_history" or bundle.reason not in STATE_REASONS.get( + bundle.state, () + ): + raise ValueError("Invalid output state, reason or source") + if bundle.state == "complete": + if bool(bundle.references) != (bundle.reason == "reported"): + raise ValueError("Inconsistent complete output rows") + if bundle.state == "unavailable" and bundle.references: + raise ValueError("Unavailable output capture has rows") + if bundle.reason == "root_unavailable" and not bundle.references: + raise ValueError("Root unavailability requires output rows") + size = BUNDLE_OVERHEAD + previous = None + for reference in bundle.references: + size += _reference_size(reference) + coordinate = (reference.output_node_id, reference.image_ordinal) + if previous is not None and coordinate <= previous: + raise ValueError("Output coordinates must be unique and sorted") + previous = coordinate + if bundle.state == "complete" and reference.root_key is None: + raise ValueError("Complete output rows require root keys") + if size > MAX_BUNDLE_BYTES: + raise ValueError("Output bundle payload limit reached") + return size