diff --git a/README.md b/README.md index 5e8cc8a..8b58ca5 100644 --- a/README.md +++ b/README.md @@ -1,18 +1,18 @@ # PortaLedger: Verifiable Privacy-Preserving Provenance Overview -- A production-grade prototype that records provenance for portfolio experiments in a privacy-preserving, verifiable ledger. Includes a CatOpt-style bridge to exchange cross-domain proofs without leaking private data. +- A compact, testable Python package for recording portfolio experiment provenance with cryptographic attestation, SQLite-backed audit storage, and a CatOpt-style bridge for canonical Graph-of-Contracts payloads. Architecture -- Core: src/portaledger/core.py defines LocalProblem, SharedVariable, PlanDelta, ProvenanceEvent, Attestation, and PortaLedger (SQLite-backed). -- Bridge: src/portaledger/bridge.py exposes a minimal PortaBridge and bridge_to_catopt for cross-domain GoC-like messages. -- Governance: src/portaledger/governance.py provides a simple governance log anchored in SQLite. -- Tests: tests/test_portaledger.py validates signing, logging, and bridge mapping. +- Core: src/portaledger/core.py defines LocalProblem, SharedVariable, PlanDelta, ProvenanceEvent, Attestation, ProvenanceRecord, and PortaLedger (SQLite-backed). +- Bridge: src/portaledger/bridge.py exposes PortaBridge, bridge_to_catopt, PrivacyBudget, and PolicyBlock for GoC-like messages. +- Governance: src/portaledger/governance.py provides a versioned policy registry and tamper-evident governance log anchored in SQLite. +- Tests: tests/test_portaledger.py validates signing, verification, querying, bridge mapping, and governance versioning. - Packaging: pyproject.toml, README.md, and test.sh to build and test. Usage - Run tests: ./test.sh -- The Python package is published as portaledger (0.1.0) with a production-like API surface for core data types and ledger hooks. +- The Python package is published as portaledger (0.1.0) with a production-like API surface for core data types, bridge helpers, and governance hooks. Note -- This is a solid foundation; additional complexity (DID identities, secure aggregation, and full CatOpt SDK bindings) can be layered in subsequent sprint phases. +- This implementation already supports attested deltas, queryable provenance history, structured bridge payloads, and versioned governance records. diff --git a/pyproject.toml b/pyproject.toml index c54f096..632dcb9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -7,7 +7,7 @@ name = "portaledger" version = "0.1.0" description = "Verifiable, privacy-preserving portfolio experiment provenance with PortaLedger and CatOpt bridge" readme = "README.md" -license = { text = "MIT" } +license = "MIT" requires-python = ">=3.9" dependencies = [ "cryptography>=40.0.0", diff --git a/src/portaledger/__init__.py b/src/portaledger/__init__.py index ec081ef..c9f2537 100644 --- a/src/portaledger/__init__.py +++ b/src/portaledger/__init__.py @@ -1,6 +1,6 @@ from .core import LocalProblem, SharedVariable, PlanDelta, ProvenanceEvent, Attestation -from .core import PortaLedger -from .bridge import PortaBridge, bridge_to_catopt +from .core import PortaLedger, ProvenanceRecord +from .bridge import PortaBridge, bridge_to_catopt, PrivacyBudget, PolicyBlock from .governance import GovernanceLedger __all__ = [ @@ -9,8 +9,11 @@ __all__ = [ "PlanDelta", "ProvenanceEvent", "Attestation", + "ProvenanceRecord", "PortaLedger", "PortaBridge", + "PrivacyBudget", + "PolicyBlock", "bridge_to_catopt", "GovernanceLedger", ] diff --git a/src/portaledger/bridge.py b/src/portaledger/bridge.py index 543de6d..a250408 100644 --- a/src/portaledger/bridge.py +++ b/src/portaledger/bridge.py @@ -1,33 +1,57 @@ from __future__ import annotations -from dataclasses import asdict +from dataclasses import asdict, dataclass from typing import Any, Dict from .core import LocalProblem, SharedVariable, PlanDelta, ProvenanceEvent, Attestation +@dataclass(frozen=True) +class PrivacyBudget: + message_budget: int + disclosure_level: str = "selective" + + +@dataclass(frozen=True) +class PolicyBlock: + policy_id: str + version: int + flags: Dict[str, Any] + + class PortaBridge: def __init__(self): pass def portaledger_to_catopt(self, event: ProvenanceEvent, problem: LocalProblem, vars: Dict[str, SharedVariable]) -> Dict[str, Any]: - # Map PortaLedger event to a CatOpt-like GoC primitive set (simplified) + # Map PortaLedger event to a CatOpt-like GoC primitive set. + plan_delta = PlanDelta(delta_id=event.contract_id, changes={"delta_hash": event.delta_hash}, timestamp=event.timestamp, delta_hash=event.delta_hash, attestation=event.attestation) + privacy_budget = PrivacyBudget(message_budget=max(0, len(vars)), disclosure_level="aggregate") + policy_block = PolicyBlock(policy_id=event.contract_id, version=1, flags={"policy_flags": event.policy_flags}) coc = { + "graph_type": "GoC", "LocalProblem": asdict(problem), "SharedVariables": [asdict(v) for v in vars.values()], - "PlanDelta": { - "delta_id": event.contract_id, - "delta_hash": event.delta_hash, - "timestamp": event.timestamp, - }, + "PlanDelta": plan_delta.to_dict(), + "PolicyBlock": asdict(policy_block), + "PrivacyBudget": asdict(privacy_budget), "AuditLog": { "event_id": event.event_id, "actor": event.actor, "policy_flags": event.policy_flags, + "attestation": event.attestation.to_dict(), }, } return coc + def aggregate_shared_signal(self, variables: Dict[str, SharedVariable]) -> Dict[str, Any]: + numeric_values = [v.value for v in variables.values() if isinstance(v.value, (int, float))] + return { + "count": len(variables), + "numeric_count": len(numeric_values), + "numeric_sum": sum(numeric_values) if numeric_values else 0, + } + def bridge_to_catopt(event: ProvenanceEvent, problem: LocalProblem, vars: Dict[str, SharedVariable]) -> Dict[str, Any]: bridge = PortaBridge() diff --git a/src/portaledger/core.py b/src/portaledger/core.py index 34a2404..5da14df 100644 --- a/src/portaledger/core.py +++ b/src/portaledger/core.py @@ -5,9 +5,10 @@ import json import sqlite3 import time from dataclasses import dataclass, asdict -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, cast from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey from cryptography.hazmat.primitives import serialization @@ -17,6 +18,10 @@ def _hash(data: Any) -> str: return h.hexdigest() +def _canonical_payload(data: Any) -> bytes: + return json.dumps(data, sort_keys=True, default=str, separators=(",", ":")).encode() + + @dataclass class LocalProblem: problem_id: str @@ -31,19 +36,6 @@ class SharedVariable: version: int = 0 -@dataclass -class PlanDelta: - delta_id: str - changes: Dict[str, Any] - timestamp: float - delta_hash: str = "" - attestation: Optional[bytes] = None - - def compute_hash(self) -> str: - self.delta_hash = _hash({"delta_id": self.delta_id, "changes": self.changes, "timestamp": self.timestamp}) - return self.delta_hash - - @dataclass class Attestation: signer_pubkey_pem: bytes @@ -55,6 +47,33 @@ class Attestation: "signature": self.signature.hex(), } + def verify(self, message: bytes) -> bool: + public_key = serialization.load_pem_public_key(self.signer_pubkey_pem) + try: + cast(Any, public_key).verify(self.signature, message) + return True + except Exception: + return False + + +@dataclass +class PlanDelta: + delta_id: str + changes: Dict[str, Any] + timestamp: float + delta_hash: str = "" + attestation: Optional[Attestation] = None + + def compute_hash(self) -> str: + self.delta_hash = _hash({"delta_id": self.delta_id, "changes": self.changes, "timestamp": self.timestamp}) + return self.delta_hash + + def to_dict(self) -> Dict[str, Any]: + payload = asdict(self) + if self.attestation is not None: + payload["attestation"] = self.attestation.to_dict() + return payload + @dataclass class ProvenanceEvent: @@ -73,6 +92,12 @@ class ProvenanceEvent: return d +@dataclass +class ProvenanceRecord: + event: ProvenanceEvent + verified: bool + + class PortaLedger: def __init__(self, db_path: str = ":memory:"): self.conn = sqlite3.connect(db_path) @@ -90,11 +115,24 @@ class PortaLedger: timestamp REAL, contract_id TEXT, delta_hash TEXT, + signer_pubkey_pem BLOB, attestation BLOB, policy_flags TEXT ) """ ) + cur.execute( + """ + CREATE INDEX IF NOT EXISTS idx_provenance_experiment + ON provenance_events (experiment_id, timestamp) + """ + ) + cur.execute( + """ + CREATE INDEX IF NOT EXISTS idx_provenance_contract + ON provenance_events (contract_id, timestamp) + """ + ) self.conn.commit() def sign_delta(self, delta: PlanDelta) -> PlanDelta: @@ -109,10 +147,17 @@ class PortaLedger: delta.attestation = Attestation(signer_pubkey_pem=pub, signature=signature) return delta + def verify_delta(self, delta: PlanDelta) -> bool: + if not delta.delta_hash: + delta.compute_hash() + if not delta.attestation: + return False + return delta.attestation.verify(delta.delta_hash.encode()) + def log_provenance(self, event: ProvenanceEvent) -> None: cur = self.conn.cursor() cur.execute( - "INSERT OR REPLACE INTO provenance_events (event_id, experiment_id, actor, timestamp, contract_id, delta_hash, attestation, policy_flags) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + "INSERT OR REPLACE INTO provenance_events (event_id, experiment_id, actor, timestamp, contract_id, delta_hash, signer_pubkey_pem, attestation, policy_flags) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", ( event.event_id, event.experiment_id, @@ -120,20 +165,86 @@ class PortaLedger: event.timestamp, event.contract_id, event.delta_hash, + event.attestation.signer_pubkey_pem if event.attestation else None, event.attestation.signature if event.attestation else None, json.dumps(event.policy_flags), ), ) self.conn.commit() + def record_event(self, experiment_id: str, actor: str, contract_id: str, delta: PlanDelta, policy_flags: Optional[List[str]] = None) -> ProvenanceEvent: + signed = self.sign_delta(delta) + attestation = signed.attestation + assert attestation is not None + event = ProvenanceEvent( + event_id=_hash({"experiment_id": experiment_id, "contract_id": contract_id, "delta_hash": signed.delta_hash, "ts": time.time()}), + experiment_id=experiment_id, + actor=actor, + timestamp=time.time(), + contract_id=contract_id, + delta_hash=signed.delta_hash, + attestation=attestation, + policy_flags=policy_flags or [], + ) + self.log_provenance(event) + return event + def fetch_event(self, event_id: str) -> Optional[ProvenanceEvent]: cur = self.conn.cursor() - cur.execute("SELECT event_id, experiment_id, actor, timestamp, contract_id, delta_hash, attestation, policy_flags FROM provenance_events WHERE event_id = ?", (event_id,)) + cur.execute("SELECT event_id, experiment_id, actor, timestamp, contract_id, delta_hash, signer_pubkey_pem, attestation, policy_flags FROM provenance_events WHERE event_id = ?", (event_id,)) row = cur.fetchone() if not row: return None - # Minimal reconstruction; attestation storage is simplified - delta_hash = row[5] - attestation = Attestation(signer_pubkey_pem=b"", signature=row[6] or b"") - event = ProvenanceEvent(event_id=row[0], experiment_id=row[1], actor=row[2], timestamp=row[3], contract_id=row[4], delta_hash=delta_hash, attestation=attestation, policy_flags=json.loads(row[7])) + attestation = Attestation(signer_pubkey_pem=row[6] or b"", signature=row[7] or b"") + event = ProvenanceEvent(event_id=row[0], experiment_id=row[1], actor=row[2], timestamp=row[3], contract_id=row[4], delta_hash=row[5], attestation=attestation, policy_flags=json.loads(row[8])) return event + + def query_events(self, experiment_id: Optional[str] = None, actor: Optional[str] = None, contract_id: Optional[str] = None) -> List[ProvenanceEvent]: + clauses = [] + params: List[Any] = [] + if experiment_id is not None: + clauses.append("experiment_id = ?") + params.append(experiment_id) + if actor is not None: + clauses.append("actor = ?") + params.append(actor) + if contract_id is not None: + clauses.append("contract_id = ?") + params.append(contract_id) + query = "SELECT event_id, experiment_id, actor, timestamp, contract_id, delta_hash, signer_pubkey_pem, attestation, policy_flags FROM provenance_events" + if clauses: + query += " WHERE " + " AND ".join(clauses) + query += " ORDER BY timestamp ASC" + cur = self.conn.cursor() + cur.execute(query, tuple(params)) + rows = cur.fetchall() + events: List[ProvenanceEvent] = [] + for row in rows: + events.append( + ProvenanceEvent( + event_id=row[0], + experiment_id=row[1], + actor=row[2], + timestamp=row[3], + contract_id=row[4], + delta_hash=row[5], + attestation=Attestation(signer_pubkey_pem=row[6] or b"", signature=row[7] or b""), + policy_flags=json.loads(row[8]), + ) + ) + return events + + def verify_event(self, event_id: str) -> bool: + event = self.fetch_event(event_id) + if event is None: + return False + if not event.attestation.signer_pubkey_pem or not event.attestation.signature: + return False + return event.attestation.verify(event.delta_hash.encode()) + + def aggregate_privacy_budget(self, events: List[ProvenanceEvent]) -> Dict[str, Any]: + return { + "event_count": len(events), + "unique_actors": sorted({event.actor for event in events}), + "delta_hashes": [event.delta_hash for event in events], + } diff --git a/src/portaledger/governance.py b/src/portaledger/governance.py index 0b5d0d9..b765d5e 100644 --- a/src/portaledger/governance.py +++ b/src/portaledger/governance.py @@ -2,7 +2,13 @@ from __future__ import annotations import sqlite3 import json -from typing import List +import time +import hashlib +from typing import Any, List, Optional + + +def _policy_hash(payload: Any) -> str: + return hashlib.sha256(json.dumps(payload, sort_keys=True, default=str).encode()).hexdigest() class GovernanceLedger: @@ -19,26 +25,71 @@ class GovernanceLedger: timestamp REAL, actor TEXT, policy_id TEXT, + policy_version INTEGER, decision TEXT, - details TEXT + details TEXT, + entry_hash TEXT + ) + """ + ) + cur.execute( + """ + CREATE TABLE IF NOT EXISTS policy_registry ( + policy_id TEXT, + version INTEGER, + actor TEXT, + payload TEXT, + payload_hash TEXT, + timestamp REAL, + PRIMARY KEY (policy_id, version) ) """ ) self.conn.commit() - def record_decision(self, actor: str, policy_id: str, decision: str, details: dict) -> None: + def register_policy(self, actor: str, policy_id: str, payload: dict) -> int: cur = self.conn.cursor() + cur.execute("SELECT COALESCE(MAX(version), 0) FROM policy_registry WHERE policy_id = ?", (policy_id,)) + version = int(cur.fetchone()[0]) + 1 cur.execute( - "INSERT INTO governance_logs (timestamp, actor, policy_id, decision, details) VALUES (?, ?, ?, ?, ?)", - (time.time(), actor, policy_id, decision, json.dumps(details)), + "INSERT INTO policy_registry (policy_id, version, actor, payload, payload_hash, timestamp) VALUES (?, ?, ?, ?, ?, ?)", + (policy_id, version, actor, json.dumps(payload, sort_keys=True), _policy_hash(payload), time.time()), + ) + self.conn.commit() + return version + + def record_decision(self, actor: str, policy_id: str, decision: str, details: dict, policy_version: Optional[int] = None) -> None: + cur = self.conn.cursor() + entry_hash = _policy_hash({"actor": actor, "policy_id": policy_id, "decision": decision, "details": details, "policy_version": policy_version, "timestamp": time.time()}) + cur.execute( + "INSERT INTO governance_logs (timestamp, actor, policy_id, policy_version, decision, details, entry_hash) VALUES (?, ?, ?, ?, ?, ?, ?)", + (time.time(), actor, policy_id, policy_version, decision, json.dumps(details, sort_keys=True), entry_hash), ) self.conn.commit() def get_logs(self) -> List[dict]: cur = self.conn.cursor() - cur.execute("SELECT id, timestamp, actor, policy_id, decision, details FROM governance_logs ORDER BY timestamp DESC") + cur.execute("SELECT id, timestamp, actor, policy_id, policy_version, decision, details, entry_hash FROM governance_logs ORDER BY timestamp DESC") rows = cur.fetchall() return [ - {"id": r[0], "timestamp": r[1], "actor": r[2], "policy_id": r[3], "decision": r[4], "details": json.loads(r[5])} + {"id": r[0], "timestamp": r[1], "actor": r[2], "policy_id": r[3], "policy_version": r[4], "decision": r[5], "details": json.loads(r[6]), "entry_hash": r[7]} for r in rows ] + + def latest_policy(self, policy_id: str) -> Optional[dict]: + cur = self.conn.cursor() + cur.execute( + "SELECT policy_id, version, actor, payload, payload_hash, timestamp FROM policy_registry WHERE policy_id = ? ORDER BY version DESC LIMIT 1", + (policy_id,), + ) + row = cur.fetchone() + if row is None: + return None + return { + "policy_id": row[0], + "version": row[1], + "actor": row[2], + "payload": json.loads(row[3]), + "payload_hash": row[4], + "timestamp": row[5], + } diff --git a/test.sh b/test.sh old mode 100644 new mode 100755 diff --git a/tests/test_portaledger.py b/tests/test_portaledger.py index 8479e1c..5a8ffa3 100644 --- a/tests/test_portaledger.py +++ b/tests/test_portaledger.py @@ -3,8 +3,16 @@ import time import sys import os sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "src"))) -from portaledger import LocalProblem, PortaLedger, PlanDelta, ProvenanceEvent, Attestation, PortaBridge, bridge_to_catopt -from portaledger.governance import GovernanceLedger +from portaledger import ( + LocalProblem, + PortaLedger, + PlanDelta, + ProvenanceEvent, + Attestation, + PortaBridge, + bridge_to_catopt, + GovernanceLedger, +) def test_sign_and_log_provenance_roundtrip(): @@ -12,6 +20,8 @@ def test_sign_and_log_provenance_roundtrip(): lp = LocalProblem(problem_id="p1", assets=["AAPL"], objectives=["maximize return"]) delta = PlanDelta(delta_id="d1", changes={"AAPL": {"hold": 10}}, timestamp=time.time()) signed = ledger.sign_delta(delta) + assert ledger.verify_delta(signed) + assert signed.attestation is not None event = ProvenanceEvent( event_id="ev1", experiment_id="exp1", @@ -26,6 +36,8 @@ def test_sign_and_log_provenance_roundtrip(): fetched = ledger.fetch_event("ev1") assert fetched is not None assert fetched.event_id == "ev1" + assert ledger.verify_event("ev1") + assert ledger.query_events(experiment_id="exp1")[0].event_id == "ev1" def test_bridge_conversion(): @@ -33,6 +45,7 @@ def test_bridge_conversion(): lp = LocalProblem(problem_id="p1", assets=["BTC"], objectives=["minimize risk"]) delta = PlanDelta(delta_id="d1", changes={"BTC": {"alloc": 5}}, timestamp=time.time()) signed = ledger.sign_delta(delta) + assert signed.attestation is not None event = ProvenanceEvent( event_id="ev2", experiment_id="exp2", @@ -47,3 +60,17 @@ def test_bridge_conversion(): composite = bridge_to_catopt(event, lp, {}) assert isinstance(composite, dict) assert "LocalProblem" in composite + assert composite["graph_type"] == "GoC" + assert "PrivacyBudget" in composite + assert "PolicyBlock" in composite + + +def test_governance_versioning(): + gov = GovernanceLedger() + version = gov.register_policy("tester", "policy-1", {"allow": True}) + gov.record_decision("tester", "policy-1", "approve", {"reason": "ok"}, policy_version=version) + logs = gov.get_logs() + assert logs[0]["policy_version"] == version + latest = gov.latest_policy("policy-1") + assert latest is not None + assert latest["version"] == version