From 15d8e2ddfe42416e27925e2313b64e68bdc0c3d5 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Thu, 27 Aug 2026 22:50:36 -0400 Subject: [PATCH 1/2] test(sanitize): pin the approval an approved archive cannot carry approve_derivative computes the approval over the archive's own SHA-256, so _write_deterministic_archive must skip APPROVAL_NAME. Unzipping the archive elsewhere therefore yields the reviewed bytes without the review: load_valid_approval refuses the tree, and compile silently emits a bundle whose source_recording_sha256 is null. These tests state the contract a portable materialize path must satisfy. They fail until that path exists. --- tests/test_sanitized_artifact.py | 288 +++++++++++++++++++++++++++++++ 1 file changed, 288 insertions(+) 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" From 40dbccf43feba4a27cc5e01140ed385de96f18a8 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Thu, 27 Aug 2026 22:51:03 -0400 Subject: [PATCH 2/2] feat(sanitize): materialize an approved derivative from its archive Add `openadapt-flow materialize-approved --archive X --approval Y --out DIR`. It extracts the archive into a new empty directory, places the reviewer's existing approval record beside it, and then runs `load_valid_approval` unchanged. It approves nothing and relaxes 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` 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. The two tree hashes are functions of the extracted bytes that the archive SHA-256 already binds, so they are recomputed rather than transported, and a derivative whose manifest records no approval rescan is refused outright. An envelope recording an automatic approval must carry its policy key id and MAC; the MAC is verified wherever OPENADAPT_SANITIZATION_POLICY_KEY is configured, reusing the same canonicalization `build_ingest_manifest` signs. Extraction trusts no member name: absolute paths, `..` segments, backslashes, directory entries, symlinks, and any member named `.openadapt-approval.json` are refused before a byte is written. `--expect-archive-sha256` pins the archive against a digest the caller already trusts, independently of the record. A failure leaves no approval file behind. The record stays unsigned, exactly as it is on the reviewing machine, so it is only as trustworthy as the channel that delivered it. That channel should be the one that already delivers the archive. --- docs/SANITIZED_ARTIFACTS.md | 29 +++ openadapt_flow/__main__.py | 60 ++++++ openadapt_flow/sanitized_artifact.py | 298 +++++++++++++++++++++++++-- 3 files changed, 366 insertions(+), 21 deletions(-) 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()