"""Intake validation (INFD-WP-0005-T02). Offline; uses the real p01 draft.""" import copy import json from pathlib import Path import pytest from informed_decision.intake import (IntakeRefused, SubmitterGrant, idempotency_outcome, scan_secret_markers, submission_digest, validate_submission) ROOT = Path(__file__).resolve().parents[1] SITTING = ROOT / "docs" / "batches" / "2026-10-03-provisioning-window" / "provisioning" GRANT = SubmitterGrant("state-hub", frozenset({"credential-lane"}), frozenset({"net-kingdom-admins"})) def envelope(): memo = json.loads((SITTING / "infd-20261003-p01.memo.json").read_text()) content = (SITTING / "infd-20261003-p01.packet.md").read_text() return { "kind": "informed-decision.memo-intake", "version": 1, "idempotency_key": "d1-p01-2026-10-03", "memo_class": "credential-lane", "review_group": "net-kingdom-admins", "drafted_by": "claude@the-custodian", "serves": {"hub_decision_id": "62aa95e4-3ccc-4867-8806-5e3db005b483", "resolves": "never"}, "approval_intent": { "action": "accept", "principal": "railiance-platform", "purpose": "Accept the 2026-09-13 scoped delivery proof", "target": {"id": "ccr:CCR-2026-0019-delivery-proof", "type": "credential-lane", "system": "railiance-platform"}, }, "memo": memo, "packet": [{"item_id": "infd-20261003-p01-packet", "media_type": "text/markdown", "content": content}], } def refused(env, reason, grant=GRANT, submitter="state-hub"): with pytest.raises(IntakeRefused) as exc: validate_submission(env, submitter_id=submitter, grant=grant) assert exc.value.reason == reason, exc.value return exc.value def test_real_draft_is_accepted_and_actor_is_forced(): out = validate_submission(envelope(), submitter_id="state-hub", grant=GRANT) assert out.memo.id == "infd-20261003-p01" assert out.approval_binding["actor"] == "informed-decision" assert out.submission_digest.startswith("sha256:") assert out.packet["infd-20261003-p01-packet"] def test_submitter_comes_from_grant_not_body(): refused(envelope(), "submitter_unknown", grant=None) refused(envelope(), "submitter_unknown", submitter="someone-else") env = envelope() env["submitter_id"] = "state-hub" refused(env, "unknown_field") def test_actor_cannot_be_supplied(): env = envelope() env["approval_intent"]["actor"] = "agent" refused(env, "unknown_field") @pytest.mark.parametrize("field,value,reason", [ ("approval_id", "ba5ce2d8-8b6d-40be-af89-2e8c147029a3", "approval_pinned_by_submitter"), ("sealed", True, "draft_must_be_unsealed"), ("version", 2, "draft_version_must_be_1"), ]) def test_draft_cannot_be_prebound(field, value, reason): env = envelope() env["memo"][field] = value refused(env, reason) def test_principal_must_be_pending(): env = envelope() env["memo"]["binding"]["principal"]["id"] = "some-keycape-sub" refused(env, "principal_prebound") @pytest.mark.parametrize("question,reason", [ ("Accept the proof? And deploy the lane?", "not_one_question"), ("Accept the proof.", "not_one_question"), ("Accept the\nproof?", "question_multiline"), ("A" * 400 + "?", "question_too_long"), ]) def test_one_question_per_memo(question, reason): env = envelope() env["memo"]["question"] = question refused(env, reason) def test_bundled_work_records_refused(): env = envelope() extra = copy.deepcopy(env["memo"]["highlights"][0]) extra["id"] = "h2" extra["locator"] = {"kind": "work_record", "workplan": "MASON-WP-0007", "task": "MASON-WP-0007-T05"} env["memo"]["highlights"].append(extra) refused(env, "bundled_acts") def test_packet_digest_must_match(): env = envelope() env["packet"][0]["content"] += "\nedited after hashing" refused(env, "packet_digest_mismatch") def test_orphan_and_missing_packet_content(): env = envelope() env["packet"].append({"item_id": "extra", "media_type": "text/plain", "content": "x"}) refused(env, "orphan_packet_content") env = envelope() env["packet"][0]["item_id"] = "other" refused(env, "packet_content_missing") @pytest.mark.parametrize("leak", [ "-----BEGIN OPENSSH PRIVATE KEY-----", "Authorization: Bearer abcdefghijklmnop1234", "eyJhbGciOiJSUzI1NiJ9.eyJzdWIiOiJ4eHh4In0.c2lnbmF0dXJlMTIz", "hvs.CAESIAbcdefghijklmnopqrstuv", "key sk-or-v1-0123456789abcdef0123456789", "AKIAABCDEFGHIJKLMNOP", "client_secret=Zx8f9a7b6c5d4e3f", "https://user:pw@forgejo.example/x", ]) def test_secret_markers_refuse_without_echo(leak): env = envelope() env["packet"][0]["content"] += "\n" + leak # Recompute the hash so only the secret scan can refuse. import hashlib env["memo"]["packet"][0]["hash"] = "sha256:" + hashlib.sha256(env["packet"][0]["content"].encode()).hexdigest() err = refused(env, err_reason(env)) assert leak not in str(err) assert err.path == "$.packet[0].content" def err_reason(env): try: validate_submission(env, submitter_id="state-hub", grant=GRANT) except IntakeRefused as exc: assert exc.reason.startswith("secret_marker:") return exc.reason raise AssertionError("not refused") def test_custody_paths_and_prose_are_not_secrets(): scan_secret_markers({"brief": "Client secret: platform/workloads/informed-decision/sitting-requester; " "token lifetime 15m; access_token: provided by key-cape"}) def test_class_and_group_must_be_admitted(): env = envelope() env["memo_class"] = "adr-accept" refused(env, "memo_class_not_admitted") env = envelope() env["review_group"] = "everyone" refused(env, "review_group_not_admitted") def test_serves_is_closed(): env = envelope() env["serves"]["resolves"] = "always" refused(env, "unknown_enum") env = envelope() env["serves"]["hub_decision_id"] = "D1" refused(env, "invalid_format") def test_idempotency(): env = envelope() digest = submission_digest("state-hub", env) assert digest == submission_digest("state-hub", {**env, "idempotency_key": "another-key-1"}) assert digest != submission_digest("activity-core", env) assert idempotency_outcome(None, digest) == "new" assert idempotency_outcome(digest, digest) == "replay" with pytest.raises(IntakeRefused) as exc: idempotency_outcome("sha256:" + "0" * 64, digest) assert exc.value.reason == "idempotency_conflict" def test_schema_documents_the_example(): jsonschema = pytest.importorskip("jsonschema") schema = json.loads((ROOT / "schemas" / "memo-intake.schema.json").read_text()) jsonschema.validate(envelope(), schema)