diff --git a/.github/workflows/authorization-record-export.yml b/.github/workflows/authorization-record-export.yml new file mode 100644 index 0000000..20edb85 --- /dev/null +++ b/.github/workflows/authorization-record-export.yml @@ -0,0 +1,39 @@ +name: Authorization Record Export + +on: + pull_request: + paths: + - "conformance/authorization-record-*.json" + - "docs/interop/PROOFPATH_AUTHORIZATION_RECORD_EXPORT_V0_1.md" + - "schemas/proofpath-authorization-record-v0.1.schema.json" + - "scripts/check_authorization_record_export.py" + - "scripts/verify_authorization_record_export.py" + - ".github/workflows/authorization-record-export.yml" + push: + branches: + - main + paths: + - "conformance/authorization-record-*.json" + - "docs/interop/PROOFPATH_AUTHORIZATION_RECORD_EXPORT_V0_1.md" + - "schemas/proofpath-authorization-record-v0.1.schema.json" + - "scripts/check_authorization_record_export.py" + - "scripts/verify_authorization_record_export.py" + - ".github/workflows/authorization-record-export.yml" + +permissions: + contents: read + +jobs: + verify: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + with: + persist-credentials: false + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + - name: Reproduce canonical authorization exports + run: python3 scripts/check_authorization_record_export.py + - name: Independently verify lifecycle and handoff semantics + run: python3 scripts/verify_authorization_record_export.py diff --git a/conformance/authorization-record-export-v0.1.json b/conformance/authorization-record-export-v0.1.json new file mode 100644 index 0000000..ba5c681 --- /dev/null +++ b/conformance/authorization-record-export-v0.1.json @@ -0,0 +1,425 @@ +{ + "fixture_version": "0.1", + "revision": "2026-06-29.3", + "title": "ProofPath provider-neutral authorization record export fixtures", + "canonicalization": "RFC8785-JCS", + "hash_algorithm": "sha256", + "schema_path": "schemas/proofpath-authorization-record-v0.1.schema.json", + "profile": "org.proofpath.authorization-export.v0.1", + "cases": [ + { + "case_id": "accepted_reversible", + "input": { + "transition_id": "transition-accepted_reversible", + "subject_id": "agent:payment", + "action_context": { + "caller_id": "agent:payment", + "intent_id": "intent-payment-001", + "causal_parent": "user-approved-payment-plan-001", + "scope": "payment.transfer", + "reversibility": "reversible" + }, + "binding_context": { + "request_method": "POST", + "request_target": "/v1/payment-proposals/evaluate", + "arguments_digest": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "policy_ref": "policy:payment/v1", + "recipient_scope": "merchant:coffee-shop", + "budget_scope": "daily:USD:50", + "approval_ref": null + }, + "decision_context": { + "decision": "ACCEPT", + "reason_codes": [ + "POLICY_MATCH", + "REVERSIBLE_ACTION" + ], + "issued_at": "2030-01-01T00:00:00Z", + "expires_at": "2030-01-01T00:10:00Z", + "consumption_state": "UNCONSUMED", + "continuation_state": "NOT_REQUIRED", + "current_state": "ACTIVE", + "policy_ref": "policy:payment/v1", + "proofpath_evidence_refs": [ + "audit:proofpath/record-001" + ] + } + }, + "expected": { + "execution_allowed": true, + "current_state": "ACTIVE", + "authority_state_at_report": "ACTIVE", + "expected_additional_side_effects": 1, + "authorization_ref": "sha256:96e2bdcba680a7a18bf8753531adbb43acad0370bd4f218a420bfc5186765ff7" + }, + "handoff": null + }, + { + "case_id": "accepted_irreversible_with_approval", + "input": { + "transition_id": "transition-accepted_irreversible_with_approval", + "subject_id": "agent:payment", + "action_context": { + "caller_id": "agent:payment", + "intent_id": "intent-payment-001", + "causal_parent": "user-approved-payment-plan-001", + "scope": "payment.transfer", + "reversibility": "irreversible" + }, + "binding_context": { + "request_method": "POST", + "request_target": "/v1/payment-proposals/evaluate", + "arguments_digest": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "policy_ref": "policy:payment/v1", + "recipient_scope": "merchant:coffee-shop", + "budget_scope": "daily:USD:50", + "approval_ref": "approval:human:001" + }, + "decision_context": { + "decision": "ACCEPT", + "reason_codes": [ + "POLICY_MATCH", + "HUMAN_APPROVAL_VERIFIED" + ], + "issued_at": "2030-01-01T00:00:00Z", + "expires_at": "2030-01-01T00:10:00Z", + "consumption_state": "UNCONSUMED", + "continuation_state": "NOT_REQUIRED", + "current_state": "ACTIVE", + "policy_ref": "policy:payment/v1", + "proofpath_evidence_refs": [ + "audit:proofpath/record-001" + ] + } + }, + "expected": { + "execution_allowed": true, + "current_state": "ACTIVE", + "authority_state_at_report": "ACTIVE", + "expected_additional_side_effects": 1, + "authorization_ref": "sha256:281f23b69fe2d43893e04712593901fed3d71ef217901d3cf8b3216df5a1638e" + }, + "handoff": null + }, + { + "case_id": "held_pending_approval", + "input": { + "transition_id": "transition-held_pending_approval", + "subject_id": "agent:payment", + "action_context": { + "caller_id": "agent:payment", + "intent_id": "intent-payment-001", + "causal_parent": "user-approved-payment-plan-001", + "scope": "payment.transfer", + "reversibility": "irreversible" + }, + "binding_context": { + "request_method": "POST", + "request_target": "/v1/payment-proposals/evaluate", + "arguments_digest": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "policy_ref": "policy:payment/v1", + "recipient_scope": "merchant:coffee-shop", + "budget_scope": "daily:USD:50", + "approval_ref": null + }, + "decision_context": { + "decision": "HOLD", + "reason_codes": [ + "HUMAN_APPROVAL_REQUIRED" + ], + "issued_at": "2030-01-01T00:00:00Z", + "expires_at": "2030-01-01T00:10:00Z", + "consumption_state": "UNCONSUMED", + "continuation_state": "AWAITING_APPROVAL", + "current_state": "PENDING_APPROVAL", + "policy_ref": "policy:payment/v1", + "proofpath_evidence_refs": [ + "audit:proofpath/record-001" + ] + } + }, + "expected": { + "execution_allowed": false, + "current_state": "PENDING_APPROVAL", + "authority_state_at_report": "PENDING_APPROVAL", + "expected_additional_side_effects": 0, + "authorization_ref": "sha256:406ee3e4a5c897058999334c3d3490eb2560013fd26002e394e3e634156553b9" + }, + "handoff": null + }, + { + "case_id": "blocked_expired_intent", + "input": { + "transition_id": "transition-blocked_expired_intent", + "subject_id": "agent:payment", + "action_context": { + "caller_id": "agent:payment", + "intent_id": "intent-payment-001", + "causal_parent": "user-approved-payment-plan-001", + "scope": "payment.transfer", + "reversibility": "irreversible" + }, + "binding_context": { + "request_method": "POST", + "request_target": "/v1/payment-proposals/evaluate", + "arguments_digest": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "policy_ref": "policy:payment/v1", + "recipient_scope": "merchant:coffee-shop", + "budget_scope": "daily:USD:50", + "approval_ref": "approval:human:001" + }, + "decision_context": { + "decision": "BLOCK", + "reason_codes": [ + "INTENT_EXPIRED" + ], + "issued_at": "2030-01-01T00:00:00Z", + "expires_at": "2029-12-31T23:59:59Z", + "consumption_state": "UNCONSUMED", + "continuation_state": "NOT_REQUIRED", + "current_state": "EXPIRED", + "policy_ref": "policy:payment/v1", + "proofpath_evidence_refs": [ + "audit:proofpath/record-001" + ] + } + }, + "expected": { + "execution_allowed": false, + "current_state": "EXPIRED", + "authority_state_at_report": "EXPIRED", + "expected_additional_side_effects": 0, + "authorization_ref": "sha256:5c66eca9ddea2ee216bb265863e8943ede3602d9c14de804c5c8a3f14701406b" + }, + "handoff": null + }, + { + "case_id": "blocked_replayed_consumed_intent", + "input": { + "transition_id": "transition-blocked_replayed_consumed_intent", + "subject_id": "agent:payment", + "action_context": { + "caller_id": "agent:payment", + "intent_id": "intent-payment-001", + "causal_parent": "user-approved-payment-plan-001", + "scope": "payment.transfer", + "reversibility": "irreversible" + }, + "binding_context": { + "request_method": "POST", + "request_target": "/v1/payment-proposals/evaluate", + "arguments_digest": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "policy_ref": "policy:payment/v1", + "recipient_scope": "merchant:coffee-shop", + "budget_scope": "daily:USD:50", + "approval_ref": "approval:human:001" + }, + "decision_context": { + "decision": "BLOCK", + "reason_codes": [ + "INTENT_REPLAYED" + ], + "issued_at": "2030-01-01T00:00:00Z", + "expires_at": "2030-01-01T00:10:00Z", + "consumption_state": "CONSUMED", + "continuation_state": "NOT_REQUIRED", + "current_state": "CONSUMED", + "policy_ref": "policy:payment/v1", + "proofpath_evidence_refs": [ + "audit:proofpath/record-001" + ] + } + }, + "expected": { + "execution_allowed": false, + "current_state": "CONSUMED", + "authority_state_at_report": "CONSUMED", + "expected_additional_side_effects": 0, + "authorization_ref": "sha256:456d7746e5f71059dce1ce7a9be64b8b92962720001227d96bf378a7c3a10a84" + }, + "handoff": null + }, + { + "case_id": "accepted_with_matching_observation", + "input": { + "transition_id": "transition-accepted_with_matching_observation", + "subject_id": "agent:payment", + "action_context": { + "caller_id": "agent:payment", + "intent_id": "intent-payment-001", + "causal_parent": "user-approved-payment-plan-001", + "scope": "payment.transfer", + "reversibility": "irreversible" + }, + "binding_context": { + "request_method": "POST", + "request_target": "/v1/payment-proposals/evaluate", + "arguments_digest": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "policy_ref": "policy:payment/v1", + "recipient_scope": "merchant:coffee-shop", + "budget_scope": "daily:USD:50", + "approval_ref": "approval:human:001" + }, + "decision_context": { + "decision": "ACCEPT", + "reason_codes": [ + "POLICY_MATCH", + "HUMAN_APPROVAL_VERIFIED" + ], + "issued_at": "2030-01-01T00:00:00Z", + "expires_at": "2030-01-01T00:10:00Z", + "consumption_state": "UNCONSUMED", + "continuation_state": "NOT_REQUIRED", + "current_state": "ACTIVE", + "policy_ref": "policy:payment/v1", + "proofpath_evidence_refs": [ + "audit:proofpath/record-001" + ] + } + }, + "expected": { + "execution_allowed": true, + "current_state": "ACTIVE", + "authority_state_at_report": "ACTIVE", + "expected_additional_side_effects": 1, + "authorization_ref": "sha256:7803d1bb97b82f38db629985e8ee21c4c8547df1b380564bf75411f658e59654" + }, + "handoff": { + "observation": { + "execution_status": "EXECUTED", + "observed_at": "2030-01-01T00:02:00Z", + "result_digest": "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "issuer": "runtime:mock-payment-rail" + }, + "expected_join": "MATCH" + } + }, + { + "case_id": "accepted_with_contradicted_response", + "input": { + "transition_id": "transition-accepted_with_contradicted_response", + "subject_id": "agent:payment", + "action_context": { + "caller_id": "agent:payment", + "intent_id": "intent-payment-001", + "causal_parent": "user-approved-payment-plan-001", + "scope": "payment.transfer", + "reversibility": "irreversible" + }, + "binding_context": { + "request_method": "POST", + "request_target": "/v1/payment-proposals/evaluate", + "arguments_digest": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "policy_ref": "policy:payment/v1", + "recipient_scope": "merchant:coffee-shop", + "budget_scope": "daily:USD:50", + "approval_ref": "approval:human:001" + }, + "decision_context": { + "decision": "ACCEPT", + "reason_codes": [ + "POLICY_MATCH", + "HUMAN_APPROVAL_VERIFIED" + ], + "issued_at": "2030-01-01T00:00:00Z", + "expires_at": "2030-01-01T00:10:00Z", + "consumption_state": "UNCONSUMED", + "continuation_state": "NOT_REQUIRED", + "current_state": "ACTIVE", + "policy_ref": "policy:payment/v1", + "proofpath_evidence_refs": [ + "audit:proofpath/record-001" + ] + } + }, + "expected": { + "execution_allowed": true, + "current_state": "ACTIVE", + "authority_state_at_report": "ACTIVE", + "expected_additional_side_effects": 1, + "authorization_ref": "sha256:675290b1e2a03013bd99f0114e83e648d5fc3e732071c8ea80f15c94db237706" + }, + "handoff": { + "observation": { + "execution_status": "EXECUTED", + "observed_at": "2030-01-01T00:02:00Z", + "result_digest": "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "issuer": "runtime:mock-payment-rail" + }, + "expected_join": "MATCH_WITH_INTEGRITY_FAILURE", + "response_integrity": { + "overall_verdict": "FAILED", + "verifier": "ltp:example", + "claim_boundary": "External verifier result; ProofPath did not author this semantic verdict." + } + } + }, + { + "case_id": "stale_authorization_honest_report", + "input": { + "transition_id": "transition-stale_authorization_honest_report", + "subject_id": "agent:payment", + "action_context": { + "caller_id": "agent:payment", + "intent_id": "intent-payment-001", + "causal_parent": "user-approved-payment-plan-001", + "scope": "payment.transfer", + "reversibility": "irreversible" + }, + "binding_context": { + "request_method": "POST", + "request_target": "/v1/payment-proposals/evaluate", + "arguments_digest": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "policy_ref": "policy:payment/v1", + "recipient_scope": "merchant:coffee-shop", + "budget_scope": "daily:USD:50", + "approval_ref": "approval:human:001" + }, + "decision_context": { + "decision": "ACCEPT", + "reason_codes": [ + "POLICY_MATCH", + "HUMAN_APPROVAL_VERIFIED" + ], + "issued_at": "2030-01-01T00:00:00Z", + "expires_at": "2030-01-01T00:05:00Z", + "consumption_state": "UNCONSUMED", + "continuation_state": "NOT_REQUIRED", + "current_state": "ACTIVE", + "policy_ref": "policy:payment/v1", + "proofpath_evidence_refs": [ + "audit:proofpath/record-001" + ] + } + }, + "expected": { + "execution_allowed": false, + "current_state": "ACTIVE", + "authority_state_at_report": "EXPIRED_AT_REPORT", + "expected_additional_side_effects": 0, + "authorization_ref": "sha256:73a76f8282b1b686d4ecef9fce2e5bcbe4ea3721c531c0c78d000f97f98d127d" + }, + "handoff": { + "observation": { + "execution_status": "EXECUTED", + "observed_at": "2030-01-01T00:04:00Z", + "result_digest": "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "issuer": "runtime:mock-payment-rail" + }, + "expected_join": "HISTORICAL_ONLY", + "response_integrity": { + "overall_verdict": "VERIFIED", + "verifier": "ltp:example", + "claim_boundary": "External verifier result; ProofPath did not author this semantic verdict." + } + } + } + ], + "invariants": [ + "Only ACCEPT records with active, fresh, unconsumed authority permit execution.", + "HOLD, BLOCK, and REJECT records expect zero additional side effects.", + "Observation and response-integrity records remain independently attributed.", + "A downstream integrity failure does not rewrite the original authorization decision.", + "An honest historical report does not reactivate expired authority." + ] +} diff --git a/conformance/authorization-record-report-context-v0.1.json b/conformance/authorization-record-report-context-v0.1.json new file mode 100644 index 0000000..68b6fee --- /dev/null +++ b/conformance/authorization-record-report-context-v0.1.json @@ -0,0 +1,14 @@ +{ + "profile": "org.proofpath.authorization-report-context.v0.1", + "cases": { + "accepted_with_matching_observation": { + "reported_at": "2030-01-01T00:03:00Z" + }, + "accepted_with_contradicted_response": { + "reported_at": "2030-01-01T00:03:00Z" + }, + "stale_authorization_honest_report": { + "reported_at": "2030-01-01T00:06:00Z" + } + } +} diff --git a/docs/interop/PROOFPATH_AUTHORIZATION_RECORD_EXPORT_V0_1.md b/docs/interop/PROOFPATH_AUTHORIZATION_RECORD_EXPORT_V0_1.md new file mode 100644 index 0000000..53c3e2a --- /dev/null +++ b/docs/interop/PROOFPATH_AUTHORIZATION_RECORD_EXPORT_V0_1.md @@ -0,0 +1,209 @@ +# ProofPath Authorization Record Export Profile v0.1 + +**Status:** Draft interoperability profile +**Issue:** [ProofPath #166](https://github.com/safal207/ProofPath/issues/166) +**Canonical ecosystem profile:** [Liminal #108](https://github.com/safal207/Liminal/issues/108) + +## Purpose + +ProofPath is a pre-execution authorization boundary. This profile defines how a +ProofPath decision can be exported as a provider-neutral `authorization_record` +without turning ProofPath into an observation source or post-response claim +verifier. + +The exported record answers: + +> Was this exact proposed action authorized, blocked, rejected, or held against +> the frozen context that existed at decision time? + +It does not answer whether a downstream tool actually executed, whether a +captured result is truthful, or whether a model later described that result +faithfully. + +## Files + +- JSON Schema: [`schemas/proofpath-authorization-record-v0.1.schema.json`](../../schemas/proofpath-authorization-record-v0.1.schema.json) +- Fixture pack: [`conformance/authorization-record-export-v0.1.json`](../../conformance/authorization-record-export-v0.1.json) +- Report-time context: [`conformance/authorization-record-report-context-v0.1.json`](../../conformance/authorization-record-report-context-v0.1.json) +- Deterministic exporter: [`scripts/check_authorization_record_export.py`](../../scripts/check_authorization_record_export.py) +- Independent semantic verifier: [`scripts/verify_authorization_record_export.py`](../../scripts/verify_authorization_record_export.py) + +Run: + +```bash +python3 scripts/check_authorization_record_export.py +python3 scripts/verify_authorization_record_export.py +``` + +## Record shape + +A portable record contains: + +```text +schema +profile +transition_id +subject_id +action_identity_profile +action_identity_digest +binding_profile +binding_digest +decision +reason_codes +issued_at +expires_at +consumption_state +continuation_state +current_state +policy_ref +proofpath_evidence_refs +claim_boundary +``` + +`current_state` is intrinsic to the exported authorization record. Its portable +values are `ACTIVE`, `PENDING_APPROVAL`, `EXPIRED`, `CONSUMED`, `BLOCKED`, +`REJECTED`, and `INVALID`. Report-derived values such as `EXPIRED_AT_REPORT` +are not serialized into the canonical authorization record. + +The fixture wraps each record with canonical bytes and a SHA-256 reference: + +```text +record_ref = "sha256:" + SHA256(RFC8785-JCS(record)) +``` + +The current fixture contains no floating-point values, so standard-library +sorted minimal JSON is equivalent to RFC 8785 JCS for the published input +domain. + +## Frozen action identity + +The action identity preimage contains: + +```text +caller_id +intent_id +causal_parent +scope +reversibility +``` + +The binding preimage contains: + +```text +request_method +request_target +arguments_digest +policy_ref +recipient_scope +budget_scope +approval_ref +``` + +A downstream observation must repeat both digests and bind to the exact +`authorization_record` reference. + +## Decision mapping + +| ProofPath decision | Portable authority meaning | +| --- | --- | +| `ACCEPT` | The exact frozen action context may proceed while fresh and unconsumed. | +| `HOLD` | Execution is not authorized yet; approval or revalidation remains open. | +| `BLOCK` | Execution is forbidden under the evaluated policy or context. | +| `REJECT` | The request is invalid for authorization. | + +No downstream observation or honest response can retroactively convert `HOLD`, +`BLOCK`, or `REJECT` into `ACCEPT`. + +## Freshness, consumption, and report time + +The immutable authorization export preserves `issued_at`, `expires_at`, +`consumption_state`, `continuation_state`, and its intrinsic `current_state`. + +Report time is deliberately supplied as separate verifier context rather than +inserted into the authorization record. The verifier derives a separate +`authority_state_at_report`, which may be `EXPIRED_AT_REPORT`, without changing +the canonical bytes or reference of the original authorization decision. + +A record that later expires may remain historically auditable, but it must not +be reused as live authority. A consumed record may remain linked to its existing +resolution, but it must not permit a second side effect. + +## Independent semantic verification + +The deterministic exporter reproduces canonical records and references. The +independent verifier does not import that exporter. It separately validates: + +- exact context shapes and SHA-256 reference syntax; +- policy binding consistency; +- decision, continuation, approval, and consumption combinations; +- intrinsic authorization state and report-time authority state; +- execution eligibility and zero-side-effect expectations; +- observation timing and authorization joins; +- response-integrity joins and their derived outcome. + +This separation prevents the fixture and exporter from silently agreeing on the +same semantic mistake. + +## Evidence handoff + +A downstream runtime may produce an `observation_record` containing: + +```text +authorization_ref +action_identity_digest +binding_digest +execution_status +observed_at +result_digest +``` + +A later verifier may produce a `response_integrity_record` containing: + +```text +authorization_ref +observation_refs +claim verdicts +verifier identity +claim boundary +``` + +ProofPath evidence bundles may carry those external artifacts by digest and +schema/profile identifier. Importing them does not mean ProofPath authored their +semantic verdicts. + +For the conformance fixtures, the semantic verifier derives one of: + +- `MATCH` — the observation joins to live authority and no supplied integrity + record reports failure; +- `MATCH_WITH_INTEGRITY_FAILURE` — authority and observation join, while the + independently attributed response-integrity verdict is not `VERIFIED`; +- `HISTORICAL_ONLY` — execution occurred within the authorization window, but + the report was evaluated after authority expired. + +## Fixture cases + +The fixture pack covers: + +1. accepted reversible action; +2. accepted irreversible action with approval; +3. held action pending approval; +4. blocked expired intent; +5. blocked replayed or consumed intent; +6. accepted action with a matching downstream observation handoff; +7. accepted action with a contradicted external response-integrity verdict; +8. stale authorization paired with an honest historical report. + +## Security and privacy boundary + +The export should use digests and references for sensitive payloads. It must not +require raw credentials, private keys, complete payment payloads, or secrets. + +This profile does not by itself prove signer identity, signature validity, +observation-source integrity, payment settlement, policy correctness, +regulatory compliance, or production security. + +## Canonical invariant + +> ProofPath exports the evidence for why an exact action could or could not +> proceed. Downstream systems may attach what happened and what the model later +> claimed, but those verdicts remain independent. diff --git a/schemas/proofpath-authorization-record-v0.1.schema.json b/schemas/proofpath-authorization-record-v0.1.schema.json new file mode 100644 index 0000000..27cabdc --- /dev/null +++ b/schemas/proofpath-authorization-record-v0.1.schema.json @@ -0,0 +1,133 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://proofpath.example/schemas/proofpath-authorization-record-v0.1.schema.json", + "title": "ProofPath Provider-Neutral Authorization Record v0.1", + "type": "object", + "required": [ + "schema", + "profile", + "transition_id", + "subject_id", + "action_identity_profile", + "action_identity_digest", + "binding_profile", + "binding_digest", + "decision", + "reason_codes", + "issued_at", + "expires_at", + "consumption_state", + "continuation_state", + "current_state", + "policy_ref", + "proofpath_evidence_refs", + "claim_boundary" + ], + "properties": { + "schema": { + "const": "org.proofpath.authorization-record.v0.1" + }, + "profile": { + "const": "org.proofpath.authorization-export.v0.1" + }, + "transition_id": { + "type": "string", + "minLength": 1 + }, + "subject_id": { + "type": "string", + "minLength": 1 + }, + "action_identity_profile": { + "const": "org.liminal.trustworthy-transition.action-identity.v0.1" + }, + "action_identity_digest": { + "$ref": "#/$defs/sha256Ref" + }, + "binding_profile": { + "const": "org.liminal.trustworthy-transition.binding.v0.1" + }, + "binding_digest": { + "$ref": "#/$defs/sha256Ref" + }, + "decision": { + "enum": [ + "ACCEPT", + "HOLD", + "BLOCK", + "REJECT" + ] + }, + "reason_codes": { + "type": "array", + "minItems": 1, + "items": { + "type": "string", + "minLength": 1 + }, + "uniqueItems": true + }, + "issued_at": { + "type": "string", + "format": "date-time" + }, + "expires_at": { + "type": [ + "string", + "null" + ], + "format": "date-time" + }, + "consumption_state": { + "enum": [ + "UNCONSUMED", + "CONSUMED", + "UNKNOWN" + ] + }, + "continuation_state": { + "enum": [ + "NOT_REQUIRED", + "AWAITING_APPROVAL", + "REQUIRES_REVALIDATION", + "RESOLVED" + ] + }, + "current_state": { + "description": "Intrinsic authorization state captured by the exported record. Report-derived states are evaluated separately.", + "enum": [ + "ACTIVE", + "PENDING_APPROVAL", + "EXPIRED", + "CONSUMED", + "BLOCKED", + "REJECTED", + "INVALID" + ] + }, + "policy_ref": { + "type": "string", + "minLength": 1 + }, + "proofpath_evidence_refs": { + "type": "array", + "minItems": 1, + "items": { + "type": "string", + "minLength": 1 + }, + "uniqueItems": true + }, + "claim_boundary": { + "type": "string", + "minLength": 1 + } + }, + "additionalProperties": false, + "$defs": { + "sha256Ref": { + "type": "string", + "pattern": "^sha256:[0-9a-f]{64}$" + } + } +} diff --git a/scripts/check_authorization_record_export.py b/scripts/check_authorization_record_export.py new file mode 100644 index 0000000..59333f2 --- /dev/null +++ b/scripts/check_authorization_record_export.py @@ -0,0 +1,335 @@ +#!/usr/bin/env python3 +"""Reproduce ProofPath authorization exports and their evidence handoffs.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import sys +from datetime import datetime +from pathlib import Path +from typing import Any + +AUTH_SCHEMA = "org.proofpath.authorization-record.v0.1" +EXPORT_PROFILE = "org.proofpath.authorization-export.v0.1" +REPORT_PROFILE = "org.proofpath.authorization-report-context.v0.1" +ACTION_PROFILE = "org.liminal.trustworthy-transition.action-identity.v0.1" +BINDING_PROFILE = "org.liminal.trustworthy-transition.binding.v0.1" +OBS_SCHEMA = "org.liminal.trustworthy-transition.observation.v0.1" +INTEGRITY_SCHEMA = "org.liminal.trustworthy-transition.response-integrity.v0.1" +CLAIM_BOUNDARY = ( + "ProofPath proves its exported authorization decision and frozen context " + "binding; it does not prove downstream execution, observation truth, or " + "response honesty." +) +INTRINSIC_STATES = { + "ACTIVE", "PENDING_APPROVAL", "EXPIRED", "CONSUMED", + "BLOCKED", "REJECTED", "INVALID", +} + + +class ExportVerificationError(ValueError): + pass + + +def canonical(value: Any) -> str: + return json.dumps( + value, ensure_ascii=False, sort_keys=True, + separators=(",", ":"), allow_nan=False, + ) + + +def digest(value: Any) -> str: + return "sha256:" + hashlib.sha256(canonical(value).encode()).hexdigest() + + +def load(path: Path) -> dict[str, Any]: + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as error: + raise ExportVerificationError(f"cannot load {path}: {error}") from error + if not isinstance(value, dict): + raise ExportVerificationError(f"{path} must contain an object") + return value + + +def timestamp(value: Any, label: str) -> datetime: + if not isinstance(value, str) or not value: + raise ExportVerificationError(f"{label} must be a timestamp") + try: + result = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError as error: + raise ExportVerificationError(f"{label} must be ISO-8601") from error + if result.tzinfo is None: + raise ExportVerificationError(f"{label} must include a timezone") + return result + + +def record_for(case: dict[str, Any]) -> dict[str, Any]: + case_id = case.get("case_id") + data = case.get("input") + if not isinstance(case_id, str) or not isinstance(data, dict): + raise ExportVerificationError("invalid case") + action = data.get("action_context") + binding = data.get("binding_context") + decision = data.get("decision_context") + if not all(isinstance(v, dict) for v in (action, binding, decision)): + raise ExportVerificationError(f"{case_id} contexts must be objects") + current_state = decision.get("current_state") + if current_state not in INTRINSIC_STATES: + raise ExportVerificationError( + f"{case_id}.current_state must be intrinsic" + ) + return { + "schema": AUTH_SCHEMA, + "profile": EXPORT_PROFILE, + "transition_id": data["transition_id"], + "subject_id": data["subject_id"], + "action_identity_profile": ACTION_PROFILE, + "action_identity_digest": digest( + {"profile_id": ACTION_PROFILE, **action} + ), + "binding_profile": BINDING_PROFILE, + "binding_digest": digest( + {"profile_id": BINDING_PROFILE, **binding} + ), + "decision": decision["decision"], + "reason_codes": decision["reason_codes"], + "issued_at": decision["issued_at"], + "expires_at": decision["expires_at"], + "consumption_state": decision["consumption_state"], + "continuation_state": decision["continuation_state"], + "current_state": current_state, + "policy_ref": decision["policy_ref"], + "proofpath_evidence_refs": decision["proofpath_evidence_refs"], + "claim_boundary": CLAIM_BOUNDARY, + } + + +def intrinsic_state(record: dict[str, Any]) -> str: + if record["consumption_state"] == "CONSUMED": + return "CONSUMED" + if record["consumption_state"] == "UNKNOWN": + return "INVALID" + if record["decision"] == "HOLD": + return "PENDING_APPROVAL" + if record["decision"] == "REJECT": + return "REJECTED" + if record["decision"] == "BLOCK": + return ( + "EXPIRED" + if "INTENT_EXPIRED" in record["reason_codes"] + else "BLOCKED" + ) + return "ACTIVE" + + +def report_state( + record: dict[str, Any], reported_at: datetime | None +) -> str: + state = intrinsic_state(record) + if state != "ACTIVE" or reported_at is None: + return state + expires_at = record.get("expires_at") + if expires_at is not None and reported_at > timestamp( + expires_at, "expires_at" + ): + return "EXPIRED_AT_REPORT" + return state + + +def handoff_for( + case: dict[str, Any], + record: dict[str, Any], + authorization_ref: str, +) -> dict[str, Any] | None: + source = case.get("handoff") + if source is None: + return None + if not isinstance(source, dict) or not isinstance( + source.get("observation"), dict + ): + raise ExportVerificationError( + f"{case['case_id']} handoff is invalid" + ) + observed = source["observation"] + observation = { + "schema": OBS_SCHEMA, + "transition_id": record["transition_id"], + "subject_id": record["subject_id"], + "authorization_ref": authorization_ref, + "action_identity_digest": record["action_identity_digest"], + "binding_digest": record["binding_digest"], + "execution_status": observed["execution_status"], + "observed_at": observed["observed_at"], + "result_digest": observed["result_digest"], + "issuer": observed["issuer"], + } + observation_ref = digest(observation) + result: dict[str, Any] = { + "observation_record": { + "record": observation, + "record_ref": observation_ref, + }, + "expected_join": source["expected_join"], + } + integrity_source = source.get("response_integrity") + if integrity_source is not None: + integrity = { + "schema": INTEGRITY_SCHEMA, + "transition_id": record["transition_id"], + "subject_id": record["subject_id"], + "authorization_ref": authorization_ref, + "observation_refs": [observation_ref], + "overall_verdict": integrity_source["overall_verdict"], + "verifier": integrity_source["verifier"], + "claim_boundary": integrity_source["claim_boundary"], + } + result["response_integrity_record"] = { + "record": integrity, + "record_ref": digest(integrity), + } + return result + + +def verify_case( + case: dict[str, Any], report_cases: dict[str, Any] +) -> dict[str, Any]: + case_id = case["case_id"] + expected = case.get("expected") + if not isinstance(expected, dict): + raise ExportVerificationError(f"{case_id}.expected is invalid") + + record = record_for(case) + authorization_ref = digest(record) + if authorization_ref != expected.get("authorization_ref"): + raise ExportVerificationError( + f"{case_id} authorization_ref mismatch: {authorization_ref}" + ) + + intrinsic = intrinsic_state(record) + if record["current_state"] != intrinsic: + raise ExportVerificationError( + f"{case_id} intrinsic current_state mismatch" + ) + if expected.get("current_state") != intrinsic: + raise ExportVerificationError( + f"{case_id} expected current_state mismatch" + ) + + entry = report_cases.get(case_id) + reported_at = None + if entry is not None: + if not isinstance(entry, dict) or set(entry) != {"reported_at"}: + raise ExportVerificationError( + f"{case_id} report context is invalid" + ) + reported_at = timestamp( + entry["reported_at"], f"{case_id}.reported_at" + ) + evaluated = report_state(record, reported_at) + if expected.get("authority_state_at_report") != evaluated: + raise ExportVerificationError( + f"{case_id} authority_state_at_report mismatch" + ) + + allowed = ( + record["decision"] == "ACCEPT" + and evaluated == "ACTIVE" + and record["consumption_state"] == "UNCONSUMED" + ) + if expected.get("execution_allowed") is not allowed: + raise ExportVerificationError( + f"{case_id} execution_allowed mismatch" + ) + if expected.get("expected_additional_side_effects") != ( + 1 if allowed else 0 + ): + raise ExportVerificationError( + f"{case_id} side-effect expectation mismatch" + ) + + handoff = handoff_for(case, record, authorization_ref) + if handoff is not None and reported_at is None: + raise ExportVerificationError( + f"{case_id} handoff lacks report context" + ) + return { + "authorization_record": { + "record": record, + "canonical_bytes_utf8": canonical(record), + "record_ref": authorization_ref, + }, + "authority_state_at_report": evaluated, + "handoff": handoff, + } + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "fixture", nargs="?", type=Path, + default=Path("conformance/authorization-record-export-v0.1.json"), + ) + parser.add_argument( + "--report-context", type=Path, + default=Path( + "conformance/authorization-record-report-context-v0.1.json" + ), + ) + parser.add_argument("--emit-case") + args = parser.parse_args() + try: + fixture = load(args.fixture) + report = load(args.report_context) + if fixture.get("profile") != EXPORT_PROFILE: + raise ExportVerificationError("fixture profile mismatch") + if report.get("profile") != REPORT_PROFILE: + raise ExportVerificationError("report profile mismatch") + cases = fixture.get("cases") + report_cases = report.get("cases") + if not isinstance(cases, list) or not isinstance(report_cases, dict): + raise ExportVerificationError("fixture shape mismatch") + + seen: set[str] = set() + selected = None + for case in cases: + case_id = case.get("case_id") + if not isinstance(case_id, str) or case_id in seen: + raise ExportVerificationError( + f"invalid or duplicate case_id: {case_id}" + ) + seen.add(case_id) + derived = verify_case(case, report_cases) + if args.emit_case == case_id: + selected = derived + record = derived["authorization_record"]["record"] + print( + f"PASS {case_id} -> {record['decision']} / " + f"{record['current_state']} / " + f"{derived['authority_state_at_report']}" + ) + + unknown = set(report_cases) - seen + if unknown: + raise ExportVerificationError( + "unknown report cases: " + ", ".join(sorted(unknown)) + ) + if args.emit_case: + if selected is None: + raise ExportVerificationError( + f"unknown case_id: {args.emit_case}" + ) + print(json.dumps(selected, ensure_ascii=False, indent=2)) + print( + f"\nProofPath authorization export fixtures passed: {len(cases)}" + ) + return 0 + except (ExportVerificationError, KeyError, TypeError) as error: + print(f"ERROR: {error}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/verify_authorization_record_export.py b/scripts/verify_authorization_record_export.py new file mode 100644 index 0000000..32b55a6 --- /dev/null +++ b/scripts/verify_authorization_record_export.py @@ -0,0 +1,488 @@ +#!/usr/bin/env python3 +"""Independently verify ProofPath authorization export semantics.""" + +from __future__ import annotations + +import hashlib +import json +import re +import sys +from datetime import datetime +from pathlib import Path +from typing import Any + +FIXTURE = Path("conformance/authorization-record-export-v0.1.json") +REPORT_CONTEXT = Path( + "conformance/authorization-record-report-context-v0.1.json" +) +AUTH_SCHEMA = "org.proofpath.authorization-record.v0.1" +EXPORT_PROFILE = "org.proofpath.authorization-export.v0.1" +REPORT_PROFILE = "org.proofpath.authorization-report-context.v0.1" +ACTION_PROFILE = "org.liminal.trustworthy-transition.action-identity.v0.1" +BINDING_PROFILE = "org.liminal.trustworthy-transition.binding.v0.1" +OBS_SCHEMA = "org.liminal.trustworthy-transition.observation.v0.1" +INTEGRITY_SCHEMA = "org.liminal.trustworthy-transition.response-integrity.v0.1" +CLAIM_BOUNDARY = ( + "ProofPath proves its exported authorization decision and frozen context " + "binding; it does not prove downstream execution, observation truth, or " + "response honesty." +) +SHA256_REF = re.compile(r"^sha256:[0-9a-f]{64}$") +DECISIONS = {"ACCEPT", "HOLD", "BLOCK", "REJECT"} +CONSUMPTION = {"UNCONSUMED", "CONSUMED", "UNKNOWN"} +CONTINUATIONS = { + "NOT_REQUIRED", "AWAITING_APPROVAL", + "REQUIRES_REVALIDATION", "RESOLVED", +} +INTRINSIC_STATES = { + "ACTIVE", "PENDING_APPROVAL", "EXPIRED", "CONSUMED", + "BLOCKED", "REJECTED", "INVALID", +} +EXECUTION_STATUSES = {"EXECUTED", "BLOCKED", "ERRORED", "REFUSED"} +JOINS = {"MATCH", "MATCH_WITH_INTEGRITY_FAILURE", "HISTORICAL_ONLY"} + + +class VerificationError(ValueError): + pass + + +def load(path: Path) -> dict[str, Any]: + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as error: + raise VerificationError(f"cannot load {path}: {error}") from error + if not isinstance(value, dict): + raise VerificationError(f"{path} must contain an object") + return value + + +def exact(value: dict[str, Any], keys: set[str], label: str) -> None: + if set(value) != keys: + raise VerificationError( + f"{label} must contain exactly {sorted(keys)}" + ) + + +def text(value: Any, label: str) -> str: + if not isinstance(value, str) or not value: + raise VerificationError(f"{label} must be a non-empty string") + return value + + +def strings(value: Any, label: str) -> list[str]: + if ( + not isinstance(value, list) + or not value + or any(not isinstance(item, str) or not item for item in value) + or len(set(value)) != len(value) + ): + raise VerificationError(f"{label} must be a unique string array") + return value + + +def sha(value: Any, label: str) -> str: + value = text(value, label) + if SHA256_REF.fullmatch(value) is None: + raise VerificationError(f"{label} must be sha256:<64 lowercase hex>") + return value + + +def time_value(value: Any, label: str) -> datetime: + value = text(value, label) + try: + result = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError as error: + raise VerificationError(f"{label} must be ISO-8601") from error + if result.tzinfo is None: + raise VerificationError(f"{label} must include a timezone") + return result + + +def canonical(value: Any) -> str: + return json.dumps( + value, ensure_ascii=False, sort_keys=True, + separators=(",", ":"), allow_nan=False, + ) + + +def digest(value: Any) -> str: + return "sha256:" + hashlib.sha256(canonical(value).encode()).hexdigest() + + +def build_record(case: dict[str, Any]) -> dict[str, Any]: + case_id = text(case.get("case_id"), "case_id") + exact(case, {"case_id", "input", "expected", "handoff"}, case_id) + data = case.get("input") + if not isinstance(data, dict): + raise VerificationError(f"{case_id}.input must be an object") + exact( + data, + { + "transition_id", "subject_id", "action_context", + "binding_context", "decision_context", + }, + f"{case_id}.input", + ) + action = data.get("action_context") + binding = data.get("binding_context") + decision = data.get("decision_context") + if not all(isinstance(v, dict) for v in (action, binding, decision)): + raise VerificationError(f"{case_id} contexts must be objects") + exact( + action, + {"caller_id", "intent_id", "causal_parent", "scope", "reversibility"}, + f"{case_id}.action_context", + ) + exact( + binding, + { + "request_method", "request_target", "arguments_digest", + "policy_ref", "recipient_scope", "budget_scope", "approval_ref", + }, + f"{case_id}.binding_context", + ) + exact( + decision, + { + "decision", "reason_codes", "issued_at", "expires_at", + "consumption_state", "continuation_state", "current_state", + "policy_ref", "proofpath_evidence_refs", + }, + f"{case_id}.decision_context", + ) + for key in ("caller_id", "intent_id", "causal_parent", "scope"): + text(action.get(key), f"{case_id}.{key}") + if action.get("reversibility") not in {"reversible", "irreversible"}: + raise VerificationError(f"{case_id}.reversibility is invalid") + for key in ( + "request_method", "request_target", "policy_ref", + "recipient_scope", "budget_scope", + ): + text(binding.get(key), f"{case_id}.{key}") + sha(binding.get("arguments_digest"), f"{case_id}.arguments_digest") + approval = binding.get("approval_ref") + if approval is not None: + text(approval, f"{case_id}.approval_ref") + + decision_value = decision.get("decision") + if decision_value not in DECISIONS: + raise VerificationError(f"{case_id}.decision is invalid") + reasons = strings(decision.get("reason_codes"), f"{case_id}.reason_codes") + time_value(decision.get("issued_at"), f"{case_id}.issued_at") + if decision.get("expires_at") is not None: + time_value(decision["expires_at"], f"{case_id}.expires_at") + if decision.get("consumption_state") not in CONSUMPTION: + raise VerificationError(f"{case_id}.consumption_state is invalid") + if decision.get("continuation_state") not in CONTINUATIONS: + raise VerificationError(f"{case_id}.continuation_state is invalid") + if decision.get("current_state") not in INTRINSIC_STATES: + raise VerificationError(f"{case_id}.current_state is not intrinsic") + if binding["policy_ref"] != decision.get("policy_ref"): + raise VerificationError(f"{case_id} policy_ref drift") + if decision_value == "HOLD": + if decision["continuation_state"] not in { + "AWAITING_APPROVAL", "REQUIRES_REVALIDATION", + }: + raise VerificationError(f"{case_id} HOLD requires continuation") + elif decision["continuation_state"] == "AWAITING_APPROVAL": + raise VerificationError(f"{case_id} non-HOLD awaits approval") + if ( + decision_value == "ACCEPT" + and action["reversibility"] == "irreversible" + and approval is None + ): + raise VerificationError( + f"{case_id} irreversible ACCEPT lacks approval" + ) + + return { + "schema": AUTH_SCHEMA, + "profile": EXPORT_PROFILE, + "transition_id": text( + data.get("transition_id"), f"{case_id}.transition_id" + ), + "subject_id": text( + data.get("subject_id"), f"{case_id}.subject_id" + ), + "action_identity_profile": ACTION_PROFILE, + "action_identity_digest": digest( + {"profile_id": ACTION_PROFILE, **action} + ), + "binding_profile": BINDING_PROFILE, + "binding_digest": digest( + {"profile_id": BINDING_PROFILE, **binding} + ), + "decision": decision_value, + "reason_codes": reasons, + "issued_at": decision["issued_at"], + "expires_at": decision["expires_at"], + "consumption_state": decision["consumption_state"], + "continuation_state": decision["continuation_state"], + "current_state": decision["current_state"], + "policy_ref": decision["policy_ref"], + "proofpath_evidence_refs": strings( + decision.get("proofpath_evidence_refs"), + f"{case_id}.proofpath_evidence_refs", + ), + "claim_boundary": CLAIM_BOUNDARY, + } + + +def intrinsic_state(record: dict[str, Any]) -> str: + if record["consumption_state"] == "CONSUMED": + return "CONSUMED" + if record["consumption_state"] == "UNKNOWN": + return "INVALID" + if record["decision"] == "HOLD": + return "PENDING_APPROVAL" + if record["decision"] == "REJECT": + return "REJECTED" + if record["decision"] == "BLOCK": + return ( + "EXPIRED" + if "INTENT_EXPIRED" in record["reason_codes"] + else "BLOCKED" + ) + return "ACTIVE" + + +def report_state( + record: dict[str, Any], reported_at: datetime | None +) -> str: + state = intrinsic_state(record) + if state != "ACTIVE" or reported_at is None: + return state + expires = record.get("expires_at") + if expires is not None and reported_at > time_value( + expires, "expires_at" + ): + return "EXPIRED_AT_REPORT" + return state + + +def verify_handoff( + case_id: str, + source: Any, + record: dict[str, Any], + authorization_ref: str, + reported_at: datetime | None, + evaluated_state: str, +) -> None: + if source is None: + if reported_at is not None: + raise VerificationError(f"{case_id} has unused report context") + return + if not isinstance(source, dict): + raise VerificationError(f"{case_id}.handoff must be an object") + if not {"observation", "expected_join"}.issubset(source): + raise VerificationError(f"{case_id}.handoff is incomplete") + if not set(source).issubset( + {"observation", "expected_join", "response_integrity"} + ): + raise VerificationError(f"{case_id}.handoff has unknown fields") + if reported_at is None: + raise VerificationError(f"{case_id} lacks report context") + observed = source.get("observation") + if not isinstance(observed, dict): + raise VerificationError(f"{case_id}.observation must be an object") + exact( + observed, + {"execution_status", "observed_at", "result_digest", "issuer"}, + f"{case_id}.observation", + ) + if observed.get("execution_status") not in EXECUTION_STATUSES: + raise VerificationError(f"{case_id}.execution_status is invalid") + observed_at = time_value( + observed.get("observed_at"), f"{case_id}.observed_at" + ) + if observed_at > reported_at: + raise VerificationError(f"{case_id} observation follows report") + issued_at = time_value(record["issued_at"], f"{case_id}.issued_at") + expires_at = ( + None + if record["expires_at"] is None + else time_value(record["expires_at"], f"{case_id}.expires_at") + ) + if observed_at < issued_at: + raise VerificationError(f"{case_id} observation predates authority") + if observed["execution_status"] == "EXECUTED": + if record["decision"] != "ACCEPT": + raise VerificationError(f"{case_id} executed without ACCEPT") + if expires_at is not None and observed_at > expires_at: + raise VerificationError(f"{case_id} executed after expiry") + + observation = { + "schema": OBS_SCHEMA, + "transition_id": record["transition_id"], + "subject_id": record["subject_id"], + "authorization_ref": authorization_ref, + "action_identity_digest": record["action_identity_digest"], + "binding_digest": record["binding_digest"], + "execution_status": observed["execution_status"], + "observed_at": observed["observed_at"], + "result_digest": sha( + observed.get("result_digest"), f"{case_id}.result_digest" + ), + "issuer": text(observed.get("issuer"), f"{case_id}.issuer"), + } + observation_ref = digest(observation) + + integrity_source = source.get("response_integrity") + integrity_verdict = None + if integrity_source is not None: + if not isinstance(integrity_source, dict): + raise VerificationError( + f"{case_id}.response_integrity must be an object" + ) + exact( + integrity_source, + {"overall_verdict", "verifier", "claim_boundary"}, + f"{case_id}.response_integrity", + ) + integrity = { + "schema": INTEGRITY_SCHEMA, + "transition_id": record["transition_id"], + "subject_id": record["subject_id"], + "authorization_ref": authorization_ref, + "observation_refs": [observation_ref], + "overall_verdict": text( + integrity_source.get("overall_verdict"), + f"{case_id}.overall_verdict", + ), + "verifier": text( + integrity_source.get("verifier"), f"{case_id}.verifier" + ), + "claim_boundary": text( + integrity_source.get("claim_boundary"), + f"{case_id}.claim_boundary", + ), + } + digest(integrity) + integrity_verdict = integrity["overall_verdict"] + + if evaluated_state == "EXPIRED_AT_REPORT": + derived_join = "HISTORICAL_ONLY" + elif integrity_verdict is not None and integrity_verdict != "VERIFIED": + derived_join = "MATCH_WITH_INTEGRITY_FAILURE" + else: + derived_join = "MATCH" + if source.get("expected_join") not in JOINS: + raise VerificationError(f"{case_id}.expected_join is invalid") + if source["expected_join"] != derived_join: + raise VerificationError( + f"{case_id} expected_join mismatch: " + f"{source['expected_join']} != {derived_join}" + ) + + +def verify_case( + case: dict[str, Any], report_cases: dict[str, Any] +) -> None: + case_id = text(case.get("case_id"), "case_id") + record = build_record(case) + authorization_ref = digest(record) + expected = case.get("expected") + if not isinstance(expected, dict): + raise VerificationError(f"{case_id}.expected must be an object") + exact( + expected, + { + "execution_allowed", "current_state", + "authority_state_at_report", + "expected_additional_side_effects", "authorization_ref", + }, + f"{case_id}.expected", + ) + if sha( + expected.get("authorization_ref"), + f"{case_id}.expected.authorization_ref", + ) != authorization_ref: + raise VerificationError( + f"{case_id} authorization_ref mismatch: {authorization_ref}" + ) + + report_entry = report_cases.get(case_id) + reported_at = None + if report_entry is not None: + if not isinstance(report_entry, dict): + raise VerificationError( + f"{case_id} report context must be an object" + ) + exact(report_entry, {"reported_at"}, f"{case_id}.report") + reported_at = time_value( + report_entry["reported_at"], f"{case_id}.reported_at" + ) + + intrinsic = intrinsic_state(record) + if record["current_state"] != intrinsic: + raise VerificationError( + f"{case_id} exported current_state mismatch: " + f"{record['current_state']} != {intrinsic}" + ) + if expected.get("current_state") != intrinsic: + raise VerificationError(f"{case_id} expected current_state mismatch") + evaluated = report_state(record, reported_at) + if expected.get("authority_state_at_report") != evaluated: + raise VerificationError( + f"{case_id} authority_state_at_report mismatch" + ) + + allowed = ( + record["decision"] == "ACCEPT" + and evaluated == "ACTIVE" + and record["consumption_state"] == "UNCONSUMED" + ) + if expected.get("execution_allowed") is not allowed: + raise VerificationError(f"{case_id} execution_allowed mismatch") + if expected.get("expected_additional_side_effects") != ( + 1 if allowed else 0 + ): + raise VerificationError( + f"{case_id} side-effect expectation mismatch" + ) + verify_handoff( + case_id, case.get("handoff"), record, authorization_ref, + reported_at, evaluated, + ) + + +def main() -> int: + try: + fixture = load(FIXTURE) + report = load(REPORT_CONTEXT) + if fixture.get("profile") != EXPORT_PROFILE: + raise VerificationError("fixture profile mismatch") + if report.get("profile") != REPORT_PROFILE: + raise VerificationError("report profile mismatch") + cases = fixture.get("cases") + report_cases = report.get("cases") + if not isinstance(cases, list) or not isinstance(report_cases, dict): + raise VerificationError("fixture shape mismatch") + + seen: set[str] = set() + for case in cases: + if not isinstance(case, dict): + raise VerificationError("case entry must be an object") + case_id = text(case.get("case_id"), "case_id") + if case_id in seen: + raise VerificationError(f"duplicate case_id: {case_id}") + seen.add(case_id) + verify_case(case, report_cases) + print(f"PASS {case_id}") + unknown = set(report_cases) - seen + if unknown: + raise VerificationError( + "unknown report cases: " + ", ".join(sorted(unknown)) + ) + print( + f"\nIndependent ProofPath authorization semantics passed: " + f"{len(cases)}" + ) + return 0 + except (VerificationError, KeyError, TypeError) as error: + print(f"ERROR: {error}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main())