diff --git a/docs/SANITIZED_ARTIFACTS.md b/docs/SANITIZED_ARTIFACTS.md index c0802cba..c7da34b4 100644 --- a/docs/SANITIZED_ARTIFACTS.md +++ b/docs/SANITIZED_ARTIFACTS.md @@ -74,6 +74,35 @@ the review, and `compile`, `validate-hosted`, and `push` all read that copy as unapproved. To move an approved derivative, copy the whole directory, hidden approval file included, next to its sibling ZIP. +Where only the archive travels -- a CI runner that downloads it from object +storage, for example -- carry the approval record beside it and rebuild the +pair with `materialize-approved`: + +```bash +openadapt-flow materialize-approved \ + --archive recording.approved.zip \ + --approval recording.approval.json \ + --out recording/ \ + --expect-archive-sha256 +``` + +It extracts the archive into a new empty directory, places the record, and runs +the same approval gate `push` runs. It approves nothing: the record must +already exist and must match the extracted bytes. `--approval` takes either the +derivative's `.openadapt-approval.json` or the +[`openadapt.sanitization/v1`](../schemas/sanitization-ingest-v1.json) ingest +envelope a deployment persisted for that artifact. The envelope carries every +field that holds authority -- archive SHA-256 and size, reviewer, approval +time, approval method -- while the two tree hashes are functions of the +extracted bytes that the archive SHA-256 already binds, so they are recomputed +rather than transported. An envelope recording an automatic approval must carry +its policy key id and MAC, and the MAC is verified wherever +`OPENADAPT_SANITIZATION_POLICY_KEY` is configured. + +The record is unsigned, exactly as it is on the reviewing machine. It is +therefore only as trustworthy as the channel that delivered it, and that +channel should be the one that already delivers the archive. + ## Destination trust Execution lane and egress destination are independent: diff --git a/openadapt_flow/__main__.py b/openadapt_flow/__main__.py index 3ba2ce04..9bc990f1 100644 --- a/openadapt_flow/__main__.py +++ b/openadapt_flow/__main__.py @@ -4320,6 +4320,30 @@ def _cmd_approve_sanitized(args: argparse.Namespace) -> int: return 0 +def _cmd_materialize_approved(args: argparse.Namespace) -> int: + from openadapt_flow.sanitized_artifact import ( + SanitizationError, + materialize_approved_derivative, + ) + + try: + approval = materialize_approved_derivative( + Path(args.archive), + approval=Path(args.approval), + destination=Path(args.out), + expected_archive_sha256=args.expect_archive_sha256, + ) + except SanitizationError as e: + print(f"materialize failed: {e}") + return 1 + print( + f"Materialized the approved derivative into {args.out}; " + f"reviewer={approval['reviewer']} " + f"sha256={approval['approved_derivative_sha256']}." + ) + return 0 + + def _cmd_report_break(args: argparse.Namespace) -> int: """Emit a PHI-free break diagnostic from a halted run's ``report.json``. @@ -6524,6 +6548,42 @@ def _repair_store_flag(rp: argparse.ArgumentParser) -> None: ) p.set_defaults(func=_cmd_review_sanitized) + p = sub.add_parser( + "materialize-approved", + help="Rebuild an approved derivative from its archive and approval record", + description=( + "Approval binds the archive's own SHA-256, so the archive cannot " + "carry it and an archive unzipped elsewhere reads as unapproved. " + "This places the reviewer's existing approval record beside the " + "extracted bytes and then runs the unchanged approval gate. It " + "approves nothing." + ), + ) + p.add_argument( + "--archive", required=True, help="Approved immutable archive (.approved.zip)" + ) + p.add_argument( + "--approval", + required=True, + help=( + "Existing approval record: either the derivative's " + ".openadapt-approval.json or the openadapt.sanitization/v1 ingest " + "envelope a deployment persisted for it" + ), + ) + p.add_argument( + "--out", required=True, help="New, empty derivative directory to materialize" + ) + p.add_argument( + "--expect-archive-sha256", + default=None, + help=( + "Pin the archive to a digest the caller already trusts, checked " + "independently of the approval record" + ), + ) + p.set_defaults(func=_cmd_materialize_approved) + p = sub.add_parser( "approve-sanitized", help="Approve and freeze the exact reviewed derivative as an immutable archive", diff --git a/openadapt_flow/sanitized_artifact.py b/openadapt_flow/sanitized_artifact.py index 1229e48d..b76b806a 100644 --- a/openadapt_flow/sanitized_artifact.py +++ b/openadapt_flow/sanitized_artifact.py @@ -31,7 +31,7 @@ from datetime import datetime, timezone from functools import lru_cache from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer -from pathlib import Path +from pathlib import Path, PurePosixPath from typing import Any, Optional from urllib.parse import parse_qs @@ -1194,6 +1194,281 @@ def load_valid_approval(destination: Path) -> dict[str, Any]: return approval +def _policy_signature_message(envelope: dict[str, Any], key_id: str) -> str: + """Canonicalize the policy-approval MAC input for one ingest envelope.""" + return json.dumps( + [ + "openadapt.sanitization-policy/v1", + envelope["artifact"]["kind"], + envelope["artifact"]["sha256"], + envelope["artifact"]["size_bytes"], + envelope["artifact"]["execution_semantics"], + envelope["artifact"]["runtime_semantics_validated"], + envelope["artifact"]["trusted_boundary_required_at_runtime"], + envelope["scrubber"]["name"], + envelope["scrubber"]["version"], + envelope["scrubber"]["policy"], + sorted(envelope["coverage"]["media_types"]), + envelope["approval"]["approved_at"], + envelope["approval"]["approved_by"], + key_id, + ], + ensure_ascii=False, + separators=(",", ":"), + ) + + +def _extract_approved_archive(archive: Path, destination: Path) -> None: + """Extract an approved archive without trusting any member name.""" + root = Path(os.path.realpath(destination)) + try: + handle = zipfile.ZipFile(archive) + except (OSError, zipfile.BadZipFile) as exc: + raise SanitizationError(f"Approved archive is unreadable: {exc}") from exc + with handle as zf: + for info in zf.infolist(): + name = info.filename + rel = PurePosixPath(name) + if ( + info.is_dir() + or rel.is_absolute() + or "\\" in name + or not rel.parts + or any(part in {"", ".", ".."} for part in rel.parts) + ): + raise SanitizationError( + f"Approved archive holds an unsafe path: {name!r}" + ) + if (info.external_attr >> 16) & 0o170000 == 0o120000: + raise SanitizationError(f"Approved archive holds a symlink: {name!r}") + if rel.name == APPROVAL_NAME: + raise SanitizationError( + "Approved archive must not carry its own approval" + ) + target = root.joinpath(*rel.parts) + if os.path.commonpath([str(root), str(target)]) != str(root): + raise SanitizationError( + f"Approved archive holds an unsafe path: {name!r}" + ) + target.parent.mkdir(parents=True, exist_ok=True) + with zf.open(info) as src, open(target, "wb") as dst: + shutil.copyfileobj(src, dst) + target.chmod(0o600) + + +def _approval_from_ingest_envelope( + envelope: dict[str, Any], manifest: dict[str, Any] +) -> dict[str, Any]: + """Rebuild the reviewer's approval record from the persisted envelope. + + A cloud deployment stores ``openadapt.sanitization/v1`` -- the public + envelope that ``build_ingest_manifest`` derives from an approval -- and not + the private approval file itself. Every field that carries authority + (archive SHA-256, archive size, reviewer, approval time, method) comes from + that stored envelope. The two tree hashes are functions of the extracted + bytes, which the archive SHA-256 already binds, so they are recomputed here + rather than transported. + """ + if envelope.get("schema") != "openadapt.sanitization/v1": + raise SanitizationError("Unsupported approval record schema") + sections: dict[str, dict[str, Any]] = {} + for section in ("artifact", "approval", "scrubber", "coverage"): + value = envelope.get(section) + if not isinstance(value, dict): + raise SanitizationError(f"Ingest envelope has no {section} section") + sections[section] = value + artifact = sections["artifact"] + approval = sections["approval"] + scrubber = sections["scrubber"] + coverage = sections["coverage"] + if approval.get("status") != "approved": + raise SanitizationError("Ingest envelope does not record an approval") + method = approval.get("method") + if method not in {"human", "policy"}: + raise SanitizationError("Approval method must be human or policy") + reviewer = str(approval.get("approved_by", "")).strip() + if not reviewer: + raise SanitizationError("Reviewer identity must not be empty") + approved_at = approval.get("approved_at") + if not isinstance(approved_at, str) or not approved_at.strip(): + raise SanitizationError("Approval time must not be empty") + archive_sha256 = artifact.get("sha256") + if not isinstance(archive_sha256, str) or not re.fullmatch( + r"[a-f0-9]{64}", archive_sha256 + ): + raise SanitizationError("Ingest envelope has no artifact SHA-256") + if approval.get("artifact_sha256") != archive_sha256: + raise SanitizationError("Approval is not bound to the envelope's artifact") + size_bytes = artifact.get("size_bytes") + if not isinstance(size_bytes, int) or isinstance(size_bytes, bool): + raise SanitizationError("Ingest envelope has no artifact size") + + # Bind the envelope to this exact derivative, beyond the archive digest. + if artifact.get("kind") != manifest["kind"]: + raise SanitizationError("Ingest envelope describes a different artifact kind") + if scrubber.get("policy") != manifest["policy_version"]: + raise SanitizationError("Ingest envelope records a different policy version") + if artifact.get("execution_semantics") != manifest["execution_semantics"]: + raise SanitizationError("Ingest envelope records different execution semantics") + if bool(coverage.get("complete")) != bool(manifest["coverage_complete"]): + raise SanitizationError("Ingest envelope records different coverage") + + if method == "policy": + _verify_policy_approval_signature(envelope) + + verification = manifest.get("approval_verification") + if not ( + isinstance(verification, dict) + and verification.get("method") == "full-stable-rescan" + and verification.get("file_count") == len(manifest["files"]) + and verification.get("unresolved") == 0 + and isinstance(verification.get("verified_at"), str) + ): + raise SanitizationError( + "Derivative carries no approval rescan; it was never approved" + ) + return { + "schema_version": SCHEMA_VERSION, + "approved_at": approved_at, + "reviewer": reviewer, + "automatic": method == "policy", + "policy_version": manifest["policy_version"], + "manifest_sha256": _manifest_hash(manifest), + "derivative_tree_sha256": manifest["derivative_tree_sha256"], + "approved_derivative_sha256": archive_sha256, + "approved_archive_size_bytes": size_bytes, + "verification": { + "verified_at": verification["verified_at"], + "method": "full-stable-rescan", + "source_tree_verified": True, + "file_count": len(manifest["files"]), + "unresolved": 0, + }, + } + + +def _verify_policy_approval_signature(envelope: dict[str, Any]) -> None: + """Check the MAC that an automatic approval must carry. + + The deployment that stored the envelope verified this MAC at ingest. The + key is not present on every consumer, so the structural requirement is + unconditional and the cryptographic check runs wherever the key is + configured. + """ + approval = envelope["approval"] + key_id = approval.get("policy_key_id") + signature = approval.get("policy_signature") + if not isinstance(key_id, str) or not re.fullmatch( + r"[A-Za-z0-9][A-Za-z0-9._:+-]{0,99}", key_id + ): + raise SanitizationError("A policy approval must carry a valid policy key id") + if not isinstance(signature, str) or not re.fullmatch(r"[a-f0-9]{64}", signature): + raise SanitizationError("A policy approval must carry its approval signature") + local_key_id = os.environ.get(POLICY_KEY_ID_ENV, "").strip() + encoded_key = os.environ.get(POLICY_KEY_ENV, "").strip() + if not encoded_key: + return + if local_key_id != key_id: + raise SanitizationError( + f"A policy approval names key {key_id!r}, which {POLICY_KEY_ID_ENV} " + "does not configure" + ) + try: + key = base64.b64decode(encoded_key, validate=True) + except (ValueError, binascii.Error) as exc: + raise SanitizationError(f"{POLICY_KEY_ENV} must be base64") from exc + expected = hmac.new( + key, + _policy_signature_message(envelope, key_id).encode("utf-8"), + hashlib.sha256, + ).hexdigest() + if not hmac.compare_digest(expected, signature): + raise SanitizationError("The policy approval signature does not verify") + + +def materialize_approved_derivative( + archive: Path, + *, + approval: Path, + destination: Path, + expected_archive_sha256: Optional[str] = None, +) -> dict[str, Any]: + """Rebuild an approved derivative from its archive and its approval record. + + ``approve_derivative`` computes the approval over the archive's own bytes, + so the archive cannot contain it. Unzipping the archive somewhere else + therefore produces the reviewed bytes without the review, and ``compile``, + ``validate-hosted``, and ``push`` correctly read that copy as unapproved. + This carries the approval record beside the archive, restores the exact + on-disk shape the gate expects, and then runs the unchanged + ``load_valid_approval``. + + Nothing here approves anything. ``approval`` must already exist, as either + the ``.openadapt-approval.json`` record or the + ``openadapt.sanitization/v1`` ingest envelope a deployment persists, and it + must match the extracted bytes. Pass ``expected_archive_sha256`` to pin the + archive against a digest the caller trusts, independently of the record. + """ + archive = Path(archive) + record_path = Path(approval) + destination = Path(destination) + if not archive.is_file(): + raise SanitizationError(f"Approved archive is missing: {archive}") + if not record_path.is_file(): + raise SanitizationError(f"Approval record is missing: {record_path}") + if expected_archive_sha256 is not None: + expected = expected_archive_sha256.strip().lower() + if not re.fullmatch(r"[a-f0-9]{64}", expected): + raise SanitizationError( + "The expected sha256 must be 64 hexadecimal characters" + ) + if _sha256_file(archive) != expected: + raise SanitizationError( + "The approved archive does not match the expected sha256" + ) + try: + record = json.loads(record_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise SanitizationError(f"Invalid approval: {exc}") from exc + if not isinstance(record, dict): + raise SanitizationError("An approval record must be a JSON object") + if destination.exists(): + if not destination.is_dir(): + raise SanitizationError( + f"Materialization destination is not a directory: {destination}" + ) + if any(destination.iterdir()): + raise SanitizationError( + f"Materialization destination is not empty: {destination}" + ) + target_archive = approved_archive_path(destination) + copied_archive = os.path.realpath(target_archive) != os.path.realpath(archive) + if copied_archive and target_archive.exists(): + raise SanitizationError( + f"An archive already sits beside the destination: {target_archive}" + ) + destination.mkdir(parents=True, exist_ok=True) + destination.chmod(0o700) + try: + _extract_approved_archive(archive, destination) + if copied_archive: + shutil.copyfile(archive, target_archive) + target_archive.chmod(0o600) + if "schema" in record: + manifest = load_and_verify_derivative(destination) + record = _approval_from_ingest_envelope(record, manifest) + approval_path(destination).write_text( + json.dumps(record, indent=2, sort_keys=True) + "\n", encoding="utf-8" + ) + approval_path(destination).chmod(0o600) + return load_valid_approval(destination) + except SanitizationError: + approval_path(destination).unlink(missing_ok=True) + if copied_archive: + target_archive.unlink(missing_ok=True) + raise + + def build_ingest_manifest(destination: Path) -> dict[str, Any]: """Build the public ``openadapt.sanitization/v1`` cloud ingest envelope.""" destination = Path(destination) @@ -1253,26 +1528,7 @@ def build_ingest_manifest(destination: Path) -> dict[str, Any]: f"Automatic approval requires {POLICY_KEY_ENV} to decode to at least 32 bytes" ) envelope["approval"]["policy_key_id"] = key_id - message = json.dumps( - [ - "openadapt.sanitization-policy/v1", - envelope["artifact"]["kind"], - envelope["artifact"]["sha256"], - envelope["artifact"]["size_bytes"], - envelope["artifact"]["execution_semantics"], - envelope["artifact"]["runtime_semantics_validated"], - envelope["artifact"]["trusted_boundary_required_at_runtime"], - envelope["scrubber"]["name"], - envelope["scrubber"]["version"], - envelope["scrubber"]["policy"], - sorted(envelope["coverage"]["media_types"]), - envelope["approval"]["approved_at"], - envelope["approval"]["approved_by"], - key_id, - ], - ensure_ascii=False, - separators=(",", ":"), - ) + message = _policy_signature_message(envelope, key_id) envelope["approval"]["policy_signature"] = hmac.new( key, message.encode("utf-8"), hashlib.sha256 ).hexdigest() diff --git a/tests/test_sanitized_artifact.py b/tests/test_sanitized_artifact.py index a7162de8..e365a652 100644 --- a/tests/test_sanitized_artifact.py +++ b/tests/test_sanitized_artifact.py @@ -42,6 +42,7 @@ build_ingest_manifest, load_and_verify_derivative, load_valid_approval, + materialize_approved_derivative, render_review_html, sanitize_artifact, ) @@ -954,3 +955,290 @@ def test_emitted_manifest_approval_and_ingest_envelope_match_published_schemas( jsonschema.Draft202012Validator(json.loads((root / name).read_text())).validate( instance ) + + +def _approved_recording(tmp_path: Path) -> tuple[Path, dict, Path]: + """Approve a recording derivative and return it with its archive.""" + source = _recording(tmp_path) + dest = tmp_path / "sanitized" + sanitize_artifact(source, dest, kind="recording") + approval = approve_derivative(dest, source=source, reviewer="alice") + return dest, approval, approved_archive_path(dest) + + +def test_unzipping_an_approved_archive_alone_yields_an_unapproved_derivative(tmp_path): + """The archive cannot carry the approval computed over its own bytes.""" + dest, _, archive = _approved_recording(tmp_path) + with zipfile.ZipFile(archive) as zf: + assert APPROVAL_NAME not in zf.namelist() + + elsewhere = tmp_path / "elsewhere" / "recording" + elsewhere.mkdir(parents=True) + with zipfile.ZipFile(archive) as zf: + zf.extractall(elsewhere) + (elsewhere.parent / "recording.approved.zip").write_bytes(archive.read_bytes()) + + assert load_and_verify_derivative(elsewhere)["kind"] == "recording" + with pytest.raises(SanitizationError, match="not been approved"): + load_valid_approval(elsewhere) + + +def test_materialize_approved_restores_the_unchanged_approval_gate(tmp_path): + dest, approval, archive = _approved_recording(tmp_path) + record = tmp_path / "approval.json" + record.write_text(json.dumps(approval), encoding="utf-8") + out = tmp_path / "elsewhere" / "recording" + + materialized = materialize_approved_derivative( + archive, + approval=record, + destination=out, + expected_archive_sha256=approval["approved_derivative_sha256"], + ) + + assert materialized == approval + assert load_valid_approval(out) == approval + assert (out / APPROVAL_NAME).is_file() + assert approved_archive_path(out).read_bytes() == archive.read_bytes() + assert { + str(path.relative_to(out)) + for path in out.rglob("*") + if path.is_file() and path.name != APPROVAL_NAME + } == { + MANIFEST_NAME, + "meta.json", + "events.jsonl", + str(Path("frames") / "before.png"), + } + + +def test_materialize_approved_accepts_the_persisted_ingest_envelope(tmp_path): + """A cloud deployment persists the envelope, not the approval file.""" + dest, approval, archive = _approved_recording(tmp_path) + envelope = build_ingest_manifest(dest) + record = tmp_path / "envelope.json" + record.write_text(json.dumps(envelope), encoding="utf-8") + out = tmp_path / "elsewhere" / "recording" + + materialized = materialize_approved_derivative( + archive, approval=record, destination=out + ) + + assert materialized["reviewer"] == approval["reviewer"] == "alice" + assert materialized["approved_at"] == approval["approved_at"] + assert materialized["automatic"] is False + assert ( + materialized["approved_derivative_sha256"] + == approval["approved_derivative_sha256"] + ) + assert load_valid_approval(out)["reviewer"] == "alice" + + +def test_materialized_recording_gives_compile_its_source_provenance(tmp_path): + """The runner built a bundle with null provenance from a bare archive.""" + source = tmp_path / "recording" + frames = source / "frames" + frames.mkdir(parents=True) + before = Image.new("RGB", (320, 200), "white") + ImageDraw.Draw(before).rectangle((90, 70, 230, 130), outline="black", width=4) + after = before.copy() + ImageDraw.Draw(after).rectangle((250, 20, 290, 50), fill="green") + before.save(frames / "0000_before.png", format="PNG") + after.save(frames / "0000_after.png", format="PNG") + (source / "events.jsonl").write_text( + json.dumps({"i": 0, "kind": "click", "x": 160, "y": 100, "t": 1.0}) + "\n" + ) + (source / "meta.json").write_text( + json.dumps( + { + "id": "materialized-recording", + "created_at": "2026-07-15T00:00:00+00:00", + "viewport": [320, 200], + "app_url": "http://example.test/", + "params": {}, + } + ) + ) + dest = tmp_path / "sanitized" + sanitize_artifact(source, dest, kind="recording") + approval = approve_derivative(dest, source=source, reviewer="alice") + + bare = tmp_path / "bare" / "recording" + bare.mkdir(parents=True) + with zipfile.ZipFile(approved_archive_path(dest)) as zf: + zf.extractall(bare) + bare_workflow = compile_recording(bare, tmp_path / "bare-bundle", name="bare") + assert bare_workflow.manifest is not None + assert bare_workflow.manifest.provenance.source_recording_sha256 is None + + out = tmp_path / "elsewhere" / "recording" + record = tmp_path / "approval.json" + record.write_text(json.dumps(build_ingest_manifest(dest)), encoding="utf-8") + materialize_approved_derivative( + approved_archive_path(dest), approval=record, destination=out + ) + workflow = compile_recording(out, tmp_path / "bundle", name="materialized") + + assert workflow.manifest is not None + assert ( + workflow.manifest.provenance.source_recording_sha256 + == approval["approved_derivative_sha256"] + ) + + +def test_materialize_approved_refuses_a_tampered_archive(tmp_path): + dest, approval, archive = _approved_recording(tmp_path) + record = tmp_path / "approval.json" + record.write_text(json.dumps(approval), encoding="utf-8") + tampered = tmp_path / "tampered.approved.zip" + with zipfile.ZipFile(archive) as src, zipfile.ZipFile(tampered, "w") as dst: + for info in src.infolist(): + payload = src.read(info.filename) + if info.filename == "events.jsonl": + payload = b'{"text":"substituted"}\n' + dst.writestr(info, payload) + + with pytest.raises(SanitizationError): + materialize_approved_derivative( + tampered, approval=record, destination=tmp_path / "out" + ) + assert not (tmp_path / "out" / APPROVAL_NAME).exists() + + +def test_materialize_approved_refuses_an_unexpected_archive_digest(tmp_path): + dest, approval, archive = _approved_recording(tmp_path) + record = tmp_path / "approval.json" + record.write_text(json.dumps(approval), encoding="utf-8") + + with pytest.raises(SanitizationError, match="expected sha256"): + materialize_approved_derivative( + archive, + approval=record, + destination=tmp_path / "out", + expected_archive_sha256="0" * 64, + ) + assert not (tmp_path / "out").exists() or not any((tmp_path / "out").iterdir()) + + +def test_materialize_approved_refuses_an_approval_for_another_artifact(tmp_path): + dest, approval, archive = _approved_recording(tmp_path) + other = dict(approval) + other["approved_derivative_sha256"] = "f" * 64 + record = tmp_path / "approval.json" + record.write_text(json.dumps(other), encoding="utf-8") + + with pytest.raises(SanitizationError, match="archive"): + materialize_approved_derivative( + archive, approval=record, destination=tmp_path / "out" + ) + + +def test_materialize_approved_refuses_a_traversing_archive_member(tmp_path): + dest, approval, archive = _approved_recording(tmp_path) + record = tmp_path / "approval.json" + record.write_text(json.dumps(approval), encoding="utf-8") + hostile = tmp_path / "hostile.approved.zip" + with zipfile.ZipFile(hostile, "w") as zf: + zf.writestr("../escaped.json", "{}") + + with pytest.raises(SanitizationError, match="unsafe path"): + materialize_approved_derivative( + hostile, approval=record, destination=tmp_path / "out" + ) + assert not (tmp_path / "escaped.json").exists() + + +def test_materialize_approved_refuses_an_archive_carrying_its_own_approval(tmp_path): + dest, approval, archive = _approved_recording(tmp_path) + record = tmp_path / "approval.json" + record.write_text(json.dumps(approval), encoding="utf-8") + hostile = tmp_path / "hostile.approved.zip" + with zipfile.ZipFile(archive) as src, zipfile.ZipFile(hostile, "w") as dst: + for info in src.infolist(): + dst.writestr(info, src.read(info.filename)) + dst.writestr(APPROVAL_NAME, json.dumps(approval)) + + with pytest.raises(SanitizationError, match="own approval"): + materialize_approved_derivative( + hostile, approval=record, destination=tmp_path / "out" + ) + + +def test_materialize_approved_refuses_a_non_empty_destination(tmp_path): + dest, approval, archive = _approved_recording(tmp_path) + record = tmp_path / "approval.json" + record.write_text(json.dumps(approval), encoding="utf-8") + out = tmp_path / "out" + out.mkdir() + (out / "prior.json").write_text("{}", encoding="utf-8") + + with pytest.raises(SanitizationError, match="not empty"): + materialize_approved_derivative(archive, approval=record, destination=out) + assert (out / "prior.json").read_text() == "{}" + + +def test_materialize_approved_refuses_a_policy_envelope_without_its_signature(tmp_path): + dest, approval, archive = _approved_recording(tmp_path) + envelope = build_ingest_manifest(dest) + envelope["approval"]["method"] = "policy" + record = tmp_path / "envelope.json" + record.write_text(json.dumps(envelope), encoding="utf-8") + + with pytest.raises(SanitizationError, match="policy"): + materialize_approved_derivative( + archive, approval=record, destination=tmp_path / "out" + ) + + +def test_materialize_approved_verifies_a_policy_signature_when_the_key_is_present( + tmp_path, monkeypatch +): + source = _recording(tmp_path) + dest = tmp_path / "sanitized" + sanitize_artifact(source, dest, kind="recording") + approve_derivative(dest, source=source, reviewer="policy:x", automatic=True) + key = base64.b64encode(b"k" * 32).decode("ascii") + monkeypatch.setenv("OPENADAPT_SANITIZATION_POLICY_KEY_ID", "policy-key-1") + monkeypatch.setenv("OPENADAPT_SANITIZATION_POLICY_KEY", key) + envelope = build_ingest_manifest(dest) + record = tmp_path / "envelope.json" + record.write_text(json.dumps(envelope), encoding="utf-8") + + materialized = materialize_approved_derivative( + approved_archive_path(dest), approval=record, destination=tmp_path / "good" + ) + assert materialized["automatic"] is True + + envelope["approval"]["policy_signature"] = "a" * 64 + record.write_text(json.dumps(envelope), encoding="utf-8") + with pytest.raises(SanitizationError, match="policy approval signature"): + materialize_approved_derivative( + approved_archive_path(dest), approval=record, destination=tmp_path / "bad" + ) + + +def test_materialize_approved_cli_round_trips(tmp_path, capsys): + from openadapt_flow.__main__ import main + + dest, approval, archive = _approved_recording(tmp_path) + record = tmp_path / "approval.json" + record.write_text(json.dumps(approval), encoding="utf-8") + out = tmp_path / "elsewhere" / "recording" + + code = main( + [ + "materialize-approved", + "--archive", + str(archive), + "--approval", + str(record), + "--out", + str(out), + "--expect-archive-sha256", + approval["approved_derivative_sha256"], + ] + ) + + assert code == 0 + assert "alice" in capsys.readouterr().out + assert load_valid_approval(out)["reviewer"] == "alice"