tegwick
9c9528f5b2
Contracts first (T02–T04): approval claim schema with issuer, freshness,
and binding digest; local transactional outbox wire; load-bearing cadence
as heartbeat or reconciliation (layer.yaml declared).
Then the object (T06–T08): SQLite closed state machine, CAS supersession,
distinct-approver fail-closed, revocation without holder cooperation,
outbox insert in the same transaction. Tests fail the mutation when
emission fails, and revoke while the drain sink is down.
Introspection GET /v1/approvals/{id}/claim is a PIP fact, not a decision.
No public consume (T05 waits on GH-WP-0002-T06). Canon T-06 coverage for
wrong binding, expiry, revoke, and supersede.
FLEX-WP-0017 T03 is unblocked on this object; T05 remains blocked only on
consumption ordering.
Assistant: grok
Assistant-Session: 01a04ceb-2057-7e20-b0f9-c282964d5dd9
45 lines
1.4 KiB
Python
45 lines
1.4 KiB
Python
"""Canonical binding and native digest.
|
|
|
|
The native digest is SHA-256 over sorted-key JSON of five fields. It is not
|
|
Go json.Marshal of a flex-auth CheckRequest — that value, when known at
|
|
issue, is stored separately as pdp_digest.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
from typing import Any
|
|
|
|
DIGEST_RE = re.compile(r"^sha256:[0-9a-f]{64}$")
|
|
|
|
BINDING_FIELDS = ("action", "actor", "principal", "purpose", "target")
|
|
|
|
|
|
def canonical_binding(binding: dict[str, Any]) -> dict[str, Any]:
|
|
missing = [k for k in BINDING_FIELDS if k not in binding or binding[k] in (None, "")]
|
|
if missing:
|
|
raise ValueError(f"binding missing {missing}")
|
|
if not isinstance(binding["target"], dict):
|
|
raise ValueError("binding.target must be an object")
|
|
return {k: binding[k] for k in BINDING_FIELDS}
|
|
|
|
|
|
def canonical_json(value: Any) -> bytes:
|
|
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode(
|
|
"utf-8"
|
|
)
|
|
|
|
|
|
def binding_digest(binding: dict[str, Any]) -> str:
|
|
payload = canonical_json(canonical_binding(binding))
|
|
return "sha256:" + hashlib.sha256(payload).hexdigest()
|
|
|
|
|
|
def require_digest(value: str | None) -> str | None:
|
|
if value is None:
|
|
return None
|
|
if not DIGEST_RE.match(value):
|
|
raise ValueError("digest must match sha256:<64 lowercase hex>")
|
|
return value
|