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