receipt-core: CoreDecisionOutputV1 semantic core (valid_until binding, normalize-before-hash)

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("_")]

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)}

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]}")

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}