"""Plugin-Manifest und Validierung (PLAN.md §14.2–14.6, §27.2). Sicherheitsgrenzen: - Pfadsicherheit: keine absoluten Pfade, kein '..' in Manifest und ZIP - ZIP-Bomb-Limits, Dateigrößenlimits, erlaubte Dateitypen - eindeutige Plugin-ID (reverse-dns), SemVer, api_version - Shader-Dateien müssen je deklariertem Backend existieren - max. 8 generische DMX-Slots je Effektinstanz (§14.7) """ from __future__ import annotations import json import zipfile from enum import StrEnum from pathlib import Path, PurePosixPath from typing import Any MANIFEST_SCHEMA_VERSION = 1 MAX_PLUGIN_FILES = 512 MAX_TOTAL_UNPACKED = 32 * 1024 * 1024 MAX_FILE_SIZE = 8 * 1024 * 1024 _ALLOWED_SUFFIXES = { ".json", ".hlsl", ".frag", ".vert", ".glsl", ".png", ".md", ".txt", ".toml", ".csv", } _ALLOWED_BACKENDS = {"d3d11", "gl", "gles"} class PluginKind(StrEnum): SOURCE = "source" GENERATOR = "generator" FILTER = "filter" TRANSITION = "transition" MIXER = "mixer" OUTPUT = "output" CONTROL = "control" AUTOMATION = "automation" def _safe_relative(raw: str) -> PurePosixPath | None: """Prüft Pfadsicherheit; None wenn unsicher (absolut oder Traversal).""" if not raw: return None p = PurePosixPath(raw) if p.is_absolute() or ".." in p.parts: return None return p def _validate_parameters(params: list[dict[str, Any]]) -> list[str]: errors: list[str] = [] seen: set[str] = set() total_dmx_slots = 0 for param in params: pid = param.get("id") if not pid or not isinstance(pid, str): errors.append("parameter without id") continue if pid in seen: errors.append(f"duplicate parameter id: {pid}") seen.add(pid) ptype = param.get("type") if ptype not in {"float", "int", "enum", "bool", "color"}: errors.append(f"parameter {pid}: invalid type {ptype!r}") if ptype == "float": for key in ("minimum", "maximum", "default"): if key not in param: errors.append(f"parameter {pid}: missing {key}") slots = param.get("dmx_slots", []) if not isinstance(slots, list) or any(not isinstance(s, int) for s in slots): errors.append(f"parameter {pid}: dmx_slots must be int list") slots = [] total_dmx_slots += len(slots) if total_dmx_slots > 8: errors.append(f"dmx slot footprint {total_dmx_slots} exceeds 8 (§14.7)") return errors def _valid_plugin_id(pid: str) -> bool: if ".." in pid or len(pid) < 5: return False parts = pid.split(".") if len(parts) < 2: return False allowed = set("abcdefghijklmnopqrstuvwxyz0123456789._-") return all(c in allowed for c in pid) def _valid_semver(version: str) -> bool: parts = version.split(".") if len(parts) != 3: return False try: for p in parts: int(p) except ValueError: return False return True def validate_manifest( manifest: dict[str, Any], plugin_root: Path | None = None ) -> list[str]: """Validiert ein geparstes Manifest; leere Fehlerliste = gültig. plugin_root: wenn gesetzt, werden deklarierte Shader auf Existenz geprüft. """ errors: list[str] = [] if manifest.get("schema_version") != MANIFEST_SCHEMA_VERSION: errors.append(f"schema_version must be {MANIFEST_SCHEMA_VERSION}") pid = manifest.get("id", "") if not isinstance(pid, str) or not _valid_plugin_id(pid): errors.append(f"invalid plugin id: {pid!r} (expected reverse-dns)") for key in ("name", "version", "vendor"): value = manifest.get(key) if not isinstance(value, str) or not value: errors.append(f"missing or empty {key}") if not _valid_semver(manifest.get("version", "")): errors.append("version must be semantic (X.Y.Z)") if manifest.get("api_version") != MANIFEST_SCHEMA_VERSION: errors.append(f"api_version must be {MANIFEST_SCHEMA_VERSION}") if manifest.get("kind") not in {k.value for k in PluginKind}: errors.append(f"invalid kind: {manifest.get('kind')!r}") entrypoints = manifest.get("entrypoints", {}) if not isinstance(entrypoints, dict) or not entrypoints: errors.append("entrypoints required") else: supported = set(manifest.get("capabilities", {}).get("supported_backends", [])) unknown = supported - _ALLOWED_BACKENDS if unknown: errors.append(f"unsupported backends: {sorted(unknown)}") for backend, entry in entrypoints.items(): if backend not in _ALLOWED_BACKENDS: errors.append(f"entrypoint backend {backend!r} not allowed") continue if backend in supported: passes = entry.get("passes", []) if not passes: errors.append(f"entrypoint {backend}: no passes") for pas in passes: shader_key = "pixel_shader" if "pixel_shader" in pas else "fragment" shader_rel = pas.get(shader_key) if not shader_rel: errors.append(f"entrypoint {backend}: pass without shader") continue sp = _safe_relative(shader_rel) if sp is None: errors.append(f"unsafe shader path: {shader_rel!r}") continue if plugin_root is not None and not (plugin_root / sp).is_file(): errors.append(f"missing shader file: {shader_rel}") params = manifest.get("parameters", []) if not isinstance(params, list): errors.append("parameters must be a list") else: errors.extend(_validate_parameters(params)) if manifest.get("failure_mode") not in {"bypass", "hold", "black"}: errors.append("failure_mode must be bypass|hold|black") return errors def validate_plugin_zip(zip_path: Path) -> list[str]: """Prüft ein Plugin-ZIP: Pfadsicherheit, Limits, Typen, Manifest (§27.2).""" errors: list[str] = [] try: with zipfile.ZipFile(zip_path) as zf: names = zf.namelist() if len(names) > MAX_PLUGIN_FILES: errors.append(f"too many files: {len(names)} > {MAX_PLUGIN_FILES}") total = 0 for info in zf.infolist(): if info.is_dir(): continue total += info.file_size if info.file_size > MAX_FILE_SIZE: errors.append(f"file too large: {info.filename}") if _safe_relative(info.filename) is None: errors.append(f"unsafe path in zip: {info.filename!r}") if Path(info.filename).suffix.lower() not in _ALLOWED_SUFFIXES: errors.append(f"disallowed file type: {info.filename}") if total > MAX_TOTAL_UNPACKED: errors.append(f"zip too large unpacked: {total} > {MAX_TOTAL_UNPACKED}") manifest_name = next( (n for n in names if n.endswith("plugin.json") and n.count("/") == 1), None, ) if manifest_name is None: errors.append("plugin.json not found at package root") else: manifest = json.loads(zf.read(manifest_name)) errors.extend(validate_manifest(manifest)) except zipfile.BadZipFile: errors.append("not a valid zip file") except json.JSONDecodeError as exc: errors.append(f"plugin.json invalid JSON: {exc}") return errors def load_manifest(plugin_dir: Path) -> tuple[dict[str, Any], list[str]]: """Lädt und validiert plugin.json aus einem Plugin-Verzeichnis.""" manifest_path = plugin_dir / "plugin.json" if not manifest_path.is_file(): return {}, ["plugin.json missing"] try: manifest = json.loads(manifest_path.read_text(encoding="utf-8")) except json.JSONDecodeError as exc: return {}, [f"plugin.json invalid JSON: {exc}"] return manifest, validate_manifest(manifest, plugin_root=plugin_dir)