274 lines
9.1 KiB
Python
274 lines
9.1 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, asdict, field
|
|
from typing import Any, Dict, Iterable, List, Optional
|
|
import hashlib
|
|
import json
|
|
import time
|
|
|
|
from .core import AttestationHint, AuditLog, PlanDelta, Policy, _canonical_json
|
|
|
|
|
|
def _hash(payload: Any) -> str:
|
|
return hashlib.sha256(_canonical_json(payload).encode("utf-8")).hexdigest()
|
|
|
|
|
|
@dataclass
|
|
class ContractRecord:
|
|
contract_id: str
|
|
kind: str
|
|
name: str
|
|
version: str
|
|
definition: Dict[str, Any] = field(default_factory=dict)
|
|
attestation: Optional[AttestationHint] = None
|
|
created_at: float = field(default_factory=lambda: time.time())
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
data = asdict(self)
|
|
if self.attestation is not None:
|
|
data["attestation"] = self.attestation.to_dict()
|
|
return data
|
|
|
|
|
|
@dataclass
|
|
class ContractLink:
|
|
source_id: str
|
|
target_id: str
|
|
relation: str
|
|
created_at: float = field(default_factory=lambda: time.time())
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
|
|
class GraphOfContracts:
|
|
def __init__(self) -> None:
|
|
self._records: Dict[str, ContractRecord] = {}
|
|
self._links: List[ContractLink] = []
|
|
|
|
def register_schema(
|
|
self,
|
|
name: str,
|
|
version: str,
|
|
definition: Dict[str, Any],
|
|
attestation: Optional[AttestationHint] = None,
|
|
) -> ContractRecord:
|
|
contract_id = f"schema:{name}:{version}"
|
|
record = ContractRecord(
|
|
contract_id=contract_id,
|
|
kind="schema",
|
|
name=name,
|
|
version=version,
|
|
definition=dict(definition),
|
|
attestation=attestation,
|
|
)
|
|
self._records[contract_id] = record
|
|
return record
|
|
|
|
def register_adapter(
|
|
self,
|
|
name: str,
|
|
version: str,
|
|
definition: Dict[str, Any],
|
|
attestation: Optional[AttestationHint] = None,
|
|
) -> ContractRecord:
|
|
contract_id = f"adapter:{name}:{version}"
|
|
record = ContractRecord(
|
|
contract_id=contract_id,
|
|
kind="adapter",
|
|
name=name,
|
|
version=version,
|
|
definition=dict(definition),
|
|
attestation=attestation,
|
|
)
|
|
self._records[contract_id] = record
|
|
return record
|
|
|
|
def link(self, source_id: str, target_id: str, relation: str = "implements") -> ContractLink:
|
|
if source_id not in self._records:
|
|
raise KeyError(f"Unknown source contract: {source_id}")
|
|
if target_id not in self._records:
|
|
raise KeyError(f"Unknown target contract: {target_id}")
|
|
link = ContractLink(source_id=source_id, target_id=target_id, relation=relation)
|
|
self._links.append(link)
|
|
return link
|
|
|
|
def resolve(self, contract_id: str) -> ContractRecord:
|
|
return self._records[contract_id]
|
|
|
|
def records(self) -> List[ContractRecord]:
|
|
return list(self._records.values())
|
|
|
|
def links(self) -> List[ContractLink]:
|
|
return list(self._links)
|
|
|
|
def validate(self) -> List[str]:
|
|
issues: List[str] = []
|
|
for link in self._links:
|
|
if link.source_id not in self._records:
|
|
issues.append(f"missing source {link.source_id}")
|
|
if link.target_id not in self._records:
|
|
issues.append(f"missing target {link.target_id}")
|
|
return issues
|
|
|
|
def snapshot(self) -> Dict[str, Any]:
|
|
return {
|
|
"records": [record.to_dict() for record in self.records()],
|
|
"links": [link.to_dict() for link in self.links()],
|
|
"fingerprint": self.fingerprint(),
|
|
}
|
|
|
|
@classmethod
|
|
def from_snapshot(cls, snapshot: Dict[str, Any]) -> "GraphOfContracts":
|
|
graph = cls()
|
|
for record_data in snapshot.get("records", []):
|
|
attestation = record_data.get("attestation")
|
|
hint = AttestationHint(**attestation) if attestation else None
|
|
record = ContractRecord(
|
|
contract_id=record_data["contract_id"],
|
|
kind=record_data["kind"],
|
|
name=record_data["name"],
|
|
version=record_data["version"],
|
|
definition=dict(record_data.get("definition", {})),
|
|
attestation=hint,
|
|
created_at=record_data.get("created_at", time.time()),
|
|
)
|
|
graph._records[record.contract_id] = record
|
|
for link_data in snapshot.get("links", []):
|
|
graph._links.append(
|
|
ContractLink(
|
|
source_id=link_data["source_id"],
|
|
target_id=link_data["target_id"],
|
|
relation=link_data["relation"],
|
|
created_at=link_data.get("created_at", time.time()),
|
|
)
|
|
)
|
|
return graph
|
|
|
|
def fingerprint(self) -> str:
|
|
payload = {
|
|
"records": [record.to_dict() for record in sorted(self._records.values(), key=lambda item: item.contract_id)],
|
|
"links": [link.to_dict() for link in self._links],
|
|
}
|
|
return _hash(payload)
|
|
|
|
|
|
@dataclass
|
|
class LedgerEntry:
|
|
entry_id: str
|
|
event_type: str
|
|
payload: Dict[str, Any]
|
|
actor: str
|
|
timestamp: float
|
|
previous_hash: str
|
|
entry_hash: str
|
|
signature: Optional[str] = None
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
|
|
class GovernanceLedger:
|
|
def __init__(self) -> None:
|
|
self._entries: List[LedgerEntry] = []
|
|
|
|
def append(
|
|
self,
|
|
event_type: str,
|
|
payload: Dict[str, Any],
|
|
actor: str = "system",
|
|
signature: Optional[str] = None,
|
|
timestamp: Optional[float] = None,
|
|
) -> LedgerEntry:
|
|
ts = time.time() if timestamp is None else timestamp
|
|
previous_hash = self._entries[-1].entry_hash if self._entries else "genesis"
|
|
entry_id = payload.get("entry_id") or f"{event_type}-{len(self._entries) + 1}"
|
|
basis = {
|
|
"entry_id": entry_id,
|
|
"event_type": event_type,
|
|
"payload": payload,
|
|
"actor": actor,
|
|
"timestamp": ts,
|
|
"previous_hash": previous_hash,
|
|
}
|
|
entry_hash = _hash(basis)
|
|
entry = LedgerEntry(
|
|
entry_id=entry_id,
|
|
event_type=event_type,
|
|
payload=dict(payload),
|
|
actor=actor,
|
|
timestamp=ts,
|
|
previous_hash=previous_hash,
|
|
entry_hash=entry_hash,
|
|
signature=signature or entry_hash,
|
|
)
|
|
self._entries.append(entry)
|
|
return entry
|
|
|
|
def entries(self) -> List[LedgerEntry]:
|
|
return list(self._entries)
|
|
|
|
def verify(self) -> List[str]:
|
|
issues: List[str] = []
|
|
previous_hash = "genesis"
|
|
for entry in self._entries:
|
|
basis = {
|
|
"entry_id": entry.entry_id,
|
|
"event_type": entry.event_type,
|
|
"payload": entry.payload,
|
|
"actor": entry.actor,
|
|
"timestamp": entry.timestamp,
|
|
"previous_hash": previous_hash,
|
|
}
|
|
expected_hash = _hash(basis)
|
|
if expected_hash != entry.entry_hash:
|
|
issues.append(f"entry {entry.entry_id} hash mismatch")
|
|
if entry.previous_hash != previous_hash:
|
|
issues.append(f"entry {entry.entry_id} chain mismatch")
|
|
previous_hash = entry.entry_hash
|
|
return issues
|
|
|
|
def snapshot(self) -> Dict[str, Any]:
|
|
return {"entries": [entry.to_dict() for entry in self._entries], "fingerprint": self.fingerprint()}
|
|
|
|
@classmethod
|
|
def from_snapshot(cls, snapshot: Dict[str, Any]) -> "GovernanceLedger":
|
|
ledger = cls()
|
|
for entry_data in snapshot.get("entries", []):
|
|
ledger._entries.append(
|
|
LedgerEntry(
|
|
entry_id=entry_data["entry_id"],
|
|
event_type=entry_data["event_type"],
|
|
payload=dict(entry_data.get("payload", {})),
|
|
actor=entry_data["actor"],
|
|
timestamp=entry_data["timestamp"],
|
|
previous_hash=entry_data["previous_hash"],
|
|
entry_hash=entry_data["entry_hash"],
|
|
signature=entry_data.get("signature"),
|
|
)
|
|
)
|
|
return ledger
|
|
|
|
def fingerprint(self) -> str:
|
|
return _hash({"entries": [entry.to_dict() for entry in self._entries]})
|
|
|
|
|
|
def validate_plan_deltas(deltas: Iterable[PlanDelta]) -> List[str]:
|
|
issues: List[str] = []
|
|
for delta in deltas:
|
|
if not delta.actions:
|
|
issues.append(f"{delta.delta_id}: empty actions")
|
|
if delta.signature is not None and not delta.signature.strip():
|
|
issues.append(f"{delta.delta_id}: blank signature")
|
|
return issues
|
|
|
|
|
|
def policy_summary(policy: Policy) -> str:
|
|
return json.dumps(policy.to_dict(), sort_keys=True)
|
|
|
|
|
|
def audit_plan_delta(delta: PlanDelta, actor: str, message: str) -> AuditLog:
|
|
payload = {"delta_id": delta.delta_id, "actor": actor, "message": message, "fingerprint": delta.fingerprint()}
|
|
signature = _hash(payload)
|
|
return AuditLog(entry_id=f"audit-{delta.delta_id}", message=message, signature=signature, anchor=payload["fingerprint"])
|