DataOps-platform/app/core/plugins/repository.py

"""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"]