From e71b2d16688780ca0172adf0da8e9b6b1222cf05 Mon Sep 17 00:00:00 2001 From: Dmitry Dementiev Date: Wed, 2 Sep 2026 03:20:42 +0300 Subject: [PATCH] =?UTF-8?q?feat(experiment):=20=D0=B4=D0=BE=D0=B1=D0=B0?= =?UTF-8?q?=D0=B2=D0=BB=D0=B5=D0=BD=20gate=20=D0=BF=D1=80=D0=B8=D0=BD?= =?UTF-8?q?=D1=8F=D1=82=D0=B8=D1=8F=20=D0=BE=D0=BF=D0=BE=D1=80=D1=8B=20?= =?UTF-8?q?=D0=B2=D0=BB=D0=B0=D0=B4=D0=B5=D0=BB=D1=8C=D1=86=D0=B5=D0=BC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Зачем: - не допустить загрузку W0 и Sortformer без отдельного принятия трёх референсов владельцем - связать manifest с точными байтами listening package и owner-authored record Что: - добавлены строгие схемы трёх reference rows и четырёх решений Step 3 - все внешние файлы проверяются через held fd без symlink и TOCTOU - добавлены negative tests для подмен, неточных JSON-типов и небезопасных путей Проверка: - uv run pytest -q (532 passed, 2 skipped) - uv run ruff check src/local_transcriber/streaming_sortformer_experiment.py tests/test_streaming_sortformer_experiment.py - git diff --check --- .../streaming_sortformer_experiment.py | 397 +++++++- tests/test_streaming_sortformer_experiment.py | 933 +++++++++++++++++- 2 files changed, 1296 insertions(+), 34 deletions(-) diff --git a/src/local_transcriber/streaming_sortformer_experiment.py b/src/local_transcriber/streaming_sortformer_experiment.py index 47f7f2b..efefe5b 100644 --- a/src/local_transcriber/streaming_sortformer_experiment.py +++ b/src/local_transcriber/streaming_sortformer_experiment.py @@ -5,11 +5,13 @@ from __future__ import annotations import argparse import copy import ctypes +import errno import hashlib import json import math import os import re +import stat import statistics import subprocess import sys @@ -44,10 +46,51 @@ TOP_LEVEL_KEYS = { "schedule", "rss_sample_interval_ms", } -FRESH_BASIS_KEYS = {"marker", "approval"} -APPROVAL_KEYS = {"decision", "timestamp"} +FRESH_BASIS_KEYS = {"marker", "owner_package", "approval"} +OWNER_PACKAGE_LINK_KEYS = {"path", "sha256", "generated_at"} +APPROVAL_LINK_KEYS = { + "path", + "sha256", + "actor", + "status", + "decision", + "timestamp", +} +OWNER_PACKAGE_SCHEMA = "local-transcriber.owner-listening-package.v1" +OWNER_PACKAGE_KEYS = {"schema", "generated_at", "references"} +OWNER_PACKAGE_REFERENCE_KEYS = { + "neutral_alias", + "canonical_fragment_audio_sha256", + "final_reference_sha256", + "full_recording_sha256", +} +REFERENCE_ACCEPTANCE_SCHEMA = "local-transcriber.reference-owner-acceptance.v1" +REFERENCE_ACCEPTANCE_KEYS = { + "schema", + "actor", + "status", + "decision", + "timestamp", + "owner_package_sha256", + "references", + "step3", +} +REFERENCE_ACCEPTANCE_ROW_KEYS = OWNER_PACKAGE_REFERENCE_KEYS | { + "fragment_expected_cluster_count", + "full_recording_expected_cluster_count", + "fragment_review_complete", + "full_recording_count_review_complete", + "reference_accepted", +} +STEP3_KEYS = { + "recording_use_approved", + "inputs_and_counts_approved", + "full_recordings_available", + "final_hearing_gate_committed", +} SOURCE_KEYS = {"commit"} INPUT_KEYS = { + "neutral_alias", "source_path", "source_sha256", "source_size_bytes", @@ -312,45 +355,294 @@ def build_full_recording_schedule( return result -def _resolve(root: Path, value: Any) -> Path: +def _aware_datetime(value: Any) -> datetime: + if not isinstance(value, str): + raise PreflightError("approval-timestamp") + try: + parsed = datetime.fromisoformat(value) + except ValueError as exc: + raise PreflightError("approval-timestamp") from exc + if parsed.tzinfo is None or parsed.utcoffset() is None: + raise PreflightError("approval-timestamp") + return parsed + + +def _safe_relative_parts(value: Any) -> tuple[str, ...]: if not isinstance(value, str) or not value: raise PreflightError("unsafe-relative-path") pure = PurePosixPath(value) - if pure.is_absolute() or ".." in pure.parts or "\\" in value: + if ( + pure.is_absolute() + or not pure.parts + or ".." in pure.parts + or "\\" in value + or "\x00" in value + ): raise PreflightError("path-escape") - base = root.resolve() - resolved = (base / Path(value)).resolve() - if not resolved.is_relative_to(base): - raise PreflightError("path-escape") - return resolved + return pure.parts + + +def _same_file_state(before: os.stat_result, after: os.stat_result) -> bool: + return ( + before.st_dev, + before.st_ino, + before.st_mode, + before.st_nlink, + before.st_size, + before.st_mtime_ns, + before.st_ctime_ns, + ) == ( + after.st_dev, + after.st_ino, + after.st_mode, + after.st_nlink, + after.st_size, + after.st_mtime_ns, + after.st_ctime_ns, + ) + + +@dataclass(frozen=True) +class _BoundFile: + sha256: str + size_bytes: int + payload: bytes | None + + +def _read_bound_regular_file( + root: Path, path_value: Any, *, capture_payload: bool +) -> _BoundFile: + """Читает файл относительно доверенного неизменяемого root dirfd.""" + parts = _safe_relative_parts(path_value) + directory_flags = ( + os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC + ) + opened_fds: list[int] = [] + fd_states: list[tuple[int, os.stat_result]] = [] + namespace_entries: list[tuple[int, str, os.stat_result]] = [] + try: + directory_fd = os.open(root, directory_flags) + opened_fds.append(directory_fd) + root_state = os.fstat(directory_fd) + if not stat.S_ISDIR(root_state.st_mode): + raise PreflightError("file-binding") + fd_states.append((directory_fd, root_state)) + for part in parts[:-1]: + parent_fd = directory_fd + directory_fd = os.open(part, directory_flags, dir_fd=parent_fd) + opened_fds.append(directory_fd) + directory_state = os.fstat(directory_fd) + if not stat.S_ISDIR(directory_state.st_mode): + raise PreflightError("file-binding") + fd_states.append((directory_fd, directory_state)) + namespace_entries.append((parent_fd, part, directory_state)) + file_fd = os.open( + parts[-1], + os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC, + dir_fd=directory_fd, + ) + opened_fds.append(file_fd) + file_state = os.fstat(file_fd) + if not stat.S_ISREG(file_state.st_mode) or file_state.st_nlink != 1: + raise PreflightError("file-binding") + fd_states.append((file_fd, file_state)) + namespace_entries.append((directory_fd, parts[-1], file_state)) + + digest = hashlib.sha256() + size_bytes = 0 + chunks: list[bytes] | None = [] if capture_payload else None + while chunk := os.read(file_fd, 64 * 1024): + digest.update(chunk) + size_bytes += len(chunk) + if chunks is not None: + chunks.append(chunk) + + for file_descriptor, original_state in fd_states: + if not _same_file_state(original_state, os.fstat(file_descriptor)): + raise PreflightError("file-binding") + for parent_fd, name, original_state in namespace_entries: + current_state = os.stat(name, dir_fd=parent_fd, follow_symlinks=False) + if not _same_file_state(original_state, current_state): + raise PreflightError("file-binding") + if size_bytes != file_state.st_size: + raise PreflightError("file-binding") + return _BoundFile( + sha256=digest.hexdigest(), + size_bytes=size_bytes, + payload=b"".join(chunks) if chunks is not None else None, + ) + except PreflightError: + raise + except ValueError as exc: + raise PreflightError("path-escape") from exc + except OSError as exc: + reason = ( + "path-escape" + if exc.errno in {errno.ELOOP, errno.ENOTDIR} + else "file-binding" + ) + raise PreflightError(reason) from exc + finally: + for opened_fd in reversed(opened_fds): + try: + os.close(opened_fd) + except OSError: + pass def _verify(root: Path, path_value: Any, expected_hash: Any, size: Any | None) -> None: - path = _resolve(root, path_value) expected = _sha(expected_hash, "invalid-sha256") - if not path.is_file(): - raise PreflightError("file-unavailable") - if size is not None and path.stat().st_size != _positive_int(size, "invalid-size"): + expected_size = None if size is None else _positive_int(size, "invalid-size") + bound = _read_bound_regular_file(root, path_value, capture_payload=False) + if expected_size is not None and bound.size_bytes != expected_size: raise PreflightError("size-mismatch") - if file_sha256(path) != expected: + if bound.sha256 != expected: raise PreflightError("hash-mismatch") -def _validate_approval(value: Any) -> bool: - if value is None: - return False - approval = _exact(value, APPROVAL_KEYS, "approval-schema") - if approval["decision"] != "approved": - return False - timestamp = approval["timestamp"] - if not isinstance(timestamp, str): - raise PreflightError("approval-timestamp") +def _strict_json_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + value: dict[str, Any] = {} + for key, item in pairs: + if key in value: + raise PreflightError("owner-acceptance-record") + value[key] = item + return value + + +def _reject_json_constant(_value: str) -> None: + raise PreflightError("owner-acceptance-record") + + +def _bound_json( + root: Path, link: Mapping[str, Any], link_keys: set[str] +) -> Mapping[str, Any]: + value = _exact(link, link_keys, "owner-acceptance-link") + expected_sha256 = _sha(value["sha256"], "owner-acceptance-hash") + bound = _read_bound_regular_file(root, value["path"], capture_payload=True) + if bound.sha256 != expected_sha256: + raise PreflightError("owner-acceptance-hash") + if bound.payload is None: + raise PreflightError("owner-acceptance-record") try: - parsed = datetime.fromisoformat(timestamp) - except ValueError as exc: - raise PreflightError("approval-timestamp") from exc - if parsed.tzinfo is None: - raise PreflightError("approval-timestamp") + loaded = json.loads( + bound.payload, + object_pairs_hook=_strict_json_object, + parse_constant=_reject_json_constant, + ) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise PreflightError("owner-acceptance-record") from exc + if not isinstance(loaded, Mapping): + raise PreflightError("owner-acceptance-record") + return loaded + + +def _validate_owner_package_row(row: Any) -> None: + value = _exact(row, OWNER_PACKAGE_REFERENCE_KEYS, "owner-package-reference") + alias = value["neutral_alias"] + if ( + not isinstance(alias, str) + or re.fullmatch(r"[a-z][a-z0-9-]{0,63}", alias) is None + ): + raise PreflightError("owner-package-reference") + for key in OWNER_PACKAGE_REFERENCE_KEYS - {"neutral_alias"}: + _sha(value[key], "owner-package-reference") + + +def _validate_acceptance_row(row: Any) -> None: + value = _exact( + row, + REFERENCE_ACCEPTANCE_ROW_KEYS, + "owner-acceptance-reference", + ) + _validate_owner_package_row( + {key: value[key] for key in OWNER_PACKAGE_REFERENCE_KEYS} + ) + for key in ( + "fragment_expected_cluster_count", + "full_recording_expected_cluster_count", + ): + count = value[key] + if type(count) is not int or not 2 <= count <= 4: + raise PreflightError("owner-acceptance-reference") + for key in ( + "fragment_review_complete", + "full_recording_count_review_complete", + "reference_accepted", + ): + if value[key] is not True: + raise PreflightError("owner-acceptance-reference") + + +def _validate_reference_owner_acceptance( + *, + root: Path, + basis: Mapping[str, Any], + expected_package_rows: Sequence[Mapping[str, Any]], + expected_acceptance_rows: Sequence[Mapping[str, Any]], +) -> bool: + try: + package_link = _exact( + basis["owner_package"], + OWNER_PACKAGE_LINK_KEYS, + "owner-package-link", + ) + package = _exact( + _bound_json(root, package_link, OWNER_PACKAGE_LINK_KEYS), + OWNER_PACKAGE_KEYS, + "owner-package-schema", + ) + if package["schema"] != OWNER_PACKAGE_SCHEMA: + raise PreflightError("owner-package-schema") + package_generated_at = _aware_datetime(package["generated_at"]) + if package["generated_at"] != package_link["generated_at"]: + raise PreflightError("owner-package-timestamp") + package_rows = package["references"] + if not isinstance(package_rows, list) or len(package_rows) != 3: + raise PreflightError("owner-package-references") + for row in package_rows: + _validate_owner_package_row(row) + if package_rows != list(expected_package_rows): + raise PreflightError("owner-package-binding") + + approval_link = _exact( + basis["approval"], APPROVAL_LINK_KEYS, "owner-acceptance-link" + ) + acceptance = _exact( + _bound_json(root, approval_link, APPROVAL_LINK_KEYS), + REFERENCE_ACCEPTANCE_KEYS, + "owner-acceptance-schema", + ) + if acceptance["schema"] != REFERENCE_ACCEPTANCE_SCHEMA: + raise PreflightError("owner-acceptance-schema") + for key in ("actor", "status", "decision", "timestamp"): + if acceptance[key] != approval_link[key]: + raise PreflightError("owner-acceptance-link-mismatch") + if ( + acceptance["actor"] != "owner" + or acceptance["status"] != "approved" + or acceptance["decision"] != "approved" + ): + raise PreflightError("owner-acceptance-decision") + accepted_at = _aware_datetime(acceptance["timestamp"]) + if accepted_at <= package_generated_at: + raise PreflightError("owner-acceptance-timestamp") + if ( + _sha(acceptance["owner_package_sha256"], "owner-package-binding") + != package_link["sha256"] + ): + raise PreflightError("owner-package-binding") + acceptance_rows = acceptance["references"] + if not isinstance(acceptance_rows, list) or len(acceptance_rows) != 3: + raise PreflightError("owner-acceptance-references") + for row in acceptance_rows: + _validate_acceptance_row(row) + if acceptance_rows != list(expected_acceptance_rows): + raise PreflightError("owner-acceptance-binding") + step3 = _exact(acceptance["step3"], STEP3_KEYS, "owner-acceptance-step3") + if any(step3[key] is not True for key in STEP3_KEYS): + raise PreflightError("owner-acceptance-step3") + except (KeyError, OSError, TypeError, PreflightError): + return False return True @@ -457,10 +749,12 @@ def validate_manifest( raise PreflightError("legacy-identity") raise PreflightError("legacy-root") - basis = _exact(value["fresh_basis"], FRESH_BASIS_KEYS, "fresh-basis-schema") - if basis["marker"] != FRESH_BASIS_MARKER: + basis = value["fresh_basis"] + if not isinstance(basis, Mapping): + raise PreflightError("fresh-basis-schema") + if basis.get("marker") != FRESH_BASIS_MARKER: raise PreflightError("fresh-basis-marker") - approved = _validate_approval(basis["approval"]) + basis_schema_valid = set(basis) == FRESH_BASIS_KEYS source = _exact(value["source"], SOURCE_KEYS, "source-schema") source_commit = _commit(source["commit"], "source-commit") if set(environment) != {"commit", "clean"}: @@ -486,8 +780,18 @@ def validate_manifest( raise PreflightError("source-hashes-not-unique") fragment_hashes = [] fragment_sidecars = [] + package_rows = [] + acceptance_rows = [] + aliases = [] for item in inputs: entry = _exact(item, INPUT_KEYS, "input-schema") + alias = entry["neutral_alias"] + if ( + not isinstance(alias, str) + or re.fullmatch(r"[a-z][a-z0-9-]{0,63}", alias) is None + ): + raise PreflightError("neutral-alias") + aliases.append(alias) if verify_files: _verify( root, @@ -522,11 +826,40 @@ def validate_manifest( record["reference_sha256"], None, ) + package_row = { + "neutral_alias": alias, + "canonical_fragment_audio_sha256": entry["fragment"]["sha256"], + "final_reference_sha256": entry["fragment"]["reference_sha256"], + "full_recording_sha256": entry["full_recording"]["sha256"], + } + package_rows.append(package_row) + acceptance_rows.append( + { + **package_row, + "fragment_expected_cluster_count": entry["fragment"][ + "expected_cluster_count" + ], + "full_recording_expected_cluster_count": entry["full_recording"][ + "expected_cluster_count" + ], + "fragment_review_complete": True, + "full_recording_count_review_complete": True, + "reference_accepted": True, + } + ) + if len(set(aliases)) != 3: + raise PreflightError("neutral-alias") _validate_schedule(value["schedule"], fragment_hashes, fragment_sidecars) _positive_int(value["rss_sample_interval_ms"], "rss-sample-interval") + approved = basis_schema_valid and _validate_reference_owner_acceptance( + root=root, + basis=basis, + expected_package_rows=package_rows, + expected_acceptance_rows=acceptance_rows, + ) return PreflightOutcome( state="valid" if approved else "needs-user", - reason=None if approved else "owner-approval-required", + reason=None if approved else "reference-owner-acceptance", fragment_count=3, full_recording_count=3, source_commit_sha256=hashlib.sha256(source_commit.encode("ascii")).hexdigest(), diff --git a/tests/test_streaming_sortformer_experiment.py b/tests/test_streaming_sortformer_experiment.py index 93d15df..071d63d 100644 --- a/tests/test_streaming_sortformer_experiment.py +++ b/tests/test_streaming_sortformer_experiment.py @@ -2,6 +2,7 @@ import copy import ctypes import hashlib import json +import os import tomllib from datetime import UTC, datetime from pathlib import Path @@ -32,6 +33,14 @@ def _write(root: Path, name: str, payload: bytes) -> dict[str, object]: } +def _write_json(root: Path, name: str, value: object) -> dict[str, object]: + return _write( + root, + name, + (json.dumps(value, ensure_ascii=True, sort_keys=True) + "\n").encode(), + ) + + def _record( root: Path, prefix: str, *, duration: float, reference: bool ) -> dict[str, object]: @@ -61,6 +70,7 @@ def _manifest(root: Path) -> dict[str, object]: source = _write(root, f"sources/{index}.wav", f"source-{index}".encode()) inputs.append( { + "neutral_alias": f"control-{index + 1}", "source_path": source["path"], "source_sha256": source["sha256"], "source_size_bytes": source["size_bytes"], @@ -72,13 +82,74 @@ def _manifest(root: Path) -> dict[str, object]: ), } ) + package_generated_at = "2026-09-01T08:00:00+00:00" + package_rows = [ + { + "neutral_alias": item["neutral_alias"], + "canonical_fragment_audio_sha256": item["fragment"]["sha256"], + "final_reference_sha256": item["fragment"]["reference_sha256"], + "full_recording_sha256": item["full_recording"]["sha256"], + } + for item in inputs + ] + package = _write_json( + root, + "owner/listening-package.json", + { + "schema": "local-transcriber.owner-listening-package.v1", + "generated_at": package_generated_at, + "references": package_rows, + }, + ) + approval_timestamp = "2026-09-01T09:00:00+00:00" + acceptance = { + "schema": "local-transcriber.reference-owner-acceptance.v1", + "actor": "owner", + "status": "approved", + "decision": "approved", + "timestamp": approval_timestamp, + "owner_package_sha256": package["sha256"], + "references": [ + { + **package_row, + "fragment_expected_cluster_count": item["fragment"][ + "expected_cluster_count" + ], + "full_recording_expected_cluster_count": item["full_recording"][ + "expected_cluster_count" + ], + "fragment_review_complete": True, + "full_recording_count_review_complete": True, + "reference_accepted": True, + } + for item, package_row in zip(inputs, package_rows, strict=True) + ], + "step3": { + "recording_use_approved": True, + "inputs_and_counts_approved": True, + "full_recordings_available": True, + "final_hearing_gate_committed": True, + }, + } + acceptance_file = _write_json( + root, "owner/reference-acceptance.json", acceptance + ) return { "schema": "local-transcriber.streaming-sortformer-experiment.v1", "fresh_basis": { "marker": "issue-46-fresh-owner-reviewed-basis", + "owner_package": { + "path": package["path"], + "sha256": package["sha256"], + "generated_at": package_generated_at, + }, "approval": { + "path": acceptance_file["path"], + "sha256": acceptance_file["sha256"], + "actor": "owner", + "status": "approved", "decision": "approved", - "timestamp": "2026-09-01T09:00:00+00:00", + "timestamp": approval_timestamp, }, }, "source": {"commit": "a" * 40}, @@ -194,6 +265,232 @@ def _environment() -> dict[str, object]: return {"commit": "a" * 40, "clean": True} +def _read_linked_json(root: Path, link: dict[str, object]) -> dict[str, object]: + return json.loads((root / str(link["path"])).read_text(encoding="utf-8")) + + +def _rewrite_linked_json( + root: Path, link: dict[str, object], value: object +) -> None: + payload = (json.dumps(value, ensure_ascii=True, sort_keys=True) + "\n").encode() + (root / str(link["path"])).write_bytes(payload) + link["sha256"] = hashlib.sha256(payload).hexdigest() + + +def _rewrite_linked_bytes( + root: Path, link: dict[str, object], payload: bytes +) -> None: + (root / str(link["path"])).write_bytes(payload) + link["sha256"] = hashlib.sha256(payload).hexdigest() + + +def _media_path(manifest: dict[str, object], binding: str) -> str: + item = manifest["inputs"][0] + paths = { + "source": item["source_path"], + "fragment": item["fragment"]["path"], + "fragment-sidecar": item["fragment"]["sidecar_path"], + "reference": item["fragment"]["reference_path"], + "full-recording": item["full_recording"]["path"], + "full-sidecar": item["full_recording"]["sidecar_path"], + } + return str(paths[binding]) + + +def _set_binding_path( + manifest: dict[str, object], binding: str, value: str +) -> None: + if binding in {"owner-package", "owner-acceptance"}: + link = "owner_package" if binding == "owner-package" else "approval" + manifest["fresh_basis"][link]["path"] = value + return + item = manifest["inputs"][0] + if binding == "source": + item["source_path"] = value + elif binding == "fragment": + item["fragment"]["path"] = value + elif binding == "fragment-sidecar": + item["fragment"]["sidecar_path"] = value + elif binding == "reference": + item["fragment"]["reference_path"] = value + elif binding == "full-recording": + item["full_recording"]["path"] = value + elif binding == "full-sidecar": + item["full_recording"]["sidecar_path"] = value + else: + raise AssertionError(f"unknown binding: {binding}") + + +def _mutate_owner_acceptance( + manifest: dict[str, object], root: Path, case: str +) -> None: + basis = manifest["fresh_basis"] + approval_link = basis["approval"] + package_link = basis["owner_package"] + acceptance = _read_linked_json(root, approval_link) + package = _read_linked_json(root, package_link) + + def write_acceptance() -> None: + _rewrite_linked_json(root, approval_link, acceptance) + + def write_package(*, sync_acceptance: bool = True) -> None: + _rewrite_linked_json(root, package_link, package) + if sync_acceptance: + acceptance["owner_package_sha256"] = package_link["sha256"] + write_acceptance() + + if case == "approval-missing": + del basis["approval"] + elif case == "approval-null": + basis["approval"] = None + elif case == "approval-link-unknown": + approval_link["count_review_status"] = "approved" + elif case == "record-missing": + (root / str(approval_link["path"])).unlink() + elif case == "record-path-missing": + approval_link["path"] = "owner/missing-acceptance.json" + elif case == "record-hash-mismatch": + approval_link["sha256"] = "f" * 64 + elif case == "record-json-invalid": + payload = b"{\n" + (root / str(approval_link["path"])).write_bytes(payload) + approval_link["sha256"] = hashlib.sha256(payload).hexdigest() + elif case == "record-unknown-key": + acceptance["count_review_status"] = "approved" + write_acceptance() + elif case == "record-count-review-schema": + acceptance = { + "schema": "local-transcriber.reference-count-review.v1", + "status": "approved", + } + write_acceptance() + elif case in {"actor", "status", "decision"}: + replacement = { + "actor": "automation", + "status": "draft", + "decision": "rejected", + }[case] + approval_link[case] = replacement + acceptance[case] = replacement + write_acceptance() + elif case == "timestamp-naive": + approval_link["timestamp"] = "2026-09-01T09:00:00" + acceptance["timestamp"] = approval_link["timestamp"] + write_acceptance() + elif case == "timestamp-early": + approval_link["timestamp"] = package_link["generated_at"] + acceptance["timestamp"] = approval_link["timestamp"] + write_acceptance() + elif case == "timestamp-link-mismatch": + acceptance["timestamp"] = "2026-09-01T10:00:00+00:00" + write_acceptance() + elif case == "package-missing": + del basis["owner_package"] + elif case == "package-link-unknown": + package_link["status"] = "approved" + elif case == "package-file-missing": + (root / str(package_link["path"])).unlink() + elif case == "package-hash-mismatch": + package_link["sha256"] = "e" * 64 + elif case == "package-timestamp-mismatch": + package_link["generated_at"] = "2026-09-01T07:00:00+00:00" + elif case == "package-unknown-key": + package["status"] = "approved" + write_package() + elif case == "package-row-binding": + package["references"][0]["final_reference_sha256"] = "d" * 64 + write_package() + elif case == "package-row-omitted": + package["references"].pop() + write_package() + elif case == "package-row-duplicate": + package["references"][2] = copy.deepcopy(package["references"][1]) + write_package() + elif case == "acceptance-row-omitted": + acceptance["references"].pop() + write_acceptance() + elif case == "acceptance-row-duplicate": + acceptance["references"][2] = copy.deepcopy(acceptance["references"][1]) + write_acceptance() + elif case in { + "neutral_alias", + "canonical_fragment_audio_sha256", + "final_reference_sha256", + "full_recording_sha256", + }: + acceptance["references"][0][case] = ( + "different-control" if case == "neutral_alias" else "c" * 64 + ) + write_acceptance() + elif case in { + "fragment_expected_cluster_count", + "full_recording_expected_cluster_count", + }: + acceptance["references"][0][case] = 4 + write_acceptance() + elif case == "fragment-count-float": + acceptance["references"][0]["fragment_expected_cluster_count"] = 2.0 + write_acceptance() + elif case == "fragment-count-bool": + acceptance["references"][0]["fragment_expected_cluster_count"] = True + write_acceptance() + elif case == "fragment-count-string": + acceptance["references"][0]["fragment_expected_cluster_count"] = "2" + write_acceptance() + elif case == "full-count-float": + acceptance["references"][0][ + "full_recording_expected_cluster_count" + ] = 2.0 + write_acceptance() + elif case == "package-alias-type": + package["references"][0]["neutral_alias"] = 1 + write_package() + elif case == "package-hash-type": + package["references"][0]["canonical_fragment_audio_sha256"] = 1 + write_package() + elif case == "acceptance-hash-type": + acceptance["references"][0]["canonical_fragment_audio_sha256"] = 1 + write_acceptance() + elif case in { + "fragment_review_complete", + "full_recording_count_review_complete", + "reference_accepted", + }: + acceptance["references"][0][case] = False + write_acceptance() + elif case == "fragment-review-int": + acceptance["references"][0]["fragment_review_complete"] = 1 + write_acceptance() + elif case == "full-review-int": + acceptance["references"][0][ + "full_recording_count_review_complete" + ] = 1 + write_acceptance() + elif case == "reference-accepted-int": + acceptance["references"][0]["reference_accepted"] = 1 + write_acceptance() + elif case == "acceptance-row-unknown": + acceptance["references"][0]["status"] = "approved" + write_acceptance() + elif case == "owner-package-binding": + acceptance["owner_package_sha256"] = "b" * 64 + write_acceptance() + elif case == "step3-int": + acceptance["step3"]["recording_use_approved"] = 1 + write_acceptance() + elif case.startswith("step3-"): + acceptance["step3"][case.removeprefix("step3-")] = False + write_acceptance() + elif case == "step3-missing": + del acceptance["step3"]["recording_use_approved"] + write_acceptance() + elif case == "step3-unknown": + acceptance["step3"]["count_review_approved"] = True + write_acceptance() + else: + raise AssertionError(f"unknown case: {case}") + + def test_preflight_validates_fresh_exact_schema_before_loading(tmp_path): manifest = _manifest(tmp_path) calls = [] @@ -213,6 +510,605 @@ def test_preflight_validates_fresh_exact_schema_before_loading(tmp_path): assert calls == [] +def test_complete_reference_owner_acceptance_is_required_before_loading(tmp_path): + manifest = _manifest(tmp_path) + calls = [] + + outcome = experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + + assert outcome.state == "valid" + assert calls == ["w0", "candidate"] + + +@pytest.mark.parametrize( + "case", + [ + "approval-missing", + "approval-null", + "approval-link-unknown", + "record-missing", + "record-path-missing", + "record-hash-mismatch", + "record-json-invalid", + "record-unknown-key", + "record-count-review-schema", + "actor", + "status", + "decision", + "timestamp-naive", + "timestamp-early", + "timestamp-link-mismatch", + "package-missing", + "package-link-unknown", + "package-file-missing", + "package-hash-mismatch", + "package-timestamp-mismatch", + "package-unknown-key", + "package-row-binding", + "package-row-omitted", + "package-row-duplicate", + "acceptance-row-omitted", + "acceptance-row-duplicate", + "neutral_alias", + "canonical_fragment_audio_sha256", + "final_reference_sha256", + "full_recording_sha256", + "fragment_expected_cluster_count", + "full_recording_expected_cluster_count", + "fragment-count-float", + "fragment-count-bool", + "fragment-count-string", + "full-count-float", + "package-alias-type", + "package-hash-type", + "acceptance-hash-type", + "fragment_review_complete", + "full_recording_count_review_complete", + "reference_accepted", + "fragment-review-int", + "full-review-int", + "reference-accepted-int", + "acceptance-row-unknown", + "owner-package-binding", + "step3-recording_use_approved", + "step3-inputs_and_counts_approved", + "step3-full_recordings_available", + "step3-final_hearing_gate_committed", + "step3-int", + "step3-missing", + "step3-unknown", + ], +) +def test_reference_owner_acceptance_failures_need_user_before_loaders( + tmp_path, case +): + manifest = _manifest(tmp_path) + _mutate_owner_acceptance(manifest, tmp_path, case) + calls = [] + + outcome = experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + + assert outcome.state == "needs-user" + assert outcome.reason == "reference-owner-acceptance" + assert calls == [] + + +@pytest.mark.parametrize( + ("record", "original", "replacement"), + [ + ( + "approval", + b'"status": "approved"', + b'"status": "draft", "status": "approved"', + ), + ( + "approval", + b'"reference_accepted": true', + b'"reference_accepted": false, "reference_accepted": true', + ), + ( + "approval", + b'"recording_use_approved": true', + b'"recording_use_approved": false, "recording_use_approved": true', + ), + ( + "owner_package", + b'"neutral_alias": "control-1"', + b'"neutral_alias": "other", "neutral_alias": "control-1"', + ), + ( + "approval", + b'"fragment_expected_cluster_count": 2', + b'"fragment_expected_cluster_count": NaN', + ), + ], +) +def test_bound_owner_json_rejects_ambiguous_json_before_loaders( + tmp_path, record, original, replacement +): + manifest = _manifest(tmp_path) + link = manifest["fresh_basis"][record] + path = tmp_path / str(link["path"]) + payload = path.read_bytes().replace(original, replacement, 1) + assert payload != path.read_bytes() + _rewrite_linked_bytes(tmp_path, link, payload) + if record == "owner_package": + acceptance_link = manifest["fresh_basis"]["approval"] + acceptance = _read_linked_json(tmp_path, acceptance_link) + acceptance["owner_package_sha256"] = link["sha256"] + _rewrite_linked_json(tmp_path, acceptance_link, acceptance) + calls = [] + + outcome = experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + + assert outcome.state == "needs-user" + assert outcome.reason == "reference-owner-acceptance" + assert calls == [] + + +@pytest.mark.parametrize("symlink_kind", ["final", "intermediate"]) +def test_bound_owner_json_rejects_symlinks_before_loaders( + tmp_path, symlink_kind +): + manifest = _manifest(tmp_path) + if symlink_kind == "final": + link = manifest["fresh_basis"]["approval"] + path = tmp_path / str(link["path"]) + target = path.with_name("acceptance-target.json") + path.rename(target) + path.symlink_to(target.name) + else: + owner = tmp_path / "owner" + target = tmp_path / "owner-target" + owner.rename(target) + owner.symlink_to(target.name, target_is_directory=True) + calls = [] + + outcome = experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + + assert outcome.state == "needs-user" + assert outcome.reason == "reference-owner-acceptance" + assert calls == [] + + +def test_bound_owner_json_cannot_hash_one_version_and_parse_another( + tmp_path, monkeypatch +): + manifest = _manifest(tmp_path) + link = manifest["fresh_basis"]["approval"] + path = tmp_path / str(link["path"]) + approved_payload = path.read_bytes() + rejected_payload = approved_payload.replace( + b'"status": "approved"', b'"status": "draft"', 1 + ) + path.write_bytes(rejected_payload) + link["sha256"] = hashlib.sha256(rejected_payload).hexdigest() + original_file_sha256 = experiment.file_sha256 + replaced = False + + def replace_after_hash(candidate): + nonlocal replaced + digest = original_file_sha256(candidate) + if candidate == path and not replaced: + path.write_bytes(approved_payload) + replaced = True + return digest + + monkeypatch.setattr(experiment, "file_sha256", replace_after_hash) + calls = [] + + outcome = experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + + assert outcome.state == "needs-user" + assert outcome.reason == "reference-owner-acceptance" + assert calls == [] + + +def test_bound_owner_json_rejects_path_replacement_during_held_read( + tmp_path, monkeypatch +): + manifest = _manifest(tmp_path) + link = manifest["fresh_basis"]["approval"] + path = tmp_path / str(link["path"]) + target_inode = path.stat().st_ino + replacement = path.with_name("replacement.json") + payload = path.read_bytes() + original_read = os.read + replaced = False + + def replace_path_after_read(file_descriptor, count): + nonlocal replaced + chunk = original_read(file_descriptor, count) + if os.fstat(file_descriptor).st_ino == target_inode and not replaced: + replacement.write_bytes(payload) + os.replace(replacement, path) + replaced = True + return chunk + + monkeypatch.setattr(experiment.os, "read", replace_path_after_read) + calls = [] + + outcome = experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + + assert replaced is True + assert outcome.state == "needs-user" + assert outcome.reason == "reference-owner-acceptance" + assert calls == [] + + +def test_bound_owner_json_rejects_in_place_mutation_during_held_read( + tmp_path, monkeypatch +): + manifest = _manifest(tmp_path) + link = manifest["fresh_basis"]["approval"] + path = tmp_path / str(link["path"]) + json_size = len(path.read_bytes().rstrip(b"\n")) + payload = path.read_bytes().rstrip(b"\n") + (b" " * 70_000) + b"\n" + _rewrite_linked_bytes(tmp_path, link, payload) + target_inode = path.stat().st_ino + original_read = os.read + mutated = False + + def mutate_after_first_chunk(file_descriptor, count): + nonlocal mutated + chunk = original_read(file_descriptor, count) + if os.fstat(file_descriptor).st_ino == target_inode and not mutated: + writer = os.open(path, os.O_WRONLY) + try: + os.pwrite(writer, b"\t", json_size + 66_000) + os.fsync(writer) + finally: + os.close(writer) + mutated = True + return chunk + + monkeypatch.setattr(experiment.os, "read", mutate_after_first_chunk) + calls = [] + + outcome = experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + + assert mutated is True + assert outcome.state == "needs-user" + assert outcome.reason == "reference-owner-acceptance" + assert calls == [] + + +def test_bound_owner_json_rejects_intermediate_directory_swap_during_read( + tmp_path, monkeypatch +): + manifest = _manifest(tmp_path) + link = manifest["fresh_basis"]["owner_package"] + path = tmp_path / str(link["path"]) + target_inode = path.stat().st_ino + parent = path.parent + moved_parent = parent.with_name("owner-original") + original_read = os.read + swapped = False + + def swap_directory_after_read(file_descriptor, count): + nonlocal swapped + chunk = original_read(file_descriptor, count) + if os.fstat(file_descriptor).st_ino == target_inode and not swapped: + parent.rename(moved_parent) + parent.mkdir() + (parent / path.name).write_bytes(b"{}\n") + swapped = True + return chunk + + monkeypatch.setattr(experiment.os, "read", swap_directory_after_read) + calls = [] + + outcome = experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + + assert swapped is True + assert outcome.state == "needs-user" + assert outcome.reason == "reference-owner-acceptance" + assert calls == [] + + +@pytest.mark.parametrize( + "binding", + [ + "source", + "fragment", + "fragment-sidecar", + "reference", + "full-recording", + "full-sidecar", + ], +) +@pytest.mark.parametrize("symlink_kind", ["final", "intermediate"]) +def test_manifest_file_bindings_reject_internal_symlinks_before_loaders( + tmp_path, binding, symlink_kind +): + manifest = _manifest(tmp_path) + path = tmp_path / _media_path(manifest, binding) + if symlink_kind == "final": + target = path.with_name(f"{path.name}.target") + path.rename(target) + path.symlink_to(target.name) + else: + parent = path.parent + target = parent.with_name(f"{parent.name}-target") + parent.rename(target) + parent.symlink_to(target.name, target_is_directory=True) + calls = [] + + with pytest.raises(experiment.PreflightError) as raised: + experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + + assert str(tmp_path) not in str(raised.value) + assert calls == [] + + +def test_manifest_file_binding_rejects_final_replacement_during_read( + tmp_path, monkeypatch +): + manifest = _manifest(tmp_path) + path = tmp_path / _media_path(manifest, "fragment") + target_inode = path.stat().st_ino + replacement = path.with_name("replacement.wav") + original_read = os.read + replaced = False + + def replace_path_after_read(file_descriptor, count): + nonlocal replaced + chunk = original_read(file_descriptor, count) + if os.fstat(file_descriptor).st_ino == target_inode and not replaced: + replacement.write_bytes(b"different-fragment-bytes") + os.replace(replacement, path) + replaced = True + return chunk + + monkeypatch.setattr(experiment.os, "read", replace_path_after_read) + calls = [] + + with pytest.raises(experiment.PreflightError): + experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + + assert replaced is True + assert path.read_bytes() == b"different-fragment-bytes" + assert hashlib.sha256(path.read_bytes()).hexdigest() != manifest["inputs"][0][ + "fragment" + ]["sha256"] + assert calls == [] + + +def test_manifest_file_binding_rejects_intermediate_directory_swap_during_read( + tmp_path, monkeypatch +): + manifest = _manifest(tmp_path) + path = tmp_path / _media_path(manifest, "fragment") + target_inode = path.stat().st_ino + parent = path.parent + moved_parent = parent.with_name("media-original") + original_read = os.read + swapped = False + + def swap_directory_after_read(file_descriptor, count): + nonlocal swapped + chunk = original_read(file_descriptor, count) + if os.fstat(file_descriptor).st_ino == target_inode and not swapped: + parent.rename(moved_parent) + parent.mkdir() + (parent / path.name).write_bytes(b"replacement-fragment") + swapped = True + return chunk + + monkeypatch.setattr(experiment.os, "read", swap_directory_after_read) + calls = [] + + with pytest.raises(experiment.PreflightError): + experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + + assert swapped is True + assert calls == [] + + +def test_manifest_file_binding_rejects_in_place_mutation_during_read( + tmp_path, monkeypatch +): + manifest = _manifest(tmp_path) + path = tmp_path / _media_path(manifest, "fragment") + target_inode = path.stat().st_ino + original_read = os.read + mutated = False + + def mutate_file_after_read(file_descriptor, count): + nonlocal mutated + chunk = original_read(file_descriptor, count) + if os.fstat(file_descriptor).st_ino == target_inode and not mutated: + path.write_bytes(b"mutated-fragment-with-different-size") + mutated = True + return chunk + + monkeypatch.setattr(experiment.os, "read", mutate_file_after_read) + calls = [] + + with pytest.raises(experiment.PreflightError): + experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + + assert mutated is True + assert calls == [] + + +def test_bound_file_verification_closes_descriptors_on_success_and_failure( + tmp_path +): + proc_fds = Path("/proc/self/fd") + if not proc_fds.is_dir(): + pytest.skip("fd accounting requires procfs") + + manifest = _manifest(tmp_path) + before_success = len(list(proc_fds.iterdir())) + for _ in range(20): + assert experiment.validate_manifest( + manifest, root=tmp_path, environment=_environment() + ).state == "valid" + assert len(list(proc_fds.iterdir())) == before_success + + failed_root = tmp_path / "failed" + failed_manifest = _manifest(failed_root) + path = failed_root / _media_path(failed_manifest, "fragment") + target = path.with_name("fragment-target.wav") + path.rename(target) + path.symlink_to(target.name) + before_failure = len(list(proc_fds.iterdir())) + for _ in range(20): + with pytest.raises(experiment.PreflightError): + experiment.validate_manifest( + failed_manifest, + root=failed_root, + environment=_environment(), + ) + assert len(list(proc_fds.iterdir())) == before_failure + + +@pytest.mark.parametrize( + "binding", + [ + "source", + "fragment", + "fragment-sidecar", + "reference", + "full-recording", + "full-sidecar", + "owner-package", + "owner-acceptance", + ], +) +def test_manifest_file_bindings_reject_nul_paths_before_loaders( + tmp_path, binding +): + manifest = _manifest(tmp_path) + _set_binding_path(manifest, binding, "private\x00binding") + calls = [] + + if binding.startswith("owner-"): + outcome = experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + assert outcome.state == "needs-user" + assert outcome.reason == "reference-owner-acceptance" + else: + with pytest.raises(experiment.PreflightError, match="path-escape"): + experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + + assert calls == [] + + +def test_manifest_file_binding_redacts_path_layer_value_error( + tmp_path, monkeypatch +): + manifest = _manifest(tmp_path) + original_open = os.open + + def reject_component(path, flags, *, dir_fd=None): + if path == "sources": + raise ValueError("private path detail") + if dir_fd is None: + return original_open(path, flags) + return original_open(path, flags, dir_fd=dir_fd) + + monkeypatch.setattr(experiment.os, "open", reject_component) + calls = [] + + with pytest.raises(experiment.PreflightError) as raised: + experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + + assert str(raised.value) == "path-escape" + assert "private" not in str(raised.value) + assert calls == [] + + def test_missing_owner_approval_is_needs_user_and_never_loads(tmp_path): manifest = _manifest(tmp_path) manifest["fresh_basis"]["approval"] = None @@ -227,7 +1123,7 @@ def test_missing_owner_approval_is_needs_user_and_never_loads(tmp_path): ) assert outcome.state == "needs-user" - assert outcome.reason == "owner-approval-required" + assert outcome.reason == "reference-owner-acceptance" assert calls == [] @@ -244,6 +1140,10 @@ def test_missing_owner_approval_is_needs_user_and_never_loads(tmp_path): lambda value: value["inputs"][0]["full_recording"].update( expected_cluster_count=5 ), + lambda value: value["inputs"][0].update(neutral_alias=""), + lambda value: value["inputs"][1].update( + neutral_alias=value["inputs"][0]["neutral_alias"] + ), lambda value: value["source"].update(commit="b" * 40), ], ) @@ -1307,6 +2207,35 @@ def test_invalid_cli_output_has_one_trailing_newline(tmp_path, capsys): assert not output.endswith("\n\n") +def test_validate_only_cli_redacts_nul_path_failure( + tmp_path, capsys, monkeypatch +): + manifest = _manifest(tmp_path) + manifest["inputs"][0]["source_path"] = "private\x00source.wav" + manifest_path = tmp_path / "manifest.json" + manifest_path.write_text(json.dumps(manifest), encoding="utf-8") + monkeypatch.setattr( + experiment, + "git_environment_probe", + lambda _repository_root: _environment(), + ) + + with pytest.raises(SystemExit) as raised: + experiment.main(["--manifest", str(manifest_path), "--validate-only"]) + + captured = capsys.readouterr() + assert raised.value.code == 2 + assert json.loads(captured.out) == { + "reason": "path-escape", + "state": "invalid", + } + assert captured.err == "" + assert "Traceback" not in captured.out + assert "\x00" not in captured.out + assert "private" not in captured.out + assert str(tmp_path) not in captured.out + + def test_child_process_request_result_seam_uses_atomic_private_files(tmp_path): request = { "semantic_id": "cell-safe",