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

"""Closed, local-only manifest contract for governed plugins.

This module deliberately contains no loader, dynamic import, evaluator, shell,
subprocess, network client or user-selected filesystem path.  A manifest is
metadata for a fixed built-in fixture handler, never an executable package.
"""

from __future__ import annotations

import copy
import hashlib
import json
import re
import unicodedata
from typing import Any


class PluginManifestError(ValueError):
    """A plugin manifest did not meet the closed local-baseline contract."""


_TOP_LEVEL = frozenset(
    {
        "schema_version",
        "plugin_uid",
        "name",
        "version",
        "api_version",
        "type",
        "capabilities",
        "permissions",
        "resource",
        "compatibility",
        "distribution",
    }
)
_TYPES = frozenset({"connector", "parser", "quality", "notification", "approval", "agent_mcp"})
_CAPABILITIES = {
    "connector": frozenset({"discover", "snapshot", "evidence"}),
    "parser": frozenset({"parse", "evidence"}),
    "quality": frozenset({"evaluate", "evidence"}),
    "notification": frozenset({"prepare", "evidence"}),
    "approval": frozenset({"evaluate", "evidence"}),
    "agent_mcp": frozenset({"read", "suggest", "evidence"}),
}
_PERMISSIONS = frozenset({"network", "file", "secret", "child_process"})
_RESOURCE = frozenset({"timeout_ms", "max_output_bytes", "max_concurrency", "max_retries"})
_COMPATIBILITY = frozenset({"platform_api"})
_DISTRIBUTION = frozenset({"kind", "fixture_id", "artifact_digest"})
_SENSITIVE = frozenset(
    {
        "url", "uri", "endpoint", "path", "file", "file_path", "filename", "command", "shell",
        "script", "code", "module", "import", "eval", "exec", "secret", "token", "password",
        "credential", "api_key", "authorization", "raw_rows", "rows", "records", "payload", "sql",
    }
)
_ID = re.compile(r"^[a-z][a-z0-9-]{2,62}$")
_SEMVER = re.compile(r"^[0-9]+\.[0-9]+\.[0-9]+(?:-[0-9A-Za-z.-]+)?$")
_HEX = re.compile(r"^[a-f0-9]{64}$")


def canonical_digest(value: dict[str, Any]) -> str:
    return hashlib.sha256(
        json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8")
    ).hexdigest()


def fixture_signature_digest(artifact_digest: str) -> str:
    """Bind the sole local fixture artifact to the platform-owned trust key.

    This is deliberately not a payload supplied public key or a general
    signature verifier.  Enterprise CA and registry signing remain disabled.
    """

    if not isinstance(artifact_digest, str) or not _HEX.fullmatch(artifact_digest):
        raise PluginManifestError("artifact_digest_invalid")
    return hashlib.sha256(f"local-fixture-key-v1|{artifact_digest}".encode("ascii")).hexdigest()


def _closed(value: Any, allowed: frozenset[str], label: str) -> dict[str, Any]:
    if not isinstance(value, dict) or len(value) > len(allowed):
        raise PluginManifestError(f"{label}_closed")
    for key in value:
        if not isinstance(key, str) or unicodedata.normalize("NFKC", key) != key:
            raise PluginManifestError("manifest_unicode_rejected")
        # The four declared capability switches are metadata, not values.  All
        # other sensitive vocabulary remains forbidden even before unknown-key
        # processing so it cannot become an accidental extension point.
        if key.casefold() in _SENSITIVE and key not in allowed:
            raise PluginManifestError("manifest_sensitive_key")
    if set(value) != allowed:
        raise PluginManifestError(f"{label}_closed")
    return copy.deepcopy(value)


def _ascii_text(value: Any, label: str, maximum: int = 120) -> str:
    if not isinstance(value, str) or not value or len(value) > maximum:
        raise PluginManifestError(f"{label}_invalid")
    if unicodedata.normalize("NFKC", value) != value or not value.isascii():
        raise PluginManifestError(f"{label}_unicode_rejected")
    return value


def _positive_int(value: Any, label: str, maximum: int) -> int:
    if isinstance(value, bool) or not isinstance(value, int) or not 0 < value <= maximum:
        raise PluginManifestError(f"{label}_invalid")
    return value


def normalize_plugin_manifest(value: Any) -> dict[str, Any]:
    """Normalize a v1 manifest for the sole fixed engineering fixture.

    Enterprise registry artifacts are intentionally rejected until an external
    trust model is approved.  This validation makes that inactivity executable.
    """

    body = _closed(value, _TOP_LEVEL, "manifest")
    if body["schema_version"] != 1 or body["api_version"] != "1":
        raise PluginManifestError("manifest_version_invalid")
    plugin_uid = _ascii_text(body["plugin_uid"], "plugin_uid", 63)
    name = _ascii_text(body["name"], "name", 63)
    if not _ID.fullmatch(plugin_uid) or not _ID.fullmatch(name) or plugin_uid != name:
        raise PluginManifestError("plugin_identity_invalid")
    version = _ascii_text(body["version"], "version", 80)
    if not _SEMVER.fullmatch(version):
        raise PluginManifestError("plugin_version_invalid")
    plugin_type = _ascii_text(body["type"], "plugin_type", 20)
    if plugin_type not in _TYPES:
        raise PluginManifestError("plugin_type_invalid")
    capabilities = body["capabilities"]
    if not isinstance(capabilities, list) or not 1 <= len(capabilities) <= 8:
        raise PluginManifestError("capabilities_invalid")
    normalized_capabilities = tuple(_ascii_text(item, "capability", 40) for item in capabilities)
    if len(set(normalized_capabilities)) != len(normalized_capabilities) or set(normalized_capabilities) - _CAPABILITIES[plugin_type]:
        raise PluginManifestError("capabilities_invalid")
    permissions = _closed(body["permissions"], _PERMISSIONS, "permissions")
    if any(item is not False for item in permissions.values()):
        raise PluginManifestError("permissions_denied")
    resource = _closed(body["resource"], _RESOURCE, "resource")
    normalized_resource = {
        "timeout_ms": _positive_int(resource["timeout_ms"], "timeout_ms", 5000),
        "max_output_bytes": _positive_int(resource["max_output_bytes"], "max_output_bytes", 65536),
        "max_concurrency": _positive_int(resource["max_concurrency"], "max_concurrency", 1),
        "max_retries": _positive_int(resource["max_retries"], "max_retries", 3),
    }
    if normalized_resource["max_output_bytes"] < 64:
        raise PluginManifestError("max_output_bytes_invalid")
    compatibility = _closed(body["compatibility"], _COMPATIBILITY, "compatibility")
    if compatibility["platform_api"] != "1":
        raise PluginManifestError("compatibility_invalid")
    distribution = _closed(body["distribution"], _DISTRIBUTION, "distribution")
    if distribution["kind"] != "builtin_fixture" or distribution["fixture_id"] != "ENGINEERING_EVIDENCE_ONLY":
        raise PluginManifestError("distribution_denied")
    artifact_digest = _ascii_text(distribution["artifact_digest"], "artifact_digest", 64)
    if not _HEX.fullmatch(artifact_digest):
        raise PluginManifestError("artifact_digest_invalid")
    normalized = {
        "schema_version": 1,
        "plugin_uid": plugin_uid,
        "name": name,
        "version": version,
        "api_version": "1",
        "type": plugin_type,
        "capabilities": list(normalized_capabilities),
        "permissions": dict.fromkeys(sorted(_PERMISSIONS), False),
        "resource": normalized_resource,
        "compatibility": {"platform_api": "1"},
        "distribution": {
            "kind": "builtin_fixture",
            "fixture_id": "ENGINEERING_EVIDENCE_ONLY",
            "artifact_digest": artifact_digest,
        },
    }
    normalized["manifest_digest"] = canonical_digest(normalized)
    return normalized


__all__ = ["PluginManifestError", "canonical_digest", "fixture_signature_digest", "normalize_plugin_manifest"]