From 2fa2ccd87c78d7912cb4b600cbec292f2c74b500 Mon Sep 17 00:00:00 2001 From: agent-b883b4bc188823a2 Date: Sun, 26 Apr 2026 22:27:36 +0200 Subject: [PATCH] build(agent): c3po#b883b4 iteration --- AGENTS.md | 1 + README.md | 2 + .../__init__.py | 6 + .../admission.py | 176 ++++++++++++++++++ .../replicator.py | 97 +++++++++- tests/test_core.py | 95 ++++++++++ 6 files changed, 369 insertions(+), 8 deletions(-) create mode 100644 src/idea187_autopoiesis_verifiable_service/admission.py diff --git a/AGENTS.md b/AGENTS.md index 1d5862e..d16d1ba 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -8,6 +8,7 @@ This repository contains a Python reference implementation for the Autopoiesis s - `genome.py`: genome schema, canonical hashing, Ed25519 signing, signed artifact verification. - `policy.py`: policy templates, deterministic smoke-test evaluation, safety attestation. +- `admission.py`: signed capability tokens and deterministic admission decisions. - `ledger.py`: append-only provenance ledger, Merkle root computation, replay trace models. - `replicator.py`: deterministic replication planner and lightweight ReplicaWeave intent model. diff --git a/README.md b/README.md index 53e02c5..2791b81 100644 --- a/README.md +++ b/README.md @@ -7,6 +7,7 @@ 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. +- `AdmissionTokenIssuer` and `AdmissionGate` sign and verify lightweight capability tokens for constrained edges. - `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. @@ -20,6 +21,7 @@ This repository ships a production-shaped core for the autonomy fabric: - content addressing via canonical JSON hashing - signed artifact creation and verification - deterministic smoke-test evaluation +- signed admission tokens for budgeted edge installs - provenance logging with Merkle aggregation - replayable replication traces - a small test suite and buildable Python package diff --git a/src/idea187_autopoiesis_verifiable_service/__init__.py b/src/idea187_autopoiesis_verifiable_service/__init__.py index c6f43f3..2a13acf 100644 --- a/src/idea187_autopoiesis_verifiable_service/__init__.py +++ b/src/idea187_autopoiesis_verifiable_service/__init__.py @@ -1,3 +1,4 @@ +from .admission import AdmissionDecision, AdmissionGate, AdmissionTokenClaims, AdmissionTokenIssuer, SignedAdmissionToken from .genome import ( CapabilityVector, GenomeSigner, @@ -13,6 +14,10 @@ from .policy import PolicySandbox, PolicyTemplate, SafetyAttestation, SmokeTestR from .replicator import ReplicaWeaveIntent, ReplayHarness, ReplicatorPlanner, ReplicationPlan, RolloutStage __all__ = [ + "AdmissionDecision", + "AdmissionGate", + "AdmissionTokenClaims", + "AdmissionTokenIssuer", "CapabilityVector", "GenomeSigner", "LedgerEntry", @@ -27,6 +32,7 @@ __all__ = [ "ResourceEnvelope", "ReplayTrace", "RolloutStage", + "SignedAdmissionToken", "SafetyAttestation", "SafetyInvariants", "ServiceGenome", diff --git a/src/idea187_autopoiesis_verifiable_service/admission.py b/src/idea187_autopoiesis_verifiable_service/admission.py new file mode 100644 index 0000000..bbb1446 --- /dev/null +++ b/src/idea187_autopoiesis_verifiable_service/admission.py @@ -0,0 +1,176 @@ +from __future__ import annotations + +import base64 +import hashlib +import json +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any + +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey, Ed25519PublicKey +from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat +from pydantic import BaseModel, ConfigDict, Field, model_validator + +from .genome import ResourceEnvelope, ServiceGenome + + +def _canonical_json(value: Any) -> bytes: + return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8") + + +class AdmissionTokenClaims(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + issuer_did: str + subject_did: str + audience: str + genome_hash: str + allowed_intents: list[str] = Field(default_factory=list) + safety_labels: list[str] = Field(default_factory=list) + resource_budget: ResourceEnvelope + issued_at: datetime + not_before: datetime + expires_at: datetime + nonce: str + + @model_validator(mode="after") + def _validate_timestamps(self) -> "AdmissionTokenClaims": + if not self.issuer_did.strip(): + raise ValueError("issuer_did cannot be empty") + if not self.subject_did.strip(): + raise ValueError("subject_did cannot be empty") + if not self.audience.strip(): + raise ValueError("audience cannot be empty") + if self.not_before > self.expires_at: + raise ValueError("not_before must be before expires_at") + if self.issued_at > self.expires_at: + raise ValueError("issued_at must be before expires_at") + return self + + +@dataclass(frozen=True) +class SignedAdmissionToken: + claims: AdmissionTokenClaims + signature_b64: str + public_key_b64: str + + def canonical_payload(self) -> bytes: + return _canonical_json(self.claims.model_dump(mode="json", exclude_none=True)) + + def digest(self) -> str: + payload = { + "claims": self.claims.model_dump(mode="json", exclude_none=True), + "public_key_b64": self.public_key_b64, + "signature_b64": self.signature_b64, + } + return hashlib.sha256(_canonical_json(payload)).hexdigest() + + +@dataclass(frozen=True) +class AdmissionDecision: + approved: bool + reasons: tuple[str, ...] + token_digest: str + + def digest(self) -> str: + return hashlib.sha256( + _canonical_json({"approved": self.approved, "reasons": list(self.reasons), "token_digest": self.token_digest}) + ).hexdigest() + + +class AdmissionTokenIssuer: + def __init__(self, private_key: Ed25519PrivateKey) -> None: + self._private_key = private_key + + @classmethod + def generate(cls) -> "AdmissionTokenIssuer": + return cls(Ed25519PrivateKey.generate()) + + @property + def public_key(self) -> Ed25519PublicKey: + return self._private_key.public_key() + + def issue(self, claims: AdmissionTokenClaims) -> SignedAdmissionToken: + signature = self._private_key.sign(_canonical_json(claims.model_dump(mode="json", exclude_none=True))) + return SignedAdmissionToken( + claims=claims, + signature_b64=base64.b64encode(signature).decode("ascii"), + public_key_b64=base64.b64encode( + self.public_key.public_bytes(encoding=Encoding.Raw, format=PublicFormat.Raw) + ).decode("ascii"), + ) + + +class AdmissionGate: + def authorize( + self, + genome: ServiceGenome, + token: SignedAdmissionToken, + *, + intent: Any | None = None, + at: datetime | None = None, + ) -> AdmissionDecision: + reasons: list[str] = [] + now = at or datetime.now(timezone.utc) + + try: + self.verify_signature(token) + except ValueError as exc: + reasons.append(str(exc)) + return AdmissionDecision(approved=False, reasons=tuple(reasons), token_digest=token.digest()) + + claims = token.claims + token_digest = token.digest() + + if claims.genome_hash != genome.content_hash(): + reasons.append("genome hash does not match admission token") + if now < claims.not_before: + reasons.append("admission token is not yet valid") + if now > claims.expires_at: + reasons.append("admission token has expired") + + envelope = genome.resource_envelope + budget = claims.resource_budget + if envelope.cpu_millicores > budget.cpu_millicores: + reasons.append("cpu budget exceeds admission token") + if envelope.memory_mib > budget.memory_mib: + reasons.append("memory budget exceeds admission token") + if envelope.bandwidth_mbps > budget.bandwidth_mbps: + reasons.append("bandwidth budget exceeds admission token") + if envelope.cost_usd_per_hour > budget.cost_usd_per_hour: + reasons.append("cost budget exceeds admission token") + if envelope.max_replicas > budget.max_replicas: + reasons.append("replica budget exceeds admission token") + + if intent is not None: + intent_kind = getattr(intent, "intent_kind", None) + safety_labels = list(getattr(intent, "safety_labels", []) or []) + budget_hint = getattr(intent, "budget_hint", {}) or {} + + if claims.allowed_intents and intent_kind not in claims.allowed_intents: + reasons.append(f"intent kind not allowed: {intent_kind}") + + missing_labels = sorted(set(safety_labels) - set(claims.safety_labels)) + if missing_labels: + reasons.append(f"missing safety label(s): {', '.join(missing_labels)}") + + budget_checks = ( + ("cpu_millicores", "cpu budget hint exceeds admission token"), + ("memory_mib", "memory budget hint exceeds admission token"), + ("bandwidth_mbps", "bandwidth budget hint exceeds admission token"), + ("cost_usd_per_hour", "cost budget hint exceeds admission token"), + ("max_replicas", "replica budget hint exceeds admission token"), + ) + for field_name, reason in budget_checks: + hinted = budget_hint.get(field_name) + if hinted is not None and hinted > getattr(budget, field_name): + reasons.append(reason) + + return AdmissionDecision(approved=not reasons, reasons=tuple(reasons), token_digest=token_digest) + + @staticmethod + def verify_signature(token: SignedAdmissionToken) -> bool: + public_key = Ed25519PublicKey.from_public_bytes(base64.b64decode(token.public_key_b64)) + signature = base64.b64decode(token.signature_b64) + public_key.verify(signature, token.canonical_payload()) + return True diff --git a/src/idea187_autopoiesis_verifiable_service/replicator.py b/src/idea187_autopoiesis_verifiable_service/replicator.py index aa9da31..da18da6 100644 --- a/src/idea187_autopoiesis_verifiable_service/replicator.py +++ b/src/idea187_autopoiesis_verifiable_service/replicator.py @@ -7,6 +7,7 @@ from typing import Any from pydantic import BaseModel, ConfigDict, Field, model_validator +from .admission import AdmissionDecision, AdmissionGate, SignedAdmissionToken from .genome import ServiceGenome from .ledger import ReplayTrace, TraceDelta from .policy import SafetyAttestation @@ -44,6 +45,7 @@ class ReplicationPlan(BaseModel): stages: list[RolloutStage] = Field(default_factory=list) trace: ReplayTrace rationale: list[str] = Field(default_factory=list) + admission_decision: AdmissionDecision | None = None def digest(self) -> str: payload = self.model_dump(mode="json", exclude_none=True) @@ -62,25 +64,89 @@ def _rollout_stages(max_replicas: int, score: float) -> list[RolloutStage]: class ReplicatorPlanner: - def plan(self, genome: ServiceGenome, attestation: SafetyAttestation) -> ReplicationPlan: + def plan( + self, + genome: ServiceGenome, + attestation: SafetyAttestation, + *, + intent: ReplicaWeaveIntent | None = None, + admission_token: SignedAdmissionToken | None = None, + ) -> ReplicationPlan: genome_hash = genome.content_hash() trace_seed = int(genome_hash[:16], 16) rationale = list(attestation.reasons) + admission_decision = None + + if admission_token is not None: + admission_decision = AdmissionGate().authorize(genome, admission_token, intent=intent) + if not admission_decision.approved: + rationale.extend(admission_decision.reasons) + trace = ReplayTrace( + seed=trace_seed, + deltas=[ + TraceDelta( + stage="blocked", + operation="deny", + payload={ + "policy": attestation.policy_name, + "admission_token": admission_decision.token_digest, + "reasons": list(admission_decision.reasons), + }, + ) + ], + ) + return ReplicationPlan( + genome_hash=genome_hash, + approved=False, + stages=[], + trace=trace, + rationale=rationale, + admission_decision=admission_decision, + ) + if not attestation.approved: - 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) + 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, + admission_decision=admission_decision, + ) 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) + signatures = [genome_hash, attestation.digest()] + if admission_decision is not None: + signatures.append(admission_decision.digest()) + trace = ReplayTrace(seed=trace_seed, deltas=deltas, signatures=signatures) + return ReplicationPlan( + genome_hash=genome_hash, + approved=True, + stages=stages, + trace=trace, + rationale=rationale, + admission_decision=admission_decision, + ) class ReplayHarness: - def verify(self, genome: ServiceGenome, attestation: SafetyAttestation, plan: ReplicationPlan) -> ReplayTrace: + def verify( + self, + genome: ServiceGenome, + attestation: SafetyAttestation, + plan: ReplicationPlan, + *, + intent: ReplicaWeaveIntent | None = None, + admission_token: SignedAdmissionToken | None = None, + ) -> ReplayTrace: genome_hash = genome.content_hash() if plan.genome_hash != genome_hash: raise ValueError("plan genome hash does not match genome content") @@ -91,8 +157,20 @@ class ReplayHarness: if plan.approved != attestation.approved: raise ValueError("plan approval does not match attestation") + expected_admission = None + if admission_token is not None: + expected_admission = AdmissionGate().authorize(genome, admission_token, intent=intent) + if plan.admission_decision != expected_admission: + raise ValueError("plan admission decision is not reproducible from the token and intent") + if expected_admission.approved and expected_admission.digest() not in plan.trace.signatures: + raise ValueError("trace signatures do not include the admission decision") + if not plan.approved: - if plan.trace.deltas != [TraceDelta(stage="blocked", operation="deny", payload={"policy": attestation.policy_name})]: + expected_payload: dict[str, object] = {"policy": attestation.policy_name} + if expected_admission is not None and not expected_admission.approved: + expected_payload["admission_token"] = expected_admission.token_digest + expected_payload["reasons"] = list(expected_admission.reasons) + if plan.trace.deltas != [TraceDelta(stage="blocked", operation="deny", payload=expected_payload)]: raise ValueError("blocked trace does not match attestation") return plan.trace @@ -106,6 +184,9 @@ class ReplayHarness: 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()]: + expected_signatures = [genome_hash, attestation.digest()] + if expected_admission is not None: + expected_signatures.append(expected_admission.digest()) + if plan.trace.signatures != expected_signatures: 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 4a56d15..4c27f2e 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -3,6 +3,9 @@ from __future__ import annotations from datetime import datetime, timezone from idea187_autopoiesis_verifiable_service import ( + AdmissionGate, + AdmissionTokenClaims, + AdmissionTokenIssuer, GenomeSigner, CapabilityVector, PolicySandbox, @@ -12,6 +15,7 @@ from idea187_autopoiesis_verifiable_service import ( ReplicatorPlanner, MutationOperator, ResourceEnvelope, + ReplicaWeaveIntent, SafetyInvariants, ServiceGenome, TestContract, @@ -154,3 +158,94 @@ def test_audit_verification_rejects_tampering() -> None: assert "seed" in str(exc) else: raise AssertionError("expected trace verification to fail") + + +def test_admission_token_binds_genome_budget_and_intent() -> None: + genome = build_genome() + attestation = PolicySandbox().evaluate( + genome, + 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"], + ), + ) + issuer = AdmissionTokenIssuer.generate() + claims = AdmissionTokenClaims( + issuer_did="did:key:zissuer", + subject_did="did:key:zsubject", + audience="edge-fleet-a", + genome_hash=genome.content_hash(), + allowed_intents=["fork"], + safety_labels=["read-only", "deterministic"], + resource_budget=ResourceEnvelope( + cpu_millicores=500, + memory_mib=512, + bandwidth_mbps=100, + cost_usd_per_hour=1.0, + max_replicas=3, + ), + issued_at=datetime(2024, 1, 1, tzinfo=timezone.utc), + not_before=datetime(2024, 1, 1, tzinfo=timezone.utc), + expires_at=datetime(2030, 1, 2, tzinfo=timezone.utc), + nonce="nonce-1", + ) + token = issuer.issue(claims) + intent = ReplicaWeaveIntent( + source="catopt", + target="edge-01", + intent_kind="fork", + budget_hint={"cpu_millicores": 250, "max_replicas": 2}, + safety_labels=["read-only"], + ) + + decision = AdmissionGate().authorize(genome, token, intent=intent, at=datetime(2024, 1, 1, 12, tzinfo=timezone.utc)) + assert decision.approved is True + + plan = ReplicatorPlanner().plan(genome, attestation, intent=intent, admission_token=token) + assert plan.approved is True + assert plan.admission_decision == decision + assert decision.digest() in plan.trace.signatures + assert ReplayHarness().verify(genome, attestation, plan, intent=intent, admission_token=token) == plan.trace + + +def test_admission_token_rejects_label_mismatch() -> None: + genome = build_genome() + issuer = AdmissionTokenIssuer.generate() + token = issuer.issue( + AdmissionTokenClaims( + issuer_did="did:key:zissuer", + subject_did="did:key:zsubject", + audience="edge-fleet-a", + genome_hash=genome.content_hash(), + allowed_intents=["fork"], + safety_labels=["read-only"], + resource_budget=ResourceEnvelope( + cpu_millicores=500, + memory_mib=512, + bandwidth_mbps=100, + cost_usd_per_hour=1.0, + max_replicas=3, + ), + issued_at=datetime(2024, 1, 1, tzinfo=timezone.utc), + not_before=datetime(2024, 1, 1, tzinfo=timezone.utc), + expires_at=datetime(2030, 1, 2, tzinfo=timezone.utc), + nonce="nonce-2", + ) + ) + intent = ReplicaWeaveIntent( + source="catopt", + target="edge-01", + intent_kind="fork", + budget_hint={"cpu_millicores": 250}, + safety_labels=["read-only", "requires-attestation"], + ) + + decision = AdmissionGate().authorize(genome, token, intent=intent, at=datetime(2024, 1, 1, 12, tzinfo=timezone.utc)) + assert decision.approved is False + assert any("missing safety label" in reason for reason in decision.reasons)