"""Plugin system for ObsiGate — backend API and manifest validation. Plugins extend ObsiGate with custom renderers, search filters, and editor actions. They run sandboxed in a Web Worker on the frontend; the backend handles manifest validation, storage, and lifecycle (install / uninstall / enable / disable), all scoped to a single vault. """ import json import logging import re import zipfile from dataclasses import dataclass, field from pathlib import Path from typing import Any from fastapi import APIRouter, Depends, File, HTTPException, UploadFile from fastapi.responses import JSONResponse from backend.auth.middleware import check_vault_access, require_admin, require_auth from backend.indexer import get_vault_data logger = logging.getLogger("obsigate.plugins") router = APIRouter(prefix="/api/plugins", tags=["plugins"]) # ── Constants ────────────────────────────────────────────────────────────── PLUGINS_DIR_NAME = ".obsigate-plugins" MANIFEST_FILENAME = "plugin.json" MAX_PLUGIN_SIZE = 500_000 # 500 KB per plugin file MAX_PLUGINS_PER_VAULT = 50 MAX_PLUGIN_FILES = 100 # max files in an installed plugin zip ALLOWED_HOOKS = frozenset([ "onFileRender", "onSearchFilter", "onEditorAction", "onSidebarItem", "onFileCreate", "onFileDelete", "onVaultMount", ]) ALLOWED_PERMISSIONS = frozenset([ "read_files", "write_files", "read_vault_metadata", "network_request", "ui_notify", "access_clipboard", ]) # ── Manifest ─────────────────────────────────────────────────────────────── @dataclass class PluginManifest: """Validated plugin manifest from plugin.json.""" name: str version: str description: str author: str main: str hooks: dict[str, str] = field(default_factory=dict) permissions: list[str] = field(default_factory=list) min_obsigate_version: str = "2.1.0" homepage: str | None = None repository: str | None = None license: str = "MIT" @classmethod def from_dict(cls, data: dict) -> "PluginManifest": """Validate and build a manifest from a raw dict (JSON object).""" required = ["name", "version", "description", "author", "main"] for field_name in required: if field_name not in data or not data[field_name]: raise ValueError(f"Missing required field: {field_name}") name = data["name"] if not re.fullmatch(r"[a-z0-9]([a-z0-9-]*[a-z0-9])?", name): raise ValueError( "Plugin name must be lowercase alphanumeric with hyphens (e.g., 'my-plugin')" ) version = data["version"] if not re.fullmatch(r"\d+\.\d+\.\d+(-[a-zA-Z0-9.-]+)?", version): raise ValueError("Version must be semantic version (e.g., '1.0.0')") hooks = data.get("hooks", {}) if not isinstance(hooks, dict): raise TypeError("'hooks' must be an object mapping hook name to handler name") for hook_name, handler in hooks.items(): if hook_name not in ALLOWED_HOOKS: raise ValueError( f"Unknown hook: {hook_name}. Allowed: {sorted(ALLOWED_HOOKS)}" ) if not isinstance(handler, str) or not handler: raise ValueError(f"Hook handler for '{hook_name}' must be a non-empty string") permissions = data.get("permissions", []) if not isinstance(permissions, list): raise TypeError("'permissions' must be a list of permission names") for perm in permissions: if perm not in ALLOWED_PERMISSIONS: raise ValueError( f"Unknown permission: {perm}. Allowed: {sorted(ALLOWED_PERMISSIONS)}" ) return cls( name=name, version=version, description=data["description"], author=data["author"], main=data["main"], hooks=hooks, permissions=permissions, min_obsigate_version=data.get("min_obsigate_version", "2.1.0"), homepage=data.get("homepage"), repository=data.get("repository"), license=data.get("license", "MIT"), ) def to_dict(self) -> dict[str, Any]: return { "name": self.name, "version": self.version, "description": self.description, "author": self.author, "main": self.main, "hooks": self.hooks, "permissions": self.permissions, "min_obsigate_version": self.min_obsigate_version, "homepage": self.homepage, "repository": self.repository, "license": self.license, } # ── Storage path helpers ─────────────────────────────────────────────────── def _vault_plugins_dir(vault_name: str) -> Path: """Return the plugins directory for a vault (creating it if absent).""" vault_data = get_vault_data(vault_name) if not vault_data: raise HTTPException(status_code=404, detail=f"Vault '{vault_name}' not found") base = Path(vault_data["path"]) pd = base / PLUGINS_DIR_NAME pd.mkdir(parents=True, exist_ok=True) return pd # ── Manager ──────────────────────────────────────────────────────────────── class PluginManager: """Manages plugin lifecycle within a single plugins directory.""" def __init__(self, plugins_dir): self.plugins_dir = Path(plugins_dir) self.plugins_dir.mkdir(parents=True, exist_ok=True) def _plugin_dir(self, name: str) -> Path: return self.plugins_dir / name # ---- install / uninstall ---- def install(self, manifest: dict, code: str) -> dict[str, Any]: """Install a plugin from a manifest dict + entry point source string.""" m = PluginManifest.from_dict(manifest) plugin_dir = self._plugin_dir(m.name) if plugin_dir.exists(): raise ValueError(f"Plugin '{m.name}' is already installed") installed = list(self.plugins_dir.iterdir()) if self.plugins_dir.exists() else [] if installed and len(installed) >= MAX_PLUGINS_PER_VAULT: raise ValueError(f"Maximum of {MAX_PLUGINS_PER_VAULT} plugins per vault reached") if len(code.encode("utf-8")) > MAX_PLUGIN_SIZE: raise ValueError(f"Plugin file exceeds {MAX_PLUGIN_SIZE} bytes") plugin_dir.mkdir(parents=True, exist_ok=True) (plugin_dir / MANIFEST_FILENAME).write_text( json.dumps(m.to_dict(), indent=2), encoding="utf-8" ) (plugin_dir / m.main).write_text(code, encoding="utf-8") logger.info("Installed plugin '%s' v%s", m.name, m.version) return { "name": m.name, "version": m.version, "description": m.description, "author": m.author, "main": m.main, "enabled": True, } def uninstall(self, name: str) -> None: plugin_dir = self._plugin_dir(name) if not plugin_dir.exists(): raise ValueError(f"Plugin '{name}' not found") import shutil shutil.rmtree(plugin_dir) logger.info("Uninstalled plugin '%s'", name) # ---- enable / disable ---- def enable(self, name: str) -> None: plugin_dir = self._plugin_dir(name) if not plugin_dir.exists(): raise ValueError(f"Plugin '{name}' not found") marker = plugin_dir / ".disabled" if marker.exists(): marker.unlink() def disable(self, name: str) -> None: plugin_dir = self._plugin_dir(name) if not plugin_dir.exists(): raise ValueError(f"Plugin '{name}' not found") (plugin_dir / ".disabled").write_text("", encoding="utf-8") def is_disabled(self, name: str) -> bool: return (self._plugin_dir(name) / ".disabled").exists() # ---- listing / metadata ---- def list_plugins(self) -> list[dict[str, Any]]: if not self.plugins_dir.exists(): return [] out = [] for d in sorted(self.plugins_dir.iterdir()): if not d.is_dir(): continue manifest_file = d / MANIFEST_FILENAME if not manifest_file.exists(): continue try: m = PluginManifest.from_dict(json.loads(manifest_file.read_text(encoding="utf-8"))) except (json.JSONDecodeError, ValueError) as exc: logger.warning("Invalid plugin dir '%s': %s", d.name, exc) continue out.append({ "name": m.name, "version": m.version, "description": m.description, "author": m.author, "main": m.main, "enabled": not self.is_disabled(m.name), "hooks": m.hooks, "permissions": m.permissions, }) return out def get_plugin(self, name: str) -> dict[str, Any] | None: plugin_dir = self._plugin_dir(name) manifest_file = plugin_dir / MANIFEST_FILENAME if not manifest_file.exists(): return None try: m = PluginManifest.from_dict(json.loads(manifest_file.read_text(encoding="utf-8"))) except (json.JSONDecodeError, ValueError): return None return { "manifest": m.to_dict(), "name": m.name, "version": m.version, "enabled": not self.is_disabled(m.name), } def get_plugin_code(self, name: str, file_name: str) -> str: plugin_dir = self._plugin_dir(name) manifest_file = plugin_dir / MANIFEST_FILENAME if not manifest_file.exists(): raise ValueError(f"Plugin '{name}' not found") target = plugin_dir / file_name # Prevent path traversal try: target.resolve().relative_to(plugin_dir.resolve()) except (ValueError, OSError): raise ValueError(f"Invalid file path: {file_name}") if not target.is_file(): raise ValueError(f"File '{file_name}' not found for plugin '{name}'") return target.read_text(encoding="utf-8") def get_hooks(self, name: str) -> dict[str, str]: plugin = self.get_plugin(name) if not plugin: raise ValueError(f"Plugin '{name}' not found") return plugin["manifest"].get("hooks", {}) class PluginRegistry: """Aggregates enabled plugins for a plugins directory.""" def get_enabled_plugins(self, plugins_dir) -> list[dict[str, Any]]: manager = PluginManager(plugins_dir) return [p for p in manager.list_plugins() if p["enabled"]] def get_plugins_by_hook(self, plugins_dir, hook: str) -> list[dict[str, Any]]: manager = PluginManager(plugins_dir) return [ p for p in manager.list_plugins() if p["enabled"] and hook in p.get("hooks", {}) ] def scan_vault(self, vault_name: str, vault_path) -> list[dict[str, Any]]: """Scan a vault directory for installed plugins (startup discovery). Returns the list of installed-but-valid plugins so the app can log how many are available per vault. Invalid/zombie plugin dirs are skipped gracefully. """ pd = Path(vault_path) / PLUGINS_DIR_NAME if not pd.is_dir(): return [] return PluginManager(pd).list_plugins() # Module-level registry singleton (startup scan + runtime lookups) _registry_singleton: PluginRegistry | None = None def get_plugin_registry() -> PluginRegistry: """Return the shared PluginRegistry instance.""" global _registry_singleton if _registry_singleton is None: _registry_singleton = PluginRegistry() return _registry_singleton # ── Validation of uploads (ZIP / directory) ──────────────────────────────── def _plugin_manifest_from_member(zf: zipfile.ZipFile) -> dict[str, Any]: try: raw = zf.read(MANIFEST_FILENAME).decode("utf-8") except KeyError: raise ValueError("Plugin zip is missing plugin.json") try: data = json.loads(raw) except json.JSONDecodeError: raise ValueError("Invalid JSON in plugin.json") m = PluginManifest.from_dict(data) return m.to_dict() def _validate_plugin_zip(zip_path) -> dict[str, Any]: """Validate an uploaded plugin zip and return its manifest dict.""" with zipfile.ZipFile(zip_path, "r") as zf: members = zf.namelist() if len(members) > MAX_PLUGIN_FILES: raise ValueError(f"Plugin contains too many files (max {MAX_PLUGIN_FILES})") for member in members: if member.endswith("/"): continue norm = member.replace("\\", "/") clean = norm.lstrip("/") if ".." in clean.split("/"): raise ValueError("Path traversal detected in plugin zip") # Reject absolute paths and any ../../ escapes if norm.startswith("/") or "/../" in f"/{norm}" or norm.endswith("/.."): raise ValueError("Path traversal detected in plugin zip") manifest = _plugin_manifest_from_member(zf) # Entry point must exist main = manifest["main"] if main not in members and f"{main}/" not in members: # accept with any leading ./ or subdir normalization found = any(m.rstrip("/") == main or m.rstrip("/") == f"./{main}" for m in members) if not found: raise ValueError(f"Missing entry point: {main}") return manifest def _validate_plugin_directory(plugin_dir) -> dict[str, Any]: """Validate an installed plugin directory and return its manifest dict.""" d = Path(plugin_dir) manifest_file = d / MANIFEST_FILENAME if not manifest_file.is_file(): raise ValueError(f"Missing {MANIFEST_FILENAME}") try: data = json.loads(manifest_file.read_text(encoding="utf-8")) except json.JSONDecodeError: raise ValueError(f"Invalid JSON in {MANIFEST_FILENAME}") m = PluginManifest.from_dict(data) if not (d / m.main).is_file(): raise ValueError(f"Missing entry point: {m.main}") return m.to_dict() # ── API Endpoints ────────────────────────────────────────────────────────── @router.get("", response_model=list[dict]) async def api_list_plugins(vault: str, current_user=Depends(require_auth)): """List installed plugins for a vault.""" if not check_vault_access(vault, current_user): raise HTTPException(status_code=403, detail=f"Accès refusé à la vault '{vault}'") plugins_dir = _vault_plugins_dir(vault) return PluginManager(plugins_dir).list_plugins() @router.get("/template") async def api_plugin_template(): """Return a starter plugin template (manifest + sample code).""" template_manifest = { "name": "my-plugin", "version": "1.0.0", "description": "Describe what your plugin does", "author": "your-name", "main": "index.js", "hooks": {"onFileRender": "render"}, "permissions": ["read_files", "ui_notify"], "license": "MIT", } template_code = ( "// ObsiGate plugin template\n" "export function render(ctx) {\n" " // ctx: { path, content, extension, vault }\n" " if (ctx.content && ctx.path.endsWith('.md')) {\n" " ctx.content = `> Plugin: ${self.name}\\n\\n` + ctx.content;\n" " }\n" " return ctx;\n" "}\n" ) return { "manifest": template_manifest, "code": template_code, } @router.get("/{plugin_name}", response_model=dict) async def api_get_plugin(vault: str, plugin_name: str, current_user=Depends(require_auth)): """Get a single plugin's metadata.""" if not check_vault_access(vault, current_user): raise HTTPException(status_code=403, detail=f"Accès refusé à la vault '{vault}'") plugins_dir = _vault_plugins_dir(vault) plugin = PluginManager(plugins_dir).get_plugin(plugin_name) if not plugin: raise HTTPException(status_code=404, detail=f"Plugin '{plugin_name}' not found") return plugin @router.get("/{plugin_name}/hooks", response_model=dict) async def api_plugin_hooks(vault: str, plugin_name: str, current_user=Depends(require_auth)): """Get the hooks a plugin registers.""" if not check_vault_access(vault, current_user): raise HTTPException(status_code=403, detail=f"Accès refusé à la vault '{vault}'") plugins_dir = _vault_plugins_dir(vault) try: return PluginManager(plugins_dir).get_hooks(plugin_name) except ValueError as exc: raise HTTPException(status_code=404, detail=str(exc)) @router.get("/{plugin_name}/code/{file_name:path}", response_model=dict) async def api_plugin_code( vault: str, plugin_name: str, file_name: str, current_user=Depends(require_auth) ): """Get a plugin source file for sandboxed execution.""" if not check_vault_access(vault, current_user): raise HTTPException(status_code=403, detail=f"Accès refusé à la vault '{vault}'") plugins_dir = _vault_plugins_dir(vault) try: code = PluginManager(plugins_dir).get_plugin_code(plugin_name, file_name) except ValueError as exc: raise HTTPException(status_code=404, detail=str(exc)) return {"name": plugin_name, "file": file_name, "code": code} @router.post("/install") async def api_install_plugin( vault: str, file: UploadFile = File(...), current_user=Depends(require_admin), ): """Install a plugin from an uploaded zip (admin only).""" if not check_vault_access(vault, current_user): raise HTTPException(status_code=403, detail=f"Accès refusé à la vault '{vault}'") content = await file.read() if len(content) > 5_000_000: # 5 MB upload cap raise HTTPException(status_code=413, detail="Plugin zip too large") import io import tempfile with tempfile.TemporaryDirectory() as tmp_dir: zip_path = Path(tmp_dir) / "plugin.zip" try: zip_path.write_bytes(content) manifest = _validate_plugin_zip(zip_path) except ValueError as exc: raise HTTPException(status_code=400, detail=f"Invalid plugin: {exc}") # Extract to a temp dir then validate the entry point exists plugins_dir = _vault_plugins_dir(vault) manager = PluginManager(plugins_dir) name = manifest["name"] with tempfile.TemporaryDirectory() as td: try: with zipfile.ZipFile(io.BytesIO(content)) as zf: zf.extractall(td) except (zipfile.BadZipFile, RuntimeError): raise HTTPException(status_code=400, detail="Invalid plugin zip") extracted = _validate_plugin_directory(Path(td) / name if (Path(td) / name).is_dir() else Path(td)) code_file = manifest["main"] with zipfile.ZipFile(io.BytesIO(content)) as zf: code = zf.read(code_file).decode("utf-8") try: result = manager.install(extracted, code) except ValueError as exc: raise HTTPException(status_code=409, detail=str(exc)) return result @router.delete("/{plugin_name}") async def api_uninstall_plugin(vault: str, plugin_name: str, current_user=Depends(require_admin)): if not check_vault_access(vault, current_user): raise HTTPException(status_code=403, detail=f"Accès refusé à la vault '{vault}'") plugins_dir = _vault_plugins_dir(vault) try: PluginManager(plugins_dir).uninstall(plugin_name) except ValueError as exc: raise HTTPException(status_code=404, detail=str(exc)) return JSONResponse({"ok": True}) @router.post("/{plugin_name}/enable") async def api_enable_plugin(vault: str, plugin_name: str, current_user=Depends(require_admin)): if not check_vault_access(vault, current_user): raise HTTPException(status_code=403, detail=f"Accès refusé à la vault '{vault}'") plugins_dir = _vault_plugins_dir(vault) try: PluginManager(plugins_dir).enable(plugin_name) except ValueError as exc: raise HTTPException(status_code=404, detail=str(exc)) return JSONResponse({"ok": True}) @router.post("/{plugin_name}/disable") async def api_disable_plugin(vault: str, plugin_name: str, current_user=Depends(require_admin)): if not check_vault_access(vault, current_user): raise HTTPException(status_code=403, detail=f"Accès refusé à la vault '{vault}'") plugins_dir = _vault_plugins_dir(vault) try: PluginManager(plugins_dir).disable(plugin_name) except ValueError as exc: raise HTTPException(status_code=404, detail=str(exc)) return JSONResponse({"ok": True}) # ── Plugin Event Bus ──────────────────────────────────────────────────────── # Lightweight pub/sub for backend events that plugins can subscribe to. # Used to dispatch onFileCreate, onFileDelete, onVaultMount to enabled plugins. import time as _time class _PluginEventBus: """In-memory event dispatcher for plugin lifecycle events.""" _listeners: dict[str, list] = {} @classmethod def subscribe(cls, event_type: str, callback) -> None: cls._listeners.setdefault(event_type, []).append(callback) @classmethod async def emit(cls, event_type: str, data: dict) -> None: for cb in cls._listeners.get(event_type, []): try: result = cb(data) if hasattr(result, "__await__"): await result except Exception as e: logger.warning(f"Plugin event '{event_type}' handler error: {e}") plugin_events = _PluginEventBus def emit_file_created(vault: str, path: str, title: str = "") -> None: """Called when a file is created via the API.""" import asyncio try: loop = asyncio.get_running_loop() loop.create_task(plugin_events.emit("onFileCreate", { "vault": vault, "path": path, "title": title, "timestamp": _time.time(), })) except RuntimeError: pass # No event loop running def emit_file_deleted(vault: str, path: str) -> None: """Called when a file is deleted via the API.""" import asyncio try: loop = asyncio.get_running_loop() loop.create_task(plugin_events.emit("onFileDelete", { "vault": vault, "path": path, "timestamp": _time.time(), })) except RuntimeError: pass def emit_vault_mounted(vault: str, path: str) -> None: """Called when a vault is loaded into the index at startup.""" import asyncio try: loop = asyncio.get_running_loop() loop.create_task(plugin_events.emit("onVaultMount", { "vault": vault, "path": path, "timestamp": _time.time(), })) except RuntimeError: pass