diff --git a/README.md b/README.md index 263ebe4..53e02c5 100644 --- a/README.md +++ b/README.md @@ -7,8 +7,9 @@ It focuses on the trust substrate needed for safe self-replication across edge a - `ServiceGenome` models immutable service blueprints with resources, capabilities, invariants, mutation operators, and test contracts. - `GenomeSigner` produces and verifies Ed25519 signatures for genome artifacts. - `PolicySandbox` checks a genome against operational budgets and deterministic smoke-test contracts. -- `ProvenanceLedger` records append-only replication events with a Merkle root for tamper evidence. +- `ProvenanceLedger` records append-only replication events with a Merkle root and chain verification for tamper evidence. - `ReplicatorPlanner` turns an approved genome into a deterministic rollout trace. +- `ReplayHarness` replays and verifies a rollout trace against the genome and attestation that produced it. - `ReplicaWeaveIntent` provides a lightweight admission contract for solver/adapter integrations. ## What is implemented diff --git a/src/idea187_autopoiesis_verifiable_service/__init__.py b/src/idea187_autopoiesis_verifiable_service/__init__.py index 0333725..c6f43f3 100644 --- a/src/idea187_autopoiesis_verifiable_service/__init__.py +++ b/src/idea187_autopoiesis_verifiable_service/__init__.py @@ -10,7 +10,7 @@ from .genome import ( ) from .ledger import LedgerEntry, ProvenanceLedger, ReplayTrace, TraceDelta from .policy import PolicySandbox, PolicyTemplate, SafetyAttestation, SmokeTestResult -from .replicator import ReplicaWeaveIntent, ReplicatorPlanner, ReplicationPlan, RolloutStage +from .replicator import ReplicaWeaveIntent, ReplayHarness, ReplicatorPlanner, ReplicationPlan, RolloutStage __all__ = [ "CapabilityVector", @@ -21,6 +21,7 @@ __all__ = [ "PolicyTemplate", "ProvenanceLedger", "ReplicaWeaveIntent", + "ReplayHarness", "ReplicatorPlanner", "ReplicationPlan", "ResourceEnvelope", diff --git a/src/idea187_autopoiesis_verifiable_service/ledger.py b/src/idea187_autopoiesis_verifiable_service/ledger.py index ae91a92..799fb3e 100644 --- a/src/idea187_autopoiesis_verifiable_service/ledger.py +++ b/src/idea187_autopoiesis_verifiable_service/ledger.py @@ -26,6 +26,12 @@ class LedgerEntry(BaseModel): previous_hash: str entry_hash: str + def payload(self) -> dict[str, Any]: + return self.model_dump(mode="json", exclude={"entry_hash"}) + + def recompute_hash(self) -> str: + return hashlib.sha256(_stable_json(self.payload())).hexdigest() + class TraceDelta(BaseModel): model_config = ConfigDict(frozen=True, extra="forbid") @@ -45,6 +51,12 @@ class ReplayTrace(BaseModel): def digest(self) -> str: return hashlib.sha256(_stable_json(self.model_dump(mode="json", exclude_none=True))).hexdigest() + def assert_valid(self, *, seed: int | None = None) -> None: + if seed is not None and self.seed != seed: + raise ValueError("trace seed does not match expected seed") + if len(self.signatures) != len(set(self.signatures)): + raise ValueError("trace signatures must be unique") + class ProvenanceLedger: def __init__(self) -> None: @@ -66,11 +78,12 @@ class ProvenanceLedger: timestamp: datetime | None = None, ) -> LedgerEntry: timestamp = timestamp or datetime.now(timezone.utc) + timestamp_json = timestamp.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") index = len(self._entries) previous_hash = self._entries[-1].entry_hash if self._entries else hashlib.sha256(b"genesis").hexdigest() payload = { "index": index, - "timestamp": timestamp.isoformat(), + "timestamp": timestamp_json, "actor_did": actor_did, "action": action, "genome_hash": genome_hash, @@ -84,6 +97,17 @@ class ProvenanceLedger: self._entries.append(entry) return entry + def verify(self) -> None: + expected_previous_hash = hashlib.sha256(b"genesis").hexdigest() + for index, entry in enumerate(self._entries): + if entry.index != index: + raise ValueError("ledger index mismatch") + if entry.previous_hash != expected_previous_hash: + raise ValueError("ledger previous hash mismatch") + if entry.recompute_hash() != entry.entry_hash: + raise ValueError("ledger entry hash mismatch") + expected_previous_hash = entry.entry_hash + def merkle_root(self) -> str: if not self._entries: return hashlib.sha256(b"").hexdigest() diff --git a/src/idea187_autopoiesis_verifiable_service/replicator.py b/src/idea187_autopoiesis_verifiable_service/replicator.py index 8b69879..aa9da31 100644 --- a/src/idea187_autopoiesis_verifiable_service/replicator.py +++ b/src/idea187_autopoiesis_verifiable_service/replicator.py @@ -50,6 +50,17 @@ class ReplicationPlan(BaseModel): return hashlib.sha256(json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8")).hexdigest() +def _rollout_stages(max_replicas: int, score: float) -> list[RolloutStage]: + canary = min(1, max_replicas) + cohort = max(1, min(max_replicas, max(2, max_replicas // 2))) + stages = [ + RolloutStage(name="canary", cohort_size=canary, percentage=5), + RolloutStage(name="cohort", cohort_size=cohort, percentage=25), + RolloutStage(name="regional", cohort_size=max_replicas, percentage=100), + ] + return stages[: 2 if score < 85 else 3] + + class ReplicatorPlanner: def plan(self, genome: ServiceGenome, attestation: SafetyAttestation) -> ReplicationPlan: genome_hash = genome.content_hash() @@ -59,18 +70,42 @@ class ReplicatorPlanner: trace = ReplayTrace(seed=trace_seed, deltas=[TraceDelta(stage="blocked", operation="deny", payload={"policy": attestation.policy_name})]) return ReplicationPlan(genome_hash=genome_hash, approved=False, stages=[], trace=trace, rationale=rationale) - max_replicas = genome.resource_envelope.max_replicas - canary = min(1, max_replicas) - cohort = max(1, min(max_replicas, max(2, max_replicas // 2))) - stages = [ - RolloutStage(name="canary", cohort_size=canary, percentage=5), - RolloutStage(name="cohort", cohort_size=cohort, percentage=25), - RolloutStage(name="regional", cohort_size=max_replicas, percentage=100), - ] - stages = stages[: 2 if attestation.score < 85 else 3] + stages = _rollout_stages(genome.resource_envelope.max_replicas, attestation.score) deltas = [ TraceDelta(stage=stage.name, operation="rollout", payload={"cohort_size": stage.cohort_size, "percentage": stage.percentage}) for stage in stages ] trace = ReplayTrace(seed=trace_seed, deltas=deltas, signatures=[genome_hash, attestation.digest()]) return ReplicationPlan(genome_hash=genome_hash, approved=True, stages=stages, trace=trace, rationale=rationale) + + +class ReplayHarness: + def verify(self, genome: ServiceGenome, attestation: SafetyAttestation, plan: ReplicationPlan) -> ReplayTrace: + genome_hash = genome.content_hash() + if plan.genome_hash != genome_hash: + raise ValueError("plan genome hash does not match genome content") + + expected_seed = int(genome_hash[:16], 16) + plan.trace.assert_valid(seed=expected_seed) + + if plan.approved != attestation.approved: + raise ValueError("plan approval does not match attestation") + + if not plan.approved: + if plan.trace.deltas != [TraceDelta(stage="blocked", operation="deny", payload={"policy": attestation.policy_name})]: + raise ValueError("blocked trace does not match attestation") + return plan.trace + + expected_stages = _rollout_stages(genome.resource_envelope.max_replicas, attestation.score) + expected_deltas = [ + TraceDelta(stage=stage.name, operation="rollout", payload={"cohort_size": stage.cohort_size, "percentage": stage.percentage}) + for stage in expected_stages + ] + + if plan.stages != expected_stages: + raise ValueError("plan stages are not reproducible from the genome and attestation") + if plan.trace.deltas != expected_deltas: + raise ValueError("trace deltas are not reproducible from the rollout stages") + if plan.trace.signatures != [genome_hash, attestation.digest()]: + raise ValueError("trace signatures do not match the expected audit chain") + return plan.trace diff --git a/tests/test_core.py b/tests/test_core.py index 8917b44..4a56d15 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -4,10 +4,13 @@ from datetime import datetime, timezone from idea187_autopoiesis_verifiable_service import ( GenomeSigner, + CapabilityVector, PolicySandbox, PolicyTemplate, ProvenanceLedger, + ReplayHarness, ReplicatorPlanner, + MutationOperator, ResourceEnvelope, SafetyInvariants, ServiceGenome, @@ -21,7 +24,7 @@ def build_genome() -> ServiceGenome: name="http-service", version="1.0.0", image_ref="ghcr.io/example/http-service:1.0.0", - capabilities={"network": ["https://internal.example"]}, + capabilities=CapabilityVector(network=["https://internal.example"]), resource_envelope=ResourceEnvelope( cpu_millicores=250, memory_mib=256, @@ -30,7 +33,7 @@ def build_genome() -> ServiceGenome: max_replicas=3, ), safety_invariants=SafetyInvariants(max_exposure=10, read_only_scopes=["/var/lib/cache"], denied_endpoints=["https://example.com"]), - mutation_operators=[{"name": "safe-fork", "kind": "fork"}], + mutation_operators=[MutationOperator(name="safe-fork", kind="fork")], test_contracts=[TestContract(name="startup", seed=7, input_fingerprint="boot", expected_output_hash=expected)], ) @@ -104,3 +107,50 @@ def test_ledger_merkle_root_and_replay_trace_are_deterministic() -> None: assert ledger.merkle_root() assert plan.approved is True assert plan.trace.digest() == plan.trace.digest() + ledger.verify() + assert ReplayHarness().verify(genome, attestation, plan) == plan.trace + + +def test_audit_verification_rejects_tampering() -> None: + genome = build_genome() + policy = PolicyTemplate( + name="default-edge", + max_cpu_millicores=500, + max_memory_mib=512, + max_bandwidth_mbps=100, + max_cost_usd_per_hour=1.0, + allow_network_egress=True, + allowed_egress_endpoints=["https://internal.example"], + allowed_mutation_operators=["safe-fork"], + ) + attestation = PolicySandbox().evaluate(genome, policy) + plan = ReplicatorPlanner().plan(genome, attestation) + + tampered_plan = plan.model_copy(update={"trace": plan.trace.model_copy(update={"seed": plan.trace.seed + 1})}) + + ledger = ProvenanceLedger() + ledger.append( + actor_did="did:key:z6Mkh1", + action="replicate", + genome_hash=genome.content_hash(), + attestation_hash=attestation.digest(), + trace_hash=plan.trace.digest(), + resource_snapshot=genome.resource_envelope.model_dump(mode="json"), + timestamp=datetime(2024, 1, 1, tzinfo=timezone.utc), + ) + + ledger._entries[0] = ledger._entries[0].model_copy(update={"previous_hash": "broken"}) + + try: + ledger.verify() + except ValueError as exc: + assert "previous hash" in str(exc) + else: + raise AssertionError("expected ledger verification to fail") + + try: + ReplayHarness().verify(genome, attestation, tampered_plan) + except ValueError as exc: + assert "seed" in str(exc) + else: + raise AssertionError("expected trace verification to fail")