diff --git a/gravityweave/receipt.py b/gravityweave/receipt.py new file mode 100644 index 0000000..c14b473 --- /dev/null +++ b/gravityweave/receipt.py @@ -0,0 +1,103 @@ +"""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", "")) diff --git a/tests/test_receipt.py b/tests/test_receipt.py new file mode 100644 index 0000000..d46c276 --- /dev/null +++ b/tests/test_receipt.py @@ -0,0 +1,39 @@ +from gravityweave.receipt import MutualFeasibilityReceipt + + +def test_receipt_create_and_verify(): + key = b"node-secret" + prop_hash = "deadbeefcafebabe" + invariants = ["fuel_bound", "power_bound"] + + r = MutualFeasibilityReceipt.create(proposal_hash=prop_hash, + invariant_ids=invariants, + decision="ok", + uncertainty=0.1, + round_nonce=1, + signer="nodeA", + key=key) + + assert r.proposal_hash == prop_hash + assert r.decision == "ok" + assert r.verify(key) is True + + ser = r.serialize() + # ensure compact size under 512 bytes + assert len(ser) <= 512 + + r2 = MutualFeasibilityReceipt.deserialize(ser) + assert r2.receipt_id == r.receipt_id + # verify with same key + assert r2.verify(key) is True + + +def test_receipt_tamper_detection(): + key = b"k" + r = MutualFeasibilityReceipt.create(proposal_hash="h", invariant_ids=["i"], decision="ok", + signer="s", key=key) + ser = r.serialize() + # tamper with serialized bytes + s2 = ser.replace(b"ok", b"deny") + r2 = MutualFeasibilityReceipt.deserialize(s2) + assert r2.verify(key) is False