DataOps-platform/app/api/system/plugin_platform_routes.py

"""Closed HTTP validation boundary for the inactive WP13 local fixture."""

from __future__ import annotations

from flask import g, jsonify, request
from sqlalchemy.exc import SQLAlchemyError

from app.api.system import bp
from app.core.plugins.governance import PluginManifestError, normalize_plugin_manifest
from app.core.plugins.persistence import (
    DatabasePluginPlatformService,
    PluginPersistenceError,
)
from app.core.plugins.repository import PluginLifecycleError, PluginPlatform
from app.core.system.permissions import AGENTS_MANAGE, require_permissions
from app.models.result import failed, success

_MAX_BODY_BYTES = 8192


def _body(required: set[str]) -> dict:
    if request.content_length is not None and request.content_length > _MAX_BODY_BYTES:
        raise ValueError("plugin_body_closed")
    value = request.get_json(silent=True)
    if not isinstance(value, dict) or set(value) != required:
        raise ValueError("plugin_body_closed")
    return value


def _response(value: object, status: int = 200):
    response = jsonify(success(value, code=status))
    response.headers["Cache-Control"] = "no-store"
    return response, status


def _failure(status: int = 400):
    response = jsonify(failed("plugin request rejected", code=status))
    response.headers["Cache-Control"] = "no-store"
    return response, status


@bp.post("/plugins/local-fixture/validate")
@require_permissions(AGENTS_MANAGE)
def validate_local_plugin_fixture():
    """Validate closed metadata only; it never stores or activates a plugin."""
    try:
        body = _body({"manifest", "registry_record"})
        platform = PluginPlatform(now=lambda: 0)
        result = platform.register(body["manifest"], registry_record=body["registry_record"])
        return _response({"plugin_uid": result["plugin_uid"], "version": result["version"], "manifest_digest": result["manifest_digest"], "mode": "ENGINEERING_EVIDENCE_ONLY", "activation": "blocked"})
    except (PluginManifestError, PluginLifecycleError, TypeError, ValueError):
        return _failure()


@bp.post("/plugins/invocations")
@require_permissions(AGENTS_MANAGE)
def invoke_plugin_fixture():
    """Invoke only an approved fixed local fixture through the DB gateway."""
    try:
        body = _body({"plugin_uid", "version", "operation", "input_digest", "idempotency_key"})
        return _response(DatabasePluginPlatformService().invoke(actor_ref=str(g.current_user["id"]), **body), 201)
    except (PluginPersistenceError, RuntimeError, TypeError, ValueError, SQLAlchemyError):
        return _failure()


@bp.post("/plugins/registry")
@require_permissions(AGENTS_MANAGE)
def register_plugin_fixture():
    try:
        body = _body({"manifest", "registry_record"})
        # Validate executable-surface metadata before allocating a DB gateway;
        # this keeps malformed URL/code/credential bodies out of all backends.
        normalize_plugin_manifest(body["manifest"])
        return _response(DatabasePluginPlatformService().register(actor_ref=str(g.current_user["id"]), **body), 201)
    except (PluginPersistenceError, RuntimeError, TypeError, ValueError, SQLAlchemyError):
        return _failure()


@bp.post("/plugins/<plugin_uid>/<version>/review")
@require_permissions(AGENTS_MANAGE)
def review_plugin_fixture(plugin_uid: str, version: str):
    try:
        _body(set())
        return _response(DatabasePluginPlatformService().review(plugin_uid=plugin_uid, version=version, actor_ref=str(g.current_user["id"])))
    except (PluginPersistenceError, RuntimeError, TypeError, ValueError, SQLAlchemyError):
        return _failure()


@bp.post("/plugins/<plugin_uid>/<version>/approvals")
@require_permissions(AGENTS_MANAGE)
def approve_plugin_fixture(plugin_uid: str, version: str):
    try:
        body = _body({"approval_action"})
        return _response(DatabasePluginPlatformService().issue_approval(plugin_uid=plugin_uid, version=version, actor_ref=str(g.current_user["id"]), **body), 201)
    except (PluginPersistenceError, RuntimeError, TypeError, ValueError, SQLAlchemyError):
        return _failure()


def _transition(plugin_uid: str, version: str, target_state: str):
    try:
        body = _body({"approval_uid", "expected_fence", "incident_uid"})
        return _response(DatabasePluginPlatformService().transition(plugin_uid=plugin_uid, version=version, target_state=target_state, actor_ref=str(g.current_user["id"]), **body))
    except (PluginPersistenceError, RuntimeError, TypeError, ValueError, SQLAlchemyError):
        return _failure()


@bp.post("/plugins/<plugin_uid>/<version>/canary")
@require_permissions(AGENTS_MANAGE)
def canary_plugin_fixture(plugin_uid: str, version: str):
    return _transition(plugin_uid, version, "canary")


@bp.post("/plugins/<plugin_uid>/<version>/activate")
@require_permissions(AGENTS_MANAGE)
def activate_plugin_fixture(plugin_uid: str, version: str):
    return _transition(plugin_uid, version, "active")


@bp.post("/plugins/<plugin_uid>/<version>/pause")
@require_permissions(AGENTS_MANAGE)
def pause_plugin_fixture(plugin_uid: str, version: str):
    return _transition(plugin_uid, version, "paused")


@bp.post("/plugins/<plugin_uid>/<version>/rollback")
@require_permissions(AGENTS_MANAGE)
def rollback_plugin_fixture(plugin_uid: str, version: str):
    return _transition(plugin_uid, version, "rolled_back")


@bp.post("/plugins/<plugin_uid>/<version>/revoke")
@require_permissions(AGENTS_MANAGE)
def revoke_plugin_fixture(plugin_uid: str, version: str):
    return _transition(plugin_uid, version, "revoked")


@bp.post("/plugins/<plugin_uid>/<version>/recover")
@require_permissions(AGENTS_MANAGE)
def recover_plugin_fixture(plugin_uid: str, version: str):
    return _transition(plugin_uid, version, "recovery")