"""In-memory reference behavior for the fixed WP13 fixture dispatcher. The production persistence boundary is supplied by the WP13 PostgreSQL gateway. This small adapter exists for deterministic unit contracts; it never loads or executes a submitted artifact. """ from __future__ import annotations import copy import hashlib import threading import uuid from collections.abc import Callable from typing import Any from .governance import ( PluginManifestError, canonical_digest, fixture_signature_digest, normalize_plugin_manifest, ) class PluginLifecycleError(ValueError): """A lifecycle transition, approval, registry record or invocation was denied.""" _RECORD = frozenset( {"artifact_digest", "signature", "sbom_digest", "license_digest", "vulnerability_digest", "provenance_digest"} ) _SIGNATURE = frozenset({"trust_store_key_id", "signature_digest"}) _TRUST_STORE = frozenset({"local-fixture-key-v1"}) _HEX = frozenset("0123456789abcdef") _ACTIONS = frozenset({"activate", "pause", "rollback", "revoke", "recover"}) def _digest(value: Any, label: str) -> str: if not isinstance(value, str) or len(value) != 64 or set(value) - _HEX: raise PluginLifecycleError(f"{label}_invalid") return value def _closed(value: Any, keys: frozenset[str], label: str) -> dict[str, Any]: if not isinstance(value, dict) or set(value) != keys: raise PluginLifecycleError(f"{label}_closed") return copy.deepcopy(value) def _scope(value: Any, *, principal: bool) -> dict[str, str]: keys = {"tenant_id", "domain"} | ({"principal_id"} if principal else set()) if not isinstance(value, dict) or set(value) != keys: raise PluginLifecycleError("scope_closed") normalized: dict[str, str] = {} for key in keys: item = value[key] if not isinstance(item, str) or not item.isascii() or not 1 <= len(item) <= 120: raise PluginLifecycleError("scope_invalid") normalized[key] = item return normalized class PluginPlatform: """Thread-safe local state machine around one fixed evidence fixture.""" def __init__(self, *, now: Callable[[], int]) -> None: self._now = now self._lock = threading.RLock() self._versions: dict[tuple[str, str], dict[str, Any]] = {} self._approvals: dict[str, dict[str, Any]] = {} self._runs: dict[tuple[str, str, str, str], dict[str, Any]] = {} def register(self, manifest: Any, *, registry_record: Any) -> dict[str, Any]: try: normalized = normalize_plugin_manifest(manifest) except PluginManifestError as exc: raise PluginLifecycleError(str(exc)) from exc record = _closed(registry_record, _RECORD, "registry_record") signature = _closed(record["signature"], _SIGNATURE, "signature") if signature["trust_store_key_id"] not in _TRUST_STORE: raise PluginLifecycleError("trust_store_key_denied") normalized_record = { "artifact_digest": _digest(record["artifact_digest"], "artifact_digest"), "signature": { "trust_store_key_id": signature["trust_store_key_id"], "signature_digest": _digest(signature["signature_digest"], "signature_digest"), }, "sbom_digest": _digest(record["sbom_digest"], "sbom_digest"), "license_digest": _digest(record["license_digest"], "license_digest"), "vulnerability_digest": _digest(record["vulnerability_digest"], "vulnerability_digest"), "provenance_digest": _digest(record["provenance_digest"], "provenance_digest"), } if normalized_record["artifact_digest"] != normalized["distribution"]["artifact_digest"]: raise PluginLifecycleError("artifact_digest_mismatch") if normalized_record["signature"]["signature_digest"] != fixture_signature_digest(normalized_record["artifact_digest"]): raise PluginLifecycleError("signature_binding_denied") key = (normalized["plugin_uid"], normalized["version"]) with self._lock: if key in self._versions: raise PluginLifecycleError("plugin_version_exists") item = normalized | {"state": "draft", "registry": normalized_record, "reviewer_id": None} self._versions[key] = item return copy.deepcopy(item) def review(self, plugin_uid: str, version: str, *, reviewer_id: str) -> dict[str, Any]: if not isinstance(reviewer_id, str) or not reviewer_id.isascii() or not reviewer_id: raise PluginLifecycleError("reviewer_invalid") with self._lock: item = self._version(plugin_uid, version) if item["state"] != "draft": raise PluginLifecycleError("review_transition_denied") item["state"] = "reviewed" item["reviewer_id"] = reviewer_id return copy.deepcopy(item) def issue_approval(self, *, plugin_uid: str, version: str, action: str, actor_id: str, reviewer_id: str, scope: Any, expires_at: int) -> dict[str, Any]: if action not in _ACTIONS or not isinstance(actor_id, str) or not isinstance(reviewer_id, str) or actor_id == reviewer_id: raise PluginLifecycleError("approval_independence_denied") if not isinstance(expires_at, int) or expires_at <= self._now(): raise PluginLifecycleError("approval_expiry_invalid") bound_scope = _scope(scope, principal=False) with self._lock: item = self._version(plugin_uid, version) if ( item["state"] != "reviewed" or item["reviewer_id"] != reviewer_id or item["reviewer_id"] == actor_id ): raise PluginLifecycleError("approval_review_denied") approval = { "approval_id": str(uuid.uuid4()), "plugin_uid": plugin_uid, "version": version, "action": action, "actor_id": actor_id, "reviewer_id": reviewer_id, "scope": bound_scope, "manifest_digest": item["manifest_digest"], "expires_at": expires_at, "consumed": False, } self._approvals[approval["approval_id"]] = approval return copy.deepcopy(approval) def activate(self, plugin_uid: str, version: str, *, actor_id: str, scope: Any, approval_id: str) -> dict[str, Any]: bound_scope = _scope(scope, principal=False) with self._lock: item = self._version(plugin_uid, version) approval = self._approvals.get(approval_id) if not approval or approval["consumed"] or approval["expires_at"] <= self._now() or approval["action"] != "activate" or approval["actor_id"] != actor_id or approval["reviewer_id"] != item["reviewer_id"] or approval["scope"] != bound_scope or approval["manifest_digest"] != item["manifest_digest"]: raise PluginLifecycleError("approval_binding_denied") if item["state"] != "reviewed": raise PluginLifecycleError("activation_transition_denied") if any(candidate["state"] == "active" for (uid, _), candidate in self._versions.items() if uid == plugin_uid): raise PluginLifecycleError("plugin_active_collision") approval["consumed"] = True item["state"] = "active" return copy.deepcopy(item) def invoke(self, *, plugin_uid: str, version: str, operation: str, scope: Any, input_digest: str, idempotency_key: str) -> dict[str, Any]: invocation_scope = _scope(scope, principal=True) digest = _digest(input_digest, "input_digest") if not isinstance(idempotency_key, str) or not idempotency_key.isascii() or not 1 <= len(idempotency_key) <= 120: raise PluginLifecycleError("idempotency_key_invalid") with self._lock: item = self._version(plugin_uid, version) if item["state"] != "active": raise PluginLifecycleError("plugin_not_active") if operation not in item["capabilities"]: raise PluginLifecycleError("capability_denied") key = (plugin_uid, version, invocation_scope["tenant_id"], idempotency_key) replay = self._runs.get(key) request = {"scope": invocation_scope, "input_digest": digest, "operation": operation} request_digest = canonical_digest(request) if replay: if replay["request_digest"] != request_digest: raise PluginLifecycleError("idempotency_conflict") return copy.deepcopy(replay["result"] | {"replay": True}) output_digest = hashlib.sha256( f"ENGINEERING_EVIDENCE_ONLY|{item['manifest_digest']}|{request_digest}".encode("ascii") ).hexdigest() result = { "plugin_uid": plugin_uid, "version": version, "run_digest": request_digest, "output_digest": output_digest, "fixture_id": "ENGINEERING_EVIDENCE_ONLY", "replay": False, } self._runs[key] = {"request_digest": request_digest, "result": result} return copy.deepcopy(result) def _version(self, plugin_uid: str, version: str) -> dict[str, Any]: item = self._versions.get((plugin_uid, version)) if item is None: raise PluginLifecycleError("plugin_version_not_found") return item __all__ = ["PluginLifecycleError", "PluginPlatform"]