idea187-autopoiesis-verifiable-service/tests/test_core.py

from __future__ import annotations

from datetime import datetime, timezone

from idea187_autopoiesis_verifiable_service import (
    AdmissionGate,
    AdmissionTokenClaims,
    AdmissionTokenIssuer,
    GenomeSigner,
    CapabilityVector,
    PolicySandbox,
    PolicyTemplate,
    ProvenanceLedger,
    ReplayHarness,
    ReplicatorPlanner,
    MutationOperator,
    ResourceEnvelope,
    ReplicaWeaveIntent,
    SafetyInvariants,
    ServiceGenome,
    TestContract,
)


def build_genome() -> ServiceGenome:
    expected = __import__("hashlib").sha256("7:boot".encode("utf-8")).hexdigest()
    return ServiceGenome(
        name="http-service",
        version="1.0.0",
        image_ref="ghcr.io/example/http-service:1.0.0",
        capabilities=CapabilityVector(network=["https://internal.example"]),
        resource_envelope=ResourceEnvelope(
            cpu_millicores=250,
            memory_mib=256,
            bandwidth_mbps=20,
            cost_usd_per_hour=0.05,
            max_replicas=3,
        ),
        safety_invariants=SafetyInvariants(max_exposure=10, read_only_scopes=["/var/lib/cache"], denied_endpoints=["https://example.com"]),
        mutation_operators=[MutationOperator(name="safe-fork", kind="fork")],
        test_contracts=[TestContract(name="startup", seed=7, input_fingerprint="boot", expected_output_hash=expected)],
    )


def test_genome_sign_and_verify() -> None:
    genome = build_genome()
    signer = GenomeSigner.generate()
    artifact = signer.sign(genome)

    assert artifact.genome_hash == genome.content_hash()
    assert GenomeSigner.verify(artifact) is True


def test_policy_sandbox_approves_safe_genome() -> 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)

    assert attestation.approved is True
    assert attestation.reasons == []
    assert all(result.passed for result in attestation.smoke_tests)


def test_ledger_merkle_root_and_replay_trace_are_deterministic() -> None:
    genome = build_genome()
    signer = GenomeSigner.generate()
    artifact = signer.sign(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)

    ledger = ProvenanceLedger()
    first = ledger.append(
        actor_did="did:key:z6Mkh1",
        action="replicate",
        genome_hash=artifact.genome_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),
    )
    second = ledger.append(
        actor_did="did:key:z6Mkh2",
        action="promote",
        genome_hash=artifact.genome_hash,
        attestation_hash=attestation.digest(),
        trace_hash=plan.trace.digest(),
        resource_snapshot=genome.resource_envelope.model_dump(mode="json"),
        timestamp=datetime(2024, 1, 1, 0, 1, tzinfo=timezone.utc),
    )

    assert first.entry_hash != second.entry_hash
    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")


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)