hms-mediaengine/_entwicklung/packages/plugin_sdk/hms_plugin_sdk/manifest.py

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