feat(autonomous-release): add owner approval verifier

services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/approval.py

@@ -0,0 +1,100 @@
+"""Owner approval verification contracts for candidate releases."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Protocol
+
+from .errors import ApprovalRequired, StateConflict
+from .models import ApprovalEnvelope, TaskPlan, canonical_json
+from .store import StateStore
+
+
+class SignatureVerifier(Protocol):
+    """Verify an approval payload without exposing private-key operations."""
+
+    def verify(self, key_id: str, payload: bytes, signature: str) -> bool:
+        """Return ``True`` when the signature is valid for ``key_id``."""
+
+
+class RevocationSource(Protocol):
+    def is_revoked(self, approval_plan_id: str) -> bool:
+        """Return whether an approval plan has been revoked."""
+
+
+@dataclass(frozen=True, slots=True)
+class ApprovalVerificationResult:
+    approval_plan_id: str
+    task_id: str
+    signing_key_id: str
+    verified_at: datetime
+
+
+class ApprovalVerifier:
+    """Verify exact approval bindings and consume nonces before execution."""
+
+    def __init__(
+        self,
+        store: StateStore,
+        signature_verifier: SignatureVerifier,
+        revocations: RevocationSource,
+    ) -> None:
+        self.store = store
+        self.signature_verifier = signature_verifier
+        self.revocations = revocations
+
+    def verify(
+        self,
+        plan: TaskPlan,
+        envelope: ApprovalEnvelope,
+        *,
+        now: datetime | None = None,
+    ) -> ApprovalVerificationResult:
+        now = now or datetime.now(timezone.utc)
+        if envelope.task_id != plan.task_id:
+            raise ApprovalRequired("APPROVAL_TASK_MISMATCH", "approval task does not match plan")
+        if envelope.task_plan_sha256 != plan.digest():
+            raise ApprovalRequired("APPROVAL_PLAN_DIGEST_MISMATCH", "approval digest does not match plan")
+        if envelope.binding != plan.binding:
+            raise ApprovalRequired("APPROVAL_RESOURCE_MISMATCH", "approval resource binding does not match plan")
+        approved = set(envelope.capabilities)
+        required = {cap for step in plan.steps for cap in step.required_capabilities}
+        if not required.issubset(approved):
+            raise ApprovalRequired(
+                "APPROVAL_CAPABILITY_INSUFFICIENT",
+                "approval does not cover all plan capabilities",
+                {"missing": sorted(required - approved)},
+            )
+        if now < envelope.not_before or now >= envelope.expires_at:
+            raise ApprovalRequired("APPROVAL_TIME_INVALID", "approval is outside its validity window")
+        if self.revocations.is_revoked(envelope.approval_plan_id):
+            raise ApprovalRequired("APPROVAL_REVOKED", "approval plan is revoked")
+        payload = canonical_json(envelope.payload()).encode("utf-8")
+        if not self.signature_verifier.verify(envelope.signing_key_id, payload, envelope.signature):
+            raise ApprovalRequired("APPROVAL_SIGNATURE_INVALID", "approval signature verification failed")
+        try:
+            self.store.consume_nonce(f"approval:{envelope.approval_plan_id}", envelope.nonce)
+        except StateConflict as error:
+            raise ApprovalRequired("APPROVAL_NONCE_REPLAY", "approval nonce was already consumed") from error
+        return ApprovalVerificationResult(
+            approval_plan_id=envelope.approval_plan_id,
+            task_id=envelope.task_id,
+            signing_key_id=envelope.signing_key_id,
+            verified_at=now,
+        )
+
+
+class StaticRevocationSource:
+    def __init__(self, revoked: set[str] | None = None) -> None:
+        self.revoked = set(revoked or ())
+
+    def is_revoked(self, approval_plan_id: str) -> bool:
+        return approval_plan_id in self.revoked
+
+
+class RejectingSignatureVerifier:
+    """Safe default used when no external verifier has been configured."""
+
+    def verify(self, key_id: str, payload: bytes, signature: str) -> bool:
+        return False