From 327e735400dc2871c0e74502e8df3fc016076777 Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Mon, 3 Aug 2026 12:54:40 +0000 Subject: [PATCH] feat: optional prebind assurance and replayable receipts - Add perseus-ledger-prebind/v1 construction, validation, persistence - Bind prebind hash into existing receipt/hash-chain path (db/metering) - Add non-mutating replay comparison API (replay_receipt_prebind) - Reject forbidden raw payload fields and invalid digests - Preserve legacy usage rows when no prebind is provided - Add receipt and replay API regression tests Closes #197 --- plutus_agent/db.py | 6 ++ plutus_agent/metering.py | 14 ++- plutus_agent/prebind.py | 147 +++++++++++++++++++++++++++++++ plutus_agent/server/api.py | 20 +++++ plutus_agent/server/app.py | 8 ++ tests/test_prebind_receipt.py | 145 ++++++++++++++++++++++++++++++ tests/test_prebind_replay_api.py | 29 ++++++ 7 files changed, 367 insertions(+), 2 deletions(-) create mode 100644 plutus_agent/prebind.py create mode 100644 tests/test_prebind_receipt.py create mode 100644 tests/test_prebind_replay_api.py diff --git a/plutus_agent/db.py b/plutus_agent/db.py index ada448a..b2e9cc1 100644 --- a/plutus_agent/db.py +++ b/plutus_agent/db.py @@ -151,6 +151,8 @@ def micros_to_usd(micros) -> float: "action_receipt_hash", "resource_constraints_version", "resource_constraints_hash", + "prebind_json", + "prebind_hash", ) @@ -572,6 +574,8 @@ def verify_checkpoints(conn, checkpoints, hmac_key: Optional[bytes] = None) -> d action_receipt_hash TEXT, resource_constraints_version TEXT, resource_constraints_hash TEXT, + prebind_json TEXT, + prebind_hash TEXT, estimated INTEGER NOT NULL DEFAULT 1, source TEXT NOT NULL DEFAULT 'api', ts REAL NOT NULL, @@ -886,6 +890,8 @@ def _migrate_add_columns(conn) -> None: ("usage_events", "action_receipt_hash", "TEXT"), ("usage_events", "resource_constraints_version", "TEXT"), ("usage_events", "resource_constraints_hash", "TEXT"), + ("usage_events", "prebind_json", "TEXT"), + ("usage_events", "prebind_hash", "TEXT"), ] for table, col, defn in additions: cols = _table_columns(conn, table) diff --git a/plutus_agent/metering.py b/plutus_agent/metering.py index 04939bd..9616a32 100644 --- a/plutus_agent/metering.py +++ b/plutus_agent/metering.py @@ -25,6 +25,7 @@ from typing import Optional from . import db, pricing +from .prebind import validate_prebind DAY = 86400 _SHA256_HEX = re.compile(r"^[0-9a-fA-F]{64}$") @@ -166,6 +167,7 @@ def record_usage(conn, org_id: str, provider: str, block_over_limit: bool = False, block_over_balance: bool = False, chain_hmac_key: Optional[bytes] = None, + prebind: Optional[dict] = None, commit: bool = True) -> MeterResult: """Meter one LLM/agent call. Returns a :class:`MeterResult`. @@ -206,6 +208,10 @@ def record_usage(conn, org_id: str, provider: str, follows the actual ``cost_usd``. """ ts = ts if ts is not None else time.time() + if prebind is not None: + valid, errors = validate_prebind(prebind) + if not valid: + raise ValueError("invalid prebind: " + ", ".join(errors)) evidence_hashes_json = _canonical_evidence_hashes(evidence_hashes) policy_version = _optional_text(policy_version, "policy_version") correction_ref = _optional_text(correction_ref, "correction_ref") @@ -475,6 +481,8 @@ def record_usage(conn, org_id: str, provider: str, "action_receipt_hash": action_receipt_hash, "resource_constraints_version": resource_constraints_version, "resource_constraints_hash": resource_constraints_hash, + "prebind_json": json.dumps(prebind, sort_keys=True, separators=(",", ":")) if prebind else None, + "prebind_hash": prebind.get("prebind_hash") if prebind else None, } row_hash = db.compute_row_hash(prev_hash, row_fields, hmac_key=chain_hmac_key) conn.execute( @@ -483,8 +491,8 @@ def record_usage(conn, org_id: str, provider: str, "baseline_micros,optimal_micros,external_ref,user_id,evidence_hashes,policy_version," "result_hash,human_review,correction_ref,agent_id,authority_manifest_ref,scope_anchor," "action_intent_hash,action_status,approval_ref,context_render_schema,context_render_hash," - "served_memory_provenance_hash,action_receipt_hash,resource_constraints_version,resource_constraints_hash,estimated,source,ts,prev_hash,row_hash) " - "VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", + "served_memory_provenance_hash,action_receipt_hash,resource_constraints_version,resource_constraints_hash,prebind_json,prebind_hash,estimated,source,ts,prev_hash,row_hash) " + "VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", (eid, org_id, workspace_id, provider, model, task_type, int(input_tokens), int(output_tokens), int(cache_read_tokens), (int(cache_write_tokens) if cache_write_tokens is not None else None), @@ -494,6 +502,8 @@ def record_usage(conn, org_id: str, provider: str, action_intent_hash, action_status, approval_ref, context_render_schema, context_render_hash, served_memory_provenance_hash, action_receipt_hash, resource_constraints_version, resource_constraints_hash, + (json.dumps(prebind, sort_keys=True, separators=(",", ":")) if prebind else None), + (prebind.get("prebind_hash") if prebind else None), int(estimated), source, ts, prev_hash, row_hash), ) diff --git a/plutus_agent/prebind.py b/plutus_agent/prebind.py new file mode 100644 index 0000000..4fad4a9 --- /dev/null +++ b/plutus_agent/prebind.py @@ -0,0 +1,147 @@ +"""Optional hash-bound pre-action assurance and pure replay comparison.""" +from __future__ import annotations + +import hashlib +import json +from collections.abc import Mapping +from typing import Any + +PREBIND_SCHEMA = "perseus-ledger-prebind/v1" +OUTCOMES = {"allow", "hold", "deny", "abstain", "interrupt", "recover"} +NON_EFFECTIVE_RESULTS = {"not_executed", "held", "denied", "abstained", "cancelled", "failed"} +FORBIDDEN_KEYS = {"prompt", "context", "content", "body", "body_json", "tool_arguments", "arguments", "result", "response", "token", "secret", "password", "api_key"} +_REQUIRED = ( + "attempted_action", "actor_ref", "authority_ref", "trusted_scope", "policy_version", + "evidence_hashes", "selected_context_digest", "resource_ref", "boundary_outcome", + "non_effective_result", "replay_id", +) + + +def _sha(value: Any) -> str: + return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()).hexdigest() + + +def _is_hash(value: Any) -> bool: + return isinstance(value, str) and len(value) == 64 and all(char in "0123456789abcdef" for char in value.lower()) + + +def _scan(value: Any, errors: list[str]) -> None: + if isinstance(value, Mapping): + for key, child in value.items(): + lowered = str(key).lower() + if lowered in FORBIDDEN_KEYS or lowered.startswith("raw_"): + errors.append(f"forbidden_field:{key}") + _scan(child, errors) + elif isinstance(value, list): + for child in value: + _scan(child, errors) + + +def prebind_digest(block: Mapping[str, Any]) -> str: + return _sha({key: value for key, value in block.items() if key != "prebind_hash"}) + + +def build_prebind(*, attempted_action: str, actor_ref: str, authority_ref: str, trusted_scope: str, + policy_version: str, evidence_hashes: list[str], selected_context_digest: str, + resource_ref: str, boundary_outcome: str, non_effective_result: str, + replay_id: str, approval_ref: str | None = None, + stage_refs: list[str] | None = None) -> dict[str, Any]: + block: dict[str, Any] = { + "schema_version": PREBIND_SCHEMA, + "attempted_action": attempted_action, + "actor_ref": actor_ref, + "authority_ref": authority_ref, + "trusted_scope": trusted_scope, + "policy_version": policy_version, + "evidence_hashes": sorted(set(evidence_hashes)), + "selected_context_digest": selected_context_digest, + "resource_ref": resource_ref, + "boundary_outcome": boundary_outcome, + "non_effective_result": non_effective_result, + "replay_id": replay_id, + "approval_ref": approval_ref, + "stage_refs": list(stage_refs or []), + } + block["prebind_hash"] = prebind_digest(block) + return block + + +def validate_prebind(block: Mapping[str, Any]) -> tuple[bool, list[str]]: + errors: list[str] = [] + if not isinstance(block, Mapping): + return False, ["prebind"] + _scan(block, errors) + if block.get("schema_version") != PREBIND_SCHEMA: + errors.append("schema_version") + for field in _REQUIRED: + if field not in block or not isinstance(block[field], str) or not block[field].strip(): + if field != "evidence_hashes": + errors.append(field) + hashes = block.get("evidence_hashes") + if not isinstance(hashes, list) or not hashes or any(not _is_hash(value) for value in hashes): + errors.append("evidence_hashes") + if not _is_hash(block.get("selected_context_digest")): + errors.append("selected_context_digest") + if block.get("boundary_outcome") not in OUTCOMES: + errors.append("boundary_outcome") + if block.get("non_effective_result") not in NON_EFFECTIVE_RESULTS: + errors.append("non_effective_result") + if block.get("boundary_outcome") == "allow" and block.get("non_effective_result") == "executed": + errors.append("outcome_result_mismatch") + if block.get("boundary_outcome") != "allow" and block.get("non_effective_result") == "executed": + errors.append("outcome_result_mismatch") + stage_refs = block.get("stage_refs") + if not isinstance(stage_refs, list) or any(not isinstance(value, str) or not value for value in stage_refs): + errors.append("stage_refs") + supplied = block.get("prebind_hash") + if not _is_hash(supplied) or supplied != prebind_digest(block): + errors.append("prebind_hash") + allowed = set(_REQUIRED) | {"schema_version", "approval_ref", "stage_refs", "prebind_hash"} + for key in set(block) - allowed: + errors.append(f"unknown_field:{key}") + return not errors, sorted(set(errors)) + + +def replay_prebind(prior: Mapping[str, Any], *, current_authority_ref: str | None = None, + current_trusted_scope: str | None = None, current_evidence_hashes: list[str] | None = None, + current_policy_version: str | None = None, current_state: Mapping[str, Any] | None = None) -> dict[str, Any]: + valid, errors = validate_prebind(prior) + if not valid: + raise ValueError("invalid prebind: " + ", ".join(errors)) + state = dict(current_state or {}) + changed: list[str] = [] + if current_authority_ref is not None and current_authority_ref != prior["authority_ref"]: + changed.append("authority_ref") + if current_trusted_scope is not None and current_trusted_scope != prior["trusted_scope"]: + changed.append("trusted_scope") + normalized = None if current_evidence_hashes is None else sorted(set(current_evidence_hashes)) + if normalized is not None and normalized != prior["evidence_hashes"]: + changed.append("evidence_hashes") + if current_policy_version is not None and current_policy_version != prior["policy_version"]: + changed.append("policy_version") + authority_ok = bool(state.get("authority_ok", not any(field in changed for field in ("authority_ref", "trusted_scope")))) + evidence_current = bool(state.get("evidence_current", not bool(state.get("evidence_stale")))) + approved = bool(state.get("approval_granted", prior.get("approval_ref") is not None)) + action_allowed = bool(state.get("action_allowed", authority_ok and evidence_current)) + admitted = authority_ok and evidence_current and approved and action_allowed + if admitted: + admission = "admitted_after_correction" if prior["boundary_outcome"] in {"hold", "deny", "abstain", "recover", "interrupt"} else "admitted" + outcome = "allow" + else: + admission = "not_admitted" + outcome = "hold" if not authority_ok or not evidence_current else "deny" + return { + "schema_version": "perseus-ledger-replay/v1", + "replay_id": prior["replay_id"], + "prior_prebind_hash": prior["prebind_hash"], + "changed_fields": changed, + "admission": admission, + "replayed_boundary_outcome": outcome, + "non_mutating": True, + "reason_codes": (["authority_changed"] if "authority_ref" in changed or "trusted_scope" in changed else []) + + (["evidence_changed"] if "evidence_hashes" in changed else []) + + (["policy_changed"] if "policy_version" in changed else []), + } + + +__all__ = ["PREBIND_SCHEMA", "build_prebind", "prebind_digest", "replay_prebind", "validate_prebind"] diff --git a/plutus_agent/server/api.py b/plutus_agent/server/api.py index 2b58b55..64bddb2 100644 --- a/plutus_agent/server/api.py +++ b/plutus_agent/server/api.py @@ -6,6 +6,7 @@ import json from .. import db, metering, pricing, savings +from ..prebind import validate_prebind def default_org_id(conn) -> str | None: @@ -64,6 +65,23 @@ def events_json(conn, org_id: str, limit: int = 50, before=None) -> dict: return _page(metering.recent_events(conn, org_id, limit=limit, before=before), limit) +def replay_receipt_prebind(conn, org_id: str, external_ref: str, **kwargs) -> dict: + """Re-evaluate a stored prebind without mutating the usage history.""" + from ..prebind import replay_prebind + + rows = db.events_by_ref(conn, org_id, external_ref) + if not rows: + raise ValueError("receipt not found") + payload = rows[0]["prebind_json"] + if payload is None: + raise ValueError("receipt has no prebind block") + prior = json.loads(payload) + valid, errors = validate_prebind(prior) + if not valid: + raise ValueError("stored prebind is invalid: " + ", ".join(errors)) + return replay_prebind(prior, **kwargs) + + _EXPORT_COLUMNS = ["id", "ts", "provider", "model", "task_type", "workspace", "input_tokens", "output_tokens", "cache_read_tokens", "cache_write_tokens", "reasoning_tokens", "user_id", @@ -119,6 +137,8 @@ def audit_json(conn, org_id: str, *, hmac_key: bytes | None = None, "served_memory_provenance_hash": row["served_memory_provenance_hash"], "action_receipt_hash": row["action_receipt_hash"], }, + "prebind": json.loads(row["prebind_json"]) + if row["prebind_json"] is not None else None, "action_authorization": { "agent_id": row["agent_id"], "authority_manifest_ref": row["authority_manifest_ref"], diff --git a/plutus_agent/server/app.py b/plutus_agent/server/app.py index a6618ce..37ad871 100644 --- a/plutus_agent/server/app.py +++ b/plutus_agent/server/app.py @@ -22,6 +22,7 @@ from .. import __version__, bridge, config as cfgmod, db, pricing from ..billing import StripeClient, BillingError, handle_webhook_event from ..utils import strict_int +from ..prebind import validate_prebind from . import api, views, auth as authmod # Paths reachable without a session when auth is enabled. @@ -830,6 +831,12 @@ def _ingest_usage(self, conn): "resource_constraints_version", "resource_constraints_hash"): if ev.get(field) is not None and not isinstance(ev[field], str): return self._json(400, {"error": f"{field} must be a string"}) + if ev.get("prebind") is not None: + if not isinstance(ev["prebind"], dict): + return self._json(400, {"error": "prebind must be an object"}) + valid, errors = validate_prebind(ev["prebind"]) + if not valid: + return self._json(400, {"error": "invalid prebind", "fields": errors}) # All valid — record the whole batch as one serialized transaction. # Fix #27/#30: db.immediate() takes the write lock up front (BEGIN @@ -893,6 +900,7 @@ def _ingest_usage(self, conn): action_receipt_hash=ev.get("action_receipt_hash"), resource_constraints_version=ev.get("resource_constraints_version"), resource_constraints_hash=ev.get("resource_constraints_hash"), + prebind=ev.get("prebind"), user_id=ev.get("user_id"), source=ev.get("source", "api"), pricing_overrides=cfg.get("pricing", {}).get("overrides"), diff --git a/tests/test_prebind_receipt.py b/tests/test_prebind_receipt.py new file mode 100644 index 0000000..4c4c777 --- /dev/null +++ b/tests/test_prebind_receipt.py @@ -0,0 +1,145 @@ +from __future__ import annotations + +import hashlib +import json + +import pytest + +from plutus_agent.prebind import ( + PREBIND_SCHEMA, + build_prebind, + prebind_digest, + replay_prebind, + validate_prebind, +) + + +def digest(value: str) -> str: + return hashlib.sha256(value.encode()).hexdigest() + + +def make_prebind(**overrides): + value = build_prebind( + attempted_action="action:deploy-42", + actor_ref="agent:hermes-prod", + authority_ref="authority:manifest-3", + trusted_scope="github:Perseus-Computing-LLC/plutus", + policy_version="policy/v3", + evidence_hashes=[digest("source")], + selected_context_digest=digest("context-selection"), + resource_ref="resource:ledger-event-42", + boundary_outcome="hold", + non_effective_result="not_executed", + replay_id="replay:deploy-42", + ) + value.update(overrides) + return value + + +def test_prebind_is_canonical_hash_bound_and_has_no_raw_payloads(): + block = make_prebind() + assert block["schema_version"] == PREBIND_SCHEMA + assert prebind_digest(block) == block["prebind_hash"] + assert validate_prebind(block) == (True, []) + serialized = json.dumps(block, sort_keys=True) + assert "prompt" not in serialized + assert "raw context" not in serialized + assert "tool_arguments" not in serialized + + +def test_prebind_rejects_missing_fields_invalid_hashes_and_raw_fields(): + missing = make_prebind() + missing.pop("selected_context_digest") + assert "selected_context_digest" in validate_prebind(missing)[1] + + invalid = make_prebind(evidence_hashes=["not-a-digest"]) + assert "evidence_hashes" in validate_prebind(invalid)[1] + + leaked = make_prebind() + leaked["prompt"] = "raw prompt" + valid, errors = validate_prebind(leaked) + assert not valid + assert "forbidden_field:prompt" in errors + + +def test_prebind_rejects_hash_tampering_and_ambiguous_outcomes(): + tampered = make_prebind() + tampered["boundary_outcome"] = "allow" + valid, errors = validate_prebind(tampered) + assert not valid + assert "prebind_hash" in errors + + ambiguous = make_prebind(boundary_outcome="allow", non_effective_result="executed") + valid, errors = validate_prebind(ambiguous) + assert not valid + assert "outcome_result_mismatch" in errors + + +def test_replay_is_pure_and_detects_scope_authority_and_evidence_changes(): + prior = make_prebind(boundary_outcome="hold", non_effective_result="not_executed") + comparison = replay_prebind( + prior, + current_authority_ref="authority:manifest-4", + current_trusted_scope="github:other/repo", + current_evidence_hashes=[digest("changed-source")], + current_state={"authority_ok": False, "evidence_current": False, "approval_granted": False}, + ) + assert comparison["replay_id"] == prior["replay_id"] + assert comparison["admission"] == "not_admitted" + assert set(comparison["changed_fields"]) >= {"authority_ref", "trusted_scope", "evidence_hashes"} + assert prior["boundary_outcome"] == "hold" + + +def test_replay_can_admit_corrected_held_attempt_without_mutating_history(): + prior = make_prebind(boundary_outcome="hold", non_effective_result="not_executed") + comparison = replay_prebind( + prior, + current_state={"authority_ok": True, "evidence_current": True, "approval_granted": True, "action_allowed": True}, + ) + assert comparison["admission"] == "admitted_after_correction" + assert comparison["replayed_boundary_outcome"] == "allow" + assert prior["non_effective_result"] == "not_executed" + + +def test_replay_rejects_tampered_prior_block(): + prior = make_prebind() + prior["replay_id"] = "replay:tampered" + with pytest.raises(ValueError, match="invalid prebind"): + replay_prebind(prior) + + +def test_prebind_is_persisted_in_hash_chain_and_receipt(tmp_path): + from plutus_agent import db, metering + from plutus_agent.server.api import audit_json + + conn = db.connect(str(tmp_path / "prebind.db")) + db.init_schema(conn) + org_id = db.create_org(conn, "prebind-integration", tier="free")["id"] + block = make_prebind(boundary_outcome="hold", non_effective_result="not_executed") + result = metering.record_usage( + conn, org_id, provider="openai", model="fixture", task_type="deploy", + external_ref="deploy-prebind", input_tokens=1, output_tokens=1, + cost_usd=0.01, prebind=block, + ) + assert result.recorded is True + receipt = audit_json(conn, org_id, external_ref="deploy-prebind") + assert receipt["events"][0]["prebind"]["prebind_hash"] == block["prebind_hash"] + assert db.verify_chain(conn, org_id)["ok"] is True + conn.close() + + +def test_legacy_usage_without_prebind_remains_unchanged_and_receipt_omits_it(tmp_path): + from plutus_agent import db, metering + from plutus_agent.server.api import audit_json + + conn = db.connect(str(tmp_path / "legacy.db")) + db.init_schema(conn) + org_id = db.create_org(conn, "legacy-no-prebind", tier="free")["id"] + metering.record_usage( + conn, org_id, provider="openai", model="fixture", external_ref="legacy", + input_tokens=1, output_tokens=1, cost_usd=0.01, + ) + receipt = audit_json(conn, org_id, external_ref="legacy") + assert receipt["events"][0]["prebind"] is None + assert db.verify_chain(conn, org_id)["ok"] is True + conn.close() diff --git a/tests/test_prebind_replay_api.py b/tests/test_prebind_replay_api.py new file mode 100644 index 0000000..5cfee61 --- /dev/null +++ b/tests/test_prebind_replay_api.py @@ -0,0 +1,29 @@ +from __future__ import annotations + +from plutus_agent import db, metering +from plutus_agent.server.api import replay_receipt_prebind +from test_prebind_receipt import make_prebind + + +def test_stored_prebind_replay_is_non_mutating(tmp_path): + conn = db.connect(str(tmp_path / "replay.db")) + db.init_schema(conn) + org_id = db.create_org(conn, "replay-prebind", tier="free")["id"] + block = make_prebind(boundary_outcome="hold", non_effective_result="not_executed") + metering.record_usage( + conn, org_id, provider="openai", model="fixture", external_ref="replay-ref", + input_tokens=1, output_tokens=1, cost_usd=0.01, prebind=block, + ) + before = conn.execute("SELECT COUNT(*) AS n FROM usage_events").fetchone()["n"] + replay = replay_receipt_prebind( + conn, org_id, "replay-ref", + current_state={"authority_ok": True, "evidence_current": True, + "approval_granted": True, "action_allowed": True}, + ) + after = conn.execute("SELECT COUNT(*) AS n FROM usage_events").fetchone()["n"] + assert replay["admission"] == "admitted_after_correction" + assert before == after == 1 + conn.close() + + +__all__ = ["test_stored_prebind_replay_is_non_mutating"]