299 lines
18 KiB
Python
299 lines
18 KiB
Python
"""Thin unattended release runner.
|
|
|
|
This module is an adapter around the existing admission and silent-release
|
|
services. It deliberately owns no release state machine and never creates
|
|
approval evidence. The runner only binds immutable source facts, delegates
|
|
authority, and validates durable readbacks before moving to the next phase.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import time
|
|
from typing import Any, Callable, Mapping
|
|
|
|
|
|
OID_RE = re.compile(r"^[0-9a-fA-F]{40}$")
|
|
SHA256_RE = re.compile(r"^[0-9a-fA-F]{64}$")
|
|
IDENT_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}$")
|
|
RETRYABLE_CODES = {408, 425, 429, 500, 502, 503, 504}
|
|
RETRYABLE_NAMES = {
|
|
"UNAVAILABLE", "BUSY", "TEMPORARY_FAILURE", "TASK_ADMISSION_IN_PROGRESS",
|
|
}
|
|
RISKS = {"low", "medium", "high"}
|
|
|
|
|
|
class UnattendedReleaseError(RuntimeError):
|
|
"""A fail-closed runner error with a stable code."""
|
|
|
|
def __init__(self, code: str, message: str | None = None):
|
|
self.code = code
|
|
super().__init__(message or code)
|
|
|
|
|
|
def _canon(value: Any) -> bytes:
|
|
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode()
|
|
|
|
|
|
def _digest(value: Any) -> str:
|
|
return hashlib.sha256(_canon(value)).hexdigest()
|
|
|
|
|
|
def _require_oid(name: str, value: Any) -> str:
|
|
value = str(value or "")
|
|
if not OID_RE.fullmatch(value):
|
|
raise UnattendedReleaseError("INVALID_" + name.upper())
|
|
return value.lower()
|
|
|
|
|
|
def _require_sha(name: str, value: Any) -> str:
|
|
value = str(value or "")
|
|
if not SHA256_RE.fullmatch(value):
|
|
raise UnattendedReleaseError("INVALID_" + name.upper())
|
|
return value.lower()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SourcePublicationIntent:
|
|
"""Immutable, deterministic source and execution binding."""
|
|
|
|
repository: str
|
|
source_commit: str
|
|
source_tree_oid: str
|
|
authority_base_oid: str
|
|
target_branch: str
|
|
file_allowlist: tuple[str, ...]
|
|
requested_capabilities: tuple[str, ...]
|
|
risk: str
|
|
execution_digest: str
|
|
generation: int = 1
|
|
event_id: str = "release"
|
|
version: str = ""
|
|
|
|
def __post_init__(self) -> None:
|
|
if not self.repository or not isinstance(self.repository, str):
|
|
raise UnattendedReleaseError("INVALID_REPOSITORY")
|
|
_require_oid("source_commit", self.source_commit)
|
|
_require_oid("source_tree_oid", self.source_tree_oid)
|
|
_require_oid("authority_base_oid", self.authority_base_oid)
|
|
if not isinstance(self.target_branch, str) or not self.target_branch or not self.target_branch.startswith("agent/") or self.target_branch.startswith("-"):
|
|
raise UnattendedReleaseError("INVALID_TARGET_BRANCH")
|
|
if ".." in self.target_branch or "//" in self.target_branch or "@{" in self.target_branch:
|
|
raise UnattendedReleaseError("INVALID_TARGET_BRANCH")
|
|
if not isinstance(self.generation, int) or isinstance(self.generation, bool) or self.generation < 1:
|
|
raise UnattendedReleaseError("INVALID_GENERATION")
|
|
_require_sha("execution_digest", self.execution_digest)
|
|
if not self.file_allowlist or any(not isinstance(p, str) or not p or p.startswith(("/", "\\")) or ".." in p or "\\" in p or p == ".git" or p.startswith(".git/") for p in self.file_allowlist):
|
|
raise UnattendedReleaseError("INVALID_FILE_ALLOWLIST")
|
|
if len(set(self.file_allowlist)) != len(self.file_allowlist):
|
|
raise UnattendedReleaseError("DUPLICATE_FILE_ALLOWLIST")
|
|
if not self.requested_capabilities or any(not isinstance(c, str) or not c or "*" in c for c in self.requested_capabilities):
|
|
raise UnattendedReleaseError("INVALID_CAPABILITIES")
|
|
if not isinstance(self.risk, str) or self.risk.lower() not in RISKS:
|
|
raise UnattendedReleaseError("INVALID_RISK")
|
|
object.__setattr__(self, "risk", self.risk.lower())
|
|
object.__setattr__(self, "file_allowlist", tuple(sorted(self.file_allowlist)))
|
|
object.__setattr__(self, "requested_capabilities", tuple(sorted(set(self.requested_capabilities))))
|
|
if not IDENT_RE.fullmatch(self.event_id):
|
|
raise UnattendedReleaseError("INVALID_EVENT_ID")
|
|
if self.version and not IDENT_RE.fullmatch(self.version):
|
|
raise UnattendedReleaseError("INVALID_VERSION")
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"repository": self.repository, "source_commit": self.source_commit.lower(),
|
|
"source_tree_oid": self.source_tree_oid.lower(), "authority_base_oid": self.authority_base_oid.lower(),
|
|
"target_branch": self.target_branch, "file_allowlist": list(self.file_allowlist),
|
|
"requested_capabilities": list(self.requested_capabilities), "risk": self.risk,
|
|
"execution_digest": self.execution_digest.lower(), "generation": self.generation,
|
|
"event_id": self.event_id, "version": self.version,
|
|
}
|
|
|
|
@property
|
|
def task_id(self) -> str:
|
|
return "release-task-" + _digest(self.as_dict())[:32]
|
|
|
|
@property
|
|
def idempotency_key(self) -> str:
|
|
return "release-intake-" + _digest({"task_id": self.task_id, "execution_digest": self.execution_digest.lower()})[:32]
|
|
|
|
def admission_request(self) -> dict[str, Any]:
|
|
"""Return the sole normalized request accepted by Task admission."""
|
|
return {"idempotency_key": self.idempotency_key, "task_id": self.task_id,
|
|
"repository": self.repository, "authority_base_oid": self.authority_base_oid,
|
|
"target_branch": self.target_branch, "generation": self.generation,
|
|
"execution_digest": self.execution_digest, "requested_capabilities": list(self.requested_capabilities),
|
|
"file_allowlist": list(self.file_allowlist), "risk": self.risk}
|
|
|
|
|
|
def _error_code(exc: BaseException) -> str:
|
|
code = getattr(exc, "code", None) or getattr(exc, "status", None) or exc.__class__.__name__
|
|
return str(code).upper().replace(" ", "_")
|
|
|
|
|
|
def _retryable(exc: BaseException) -> bool:
|
|
status = getattr(exc, "status_code", None) or getattr(exc, "http_status", None) or getattr(exc, "status", None)
|
|
if isinstance(status, int):
|
|
return status in RETRYABLE_CODES
|
|
code = _error_code(exc)
|
|
return code in RETRYABLE_NAMES
|
|
|
|
|
|
def _unknown_outcome(exc: BaseException) -> bool:
|
|
return isinstance(exc, (TimeoutError, ConnectionError, OSError)) or _error_code(exc) in {"TASK_ADMISSION_TASK_STATE_UNKNOWN", "TIMEOUT", "TIMED_OUT", "CONNECTION_LOST"}
|
|
|
|
|
|
class GovernedUnattendedReleaseRunner:
|
|
"""Phase-separated runner; preparation has no release/deployment rights."""
|
|
|
|
def __init__(self, *, admission: Any, release_service: Any = None,
|
|
publication_adapter: Any = None, owner_approval_broker: Callable[[Mapping[str, Any]], Mapping[str, Any]] | None = None,
|
|
deployment_enabled: bool = False, deployment_service: Any = None,
|
|
max_attempts: int = 3, deadline_seconds: float = 30.0,
|
|
environment_mode: str = "SIMULATION",
|
|
clock: Callable[[], float] = time.monotonic, sleep: Callable[[float], None] = time.sleep):
|
|
if admission is None:
|
|
raise ValueError("admission is required")
|
|
if max_attempts < 1 or max_attempts > 10 or deadline_seconds <= 0:
|
|
raise ValueError("invalid retry bounds")
|
|
self.admission = admission
|
|
self.environment_mode = str(environment_mode).upper()
|
|
if self.environment_mode not in {"SIMULATION", "AUTHORITY"}:
|
|
raise ValueError("environment_mode must be SIMULATION or AUTHORITY")
|
|
self.release_service = release_service
|
|
self.publication_adapter = publication_adapter
|
|
self.owner_approval_broker = owner_approval_broker
|
|
self.deployment_enabled = bool(deployment_enabled)
|
|
self.deployment_service = deployment_service
|
|
self.max_attempts = max_attempts
|
|
self.deadline_seconds = deadline_seconds
|
|
self.clock, self.sleep = clock, sleep
|
|
|
|
def intake(self, intent: SourcePublicationIntent) -> dict[str, Any]:
|
|
if not isinstance(intent, SourcePublicationIntent):
|
|
raise UnattendedReleaseError("INTENT_TYPE_REQUIRED")
|
|
return {"phase": "PREPARATION", "mode": self.environment_mode, "task_id": intent.task_id,
|
|
"idempotency_key": intent.idempotency_key, "intent": intent.as_dict(),
|
|
"release_rights": False, "deployment_rights": False}
|
|
|
|
def admit(self, intent: SourcePublicationIntent, authorization: Mapping[str, Any]) -> dict[str, Any]:
|
|
if not isinstance(authorization, Mapping):
|
|
raise UnattendedReleaseError("AUTHORIZATION_REQUIRED")
|
|
if self.environment_mode != "AUTHORITY":
|
|
return {"phase": "BLOCKED", "mode": self.environment_mode, "task_id": intent.task_id, "error": "AUTHORITY_MODE_REQUIRED"}
|
|
raw = intent.admission_request()
|
|
deadline = self.clock() + self.deadline_seconds
|
|
envelope_id = None
|
|
for attempt in range(1, self.max_attempts + 1):
|
|
if self.clock() >= deadline:
|
|
return {"phase": "RECONCILIATION_REQUIRED" if envelope_id else "BLOCKED", "mode": "AUTHORITY", "task_id": intent.task_id, "error": "ADMISSION_DEADLINE_EXCEEDED"}
|
|
try:
|
|
created = self.admission.create_envelope(raw)
|
|
envelope = created.get("envelope", created) if isinstance(created, Mapping) else {}
|
|
envelope_id = envelope.get("envelope_id") or envelope_id
|
|
auth = dict(authorization)
|
|
auth["envelope_id"] = envelope_id
|
|
if self.clock() >= deadline:
|
|
return self._reconcile(envelope_id, intent, "ADMISSION_DEADLINE_EXCEEDED")
|
|
receipt = self.admission.authorize_and_admit(auth)
|
|
if self.clock() >= deadline:
|
|
return self._reconcile(envelope_id, intent, "ADMISSION_DEADLINE_EXCEEDED")
|
|
return {"phase": "ADMITTED", "mode": "AUTHORITY", "attempt": attempt,
|
|
"envelope": created, "receipt": receipt}
|
|
except Exception as exc:
|
|
status_code = getattr(exc, "status_code", None) or getattr(exc, "http_status", None) or getattr(exc, "status", None)
|
|
if status_code == 403 or _error_code(exc) in {"403", "FORBIDDEN"}:
|
|
return {"phase": "BLOCKED", "mode": "AUTHORITY", "task_id": intent.task_id, "error": _error_code(exc)}
|
|
if envelope_id and (_unknown_outcome(exc) or not _retryable(exc) or self.clock() >= deadline or attempt >= self.max_attempts):
|
|
return self._reconcile(envelope_id, intent, _error_code(exc))
|
|
if not _retryable(exc) or self.clock() >= deadline or attempt >= self.max_attempts:
|
|
return {"phase": "BLOCKED", "mode": "AUTHORITY", "task_id": intent.task_id, "error": _error_code(exc)}
|
|
self.sleep(min(0.05 * (2 ** (attempt - 1)), max(0.0, deadline - self.clock())))
|
|
raise UnattendedReleaseError("ADMISSION_RETRY_EXHAUSTED")
|
|
|
|
def _reconcile(self, envelope_id: str | None, intent: SourcePublicationIntent, error: str) -> dict[str, Any]:
|
|
if not envelope_id:
|
|
return {"phase": "RECONCILIATION_REQUIRED", "mode": "AUTHORITY", "task_id": intent.task_id, "error": error}
|
|
try:
|
|
status = self.admission.status(envelope_id)
|
|
except Exception as exc:
|
|
return {"phase": "RECONCILIATION_REQUIRED", "mode": "AUTHORITY", "task_id": intent.task_id, "error": error, "status_error": _error_code(exc)}
|
|
state = str(status.get("status") or status.get("task_state") or "").upper() if isinstance(status, Mapping) else ""
|
|
if state in {"NOT_APPLIED", "ABSENT", "NOT_FOUND"}:
|
|
return {"phase": "RETRYABLE_RECONCILIATION", "mode": "AUTHORITY", "task_id": intent.task_id, "status": status, "error": error}
|
|
if state in {"QUEUED", "CLAIMED", "RUNNING", "ADMITTED", "SUCCEEDED", "COMPLETED"}:
|
|
return {"phase": "RECONCILED", "mode": "AUTHORITY", "task_id": intent.task_id, "status": status}
|
|
return {"phase": "RECONCILIATION_REQUIRED", "mode": "AUTHORITY", "task_id": intent.task_id, "status": status, "error": error}
|
|
|
|
def publish(self, intent: SourcePublicationIntent, admission_receipt: Mapping[str, Any]) -> dict[str, Any]:
|
|
if self.environment_mode != "AUTHORITY":
|
|
return {"phase": "BLOCKED", "mode": self.environment_mode, "error": "AUTHORITY_MODE_REQUIRED"}
|
|
if self.publication_adapter is None:
|
|
return {"phase": "BLOCKED", "mode": "AUTHORITY", "error": "PUBLICATION_BACKEND_MISSING"}
|
|
required = ("task_lease", "task_principal", "writer_lease", "commit_receipt", "source_oid", "tree_oid", "authenticated")
|
|
try:
|
|
if not isinstance(admission_receipt, Mapping) or any(str(admission_receipt.get(k, "")) != expected for k, expected in (("task_id", intent.task_id), ("authority_base_oid", intent.authority_base_oid), ("target_branch", intent.target_branch), ("execution_digest", intent.execution_digest))):
|
|
return {"phase": "BLOCKED", "mode": "AUTHORITY", "error": "ADMISSION_RECEIPT_BINDING_MISMATCH"}
|
|
result = self.publication_adapter.publish(intent, admission_receipt)
|
|
if not isinstance(result, Mapping) or result.get("authenticated") is not True or any(not result.get(k) for k in required[:-1]):
|
|
return {"phase": "BLOCKED", "mode": "AUTHORITY", "error": "PUBLICATION_LEASE_CHAIN_MISSING"}
|
|
if not OID_RE.fullmatch(str(result["source_oid"])) or str(result["tree_oid"]).lower() != intent.source_tree_oid.lower() or any(str(result.get(k, "")) != str(v) for k, v in (("task_id", intent.task_id), ("authority_base_oid", intent.authority_base_oid), ("target_branch", intent.target_branch), ("execution_digest", intent.execution_digest))):
|
|
return {"phase": "BLOCKED", "mode": "AUTHORITY", "error": "PUBLICATION_RC_BINDING_MISMATCH"}
|
|
return {"phase": "PUBLISHED", "mode": "AUTHORITY", "readback": dict(result)}
|
|
except Exception as exc:
|
|
return {"phase": "BLOCKED", "mode": "AUTHORITY", "error": _error_code(exc)}
|
|
|
|
def postpublication_rc(self, *, event_id: str, source_commit: str, version: str = "", source_readback: Mapping[str, Any] | None = None) -> dict[str, Any]:
|
|
if self.environment_mode != "AUTHORITY":
|
|
return {"phase": "BLOCKED", "mode": self.environment_mode, "error": "AUTHORITY_MODE_REQUIRED"}
|
|
if not isinstance(source_readback, Mapping) or source_readback.get("authenticated") is not True or str(source_readback.get("source_commit", "")).lower() != str(source_commit).lower():
|
|
return {"phase": "BLOCKED", "mode": "AUTHORITY", "error": "SOURCE_READBACK_REQUIRED"}
|
|
if self.release_service is None:
|
|
return {"phase": "BLOCKED", "mode": "AUTHORITY", "error": "RELEASE_BACKEND_MISSING"}
|
|
try:
|
|
intent = self.release_service.create_release_intent({"event_id": event_id, "source_commit": source_commit, **({"version": version} if version else {})})
|
|
candidate_id = str(intent.get("candidate_id") or intent.get("candidate", {}).get("candidate_id") or "")
|
|
if not candidate_id:
|
|
return {"phase": "BLOCKED", "mode": "AUTHORITY", "error": "CANDIDATE_BINDING_MISSING"}
|
|
readers = ("get_release_candidate", "get_release_plan", "get_acceptance")
|
|
if any(not callable(getattr(self.release_service, name, None)) for name in readers):
|
|
return {"phase": "BLOCKED", "mode": "AUTHORITY", "candidate_id": candidate_id, "error": "RC_READBACK_BACKEND_MISSING"}
|
|
candidate = self.release_service.get_release_candidate(candidate_id)
|
|
plan = self.release_service.get_release_plan(candidate_id)
|
|
acceptance = self.release_service.get_acceptance(candidate_id)
|
|
deployment = plan.get("deployment") if isinstance(plan, Mapping) else None
|
|
if (not isinstance(candidate, Mapping) or not isinstance(plan, Mapping) or not isinstance(deployment, Mapping) or not isinstance(acceptance, Mapping)
|
|
or str(candidate.get("candidate_id")) != candidate_id
|
|
or str(candidate.get("source_commit")) .lower() != str(source_commit).lower()
|
|
or str(plan.get("candidate_id", candidate_id)) != candidate_id
|
|
or str(deployment.get("source_commit")) .lower() != str(source_commit).lower()
|
|
or str(deployment.get("tree_oid", "")).lower() != str(source_readback.get("tree_oid", "")).lower()
|
|
or str(acceptance.get("candidate_id")) != candidate_id
|
|
or str(acceptance.get("status")) != "ACCEPTED"):
|
|
return {"phase": "BLOCKED", "mode": "AUTHORITY", "candidate_id": candidate_id, "error": "RC_READBACK_BINDING_MISMATCH"}
|
|
envelope = self.release_service.get_approval_envelope(candidate_id)
|
|
if self.owner_approval_broker is None:
|
|
return {"phase": "BLOCKED", "mode": "AUTHORITY", "candidate_id": candidate_id, "error": "OWNER_APPROVAL_BACKEND_MISSING"}
|
|
approval = self.owner_approval_broker(dict(envelope))
|
|
if not isinstance(approval, Mapping):
|
|
return {"phase": "BLOCKED", "mode": "AUTHORITY", "candidate_id": candidate_id, "error": "OWNER_APPROVAL_MISSING"}
|
|
accepted = self.release_service.submit_owner_approval(dict(approval))
|
|
return {"phase": "APPROVED", "mode": "AUTHORITY", "candidate_id": candidate_id,
|
|
"readback": {"candidate": dict(candidate), "plan": dict(plan), "acceptance": dict(acceptance)},
|
|
"approval": accepted}
|
|
except Exception as exc:
|
|
return {"phase": "BLOCKED", "mode": "AUTHORITY", "error": _error_code(exc)}
|
|
|
|
def deploy(self, candidate_id: str) -> dict[str, Any]:
|
|
if self.environment_mode != "AUTHORITY" or not self.deployment_enabled or self.deployment_service is None:
|
|
return {"phase": "BLOCKED", "mode": "AUTHORITY", "error": "DEPLOYMENT_NOT_ENABLED"}
|
|
try:
|
|
result = self.deployment_service.start_deployment(candidate_id)
|
|
if not isinstance(result, Mapping) or result.get("execution_enabled") is not True or result.get("side_effects_executed") is not True or str(result.get("status", "")).upper() not in {"COMPLETED", "COMPLETE"}:
|
|
return {"phase": "BLOCKED", "mode": "AUTHORITY", "error": "DEPLOYMENT_NOT_VERIFIED", "result": dict(result) if isinstance(result, Mapping) else {}}
|
|
return {"phase": "DEPLOYED", "mode": "AUTHORITY", "result": dict(result)}
|
|
except Exception as exc:
|
|
return {"phase": "BLOCKED", "mode": "AUTHORITY", "error": _error_code(exc)}
|