104 lines
3.7 KiB
Python
104 lines
3.7 KiB
Python
"""Mutual Feasibility Receipt (MFR).
|
|
|
|
Small, signed receipts produced in response to a PlanDelta synopsis/proposal
|
|
that state which invariants the responder believes are satisfied for a given
|
|
custody window. The receipt is deliberately compact and uses an HMAC-based
|
|
signature for tests/poC; production would use DID-based signatures.
|
|
|
|
Fields:
|
|
- receipt_id: short random id
|
|
- proposal_hash: short digest of the proposal (e.g., delta summary_hash)
|
|
- invariant_ids: list of small invariant identifiers (strings)
|
|
- decision: one of "ok", "partial", "deny"
|
|
- uncertainty: float 0..1 expressing uncertainty (0 = certain)
|
|
- round_nonce: numeric round to prevent replay
|
|
- ts: timestamp
|
|
- sig: HMAC signature over canonical payload
|
|
|
|
"""
|
|
from dataclasses import dataclass, field
|
|
from typing import List, Dict, Any, Optional
|
|
import time
|
|
import uuid
|
|
import json
|
|
import hmac
|
|
import hashlib
|
|
|
|
|
|
@dataclass
|
|
class MutualFeasibilityReceipt:
|
|
receipt_id: str
|
|
proposal_hash: str
|
|
invariant_ids: List[str]
|
|
decision: str
|
|
uncertainty: float
|
|
round_nonce: int
|
|
ts: float
|
|
signer: str = ""
|
|
sig: str = ""
|
|
|
|
@classmethod
|
|
def create(cls, proposal_hash: str, invariant_ids: List[str], decision: str,
|
|
uncertainty: float = 0.0, round_nonce: int = 0, signer: str = "",
|
|
key: Optional[bytes] = None) -> "MutualFeasibilityReceipt":
|
|
rid = uuid.uuid4().hex[:12]
|
|
ts = time.time()
|
|
r = cls(receipt_id=rid, proposal_hash=proposal_hash,
|
|
invariant_ids=invariant_ids, decision=decision,
|
|
uncertainty=float(uncertainty), round_nonce=int(round_nonce), ts=ts,
|
|
signer=signer, sig="")
|
|
if key:
|
|
r.sig = r._sign(key)
|
|
return r
|
|
|
|
def _canonical_payload(self) -> str:
|
|
# deterministic JSON over the authoritative fields
|
|
payload = {
|
|
"receipt_id": self.receipt_id,
|
|
"proposal_hash": self.proposal_hash,
|
|
"invariant_ids": self.invariant_ids,
|
|
"decision": self.decision,
|
|
"uncertainty": round(self.uncertainty, 6),
|
|
"round_nonce": self.round_nonce,
|
|
"ts": int(self.ts),
|
|
"signer": self.signer,
|
|
}
|
|
return json.dumps(payload, sort_keys=True, separators=(",", ":"))
|
|
|
|
def _sign(self, key: bytes) -> str:
|
|
payload = self._canonical_payload()
|
|
return hmac.new(key, payload.encode("utf-8"), hashlib.sha256).hexdigest()
|
|
|
|
def verify(self, key: bytes) -> bool:
|
|
if not self.sig:
|
|
return False
|
|
expected = self._sign(key)
|
|
return hmac.compare_digest(expected, self.sig)
|
|
|
|
def serialize(self) -> bytes:
|
|
obj = {
|
|
"receipt_id": self.receipt_id,
|
|
"proposal_hash": self.proposal_hash,
|
|
"invariant_ids": self.invariant_ids,
|
|
"decision": self.decision,
|
|
"uncertainty": self.uncertainty,
|
|
"round_nonce": self.round_nonce,
|
|
"ts": self.ts,
|
|
"signer": self.signer,
|
|
"sig": self.sig,
|
|
}
|
|
return json.dumps(obj, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
|
|
|
@classmethod
|
|
def deserialize(cls, b: bytes) -> "MutualFeasibilityReceipt":
|
|
obj = json.loads(b.decode("utf-8"))
|
|
return cls(receipt_id=obj.get("receipt_id", ""),
|
|
proposal_hash=obj.get("proposal_hash", ""),
|
|
invariant_ids=obj.get("invariant_ids", []),
|
|
decision=obj.get("decision", "deny"),
|
|
uncertainty=float(obj.get("uncertainty", 0.0)),
|
|
round_nonce=int(obj.get("round_nonce", 0)),
|
|
ts=float(obj.get("ts", 0.0)),
|
|
signer=obj.get("signer", ""),
|
|
sig=obj.get("sig", ""))
|