diff --git a/src/agent_passport/receipt_core/__init__.py b/src/agent_passport/receipt_core/__init__.py new file mode 100644 index 0000000..1ea9f0b --- /dev/null +++ b/src/agent_passport/receipt_core/__init__.py @@ -0,0 +1,37 @@ +"""APS v1 receipt core. + +New signed artifacts in this package use strict RFC 8785 bytes and explicit +domain separation. Legacy formats are classified, never guessed by trying +multiple canonicalizers. +""" + +from .decision_ref import ( + build_decision_ref_v1, + compute_decision_component_ref_v1, + compute_decision_ref_v1, + normalize_core_decision_output_v1, +) +from .receipt import ( + compute_receipt_id_v1, + create_receipt_v1, + receipt_id_payload_v1, + receipt_signature_payload_v1, + validate_receipt_v1, + verify_receipt_v1, +) +from .supporting_record import ( + build_evidence_bundle_body_v2, + build_evidence_bundle_proof_v2, + classify_supporting_record_format, + compute_supporting_record_id_v1, + create_supporting_record_v1, + evidence_bundle_merkle_root_v2, + supporting_record_id_payload_v1, + supporting_record_signature_payload_v1, + validate_supporting_record_v1, + verify_evidence_bundle_body_v2, + verify_evidence_bundle_proof_v2, + verify_supporting_record_v1, +) + +__all__ = [name for name in globals() if not name.startswith("_")] diff --git a/src/agent_passport/receipt_core/decision_ref.py b/src/agent_passport/receipt_core/decision_ref.py new file mode 100644 index 0000000..6729f65 --- /dev/null +++ b/src/agent_passport/receipt_core/decision_ref.py @@ -0,0 +1,93 @@ +"""Content-derived policy decision references.""" + +from __future__ import annotations + +import hashlib +import re +import unicodedata + +from .jcs import assert_exact_keys, strict_jcs +from .receipt import _is_exact_utc_milliseconds + +DECISION_REF_TAG = "APS-DECISION-REF-V1" +DECISION_COMPONENT_TAGS = { + "authority": "APS-DECISION-AUTHORITY-V1", + "policy": "APS-DECISION-POLICY-V1", + "context": "APS-DECISION-CONTEXT-V1", + "output": "APS-DECISION-OUTPUT-V1", +} +HEX64 = re.compile(r"^[0-9a-f]{64}$") + + +def _sha256_hex(value: str) -> str: + return hashlib.sha256(value.encode("utf-8")).hexdigest() + + +def compute_decision_component_ref_v1(kind: str, value) -> str: + if kind not in DECISION_COMPONENT_TAGS: + raise ValueError("unknown decision component") + return _sha256_hex(f"{DECISION_COMPONENT_TAGS[kind]}\0{strict_jcs(value)}") + + +def validate_decision_ref_input_v1(value: dict) -> None: + keys = {"profile", "action_ref", "authority_state_ref", "policy_ref", "context_ref", "decision_output_ref"} + assert_exact_keys(value, keys, keys, "DecisionRefInputV1") + strict_jcs(value) + if value["profile"] != "aps-decision-ref-v1": + raise ValueError("DecisionRefInputV1: profile") + for key in keys - {"profile"}: + if not isinstance(value[key], str) or not HEX64.fullmatch(value[key]): + raise ValueError(f"DecisionRefInputV1: {key} must be lowercase sha256 hex") + + +def compute_decision_ref_v1(value: dict) -> str: + validate_decision_ref_input_v1(value) + return _sha256_hex(f"{DECISION_REF_TAG}\0{strict_jcs(value)}") + + +def normalize_core_decision_output_v1(value: dict) -> dict: + keys = {"profile", "verdict", "effective_authority_ref", "constraints", "valid_until"} + assert_exact_keys(value, keys, keys, "CoreDecisionOutputV1") + strict_jcs(value) + if value["profile"] != "aps-core-decision-output-v1": + raise ValueError("CoreDecisionOutputV1: profile") + verdict = value["verdict"] + if verdict not in {"permit", "deny", "narrow"}: + raise ValueError("CoreDecisionOutputV1: verdict") + effective = value["effective_authority_ref"] + if effective is not None and (not isinstance(effective, str) or not HEX64.fullmatch(effective)): + raise ValueError("CoreDecisionOutputV1: effective_authority_ref") + if verdict == "deny" and effective is not None: + raise ValueError("CoreDecisionOutputV1: deny requires null effective_authority_ref") + if verdict != "deny" and effective is None: + raise ValueError("CoreDecisionOutputV1: permit/narrow require effective_authority_ref") + constraints = value["constraints"] + if not isinstance(constraints, list) or not all(isinstance(item, str) for item in constraints): + raise ValueError("CoreDecisionOutputV1: constraints") + valid_until = value["valid_until"] + if verdict == "deny": + if valid_until is not None: + raise ValueError("CoreDecisionOutputV1: deny requires null valid_until") + elif not isinstance(valid_until, str) or not _is_exact_utc_milliseconds(valid_until): + raise ValueError("CoreDecisionOutputV1: permit/narrow require valid_until as exact UTC milliseconds") + normalized = sorted({unicodedata.normalize("NFC", item) for item in constraints}) + return {**value, "constraints": normalized} + + +def build_decision_ref_v1(*, action_ref: str, authority_state, policy_input, decision_context, decision_output) -> dict: + if not isinstance(action_ref, str) or not HEX64.fullmatch(action_ref): + raise ValueError("action_ref must be lowercase sha256 hex") + if not isinstance(decision_output, dict): + raise ValueError("decision_output must be a CoreDecisionOutputV1 object") + # Normalize before hashing: the normalizer validates the closed five-member shape, so an + # unnormalized or malformed decision output cannot reach the digest through this path. + normalized_output = normalize_core_decision_output_v1(decision_output) + ref_input = { + "profile": "aps-decision-ref-v1", + "action_ref": action_ref, + "authority_state_ref": compute_decision_component_ref_v1("authority", authority_state), + "policy_ref": compute_decision_component_ref_v1("policy", policy_input), + "context_ref": compute_decision_component_ref_v1("context", decision_context), + "decision_output_ref": compute_decision_component_ref_v1("output", normalized_output), + } + return {"input": ref_input, "decision_ref": compute_decision_ref_v1(ref_input)} diff --git a/src/agent_passport/receipt_core/jcs.py b/src/agent_passport/receipt_core/jcs.py new file mode 100644 index 0000000..df08158 --- /dev/null +++ b/src/agent_passport/receipt_core/jcs.py @@ -0,0 +1,114 @@ +"""Strict new-write I-JSON validation followed by RFC 8785 canonicalization.""" + +from __future__ import annotations + +import math +import json + +from ..canonical import canonicalize_jcs + + +class IJsonValidationError(TypeError): + """An in-memory value cannot be represented as an APS new-write I-JSON value.""" + + +def _assert_scalar_string(value: str, path: str) -> None: + for char in value: + if 0xD800 <= ord(char) <= 0xDFFF: + raise IJsonValidationError(f"{path}: unpaired UTF-16 surrogate") + + +def assert_i_json(value, path: str = "$", ancestors: set[int] | None = None) -> None: + """Validate without converting custom objects, keys, dates, or missing values.""" + if ancestors is None: + ancestors = set() + if value is None or type(value) is bool: + return + if type(value) is str: + _assert_scalar_string(value, path) + return + if type(value) is int: + if abs(value) > 9_007_199_254_740_991: + raise IJsonValidationError(f"{path}: integer exceeds the interoperable IEEE 754 range") + return + if type(value) is float: + if not math.isfinite(value): + raise IJsonValidationError(f"{path}: non-finite number") + if value.is_integer() and abs(value) > 9_007_199_254_740_991: + raise IJsonValidationError(f"{path}: integer exceeds the interoperable IEEE 754 range") + return + if type(value) not in (list, dict): + raise IJsonValidationError(f"{path}: unsupported {type(value).__name__}") + identity = id(value) + if identity in ancestors: + raise IJsonValidationError(f"{path}: cyclic value") + ancestors.add(identity) + if type(value) is list: + for index, item in enumerate(value): + assert_i_json(item, f"{path}[{index}]", ancestors) + else: + for key, item in value.items(): + if not isinstance(key, str): + raise IJsonValidationError(f"{path}: object key is not a string") + _assert_scalar_string(key, f"{path} key") + assert_i_json(item, f"{path}.{key}", ancestors) + ancestors.remove(identity) + + +def strict_jcs(value) -> str: + assert_i_json(value) + return canonicalize_jcs(value) + + +def parse_strict_i_json(raw: str, max_utf8_bytes: int = 1_048_576, max_depth: int = 128): + """Parse bounded raw JSON while rejecting decoded duplicate member names.""" + if type(raw) is not str: + raise IJsonValidationError("$: raw JSON string required") + if type(max_utf8_bytes) is not int or max_utf8_bytes < 1 or len(raw.encode("utf-8", "surrogatepass")) > max_utf8_bytes: + raise IJsonValidationError("$: raw JSON size limit exceeded") + if type(max_depth) is not int or max_depth < 1: + raise IJsonValidationError("$: invalid depth limit") + + def pairs_hook(pairs): + value = {} + for key, item in pairs: + if key in value: + raise IJsonValidationError("$: duplicate object member") + value[key] = item + return value + + try: + value = json.loads( + raw, + object_pairs_hook=pairs_hook, + parse_constant=lambda token: (_ for _ in ()).throw(IJsonValidationError(f"$: invalid number {token}")), + ) + except IJsonValidationError: + raise + except (ValueError, TypeError, RecursionError) as exc: + raise IJsonValidationError("$: invalid JSON") from exc + + def check_depth(item, depth=1): + if depth > max_depth: + raise IJsonValidationError("$: JSON nesting limit exceeded") + if type(item) is list: + for child in item: + check_depth(child, depth + 1) + elif type(item) is dict: + for child in item.values(): + check_depth(child, depth + 1) + + check_depth(value) + assert_i_json(value) + return value + + +def assert_exact_keys(value: dict, allowed: set[str], required: set[str], name: str) -> None: + if not isinstance(value, dict): + raise IJsonValidationError(f"{name}: object required") + unknown = set(value) - allowed + if unknown: + raise IJsonValidationError(f"{name}: unknown field {sorted(unknown)[0]}") + missing = required - set(value) + if missing: + raise IJsonValidationError(f"{name}: missing field {sorted(missing)[0]}") diff --git a/src/agent_passport/receipt_core/receipt.py b/src/agent_passport/receipt_core/receipt.py new file mode 100644 index 0000000..48e6c2b --- /dev/null +++ b/src/agent_passport/receipt_core/receipt.py @@ -0,0 +1,146 @@ +"""Stable APS ReceiptV1 identifiers, signatures, and verification.""" + +from __future__ import annotations + +import copy +from datetime import datetime, timezone +import hashlib +import re + +from ..crypto import sign, verify +from .jcs import assert_exact_keys, strict_jcs + +RECEIPT_ID_TAG = "APS-RECEIPT-ID-V1" +RECEIPT_SIG_TAG = "APS-RECEIPT-SIG-V1" +HEX64 = re.compile(r"^[0-9a-f]{64}$") +HEX128 = re.compile(r"^[0-9a-f]{128}$") +UTC_MS = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$") + + +def _sha256_hex(value: str) -> str: + return hashlib.sha256(value.encode("utf-8")).hexdigest() + + +def _is_exact_utc_milliseconds(value: str) -> bool: + if not UTC_MS.fullmatch(value): + return False + try: + parsed = datetime.strptime(value, "%Y-%m-%dT%H:%M:%S.%fZ").replace(tzinfo=timezone.utc) + except ValueError: + return False + return parsed.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z" == value + + +def _without(receipt: dict, *keys: str) -> dict: + return {key: copy.deepcopy(value) for key, value in receipt.items() if key not in keys} + + +def receipt_id_payload_v1(receipt: dict) -> str: + return f"{RECEIPT_ID_TAG}\0{strict_jcs(_without(receipt, 'receipt_id', 'signatures'))}" + + +def compute_receipt_id_v1(receipt: dict) -> str: + return _sha256_hex(receipt_id_payload_v1(receipt)) + + +def receipt_signature_payload_v1(receipt: dict, descriptor: dict) -> str: + form = {"receipt": _without(receipt, "signatures"), "signer": copy.deepcopy(descriptor)} + return f"{RECEIPT_SIG_TAG}\0{strict_jcs(form)}" + + +def validate_receipt_v1(receipt: dict, require_values: bool = True) -> None: + allowed = {"profile", "receipt_id", "receipt_type", "issuer", "subject_agent", "action_ref", "delegation_ref", "decision_ref", "issued_at", "evidence_refs", "result", "prev", "signatures"} + required = allowed - {"decision_ref", "prev"} + assert_exact_keys(receipt, allowed, required, "ReceiptV1") + strict_jcs(receipt) + if receipt["profile"] != "aps-receipt-v1": + raise ValueError("ReceiptV1: profile") + for key in ("receipt_type", "issuer", "subject_agent", "delegation_ref"): + if not isinstance(receipt[key], str) or not receipt[key]: + raise ValueError("ReceiptV1: empty identifier") + if require_values and (not isinstance(receipt["receipt_id"], str) or not HEX64.fullmatch(receipt["receipt_id"])): + raise ValueError("ReceiptV1: receipt_id") + if not isinstance(receipt["action_ref"], str) or not HEX64.fullmatch(receipt["action_ref"]): + raise ValueError("ReceiptV1: action_ref") + for key in ("decision_ref", "prev"): + if key in receipt and (not isinstance(receipt[key], str) or not HEX64.fullmatch(receipt[key])): + raise ValueError(f"ReceiptV1: {key}") + if not isinstance(receipt["issued_at"], str) or not _is_exact_utc_milliseconds(receipt["issued_at"]): + raise ValueError("ReceiptV1: issued_at") + if not isinstance(receipt["result"], dict): + raise ValueError("ReceiptV1: result") + if not isinstance(receipt["evidence_refs"], list) or not isinstance(receipt["signatures"], list): + raise ValueError("ReceiptV1: arrays") + seen = set() + previous = None + for ref in receipt["evidence_refs"]: + assert_exact_keys(ref, {"artifact_type", "sha256"}, {"artifact_type", "sha256"}, "EvidenceRefV1") + if not isinstance(ref["artifact_type"], str) or not ref["artifact_type"] or not isinstance(ref["sha256"], str) or not HEX64.fullmatch(ref["sha256"]): + raise ValueError("EvidenceRefV1: value") + order = (ref["artifact_type"].encode("utf-8"), ref["sha256"].encode("utf-8")) + if order in seen: + raise ValueError("ReceiptV1: duplicate evidence_ref") + if previous is not None and previous >= order: + raise ValueError("ReceiptV1: evidence_refs not sorted") + seen.add(order) + previous = order + seen = set() + previous = None + for proof in receipt["signatures"]: + keys = {"signer", "key_id", "alg", "value"} + assert_exact_keys(proof, keys, keys, "ReceiptSignatureV1") + if not isinstance(proof["signer"], str) or not proof["signer"] or not isinstance(proof["key_id"], str) or not proof["key_id"] or proof["alg"] != "Ed25519": + raise ValueError("ReceiptSignatureV1: value") + if require_values and (not isinstance(proof["value"], str) or not HEX128.fullmatch(proof["value"])): + raise ValueError("ReceiptSignatureV1: value") + order = (proof["signer"].encode("utf-8"), proof["key_id"].encode("utf-8")) + if order in seen: + raise ValueError("ReceiptV1: duplicate signature") + if previous is not None and previous >= order: + raise ValueError("ReceiptV1: signatures not sorted") + seen.add(order) + previous = order + if require_values and not any(proof["signer"] == receipt["issuer"] for proof in receipt["signatures"]): + raise ValueError("ReceiptV1: issuer signature missing") + + +def create_receipt_v1(fields: dict, signers: list[dict]) -> dict: + if not signers: + raise ValueError("ReceiptV1: at least one signer") + receipt = copy.deepcopy(fields) + receipt["evidence_refs"] = sorted(receipt["evidence_refs"], key=lambda ref: (ref["artifact_type"].encode("utf-8"), ref["sha256"].encode("utf-8"))) + descriptors = sorted(signers, key=lambda item: (item["signer"].encode("utf-8"), item["key_id"].encode("utf-8"))) + receipt["receipt_id"] = "0" * 64 + receipt["signatures"] = [] + validate_receipt_v1(receipt, False) + receipt["receipt_id"] = compute_receipt_id_v1(receipt) + receipt["signatures"] = [] + for item in descriptors: + descriptor = {"signer": item["signer"], "key_id": item["key_id"], "alg": "Ed25519"} + receipt["signatures"].append({**descriptor, "value": sign(receipt_signature_payload_v1(receipt, descriptor), item["private_key"])}) + validate_receipt_v1(receipt) + return receipt + + +def verify_receipt_v1(receipt: dict, resolve_key) -> dict: + try: + validate_receipt_v1(receipt) + except (TypeError, ValueError) as exc: + return {"valid": False, "receipt_id_valid": False, "signature_results": [], "errors": [str(exc)]} + id_valid = compute_receipt_id_v1(receipt) == receipt["receipt_id"] + errors = [] if id_valid else ["receipt_id_mismatch"] + results = [] + for proof in receipt["signatures"]: + try: + public_key = resolve_key(proof["signer"], proof["key_id"], receipt["issued_at"]) + except Exception: + results.append({"signer": proof["signer"], "key_id": proof["key_id"], "valid": False, "reason": "key_resolution_error"}) + continue + if public_key is None: + results.append({"signer": proof["signer"], "key_id": proof["key_id"], "valid": False, "reason": "key_unresolved"}) + continue + descriptor = {"signer": proof["signer"], "key_id": proof["key_id"], "alg": proof["alg"]} + results.append({"signer": proof["signer"], "key_id": proof["key_id"], "valid": verify(receipt_signature_payload_v1(receipt, descriptor), proof["value"], public_key)}) + if any(not item["valid"] for item in results): + errors.append("signature_invalid") + return {"valid": not errors, "receipt_id_valid": id_valid, "signature_results": results, "errors": errors} diff --git a/src/agent_passport/receipt_core/supporting_record.py b/src/agent_passport/receipt_core/supporting_record.py new file mode 100644 index 0000000..dfa6a9d --- /dev/null +++ b/src/agent_passport/receipt_core/supporting_record.py @@ -0,0 +1,269 @@ +"""Typed supporting records and EvidenceBundle v2 content binding.""" + +from __future__ import annotations + +import copy +from datetime import datetime, timezone +import hashlib +import re + +from ..crypto import sign, verify +from .jcs import assert_exact_keys, strict_jcs + +SUPPORTING_RECORD_ID_TAG = "APS-SUPPORTING-RECORD-ID-V1" +SUPPORTING_RECORD_SIG_TAG = "APS-SUPPORTING-RECORD-SIG-V1" +HEX64 = re.compile(r"^[0-9a-f]{64}$") +HEX128 = re.compile(r"^[0-9a-f]{128}$") +UTC_MS = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$") + + +def _sha256(value: bytes) -> bytes: + return hashlib.sha256(value).digest() + + +def _is_exact_utc_milliseconds(value: str) -> bool: + if not UTC_MS.fullmatch(value): + return False + try: + parsed = datetime.strptime(value, "%Y-%m-%dT%H:%M:%S.%fZ").replace(tzinfo=timezone.utc) + except ValueError: + return False + return parsed.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z" == value + + +def _without(record: dict, *keys: str) -> dict: + return {key: copy.deepcopy(value) for key, value in record.items() if key not in keys} + + +def supporting_record_id_payload_v1(record: dict) -> str: + return f"{SUPPORTING_RECORD_ID_TAG}\0{record['record_type']}\0{strict_jcs(_without(record, 'record_id', 'sig'))}" + + +def compute_supporting_record_id_v1(record: dict) -> str: + return _sha256(supporting_record_id_payload_v1(record).encode()).hex() + + +def supporting_record_signature_payload_v1(record: dict) -> str: + return f"{SUPPORTING_RECORD_SIG_TAG}\0{record['record_type']}\0{strict_jcs(_without(record, 'sig'))}" + + +def validate_supporting_record_v1(record: dict, require_crypto: bool = True) -> None: + allowed = {"profile", "record_id", "record_type", "issuer", "issuer_key_id", "issued_at", "action_ref", "body", "sig_alg", "sig"} + required = allowed - {"action_ref"} + assert_exact_keys(record, allowed, required, "SupportingRecordV1") + strict_jcs(record) + if record["profile"] != "aps-supporting-record-v1": + raise ValueError("SupportingRecordV1: profile") + for key in ("record_type", "issuer", "issuer_key_id"): + if not isinstance(record[key], str) or not record[key]: + raise ValueError("SupportingRecordV1: identifier") + if not isinstance(record["issued_at"], str) or not _is_exact_utc_milliseconds(record["issued_at"]): + raise ValueError("SupportingRecordV1: issued_at") + if "action_ref" in record and (not isinstance(record["action_ref"], str) or not HEX64.fullmatch(record["action_ref"])): + raise ValueError("SupportingRecordV1: action_ref") + if not isinstance(record["body"], dict): + raise ValueError("SupportingRecordV1: body") + if record["sig_alg"] != "Ed25519": + raise ValueError("SupportingRecordV1: sig_alg") + if require_crypto and (not isinstance(record["record_id"], str) or not HEX64.fullmatch(record["record_id"]) or not isinstance(record["sig"], str) or not HEX128.fullmatch(record["sig"])): + raise ValueError("SupportingRecordV1: crypto encoding") + + +def create_supporting_record_v1(fields: dict, private_key: str) -> dict: + record = copy.deepcopy(fields) + record.update({"record_id": "0" * 64, "sig": ""}) + validate_supporting_record_v1(record, False) + record["record_id"] = compute_supporting_record_id_v1(record) + record["sig"] = sign(supporting_record_signature_payload_v1(record), private_key) + validate_supporting_record_v1(record) + return record + + +def verify_supporting_record_v1(record: dict, public_key: str) -> dict: + try: + validate_supporting_record_v1(record) + except (TypeError, ValueError): + return {"valid": False, "id_valid": False, "signature_valid": False} + id_valid = compute_supporting_record_id_v1(record) == record["record_id"] + signature_valid = verify(supporting_record_signature_payload_v1(record), record["sig"], public_key) + return {"valid": id_valid and signature_valid, "id_valid": id_valid, "signature_valid": signature_valid} + + +def _entry_bytes(entry: dict) -> bytes: + return strict_jcs(entry).encode("utf-8") + + +def evidence_bundle_merkle_root_v2(entries: list[dict]) -> str: + if not entries: + raise ValueError("EvidenceBundleV2: at least one member") + seen = set() + canonical_entries = [] + previous = None + for entry in entries: + keys = {"member_id", "member_type", "sha256"} + assert_exact_keys(entry, keys, keys, "EvidenceBundleMemberV2") + if not isinstance(entry["member_id"], str) or not entry["member_id"] or not isinstance(entry["member_type"], str) or not entry["member_type"] or not isinstance(entry["sha256"], str) or not HEX64.fullmatch(entry["sha256"]): + raise ValueError("EvidenceBundleMemberV2: value") + if entry["member_id"] in seen: + raise ValueError("EvidenceBundleV2: duplicate member_id") + seen.add(entry["member_id"]) + canonical = _entry_bytes(entry) + if previous is not None and previous >= canonical: + raise ValueError("EvidenceBundleV2: members not sorted") + previous = canonical + canonical_entries.append(canonical) + level = [_sha256(b"\x00" + canonical) for canonical in canonical_entries] + while len(level) > 1: + next_level = [] + for index in range(0, len(level), 2): + if index + 1 == len(level): + next_level.append(level[index]) + else: + next_level.append(_sha256(b"\x01" + level[index] + level[index + 1])) + level = next_level + return level[0].hex() + + +def build_evidence_bundle_body_v2(members: list[dict]) -> dict: + if not members: + raise ValueError("EvidenceBundleV2: at least one member") + seen = set() + entries = [] + for member in members: + assert_exact_keys(member, {"member_id", "member_type", "payload"}, {"member_id", "member_type", "payload"}, "EvidenceBundleMemberInputV2") + if not isinstance(member["member_id"], str) or not member["member_id"] or not isinstance(member["member_type"], str) or not member["member_type"]: + raise ValueError("EvidenceBundleV2: member identifier") + if member["member_id"] in seen: + raise ValueError("EvidenceBundleV2: duplicate member_id") + seen.add(member["member_id"]) + entries.append({"member_id": member["member_id"], "member_type": member["member_type"], "sha256": _sha256(strict_jcs(member["payload"]).encode()).hex()}) + entries.sort(key=_entry_bytes) + return {"members": entries, "merkle_root": evidence_bundle_merkle_root_v2(entries)} + + +def verify_evidence_bundle_body_v2(body: dict, payloads: dict | None = None) -> bool: + try: + assert_exact_keys(body, {"members", "merkle_root"}, {"members", "merkle_root"}, "EvidenceBundleBodyV2") + if not isinstance(body["members"], list) or not body["members"] or not isinstance(body["merkle_root"], str) or not HEX64.fullmatch(body["merkle_root"]): + return False + seen = set() + previous = None + for entry in body["members"]: + keys = {"member_id", "member_type", "sha256"} + assert_exact_keys(entry, keys, keys, "EvidenceBundleMemberV2") + if not isinstance(entry["member_id"], str) or not entry["member_id"] or not isinstance(entry["member_type"], str) or not entry["member_type"] or not isinstance(entry["sha256"], str) or not HEX64.fullmatch(entry["sha256"]): + return False + if entry["member_id"] in seen: + return False + seen.add(entry["member_id"]) + canonical = _entry_bytes(entry) + if previous is not None and previous >= canonical: + return False + previous = canonical + if payloads is not None and (entry["member_id"] not in payloads or _sha256(strict_jcs(payloads[entry["member_id"]]).encode()).hex() != entry["sha256"]): + return False + return evidence_bundle_merkle_root_v2(body["members"]) == body["merkle_root"] + except (TypeError, ValueError, KeyError): + return False + + +def _evidence_leaf(entry: dict) -> bytes: + return _sha256(b"\x00" + _entry_bytes(entry)) + + +def build_evidence_bundle_proof_v2(entries: list[dict], member_id: str) -> dict: + """Build a shape-checked inclusion proof; odd nodes are promoted, not copied.""" + evidence_bundle_merkle_root_v2(entries) + try: + leaf_index = next(index for index, entry in enumerate(entries) if entry["member_id"] == member_id) + except StopIteration as exc: + raise ValueError("EvidenceBundleV2: member not found") from exc + index = leaf_index + level = [_evidence_leaf(entry) for entry in entries] + path = [] + while len(level) > 1: + if index % 2 == 1: + path.append({"position": "left", "sha256": level[index - 1].hex()}) + elif index + 1 < len(level): + path.append({"position": "right", "sha256": level[index + 1].hex()}) + else: + path.append({"position": "promote"}) + next_level = [] + for offset in range(0, len(level), 2): + if offset + 1 == len(level): + next_level.append(level[offset]) + else: + next_level.append(_sha256(b"\x01" + level[offset] + level[offset + 1])) + index //= 2 + level = next_level + return { + "profile": "aps-evidence-proof-v2", + "member": copy.deepcopy(entries[leaf_index]), + "leaf_index": leaf_index, + "leaf_count": len(entries), + "path": path, + } + + +_PAYLOAD_ABSENT = object() + + +def verify_evidence_bundle_proof_v2(proof: dict, trusted_root: str, payload=_PAYLOAD_ABSENT) -> bool: + try: + keys = {"profile", "member", "leaf_index", "leaf_count", "path"} + assert_exact_keys(proof, keys, keys, "EvidenceBundleProofV2") + if proof["profile"] != "aps-evidence-proof-v2" or not isinstance(trusted_root, str) or not HEX64.fullmatch(trusted_root): + return False + member = proof["member"] + member_keys = {"member_id", "member_type", "sha256"} + assert_exact_keys(member, member_keys, member_keys, "EvidenceBundleMemberV2") + if not isinstance(member["member_id"], str) or not member["member_id"] or not isinstance(member["member_type"], str) or not member["member_type"] or not isinstance(member["sha256"], str) or not HEX64.fullmatch(member["sha256"]): + return False + leaf_index = proof["leaf_index"] + leaf_count = proof["leaf_count"] + if isinstance(leaf_index, bool) or isinstance(leaf_count, bool) or not isinstance(leaf_index, int) or not isinstance(leaf_count, int) or leaf_count < 1 or leaf_index < 0 or leaf_index >= leaf_count or not isinstance(proof["path"], list): + return False + if payload is not _PAYLOAD_ABSENT and _sha256(strict_jcs(payload).encode()).hex() != member["sha256"]: + return False + index, width = leaf_index, leaf_count + digest = _evidence_leaf(member) + path_index = 0 + while width > 1: + if path_index >= len(proof["path"]): + return False + step = proof["path"][path_index] + path_index += 1 + expected = "left" if index % 2 == 1 else "right" if index + 1 < width else "promote" + if not isinstance(step, dict) or step.get("position") != expected: + return False + if expected == "promote": + assert_exact_keys(step, {"position"}, {"position"}, "EvidenceBundleProofStepV2") + else: + assert_exact_keys(step, {"position", "sha256"}, {"position", "sha256"}, "EvidenceBundleProofStepV2") + if not isinstance(step["sha256"], str) or not HEX64.fullmatch(step["sha256"]): + return False + sibling = bytes.fromhex(step["sha256"]) + digest = _sha256(b"\x01" + sibling + digest) if expected == "left" else _sha256(b"\x01" + digest + sibling) + index //= 2 + width = (width + 1) // 2 + return path_index == len(proof["path"]) and digest.hex() == trusted_root + except (TypeError, ValueError, KeyError): + return False + + +def classify_supporting_record_format(value) -> dict: + if not isinstance(value, dict): + return {"format": "unknown", "canonicalization": "unknown", "legacy": False} + if value.get("profile") == "aps-supporting-record-v1": + return {"format": "supporting-record-v1", "canonicalization": "rfc8785", "legacy": False} + if value.get("profile") == "aps-composition-check-v0": + return {"format": "composition-check-v0", "canonicalization": "rfc8785-tagged-v0", "legacy": True} + if value.get("spec_version") == "0.1.0" and value.get("record_type") == "accountability_record": + return {"format": "accountability-record-0.1.0", "canonicalization": "rfc8785-untagged", "legacy": True} + if value.get("type") == "read_fidelity_receipt": + return {"format": "read-fidelity-unwrapped", "canonicalization": "rfc8785-untagged", "legacy": True} + if isinstance(value.get("manifest"), dict) and value["manifest"].get("profile") == "aps:evidence-bundle:v1": + return {"format": "evidence-bundle-v1", "canonicalization": "aps-legacy-null-dropping", "legacy": True} + if "authority_ref" in value and "observer_key" in value and "profile" not in value: + return {"format": "revocation-observation-unversioned", "canonicalization": "aps-legacy-null-dropping", "legacy": True} + return {"format": "unknown", "canonicalization": "unknown", "legacy": False} diff --git a/tests/test_receipt_core_v1.py b/tests/test_receipt_core_v1.py new file mode 100644 index 0000000..27f889f --- /dev/null +++ b/tests/test_receipt_core_v1.py @@ -0,0 +1,196 @@ +import copy +import re + +import pytest + +from agent_passport.crypto import public_key_from_private +from agent_passport.receipt_core import ( + build_decision_ref_v1, + compute_decision_component_ref_v1, + build_evidence_bundle_body_v2, + build_evidence_bundle_proof_v2, + classify_supporting_record_format, + compute_receipt_id_v1, + create_receipt_v1, + create_supporting_record_v1, + normalize_core_decision_output_v1, + verify_evidence_bundle_body_v2, + verify_evidence_bundle_proof_v2, + verify_receipt_v1, + verify_supporting_record_v1, +) +from agent_passport.receipt_core.jcs import IJsonValidationError, parse_strict_i_json, strict_jcs + +PRIVATE_KEY = "00" * 32 +PUBLIC_KEY = public_key_from_private(PRIVATE_KEY) +KAT = { + # Repinned: build_decision_ref_v1 now normalizes before hashing, so the decision output + # must be a valid five-member CoreDecisionOutputV1. Was 2157809a9a722314ae19dce7a242ea3b54a8948230fab2fab5d5dc15bd663dc2. + "decision_ref": "e474f27bc7e228cd515d4192936cd5525ac8f362fa95173c82eaff02059389e7", + "receipt_id": "89b0b77807e99845aab403f01bcdaa2f02949f6c9db84e1aca6c0a8449e4d023", + "receipt_sig": "83deb713568bbdf0c85e1a6d46345530e84dbe86cdefe1cb608b0f14372c176a9e69e0033db11c4c44ff84be3de3bee5e212707eb84206f6c34455206d37f90b", + "merkle_root": "03700eeba1b453086063612d3df73f711827735c3fe30cf8a8a2a6379a6f6d5f", + "record_id": "7d73684a65444088e841f2b30f0ecf139fbadbeab277d57d20d1a2ef5fe2a7b2", + "record_sig": "56a2116eb4e259a336c36a646e69322cf2e7850b7202f2093c5957e4e6a100cf4cae8048cb4647bbc557447bea531dd7e69c117e9298515cc767110cf0f7d809", + "proof_root": "dab9f2f5f3571345327f0144f2eafbb6e835ac5cbb48e9d789224e135ce16247", + "proof_left": "f3bdfcf031dea9da3129ae67bdf7f69caefc41660541d0acb015e1b49ae95470", +} + + +def hx(char): + return char * 64 + + +KAT_OUTPUT = {"profile": "aps-core-decision-output-v1", "verdict": "permit", + "effective_authority_ref": "b" * 64, "constraints": [], + "valid_until": "2026-04-08T12:00:05.000Z"} +BUILDER_PLACEHOLDERS = {"action_ref": "a" * 64, "authority_state": {"scope": ["read"], "revoked": False}, + "policy_input": {"id": "p1", "version": "1"}, "decision_context": {"tenant": "t1"}} +CASE1_OUTPUT = {"profile": "aps-core-decision-output-v1", "verdict": "permit", + "effective_authority_ref": "3f2a1c9d8e7b6a5f4e3d2c1b0a9f8e7d6c5b4a39281706f5e4d3c2b1a0987654", + "constraints": ["commerce:read"], "valid_until": "2026-04-08T12:00:05.000Z"} + + +def test_decision_ref_is_content_derived_and_normalizes_constraints(): + first = build_decision_ref_v1( + action_ref=hx("a"), authority_state={"scope": ["read"], "revoked": False}, + policy_input={"id": "p1", "version": "1"}, decision_context={"tenant": "t1"}, + decision_output=KAT_OUTPUT, + ) + assert first["decision_ref"] == KAT["decision_ref"] + reordered = build_decision_ref_v1( + action_ref=hx("a"), authority_state={"revoked": False, "scope": ["read"]}, + policy_input={"version": "1", "id": "p1"}, decision_context={"tenant": "t1"}, + decision_output={"valid_until": "2026-04-08T12:00:05.000Z", "constraints": [], + "effective_authority_ref": "b" * 64, "verdict": "permit", + "profile": "aps-core-decision-output-v1"}, + ) + assert first["decision_ref"] == reordered["decision_ref"] + output = normalize_core_decision_output_v1({"profile": "aps-core-decision-output-v1", "verdict": "narrow", "effective_authority_ref": hx("b"), "constraints": ["é", "read", "e\u0301", "read"], "valid_until": "2026-04-08T12:00:05.000Z"}) + assert output["constraints"] == ["read", "é"] + + +def test_receipt_binds_id_signer_descriptor_and_content(): + receipt = create_receipt_v1({ + "profile": "aps-receipt-v1", "receipt_type": "aps:action:v1", "issuer": "did:example:issuer", + "subject_agent": "did:example:agent", "action_ref": hx("a"), "delegation_ref": hx("b"), + "decision_ref": hx("c"), "issued_at": "2026-07-18T12:00:00.000Z", + "evidence_refs": [{"artifact_type": "z", "sha256": hx("e")}, {"artifact_type": "a", "sha256": hx("d")}], + "result": {"status": "success", "detail": None}, + }, [{"signer": "did:example:issuer", "key_id": "key-1", "private_key": PRIVATE_KEY}]) + assert receipt["receipt_id"] == compute_receipt_id_v1(receipt) + assert receipt["receipt_id"] == KAT["receipt_id"] + assert receipt["signatures"][0]["value"] == KAT["receipt_sig"] + assert verify_receipt_v1(receipt, lambda *_: PUBLIC_KEY)["valid"] + relabeled = copy.deepcopy(receipt) + relabeled["signatures"][0]["key_id"] = "key-2" + assert not verify_receipt_v1(relabeled, lambda *_: PUBLIC_KEY)["valid"] + tampered = copy.deepcopy(receipt) + tampered["result"]["status"] = "failure" + assert not verify_receipt_v1(tampered, lambda *_: PUBLIC_KEY)["valid"] + + +def test_supporting_record_and_bundle_bind_every_member_axis(): + payloads = {"m1": {"value": None}, "m2": {"value": 2}} + bundle = build_evidence_bundle_body_v2([ + {"member_id": "m2", "member_type": "two", "payload": payloads["m2"]}, + {"member_id": "m1", "member_type": "one", "payload": payloads["m1"]}, + ]) + assert bundle["merkle_root"] == KAT["merkle_root"] + assert verify_evidence_bundle_body_v2(bundle, payloads) + record = create_supporting_record_v1({ + "profile": "aps-supporting-record-v1", "record_type": "aps:evidence-bundle:v2", + "issuer": "did:example:issuer", "issuer_key_id": "key-1", "issued_at": "2026-07-18T12:00:00.000Z", + "body": bundle, "sig_alg": "Ed25519", + }, PRIVATE_KEY) + assert record["record_id"] == KAT["record_id"] + assert record["sig"] == KAT["record_sig"] + assert verify_supporting_record_v1(record, PUBLIC_KEY)["valid"] + changed = copy.deepcopy(bundle) + changed["members"][0]["member_type"] = "changed" + assert not verify_evidence_bundle_body_v2(changed, payloads) + three = build_evidence_bundle_body_v2([ + {"member_id": "m3", "member_type": "three", "payload": {"value": 3}}, + {"member_id": "m1", "member_type": "one", "payload": {"value": 1}}, + {"member_id": "m2", "member_type": "two", "payload": {"value": 2}}, + ]) + proof = build_evidence_bundle_proof_v2(three["members"], "m3") + assert three["merkle_root"] == KAT["proof_root"] + assert proof["path"] == [{"position": "promote"}, {"position": "left", "sha256": KAT["proof_left"]}] + assert any(step["position"] == "promote" for step in proof["path"]) + assert verify_evidence_bundle_proof_v2(proof, three["merkle_root"], {"value": 3}) + malformed = copy.deepcopy(proof) + malformed["leaf_count"] = 4 + assert not verify_evidence_bundle_proof_v2(malformed, three["merkle_root"], {"value": 3}) + + +def test_strict_new_write_and_explicit_legacy_dispatch(): + with pytest.raises(IJsonValidationError): + strict_jcs({"x": object()}) + with pytest.raises(IJsonValidationError): + strict_jcs({"x": "\ud800"}) + with pytest.raises(IJsonValidationError, match="IEEE 754"): + strict_jcs({"integer": 9_007_199_254_740_992}) + with pytest.raises(IJsonValidationError, match="duplicate object member"): + parse_strict_i_json('{"a":1,"\\u0061":2}') + assert classify_supporting_record_format({"manifest": {"profile": "aps:evidence-bundle:v1"}})["format"] == "evidence-bundle-v1" + assert classify_supporting_record_format({"profile": "future"})["format"] == "unknown" + + +def test_core_decision_output_binds_valid_until_into_the_hashed_preimage(): + permit = {"profile": "aps-core-decision-output-v1", "verdict": "permit", "effective_authority_ref": hx("b"), "constraints": ["commerce:read"], "valid_until": "2026-04-08T12:00:05.000Z"} + first = compute_decision_component_ref_v1("output", normalize_core_decision_output_v1(permit)) + later = compute_decision_component_ref_v1("output", normalize_core_decision_output_v1({**permit, "valid_until": "2026-04-08T12:00:06.000Z"})) + assert first != later + + deny = normalize_core_decision_output_v1({"profile": "aps-core-decision-output-v1", "verdict": "deny", "effective_authority_ref": None, "constraints": [], "valid_until": None}) + assert deny["valid_until"] is None + assert re.fullmatch(r"[0-9a-f]{64}", compute_decision_component_ref_v1("output", deny)) + + with pytest.raises(ValueError, match="valid_until"): + normalize_core_decision_output_v1({**permit, "valid_until": None}) + with pytest.raises(ValueError, match="valid_until"): + normalize_core_decision_output_v1({"profile": "aps-core-decision-output-v1", "verdict": "deny", "effective_authority_ref": None, "constraints": [], "valid_until": "2026-04-08T12:00:05.000Z"}) + with pytest.raises(Exception, match="valid_until"): + normalize_core_decision_output_v1({k: v for k, v in permit.items() if k != "valid_until"}) + with pytest.raises(ValueError, match="valid_until"): + normalize_core_decision_output_v1({**permit, "valid_until": "2026-04-08T12:00:05Z"}) + + +def test_build_decision_ref_v1_end_to_end_binds_the_normalized_five_member_output(): + built = build_decision_ref_v1(decision_output=CASE1_OUTPUT, **BUILDER_PLACEHOLDERS) + assert built["input"]["decision_output_ref"] == "4226e417c01f4d395db503bfcc00a2e022f5f864568b5fb2b88eefe4fbd0c551" + assert built["decision_ref"] == "8a9595fd9d7caf15654fca34816e22d02562f9db16dcc7d5958b906541dce8de" + + +def test_build_decision_ref_v1_rejects_every_malformed_decision_output(): + missing = {k: v for k, v in CASE1_OUTPUT.items() if k != "valid_until"} + with pytest.raises(Exception, match="valid_until"): + build_decision_ref_v1(decision_output=missing, **BUILDER_PLACEHOLDERS) + with pytest.raises(ValueError, match="valid_until"): + build_decision_ref_v1(decision_output={**CASE1_OUTPUT, "valid_until": "2026-04-08T12:00:05Z"}, **BUILDER_PLACEHOLDERS) + with pytest.raises(ValueError, match="valid_until"): + build_decision_ref_v1(decision_output={"profile": "aps-core-decision-output-v1", "verdict": "deny", + "effective_authority_ref": None, "constraints": [], + "valid_until": "2026-04-08T12:00:05.000Z"}, **BUILDER_PLACEHOLDERS) + with pytest.raises(ValueError, match="valid_until"): + build_decision_ref_v1(decision_output={**CASE1_OUTPUT, "valid_until": None}, **BUILDER_PLACEHOLDERS) + with pytest.raises(Exception, match="unknown field extra"): + build_decision_ref_v1(decision_output={**CASE1_OUTPUT, "extra": "nope"}, **BUILDER_PLACEHOLDERS) + with pytest.raises(ValueError, match="CoreDecisionOutputV1 object"): + build_decision_ref_v1(decision_output=["not", "an", "object"], **BUILDER_PLACEHOLDERS) + + +def test_decision_ref_changes_when_only_valid_until_changes(): + a = build_decision_ref_v1(decision_output=CASE1_OUTPUT, **BUILDER_PLACEHOLDERS) + b = build_decision_ref_v1(decision_output={**CASE1_OUTPUT, "valid_until": "2026-04-08T12:00:06.000Z"}, **BUILDER_PLACEHOLDERS) + assert a["decision_ref"] != b["decision_ref"] + assert b["decision_ref"] == "7c45ce5ecfa8f8aa1bd249513c73c81b31eb193b236dfe50530d58b598c388eb" + + +# Hash primitive test, not a conformance path test. compute_decision_component_ref_v1 is the +# low-level primitive and takes an already-normalized I-JSON value. +def test_compute_decision_component_ref_v1_hashes_an_arbitrary_i_json_value(): + digest = compute_decision_component_ref_v1("context", {"tenant": "t1", "nested": [1, 2, 3]}) + assert re.fullmatch(r"[0-9a-f]{64}", digest) + assert digest == compute_decision_component_ref_v1("context", {"nested": [1, 2, 3], "tenant": "t1"})