117 lines
4.4 KiB
Python
117 lines
4.4 KiB
Python
"""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")
|
|
try:
|
|
signature_valid = self.signature_verifier.verify(
|
|
envelope.signing_key_id, payload, envelope.signature
|
|
)
|
|
except Exception as error:
|
|
raise ApprovalRequired(
|
|
"BLOCKED_OWNER_APPROVAL_VERIFIER_UNAVAILABLE",
|
|
"trusted Owner approval verifier is unavailable",
|
|
) from error
|
|
if not signature_valid:
|
|
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
|
|
|
|
|
|
class UnavailableSignatureVerifier:
|
|
"""Safe runtime default when no Owner HSM/KMS verifier has been supplied."""
|
|
|
|
def verify(self, key_id: str, payload: bytes, signature: str) -> bool:
|
|
raise ConnectionError("Owner verifier unavailable")
|