diff --git a/backend/app/api/__init__.py b/backend/app/api/__init__.py index ef003a1..30490ff 100644 --- a/backend/app/api/__init__.py +++ b/backend/app/api/__init__.py @@ -16,6 +16,7 @@ def register_routers(app: FastAPI) -> None: from app.api.compose import router as compose_router from app.api.mailboxes import router as mailboxes_router from app.api.messages import router as messages_router + from app.api.plugins import router as plugins_router from app.api.recipients import router as recipients_router from app.api.saved_searches import router as saved_searches_router from app.api.search import router as search_router @@ -36,6 +37,13 @@ def register_routers(app: FastAPI) -> None: settings_router, mailboxes_router, saved_searches_router, + plugins_router, ): app.include_router(router, prefix="/api") app.include_router(ws_router, prefix="/api") + + # Mount discovered plugin routers at /api/plugins/{name} (SPEC §11). Hooks are + # subscribed in the lifespan startup so they register once per process. + from app.plugins import loader + + loader.mount_routers(app) diff --git a/backend/app/api/plugins.py b/backend/app/api/plugins.py new file mode 100644 index 0000000..1c57262 --- /dev/null +++ b/backend/app/api/plugins.py @@ -0,0 +1,23 @@ +"""Plugin platform HTTP surface (SPEC §11). + +Currently just the aggregated UI manifest the frontend loader fetches on boot. +Individual plugin routers are mounted separately at ``/api/plugins/{name}`` by +:func:`app.plugins.loader.mount_routers`. +""" + +from __future__ import annotations + +from typing import Any + +from fastapi import APIRouter + +from app.identity.deps import CurrentUser +from app.plugins import loader + +router = APIRouter(prefix="/plugins", tags=["plugins"]) + + +@router.get("/ui-manifest") +async def ui_manifest(user: CurrentUser) -> dict[str, Any]: + """Slots, pages, and capabilities declared by all installed plugins (SPEC §11.3).""" + return loader.build_ui_manifest() diff --git a/backend/app/main.py b/backend/app/main.py index d998e9f..6ff12b9 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -18,6 +18,12 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: """Warm shared resources on startup, dispose them on shutdown.""" # Touch settings early so misconfiguration fails fast. get_settings() + # Subscribe plugin hooks so events published in the API process reach them + # (worker-published events are handled by the worker's own registration). + from app.events import get_event_bus + from app.plugins import loader + + loader.register_hooks(get_event_bus()) yield await close_redis() await dispose_engine() diff --git a/backend/app/models.py b/backend/app/models.py index 44c3517..68cd7f9 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -8,6 +8,7 @@ from __future__ import annotations import datetime as dt import uuid +from typing import Any from sqlalchemy import ( BigInteger, @@ -288,6 +289,23 @@ class SavedSearch(Base): pinned: Mapped[bool] = mapped_column(nullable=False, server_default=text("true")) +class PluginKv(Base): + """Namespaced key/value store for plugins (SPEC §11 ``plugin_kv``). + + Core-owned infrastructure (not a ``plg_``-prefixed plugin table): every plugin + gets a slice keyed by (plugin, user_id, key). ``value`` is arbitrary JSON. + Invariant I-7 (user-data rows carry ``user_id``) holds here too.""" + + __tablename__ = "plugin_kv" + + plugin: Mapped[str] = mapped_column(Text, primary_key=True) + user_id: Mapped[uuid.UUID] = mapped_column( + UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), primary_key=True + ) + key: Mapped[str] = mapped_column(Text, primary_key=True) + value: Mapped[Any] = mapped_column(JSONB, nullable=False) + + class KnownRecipient(Base): __tablename__ = "known_recipients" __table_args__ = ( diff --git a/backend/app/plugins/__init__.py b/backend/app/plugins/__init__.py new file mode 100644 index 0000000..8a8f9ae --- /dev/null +++ b/backend/app/plugins/__init__.py @@ -0,0 +1,6 @@ +"""Plugin platform (SPEC §11). + +Public surface for plugin authors is :mod:`app.plugins.sdk`; the loader and +core wiring live here but plugins must not import them directly (SPEC §11: +"plugins may not import from ``app.*`` internals other than ``app.plugins.sdk``"). +""" diff --git a/backend/app/plugins/alembic/env.py b/backend/app/plugins/alembic/env.py new file mode 100644 index 0000000..f9d9ec4 --- /dev/null +++ b/backend/app/plugins/alembic/env.py @@ -0,0 +1,39 @@ +"""Shared Alembic environment for plugin migration branches (SPEC §11.1). + +Every plugin reuses this env; the runner (:mod:`app.plugins.migrate`) points +``version_locations`` at the plugin's own revision directory and gives it a +dedicated ``version_table`` (``alembic_version_plg_{name}``) so each plugin's +migration history is independent of core and of every other plugin. +""" + +from __future__ import annotations + +import asyncio + +from alembic import context +from sqlalchemy.ext.asyncio import async_engine_from_config +from sqlalchemy.pool import NullPool + +from app.config import get_settings + +config = context.config +config.set_main_option("sqlalchemy.url", get_settings().database_url) +_version_table = config.get_main_option("version_table") or "alembic_version" + + +def _do_run_migrations(connection: object) -> None: + context.configure(connection=connection, version_table=_version_table) # type: ignore[arg-type] + with context.begin_transaction(): + context.run_migrations() + + +async def _run_online() -> None: + engine = async_engine_from_config( + config.get_section(config.config_ini_section, {}), prefix="sqlalchemy.", poolclass=NullPool + ) + async with engine.connect() as connection: + await connection.run_sync(_do_run_migrations) + await engine.dispose() + + +asyncio.run(_run_online()) diff --git a/backend/app/plugins/alembic/script.py.mako b/backend/app/plugins/alembic/script.py.mako new file mode 100644 index 0000000..063ed2a --- /dev/null +++ b/backend/app/plugins/alembic/script.py.mako @@ -0,0 +1,26 @@ +"""${message} + +Revision ID: ${up_revision} +Revises: ${down_revision | comma,n} +Create Date: ${create_date} +""" +from __future__ import annotations + +from collections.abc import Sequence + +from alembic import op +import sqlalchemy as sa +${imports if imports else ""} + +revision: str = ${repr(up_revision)} +down_revision: str | None = ${repr(down_revision)} +branch_labels: str | Sequence[str] | None = ${repr(branch_labels)} +depends_on: str | Sequence[str] | None = ${repr(depends_on)} + + +def upgrade() -> None: + ${upgrades if upgrades else "pass"} + + +def downgrade() -> None: + ${downgrades if downgrades else "pass"} diff --git a/backend/app/plugins/cli.py b/backend/app/plugins/cli.py new file mode 100644 index 0000000..91d9958 --- /dev/null +++ b/backend/app/plugins/cli.py @@ -0,0 +1,58 @@ +"""Plugin management CLI (SPEC §11.1). + + python -m app.plugins.cli list + python -m app.plugins.cli install + python -m app.plugins.cli uninstall [--drop-data] + +Builtin (core) plugins refuse uninstall. +""" + +from __future__ import annotations + +import argparse + +from app.plugins import loader, migrate + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + prog="python -m app.plugins.cli", description="Manage webmail plugins" + ) + sub = parser.add_subparsers(dest="cmd", required=True) + sub.add_parser("list", help="list discovered plugins") + p_install = sub.add_parser("install", help="apply a plugin's migrations") + p_install.add_argument("name") + p_uninstall = sub.add_parser("uninstall", help="remove a plugin (refuses builtin)") + p_uninstall.add_argument("name") + p_uninstall.add_argument( + "--drop-data", action="store_true", help="also drop the plugin's tables" + ) + args = parser.parse_args(argv) + + if args.cmd == "list": + for plugin in loader.get_plugins(): + kind = "builtin" if plugin.builtin else "removable" + migr = "migrations" if plugin.migrations else "no-migrations" + print(f"{plugin.name}\t{plugin.version}\t{kind}\t{migr}") + return 0 + + if args.cmd == "install": + migrate.install(args.name) + print(f"installed {args.name}") + return 0 + + if args.cmd == "uninstall": + try: + migrate.uninstall(args.name, drop_data=args.drop_data) + except loader.PluginRemovalError as exc: + print(f"error: {exc}") + return 1 + suffix = " (data dropped)" if args.drop_data else " (data kept)" + print(f"uninstalled {args.name}{suffix}") + return 0 + + return 1 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/backend/app/plugins/db.py b/backend/app/plugins/db.py new file mode 100644 index 0000000..ece5ac9 --- /dev/null +++ b/backend/app/plugins/db.py @@ -0,0 +1,44 @@ +"""RLS-enforced database access for plugins (SPEC §11.1 + privacy hardening). + +``plugin_session(user_id)`` yields a normal SQLAlchemy session whose transaction +has been dropped to the least-privilege :data:`~app.plugins.rls.PLUGIN_ROLE` and +bound to the per-user RLS context. Both are set with transaction-local scope +(``SET LOCAL`` / ``set_config(..., is_local=true)``), so they auto-revert on +commit/rollback and cannot leak across pooled connections. + +Because the role is non-superuser / non-owner / NOBYPASSRLS, the row-level policy +from :mod:`app.plugins.rls` is actually enforced: a plugin query that forgets to +filter ``user_id`` still returns only the current user's rows, and it cannot write +rows for anyone else (policy ``WITH CHECK``). +""" + +from __future__ import annotations + +import uuid +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager + +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db import get_sessionmaker +from app.plugins.rls import PLUGIN_ROLE + + +@asynccontextmanager +async def plugin_session(user_id: uuid.UUID) -> AsyncIterator[AsyncSession]: + """A transactional session scoped to ``user_id`` under RLS (SPEC §11.1).""" + async with get_sessionmaker()() as session: + try: + # Bind the RLS user context first (while still the privileged role), then + # drop into the restricted role for the rest of the transaction. + await session.execute( + text("SELECT set_config('app.current_user_id', :uid, true)"), + {"uid": str(user_id)}, + ) + await session.execute(text(f"SET LOCAL ROLE {PLUGIN_ROLE}")) + yield session + await session.commit() + except Exception: + await session.rollback() + raise diff --git a/backend/app/plugins/jobs.py b/backend/app/plugins/jobs.py new file mode 100644 index 0000000..6d52874 --- /dev/null +++ b/backend/app/plugins/jobs.py @@ -0,0 +1,130 @@ +"""Plugin background jobs on the shared arq worker (SPEC §11.2). + +Each plugin ``JobSpec`` is registered as an arq function under a namespaced name +``plg.{plugin}.{func}`` (enqueue on demand from hooks/routes via ``sdk.enqueue``), +and, when it has a ``schedule``, also as an arq cron job. ``singleton=True`` wraps +the job in a Redis NX lock keyed by (plugin, func, user) so at most one instance +runs per user (or globally for scheduleds with no user). Jobs share the worker +pool + retry semantics with core jobs and log under the ``plg`` logger. +""" + +from __future__ import annotations + +import logging +from typing import Any + +from arq import cron +from arq.cron import CronJob +from arq.worker import Function, func + +from app.plugins import loader +from app.plugins.types import JobSpec, Plugin +from app.redis_client import get_redis + +logger = logging.getLogger("plg") + +# Safety expiry on the singleton lock so a crashed job can't wedge a user forever. +# A job that runs longer than this could overlap; acceptable for M3. +_LOCK_TTL = 3600 + +# Standard 5-field cron order and each field's inclusive value bounds. +_FIELD_ORDER = ("minute", "hour", "day", "month", "weekday") +_FIELD_BOUNDS = { + "minute": (0, 59), + "hour": (0, 23), + "day": (1, 31), + "month": (1, 12), + "weekday": (0, 6), +} + + +def job_name(plugin_name: str, spec: JobSpec) -> str: + """The arq function name a plugin job is registered under.""" + return f"plg.{plugin_name}.{spec.func.__name__}" + + +def _lock_key(plugin_name: str, func_name: str, user: object) -> str: + return f"plglock:{plugin_name}:{func_name}:{user}" + + +def _make_runner(plugin: Plugin, spec: JobSpec) -> tuple[Any, str]: + """Wrap a job func with the singleton lock (when requested) + namespaced name.""" + func_name = spec.func.__name__ + name = job_name(plugin.name, spec) + + async def runner(ctx: dict[str, Any], *args: Any, **kwargs: Any) -> Any: + if not spec.singleton: + return await spec.func(ctx, *args, **kwargs) + user = kwargs.get("user_id", "global") + key = _lock_key(plugin.name, func_name, user) + acquired = await get_redis().set(key, b"1", nx=True, ex=_LOCK_TTL) + if not acquired: + logger.info("%s skipped: singleton already running (user=%s)", name, user) + return None + try: + return await spec.func(ctx, *args, **kwargs) + finally: + await get_redis().delete(key) + + return runner, name + + +def plugin_arq_functions() -> list[Function]: + """Every plugin job as an arq function (enqueue-on-demand), namespaced.""" + funcs: list[Function] = [] + for plugin in loader.get_plugins(): + for spec in plugin.jobs: + runner, name = _make_runner(plugin, spec) + funcs.append(func(runner, name=name)) + return funcs + + +def _parse_field(token: str, lo: int, hi: int) -> set[int] | None: + """Parse one cron field to a value set (``*`` -> None = every).""" + if token == "*": + return None + values: set[int] = set() + for part in token.split(","): + if part.startswith("*/"): + values.update(range(lo, hi + 1, int(part[2:]))) + elif "-" in part: + start, end = part.split("-", 1) + values.update(range(int(start), int(end) + 1)) + else: + values.add(int(part)) + return values + + +def parse_cron(expr: str) -> dict[str, set[int] | None]: + """Parse a standard 5-field cron string into arq ``cron`` field kwargs.""" + parts = expr.split() + if len(parts) != 5: + raise ValueError(f"cron expression must have 5 fields (min hour dom mon dow): {expr!r}") + return { + name: _parse_field(token, *_FIELD_BOUNDS[name]) + for name, token in zip(_FIELD_ORDER, parts, strict=True) + } + + +def plugin_cron_jobs() -> list[CronJob]: + """Cron-scheduled plugin jobs as arq ``CronJob``s (run on the shared worker).""" + jobs: list[CronJob] = [] + for plugin in loader.get_plugins(): + for spec in plugin.jobs: + if spec.schedule is None: + continue + runner, name = _make_runner(plugin, spec) + f = parse_cron(spec.schedule) + jobs.append( + cron( + runner, + name=f"{name}.cron", + second=0, + minute=f["minute"], + hour=f["hour"], + day=f["day"], + month=f["month"], + weekday=f["weekday"], + ) + ) + return jobs diff --git a/backend/app/plugins/kv.py b/backend/app/plugins/kv.py new file mode 100644 index 0000000..fc8bc15 --- /dev/null +++ b/backend/app/plugins/kv.py @@ -0,0 +1,63 @@ +"""Namespaced key/value store handed to plugins (SPEC §11 ``plugin_kv``). + +A :class:`PluginKv` is bound to one (plugin, user_id) slice of the core +``plugin_kv`` table; values are arbitrary JSON. This is the light-state store +every plugin gets — heavy plugins use their own ``plg_``-prefixed tables (§11.1). +""" + +from __future__ import annotations + +import uuid +from typing import Any + +from sqlalchemy import delete, select +from sqlalchemy.dialects.postgresql import insert as pg_insert + +from app.db import session_scope +from app.models import PluginKv + + +class PluginKvStore: + """CRUD over one plugin's per-user KV slice. Instantiate via ``sdk.kv(user_id)``.""" + + def __init__(self, plugin: str, user_id: uuid.UUID) -> None: + self._plugin = plugin + self._user_id = user_id + + async def get(self, key: str, default: Any = None) -> Any: + async with session_scope() as session: + row = await session.get(PluginKv, (self._plugin, self._user_id, key)) + return row.value if row is not None else default + + async def set(self, key: str, value: Any) -> None: + """Upsert ``key`` -> ``value`` (JSON-serialisable).""" + stmt = ( + pg_insert(PluginKv) + .values(plugin=self._plugin, user_id=self._user_id, key=key, value=value) + .on_conflict_do_update( + index_elements=[PluginKv.plugin, PluginKv.user_id, PluginKv.key], + set_={"value": value}, + ) + ) + async with session_scope() as session: + await session.execute(stmt) + + async def delete(self, key: str) -> None: + async with session_scope() as session: + await session.execute( + delete(PluginKv).where( + PluginKv.plugin == self._plugin, + PluginKv.user_id == self._user_id, + PluginKv.key == key, + ) + ) + + async def items(self) -> dict[str, Any]: + """Return every key/value in this plugin+user slice.""" + async with session_scope() as session: + rows = await session.execute( + select(PluginKv.key, PluginKv.value).where( + PluginKv.plugin == self._plugin, PluginKv.user_id == self._user_id + ) + ) + return dict(rows.all()) # type: ignore[arg-type] diff --git a/backend/app/plugins/loader.py b/backend/app/plugins/loader.py new file mode 100644 index 0000000..526c026 --- /dev/null +++ b/backend/app/plugins/loader.py @@ -0,0 +1,148 @@ +"""Plugin discovery + core wiring (SPEC §11). + +Discovers plugins via the ``webmail.plugins`` entry-point group, then: +- mounts each plugin router at ``/api/plugins/{name}`` (API process), +- forwards §8.3 events to plugin hooks (both API and worker processes), +- aggregates UI manifests for ``GET /api/plugins/ui-manifest``. + +Plugin background jobs (§11.2) and migrations (§11.1) are wired in later M3 phases. +""" + +from __future__ import annotations + +import logging +from importlib.metadata import entry_points +from typing import TYPE_CHECKING, Any + +from app.events import EventBus +from app.plugins.types import Page, Plugin, Slot, UiManifest + +if TYPE_CHECKING: + from fastapi import FastAPI + +logger = logging.getLogger(__name__) + +ENTRYPOINT_GROUP = "webmail.plugins" + +_plugins: list[Plugin] | None = None +# Plugin names whose hooks are already subscribed on a given bus, so repeated +# create_app()/worker starts in one process don't double-register (tests, reload). +_hooks_registered: set[tuple[int, str]] = set() + + +def _coerce_plugin(obj: Any) -> Plugin: + """An entry point may resolve to a ``Plugin`` or a module exposing ``plugin``.""" + if isinstance(obj, Plugin): + return obj + candidate = getattr(obj, "plugin", None) + if isinstance(candidate, Plugin): + return candidate + raise TypeError(f"entry point {obj!r} does not expose a Plugin (SPEC §11)") + + +def discover_plugins() -> list[Plugin]: + """Load every plugin registered under the ``webmail.plugins`` group.""" + found: list[Plugin] = [] + seen: set[str] = set() + for ep in entry_points(group=ENTRYPOINT_GROUP): + try: + plugin = _coerce_plugin(ep.load()) + except Exception: # noqa: BLE001 - one bad plugin must not break the app + logger.exception("failed to load plugin entry point %s", ep.name) + continue + if plugin.name in seen: + logger.error("duplicate plugin name %r; skipping", plugin.name) + continue + seen.add(plugin.name) + found.append(plugin) + logger.info("loaded plugin %s v%s", plugin.name, plugin.version) + return found + + +def get_plugins() -> list[Plugin]: + """Return the process-wide discovered plugin list (cached).""" + global _plugins + if _plugins is None: + _plugins = discover_plugins() + return _plugins + + +def set_plugins_for_test(plugins: list[Plugin] | None) -> None: + """Override (or reset with ``None``) the plugin registry — tests only.""" + global _plugins + _plugins = plugins + _hooks_registered.clear() + + +def register_hooks(bus: EventBus) -> None: + """Subscribe each plugin's hooks to the bus (idempotent per process/bus). + + Called in both the API and the worker so hooks fire wherever the event is + published (e.g. ``message.new`` originates in the worker).""" + for plugin in get_plugins(): + marker = (id(bus), plugin.name) + if marker in _hooks_registered: + continue + for event, handler in plugin.hooks.items(): + bus.subscribe(event, handler) + _hooks_registered.add(marker) + + +def mount_routers(app: FastAPI) -> None: + """Mount each plugin's router at ``/api/plugins/{name}`` (SPEC §11).""" + for plugin in get_plugins(): + if plugin.router is not None: + app.include_router(plugin.router, prefix=f"/api/plugins/{plugin.name}") + + +def _slot_dict(slot: Slot) -> dict[str, str]: + return {"name": slot.name, "bundle": slot.bundle} + + +def _page_dict(page: Page) -> dict[str, Any]: + return { + "route": page.route, + "bundle": page.bundle, + "nav": {"icon": page.nav.icon, "label": page.nav.label, "order": page.nav.order}, + } + + +def _manifest_dict(plugin: Plugin, manifest: UiManifest) -> dict[str, Any]: + return { + "name": plugin.name, + "version": plugin.version, + "builtin": plugin.builtin, # core plugins the UI must not offer to remove + "slots": [_slot_dict(s) for s in manifest.slots], + "pages": [_page_dict(p) for p in manifest.pages], + "capabilities": dict(manifest.capabilities), + } + + +def build_ui_manifest() -> dict[str, Any]: + """Aggregate every plugin's UI manifest for the frontend loader (SPEC §11.3).""" + plugins = [ + _manifest_dict(p, p.ui_manifest) for p in get_plugins() if p.ui_manifest is not None + ] + return {"plugins": plugins} + + +def get_plugin(name: str) -> Plugin | None: + """Return the loaded plugin named ``name``, or None.""" + return next((p for p in get_plugins() if p.name == name), None) + + +class PluginRemovalError(Exception): + """Raised when trying to uninstall a builtin (core) plugin (SPEC §11 / §12).""" + + +def assert_removable(name: str) -> Plugin: + """Return the plugin, or raise if it's unknown or builtin (non-removable). + + Core plugins (calendar, contacts) ship ``builtin=True`` and may never be + uninstalled by a user; the uninstall path calls this before touching data.""" + plugin = get_plugin(name) + if plugin is None: + raise PluginRemovalError(f"unknown plugin {name!r}") + if plugin.builtin: + raise PluginRemovalError(f"plugin {name!r} is builtin and cannot be removed") + return plugin diff --git a/backend/app/plugins/migrate.py b/backend/app/plugins/migrate.py new file mode 100644 index 0000000..dcf6ed1 --- /dev/null +++ b/backend/app/plugins/migrate.py @@ -0,0 +1,98 @@ +"""Per-plugin Alembic migration runner + install/uninstall lifecycle (SPEC §11.1). + +Each plugin ships its own revision directory (``Plugin.migrations``) and runs +against a dedicated version table, so its schema history is independent of core +and of other plugins. ``install`` upgrades to head; ``uninstall --drop-data`` +runs the plugin's downgrade chain and removes the version table, leaving zero +residue (the §12 M3 acceptance round-trip). Builtin (core) plugins refuse removal. +""" + +from __future__ import annotations + +import asyncio +import logging +from pathlib import Path + +from alembic import command +from alembic.config import Config +from sqlalchemy import text +from sqlalchemy.ext.asyncio import create_async_engine +from sqlalchemy.pool import NullPool + +from app.config import get_settings +from app.plugins import loader +from app.plugins.types import Plugin + +logger = logging.getLogger(__name__) + +_SHARED_ENV = str(Path(__file__).parent / "alembic") + + +class PluginMigrationError(Exception): + """Raised when a plugin's migrations can't be run.""" + + +def _version_table(plugin: Plugin) -> str: + # plugin.name is validated [a-z][a-z0-9_]* at construction, so this is safe to + # interpolate into DDL. + return f"alembic_version_plg_{plugin.name}" + + +def _config(plugin: Plugin) -> Config: + if not plugin.migrations: + raise PluginMigrationError(f"plugin {plugin.name!r} declares no migrations") + cfg = Config() + cfg.set_main_option("script_location", _SHARED_ENV) + cfg.set_main_option("version_locations", plugin.migrations) + cfg.set_main_option("path_separator", "os") # split version_locations on os.pathsep only + cfg.set_main_option("version_table", _version_table(plugin)) + return cfg + + +def upgrade(plugin: Plugin, revision: str = "head") -> None: + """Run the plugin's migrations up to ``revision`` (default: head).""" + command.upgrade(_config(plugin), revision) + + +def downgrade(plugin: Plugin, revision: str = "base") -> None: + """Run the plugin's downgrades to ``revision`` (default: base = drop all).""" + command.downgrade(_config(plugin), revision) + + +async def _drop_version_table(plugin: Plugin) -> None: + # Fresh engine (not the shared global one) so this is safe to call from any loop. + engine = create_async_engine(get_settings().database_url, poolclass=NullPool) + try: + async with engine.begin() as conn: + await conn.execute(text(f"DROP TABLE IF EXISTS {_version_table(plugin)}")) + finally: + await engine.dispose() + + +def install(name: str) -> None: + """Apply a plugin's migrations (create its ``plg_*`` tables). No-op if none.""" + plugin = loader.get_plugin(name) + if plugin is None: + raise PluginMigrationError(f"unknown plugin {name!r}") + if plugin.migrations is None: + logger.info("plugin %s has no migrations; nothing to install", name) + return + logger.info("installing plugin %s: upgrade -> head", name) + upgrade(plugin, "head") + + +def uninstall(name: str, *, drop_data: bool = False) -> None: + """Uninstall a plugin. Refuses builtin plugins. + + Without ``drop_data`` the schema/data are kept (a reversible disable). With + ``drop_data`` the downgrade chain runs and the version table is removed, so + nothing remains in the database (SPEC §12 M3 acceptance).""" + plugin = loader.assert_removable(name) # raises PluginRemovalError for builtin/unknown + if not drop_data: + logger.info("uninstalling plugin %s (data kept)", name) + return + if plugin.migrations is None: + return + logger.info("uninstalling plugin %s --drop-data: downgrade -> base", name) + downgrade(plugin, "base") + asyncio.run(_drop_version_table(plugin)) diff --git a/backend/app/plugins/rls.py b/backend/app/plugins/rls.py new file mode 100644 index 0000000..0156b90 --- /dev/null +++ b/backend/app/plugins/rls.py @@ -0,0 +1,46 @@ +"""Row-Level Security for plugin-owned tables (SPEC §11.1 + privacy hardening). + +Per-user isolation for ``plg_*`` tables is enforced by Postgres, not by remembering +a ``WHERE user_id = ?`` in every query. Each user-data table gets RLS enabled + +forced, with a policy comparing ``user_id`` to the transaction-local GUC +``app.current_user_id``. Plugin code reaches these tables only through +``sdk.session(user_id)`` (see :mod:`app.plugins.db`), which drops to the +least-privilege :data:`PLUGIN_ROLE` and binds that GUC — so even a query that +forgets to filter cannot see or write another user's rows. + +Plugin migrations call :func:`enable_rls` right after creating a user-data table. +""" + +from __future__ import annotations + +# NOLOGIN, non-superuser, NOBYPASSRLS role plugin sessions SET LOCAL ROLE into +# (created by core migration 0014). RLS is skipped for superusers / table owners / +# BYPASSRLS roles, so plugin access must run as this restricted role to be enforced. +PLUGIN_ROLE = "webmail_plugin" + +_USER_CTX = "current_setting('app.current_user_id', true)::uuid" + + +def rls_ddl(table: str, *, user_column: str = "user_id") -> list[str]: + """SQL statements that lock ``table`` to the current user (SPEC §11.1). + + Pure/testable: returns the DDL rather than executing it. ``FORCE`` makes even + the table owner subject to the policy; ``WITH CHECK`` blocks inserting/updating + rows for a different user.""" + policy = f"{table}_user_isolation" + return [ + f"ALTER TABLE {table} ENABLE ROW LEVEL SECURITY", + f"ALTER TABLE {table} FORCE ROW LEVEL SECURITY", + f"CREATE POLICY {policy} ON {table} " + f"USING ({user_column} = {_USER_CTX}) " + f"WITH CHECK ({user_column} = {_USER_CTX})", + f"GRANT SELECT, INSERT, UPDATE, DELETE ON {table} TO {PLUGIN_ROLE}", + ] + + +def enable_rls(table: str, *, user_column: str = "user_id") -> None: + """Apply :func:`rls_ddl` via Alembic ``op`` — call from a plugin migration.""" + from alembic import op + + for statement in rls_ddl(table, user_column=user_column): + op.execute(statement) diff --git a/backend/app/plugins/sdk.py b/backend/app/plugins/sdk.py new file mode 100644 index 0000000..e96e06c --- /dev/null +++ b/backend/app/plugins/sdk.py @@ -0,0 +1,116 @@ +"""The plugin SDK — the ONLY ``app.*`` surface a plugin may import (SPEC §11). + +Everything a plugin needs is reached through :func:`sdk`, which returns a +:class:`PluginSdk` bound to the plugin's name: a plugin-scoped DB session, the +event bus (publishing under ``plugin.{name}.*``), the blob store, outbox +draft/send helpers, the settings accessor, an SSRF-guarded HTTP client +(§10.6), and the namespaced KV store (§11). + +The frozen contract types (:class:`Plugin`, :class:`UiManifest`, :class:`JobSpec`, +…) are re-exported here so plugin packages import only from ``app.plugins.sdk``. +""" + +from __future__ import annotations + +import uuid +from contextlib import AbstractAsyncContextManager +from dataclasses import dataclass +from typing import Any + +from sqlalchemy.ext.asyncio import AsyncSession + +from app.accounts.ssrf import guarded_get +from app.config import Settings, get_settings +from app.events import get_event_bus +from app.indexer import blobstore +from app.outbox import service as outbox_service +from app.plugins.db import plugin_session +from app.plugins.kv import PluginKvStore +from app.plugins.types import ( + Hook, + JobFunc, + JobSpec, + Page, + PageNav, + Plugin, + Slot, + UiManifest, +) +from app.queue import enqueue_job + +__all__ = [ + "Hook", + "JobFunc", + "JobSpec", + "Page", + "PageNav", + "Plugin", + "PluginSdk", + "Slot", + "UiManifest", + "blobstore", + "sdk", +] + + +@dataclass(frozen=True) +class PluginSdk: + """A lightweight facade over shared services, scoped to one plugin. + + Cheap to construct (wraps process-wide singletons + the plugin name), so + handlers/routes obtain it on demand via ``sdk("")``.""" + + name: str + + # --- config ------------------------------------------------------------- + @property + def settings(self) -> Settings: + return get_settings() + + # --- database ----------------------------------------------------------- + def session(self, user_id: uuid.UUID) -> AbstractAsyncContextManager[AsyncSession]: + """An RLS-enforced transactional session scoped to ``user_id`` (SPEC §11.1). + + Row-level security limits every query to this user's rows on ``plg_*`` + tables — even a query that forgets ``WHERE user_id`` cannot leak or write + across users (see :mod:`app.plugins.db`).""" + return plugin_session(user_id) + + def kv(self, user_id: uuid.UUID) -> PluginKvStore: + """The namespaced KV store for this plugin and ``user_id`` (SPEC §11).""" + return PluginKvStore(self.name, user_id) + + # --- events ------------------------------------------------------------- + async def publish(self, event: str, payload: dict[str, object]) -> None: + """Publish ``plugin.{name}.{event}`` on the bus (payload needs ``user_id``).""" + await get_event_bus().publish(f"plugin.{self.name}.{event}", payload) + + # --- background jobs ---------------------------------------------------- + async def enqueue(self, job_func: JobFunc, **kwargs: Any) -> None: + """Enqueue one of this plugin's jobs on the shared worker (SPEC §11.2). + + ``job_func`` must be a job declared in this plugin's ``jobs=[...]``. Pass + ``user_id=`` for singleton jobs so the per-user lock applies.""" + await enqueue_job(f"plg.{self.name}.{job_func.__name__}", **kwargs) + + # --- blob store --------------------------------------------------------- + @property + def blob(self) -> object: + """The blob store (``store``/``load``/``exists``/``remove``), keyed by user.""" + return blobstore + + # --- outbox ------------------------------------------------------------- + @property + def outbox(self) -> object: + """Draft/send helpers (``upsert_draft``, ``create_outbox``, …).""" + return outbox_service + + # --- outbound HTTP ------------------------------------------------------ + async def http_get(self, url: str) -> str | None: + """GET ``url`` through the §10.6 SSRF guard (blocks private IPs/oversize).""" + return await guarded_get(url) + + +def sdk(name: str) -> PluginSdk: + """Return the SDK facade bound to plugin ``name`` (SPEC §11).""" + return PluginSdk(name) diff --git a/backend/app/plugins/types.py b/backend/app/plugins/types.py new file mode 100644 index 0000000..b27ae83 --- /dev/null +++ b/backend/app/plugins/types.py @@ -0,0 +1,106 @@ +# INTERFACE VERSION: 1 +"""Frozen plugin contract types (SPEC §11). + +A plugin is a Python package ``webmail_plugin_`` exposing a module-level +``plugin = Plugin(...)``, discovered via the ``webmail.plugins`` entry-point group. + +These dataclasses are the frozen surface (SPEC §11 "freeze in M3"): fields may be +added with defaults, but existing names/meanings may not change without bumping +the INTERFACE VERSION above. +""" + +from __future__ import annotations + +import re +from collections.abc import Awaitable, Callable +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from fastapi import APIRouter + +# A hook handler receives an event payload (always carrying ``user_id``, SPEC §5.3). +Hook = Callable[[dict[str, object]], Awaitable[None]] +# A background job body: receives the arq context dict plus job-specific kwargs. +JobFunc = Callable[..., Awaitable[Any]] + +# The frozen slot catalog (SPEC §11.3). Bundles are injected into these UI points. +SLOT_NAMES: frozenset[str] = frozenset( + {"thread.toolbar", "message.actions", "message.banner", "sidebar.section", "settings.page"} +) + +_NAME_RE = re.compile(r"[a-z][a-z0-9_]*") + + +@dataclass(frozen=True) +class Slot: + """A lazy-loaded ESM bundle injected into a named mail-UI slot (SPEC §11.3).""" + + name: str + bundle: str # path served by the plugin router, resolved by the frontend loader + + def __post_init__(self) -> None: + if self.name not in SLOT_NAMES: + raise ValueError(f"unknown slot {self.name!r} (SPEC §11.3 slot catalog)") + + +@dataclass(frozen=True) +class PageNav: + """Nav entry for a page-plugin (SPEC §11.3).""" + + icon: str + label: str + order: int = 100 + + +@dataclass(frozen=True) +class Page: + """A full-content-area view mounted at ``route`` with a nav entry (SPEC §11.3).""" + + route: str + nav: PageNav + bundle: str + + +@dataclass(frozen=True) +class UiManifest: + """A plugin's declared frontend surfaces (SPEC §11.3).""" + + slots: list[Slot] = field(default_factory=list) + pages: list[Page] = field(default_factory=list) + capabilities: dict[str, str] = field(default_factory=dict) # name -> endpoint path + + +@dataclass(frozen=True) +class JobSpec: + """A background task on the shared arq worker (SPEC §11.2). + + ``schedule`` is a cron string for periodic jobs (None = enqueue on demand); + ``singleton`` caps it to one running instance per (plugin, user) via a Redis + lock. Registration onto the worker is done by the loader (M3 §11.2).""" + + func: JobFunc + schedule: str | None = None + singleton: bool = False + + +@dataclass(frozen=True) +class Plugin: + """The object a plugin package exposes as module-level ``plugin`` (SPEC §11).""" + + name: str + version: str + hooks: dict[str, Hook] = field(default_factory=dict) + router: APIRouter | None = None + ui_manifest: UiManifest | None = None + migrations: str | None = None # path to the plugin's Alembic version branch (11.1) + jobs: list[JobSpec] = field(default_factory=list) + # Core plugins (calendar, contacts) ship builtin=True: always installed, never + # offered for removal in the UI, and the uninstall path refuses --drop-data. + builtin: bool = False + + def __post_init__(self) -> None: + if not _NAME_RE.fullmatch(self.name): + raise ValueError( + f"invalid plugin name {self.name!r}: must match [a-z][a-z0-9_]* (SPEC §11)" + ) diff --git a/backend/migrations/versions/0013_plugin_kv.py b/backend/migrations/versions/0013_plugin_kv.py new file mode 100644 index 0000000..4046a29 --- /dev/null +++ b/backend/migrations/versions/0013_plugin_kv.py @@ -0,0 +1,39 @@ +"""add plugin_kv (SPEC §11 plugin namespaced key/value store) + +Core-owned table backing ``plugin_kv(plugin, user_id, key)`` — the light-state +store every plugin gets (SPEC §11). Not a ``plg_``-prefixed plugin-owned table; +plugin-owned tables ship in each plugin's own Alembic branch (§11.1). + +Revision ID: 0013 +Revises: 0012 +Create Date: 2026-07-26 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +from alembic import op + +revision: str = "0013" +down_revision: str | None = "0012" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.execute( + """ + CREATE TABLE plugin_kv ( + plugin text NOT NULL, + user_id uuid NOT NULL REFERENCES users(id) ON DELETE CASCADE, + key text NOT NULL, + value jsonb NOT NULL, + PRIMARY KEY (plugin, user_id, key) + ) + """ + ) + + +def downgrade() -> None: + op.execute("DROP TABLE IF EXISTS plugin_kv") diff --git a/backend/migrations/versions/0014_plugin_role.py b/backend/migrations/versions/0014_plugin_role.py new file mode 100644 index 0000000..dda3e4e --- /dev/null +++ b/backend/migrations/versions/0014_plugin_role.py @@ -0,0 +1,61 @@ +"""create the least-privilege plugin DB role (SPEC §11.1 RLS enforcement) + +Plugin table access runs as ``webmail_plugin`` — a NOLOGIN, NOSUPERUSER, +NOBYPASSRLS role that plugin sessions ``SET LOCAL ROLE`` into per transaction. +Row-Level Security is skipped for superusers, table owners, and BYPASSRLS roles, +so per-user isolation on ``plg_*`` tables is only real when access runs as this +restricted role. The app role is granted membership so it can SET ROLE into it +(the dev role happens to be superuser, but this keeps prod parity where it isn't). + +Per-table SELECT/INSERT/UPDATE/DELETE grants + policies are added by each plugin's +own migration via ``app.plugins.rls.enable_rls`` (§11.1); this only provisions the +role and schema usage. + +Revision ID: 0014 +Revises: 0013 +Create Date: 2026-07-26 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +from alembic import op + +revision: str = "0014" +down_revision: str | None = "0013" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.execute( + """ + DO $$ + BEGIN + IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'webmail_plugin') THEN + CREATE ROLE webmail_plugin NOLOGIN NOSUPERUSER NOBYPASSRLS NOCREATEDB NOCREATEROLE; + END IF; + END + $$; + """ + ) + op.execute("GRANT USAGE ON SCHEMA public TO webmail_plugin") + # Let the connecting application role drop into the restricted role via SET ROLE. + op.execute("GRANT webmail_plugin TO CURRENT_USER") + + +def downgrade() -> None: + op.execute( + """ + DO $$ + BEGIN + IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'webmail_plugin') THEN + EXECUTE 'REVOKE webmail_plugin FROM CURRENT_USER'; + EXECUTE 'DROP OWNED BY webmail_plugin'; + EXECUTE 'DROP ROLE webmail_plugin'; + END IF; + END + $$; + """ + ) diff --git a/backend/tests/fixtures/plugin_demo/versions/0001_note.py b/backend/tests/fixtures/plugin_demo/versions/0001_note.py new file mode 100644 index 0000000..f4c70b9 --- /dev/null +++ b/backend/tests/fixtures/plugin_demo/versions/0001_note.py @@ -0,0 +1,35 @@ +"""demo plugin: a user-owned note table with RLS (skeleton for SPEC §11.1 tests) + +Revision ID: demo0001 +Revises: +Create Date: 2026-07-26 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +from alembic import op + +from app.plugins.rls import enable_rls + +revision: str = "demo0001" +down_revision: str | None = None +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.execute( + "CREATE TABLE plg_demo_note (" + "id uuid PRIMARY KEY DEFAULT gen_random_uuid(), " + "user_id uuid NOT NULL, " + "body text NOT NULL DEFAULT '')" + ) + # Per-user isolation enforced by Postgres (SPEC §11.1); every plg_ user-data + # table calls this right after creation. + enable_rls("plg_demo_note") + + +def downgrade() -> None: + op.execute("DROP TABLE IF EXISTS plg_demo_note") diff --git a/backend/tests/plugins/__init__.py b/backend/tests/plugins/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/tests/plugins/conftest.py b/backend/tests/plugins/conftest.py new file mode 100644 index 0000000..040c697 --- /dev/null +++ b/backend/tests/plugins/conftest.py @@ -0,0 +1,16 @@ +"""Reset the plugin registry after each test so a test override never leaks into +other suites that call ``create_app`` (which mounts discovered plugins).""" + +from __future__ import annotations + +from collections.abc import Iterator + +import pytest + +from app.plugins import loader + + +@pytest.fixture(autouse=True) +def _reset_plugins() -> Iterator[None]: + yield + loader.set_plugins_for_test(None) diff --git a/backend/tests/plugins/test_builtin.py b/backend/tests/plugins/test_builtin.py new file mode 100644 index 0000000..325d8ec --- /dev/null +++ b/backend/tests/plugins/test_builtin.py @@ -0,0 +1,38 @@ +"""Builtin (non-removable core) plugin handling (SPEC §11 / §12).""" + +from __future__ import annotations + +import pytest + +from app.plugins import loader +from app.plugins.sdk import Plugin + + +def test_builtin_defaults_false() -> None: + assert Plugin(name="filters", version="1").builtin is False + assert Plugin(name="calendar", version="1", builtin=True).builtin is True + + +def test_assert_removable_blocks_builtin() -> None: + loader.set_plugins_for_test( + [ + Plugin(name="filters", version="1"), # removable + Plugin(name="calendar", version="1", builtin=True), # core, non-removable + ] + ) + assert loader.assert_removable("filters").name == "filters" + + with pytest.raises(loader.PluginRemovalError, match="builtin"): + loader.assert_removable("calendar") + with pytest.raises(loader.PluginRemovalError, match="unknown"): + loader.assert_removable("nope") + + +def test_builtin_surfaces_in_ui_manifest() -> None: + from app.plugins.sdk import UiManifest + + loader.set_plugins_for_test( + [Plugin(name="calendar", version="1", builtin=True, ui_manifest=UiManifest())] + ) + entry = loader.build_ui_manifest()["plugins"][0] + assert entry["name"] == "calendar" and entry["builtin"] is True diff --git a/backend/tests/plugins/test_jobs.py b/backend/tests/plugins/test_jobs.py new file mode 100644 index 0000000..e231430 --- /dev/null +++ b/backend/tests/plugins/test_jobs.py @@ -0,0 +1,115 @@ +"""Plugin background jobs: registration, cron parsing, singleton lock (SPEC §11.2).""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from app.plugins import jobs, loader +from app.plugins.jobs import job_name, parse_cron, plugin_arq_functions, plugin_cron_jobs +from app.plugins.sdk import JobSpec, Plugin, sdk +from app.redis_client import get_redis + + +async def _noop(ctx: dict[str, Any], **kwargs: Any) -> None: + return None + + +def test_job_name_is_namespaced() -> None: + assert job_name("calendar", JobSpec(func=_noop)) == "plg.calendar._noop" + + +def test_parse_cron() -> None: + assert parse_cron("*/15 * * * *") == { + "minute": {0, 15, 30, 45}, + "hour": None, + "day": None, + "month": None, + "weekday": None, + } + assert parse_cron("0 9-11 1 * 1,3") == { + "minute": {0}, + "hour": {9, 10, 11}, + "day": {1}, + "month": None, + "weekday": {1, 3}, + } + with pytest.raises(ValueError, match="5 fields"): + parse_cron("* * *") + + +def test_functions_and_cron_registered() -> None: + async def refresh(ctx: dict[str, Any], **kwargs: Any) -> None: + return None + + async def poll(ctx: dict[str, Any], **kwargs: Any) -> None: + return None + + loader.set_plugins_for_test( + [ + Plugin( + name="cal", + version="1", + jobs=[JobSpec(func=refresh), JobSpec(func=poll, schedule="*/5 * * * *")], + ) + ] + ) + fn_names = {f.name for f in plugin_arq_functions()} + assert fn_names == {"plg.cal.refresh", "plg.cal.poll"} # both enqueueable + + crons = plugin_cron_jobs() + assert [c.name for c in crons] == ["plg.cal.poll.cron"] # only the scheduled one + + +async def test_singleton_lock_prevents_overlap() -> None: + calls: list[str] = [] + + async def work(ctx: dict[str, Any], **kwargs: Any) -> None: + calls.append(str(kwargs.get("user_id"))) + + plugin = Plugin(name="cal", version="1", jobs=[JobSpec(func=work, singleton=True)]) + runner, _ = jobs._make_runner(plugin, plugin.jobs[0]) + + await runner({}, user_id="u1") # runs; lock acquired then released + assert calls == ["u1"] + + # Simulate an instance already holding the lock for u2 -> the run is skipped. + held = jobs._lock_key("cal", "work", "u2") + await get_redis().set(held, b"1", nx=True, ex=60) + await runner({}, user_id="u2") + assert calls == ["u1"] # unchanged + + await get_redis().delete(held) + await runner({}, user_id="u2") # lock free now -> runs + assert calls == ["u1", "u2"] + + +async def test_non_singleton_always_runs() -> None: + calls: list[int] = [] + + async def work(ctx: dict[str, Any], **kwargs: Any) -> None: + calls.append(1) + + plugin = Plugin(name="cal", version="1", jobs=[JobSpec(func=work)]) + runner, _ = jobs._make_runner(plugin, plugin.jobs[0]) + await runner({}, user_id="u1") + await runner({}, user_id="u1") + assert calls == [1, 1] # no lock, both run + + +async def test_sdk_enqueue_uses_namespaced_name(monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict[str, Any] = {} + + async def fake_enqueue(name: str, **kwargs: Any) -> None: + captured["name"] = name + captured["kwargs"] = kwargs + + monkeypatch.setattr("app.plugins.sdk.enqueue_job", fake_enqueue) + + async def refresh(ctx: dict[str, Any], **kwargs: Any) -> None: + return None + + await sdk("cal").enqueue(refresh, user_id="u1") + assert captured["name"] == "plg.cal.refresh" + assert captured["kwargs"] == {"user_id": "u1"} diff --git a/backend/tests/plugins/test_kv.py b/backend/tests/plugins/test_kv.py new file mode 100644 index 0000000..bf4a53e --- /dev/null +++ b/backend/tests/plugins/test_kv.py @@ -0,0 +1,52 @@ +"""Namespaced plugin KV store round-trips (SPEC §11 plugin_kv).""" + +from __future__ import annotations + +import uuid + +from sqlalchemy import delete +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine +from sqlalchemy.pool import NullPool + +from app.config import get_settings +from app.db import session_scope +from app.identity.passwords import hash_password +from app.models import User +from app.plugins.sdk import sdk + + +async def test_plugin_kv_crud_is_namespaced() -> None: + engine = create_async_engine(get_settings().database_url, poolclass=NullPool) + sm = async_sessionmaker(engine, expire_on_commit=False) + session = sm() + user = User(handle=f"kv{uuid.uuid4().hex[:8]}@test.local", password_hash=hash_password("pw")) + session.add(user) + await session.commit() + + try: + kv = sdk("filters").kv(user.id) + other = sdk("labels").kv(user.id) # different plugin, same user + + assert await kv.get("rules", default=[]) == [] # missing key -> default + await kv.set("rules", [{"match": "boss", "flag": "important"}]) + await kv.set("count", 3) + assert await kv.get("rules") == [{"match": "boss", "flag": "important"}] + assert await kv.get("count") == 3 + + # Namespacing: the labels plugin sees nothing filters wrote. + assert await other.get("rules") is None + assert await other.items() == {} + + assert await kv.items() == {"rules": [{"match": "boss", "flag": "important"}], "count": 3} + + await kv.set("count", 4) # upsert overwrites + assert await kv.get("count") == 4 + + await kv.delete("count") + assert await kv.get("count") is None + assert set((await kv.items()).keys()) == {"rules"} + finally: + async with session_scope() as cleanup: # cascade drops plugin_kv rows + await cleanup.execute(delete(User).where(User.id == user.id)) + await session.close() + await engine.dispose() diff --git a/backend/tests/plugins/test_loader.py b/backend/tests/plugins/test_loader.py new file mode 100644 index 0000000..a50e0e8 --- /dev/null +++ b/backend/tests/plugins/test_loader.py @@ -0,0 +1,117 @@ +"""Plugin discovery + core wiring (SPEC §11).""" + +from __future__ import annotations + +import httpx +import pytest +from fastapi import APIRouter, FastAPI +from httpx import ASGITransport + +from app.events import get_event_bus +from app.plugins import loader +from app.plugins.sdk import Page, PageNav, Plugin, Slot, UiManifest + + +class _FakeEP: + def __init__(self, name: str, obj: object) -> None: + self.name = name + self._obj = obj + + def load(self) -> object: + return self._obj + + +def test_discover_reads_entry_points(monkeypatch: pytest.MonkeyPatch) -> None: + plugin = Plugin(name="filters", version="1.2") + monkeypatch.setattr(loader, "entry_points", lambda group: [_FakeEP("filters", plugin)]) + loader.set_plugins_for_test(None) # force rediscovery through the fake + found = loader.discover_plugins() + assert [(p.name, p.version) for p in found] == [("filters", "1.2")] + + +def test_discover_accepts_module_exposing_plugin(monkeypatch: pytest.MonkeyPatch) -> None: + class _Mod: + plugin = Plugin(name="labels", version="0.1") + + monkeypatch.setattr(loader, "entry_points", lambda group: [_FakeEP("labels", _Mod())]) + loader.set_plugins_for_test(None) + assert [p.name for p in loader.discover_plugins()] == ["labels"] + + +def test_discover_skips_duplicate_names(monkeypatch: pytest.MonkeyPatch) -> None: + eps = [ + _FakeEP("a", Plugin(name="dup", version="1")), + _FakeEP("b", Plugin(name="dup", version="2")), + ] + monkeypatch.setattr(loader, "entry_points", lambda group: eps) + loader.set_plugins_for_test(None) + found = loader.discover_plugins() + assert [p.version for p in found] == ["1"] # first wins, duplicate dropped + + +async def test_register_hooks_forwards_events() -> None: + seen: list[dict[str, object]] = [] + + async def on_new(payload: dict[str, object]) -> None: + seen.append(payload) + + loader.set_plugins_for_test( + [Plugin(name="watcher", version="1", hooks={"message.new": on_new})] + ) + bus = get_event_bus() + loader.register_hooks(bus) + loader.register_hooks(bus) # idempotent: no double subscription + + await bus.publish("message.new", {"user_id": "u", "message_id": "m"}) + assert seen == [{"user_id": "u", "message_id": "m"}] + + +async def test_mount_routers_prefixes_by_name() -> None: + r = APIRouter() + + @r.get("/ping") + async def _ping() -> dict[str, str]: + return {"ok": "1"} + + loader.set_plugins_for_test([Plugin(name="demo", version="1", router=r)]) + app = FastAPI() + loader.mount_routers(app) + transport = ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client: + resp = await client.get("/api/plugins/demo/ping") + assert resp.status_code == 200 + assert resp.json() == {"ok": "1"} + + +def test_build_ui_manifest_aggregates() -> None: + manifest = UiManifest( + slots=[Slot(name="message.actions", bundle="/a.js")], + pages=[ + Page( + route="/notes", + nav=PageNav(icon="note", label="Notes", order=5), + bundle="/p.js", + ) + ], + capabilities={"recipients.suggest": "/api/plugins/contacts/suggest"}, + ) + loader.set_plugins_for_test([Plugin(name="notes", version="2.0", ui_manifest=manifest)]) + out = loader.build_ui_manifest() + assert out == { + "plugins": [ + { + "name": "notes", + "version": "2.0", + "builtin": False, + "slots": [{"name": "message.actions", "bundle": "/a.js"}], + "pages": [ + { + "route": "/notes", + "bundle": "/p.js", + "nav": {"icon": "note", "label": "Notes", "order": 5}, + } + ], + "capabilities": {"recipients.suggest": "/api/plugins/contacts/suggest"}, + } + ] + } diff --git a/backend/tests/plugins/test_migrate.py b/backend/tests/plugins/test_migrate.py new file mode 100644 index 0000000..4c5c359 --- /dev/null +++ b/backend/tests/plugins/test_migrate.py @@ -0,0 +1,89 @@ +"""Per-plugin migration runner + install/uninstall lifecycle (SPEC §11.1, §12 M3). + +The runner drives Alembic synchronously (env.py calls asyncio.run), so it's +invoked via ``asyncio.to_thread`` from these async tests to avoid a nested loop. +""" + +from __future__ import annotations + +import asyncio +from pathlib import Path + +import pytest +from sqlalchemy import text +from sqlalchemy.ext.asyncio import create_async_engine +from sqlalchemy.pool import NullPool + +from app.config import get_settings +from app.plugins import loader, migrate +from app.plugins.sdk import Plugin + +_VERSIONS = str(Path(__file__).parent.parent / "fixtures" / "plugin_demo" / "versions") + + +async def _scalar(sql: str, **params: str) -> object: + engine = create_async_engine(get_settings().database_url, poolclass=NullPool) + try: + async with engine.connect() as conn: + return (await conn.execute(text(sql), params)).scalar() + finally: + await engine.dispose() + + +async def _regclass(name: str) -> object: + return await _scalar("SELECT to_regclass(:n)", n=name) + + +async def _cleanup() -> None: + engine = create_async_engine(get_settings().database_url, poolclass=NullPool) + try: + async with engine.begin() as conn: + await conn.execute(text("DROP TABLE IF EXISTS plg_demo_note")) + await conn.execute(text("DROP TABLE IF EXISTS alembic_version_plg_demo")) + finally: + await engine.dispose() + + +async def test_install_run_uninstall_dropdata_roundtrip() -> None: + loader.set_plugins_for_test([Plugin(name="demo", version="1.0", migrations=_VERSIONS)]) + try: + await asyncio.to_thread(migrate.install, "demo") + + # Table exists, RLS is enabled on it, and the per-plugin version table is present. + assert await _regclass("plg_demo_note") == "plg_demo_note" + assert await _scalar( + "SELECT relrowsecurity FROM pg_class WHERE relname = 'plg_demo_note'" + ) is True + assert await _regclass("alembic_version_plg_demo") == "alembic_version_plg_demo" + + # --drop-data removes the table AND the version table: zero residue. + await asyncio.to_thread(migrate.uninstall, "demo", drop_data=True) + assert await _regclass("plg_demo_note") is None + assert await _regclass("alembic_version_plg_demo") is None + finally: + await _cleanup() + + +async def test_uninstall_without_drop_data_keeps_the_table() -> None: + loader.set_plugins_for_test([Plugin(name="demo", version="1.0", migrations=_VERSIONS)]) + try: + await asyncio.to_thread(migrate.install, "demo") + await asyncio.to_thread(migrate.uninstall, "demo", drop_data=False) + assert await _regclass("plg_demo_note") == "plg_demo_note" # data preserved + finally: + await asyncio.to_thread(migrate.uninstall, "demo", drop_data=True) + await _cleanup() + + +async def test_uninstall_refuses_builtin_plugin() -> None: + loader.set_plugins_for_test( + [Plugin(name="calendar", version="1", builtin=True, migrations=_VERSIONS)] + ) + with pytest.raises(loader.PluginRemovalError, match="builtin"): + await asyncio.to_thread(migrate.uninstall, "calendar", drop_data=True) + + +async def test_install_unknown_plugin_errors() -> None: + loader.set_plugins_for_test([]) + with pytest.raises(migrate.PluginMigrationError, match="unknown"): + await asyncio.to_thread(migrate.install, "nope") diff --git a/backend/tests/plugins/test_rls.py b/backend/tests/plugins/test_rls.py new file mode 100644 index 0000000..2a97b19 --- /dev/null +++ b/backend/tests/plugins/test_rls.py @@ -0,0 +1,84 @@ +"""Row-Level Security enforcement for plugin tables (SPEC §11.1 + privacy). + +Proves that per-user isolation is enforced by Postgres — not by the query — so a +plugin that forgets ``WHERE user_id`` still cannot read or write another user's +rows. Requires the ``webmail_plugin`` role (migration 0014) to be applied. +""" + +from __future__ import annotations + +import uuid + +import pytest +from sqlalchemy import text + +from app.db import session_scope +from app.plugins.db import plugin_session +from app.plugins.rls import rls_ddl +from app.plugins.sdk import sdk + +TABLE = "plg_rlsdemo_note" + + +async def _create_demo_table() -> None: + async with session_scope() as s: + await s.execute(text(f"DROP TABLE IF EXISTS {TABLE}")) + await s.execute( + text( + f"CREATE TABLE {TABLE} (" + "id uuid PRIMARY KEY DEFAULT gen_random_uuid(), " + "user_id uuid NOT NULL, body text)" + ) + ) + for stmt in rls_ddl(TABLE): + await s.execute(text(stmt)) + + +async def test_rls_enforces_per_user_isolation() -> None: + user_a, user_b = uuid.uuid4(), uuid.uuid4() + await _create_demo_table() + try: + # Seed one row per user as the privileged role (superuser bypasses RLS). + async with session_scope() as s: + await s.execute( + text(f"INSERT INTO {TABLE}(user_id, body) VALUES (:u, 'a-note')"), + {"u": str(user_a)}, + ) + b_id = ( + await s.execute( + text(f"INSERT INTO {TABLE}(user_id, body) VALUES (:u, 'b-note') RETURNING id"), + {"u": str(user_b)}, + ) + ).scalar_one() + + # Scoped to user A: the query has NO user filter, yet only A's row is visible. + async with plugin_session(user_a) as s: + assert (await s.execute(text("SELECT current_user"))).scalar_one() == "webmail_plugin" + visible = (await s.execute(text(f"SELECT user_id FROM {TABLE}"))).scalars().all() + assert set(visible) == {user_a} + # IDOR blocked: fetching B's row by its exact id returns nothing. + got = ( + await s.execute(text(f"SELECT id FROM {TABLE} WHERE id = :i"), {"i": str(b_id)}) + ).first() + assert got is None + + # WITH CHECK: A cannot insert a row owned by B (error aborts the txn). + with pytest.raises(Exception, match="row-level security"): + async with plugin_session(user_a) as s: + await s.execute( + text(f"INSERT INTO {TABLE}(user_id, body) VALUES (:u, 'forged')"), + {"u": str(user_b)}, + ) + + # A can write its own rows via the SDK session; B's data stays invisible. + async with sdk("demo").session(user_a) as s: + await s.execute( + text(f"INSERT INTO {TABLE}(user_id, body) VALUES (:u, 'a-note-2')"), + {"u": str(user_a)}, + ) + async with plugin_session(user_a) as s: + count = (await s.execute(text(f"SELECT count(*) FROM {TABLE}"))).scalar_one() + assert count == 2 # both of A's rows, none of B's + finally: + async with session_scope() as s: + await s.execute(text(f"DROP TABLE IF EXISTS {TABLE}")) diff --git a/backend/tests/plugins/test_sdk.py b/backend/tests/plugins/test_sdk.py new file mode 100644 index 0000000..3d1dcd2 --- /dev/null +++ b/backend/tests/plugins/test_sdk.py @@ -0,0 +1,51 @@ +"""Plugin SDK facade behaviour (SPEC §11).""" + +from __future__ import annotations + +import uuid + +import pytest + +from app.events import get_event_bus +from app.plugins.sdk import sdk + + +async def test_publish_namespaces_under_plugin_name() -> None: + received: list[dict[str, object]] = [] + + async def handler(payload: dict[str, object]) -> None: + received.append(payload) + + event = f"plugin.demo.thing_{uuid.uuid4().hex[:6]}" + suffix = event.removeprefix("plugin.demo.") + get_event_bus().subscribe(event, handler) + + await sdk("demo").publish(suffix, {"user_id": "u1", "x": 1}) + assert received == [{"user_id": "u1", "x": 1}] + + +async def test_publish_requires_user_id() -> None: + with pytest.raises(ValueError, match="user_id"): + await sdk("demo").publish("thing", {"x": 1}) + + +async def test_http_get_uses_ssrf_guard(monkeypatch: pytest.MonkeyPatch) -> None: + calls: list[str] = [] + + async def fake_guarded_get(url: str) -> str | None: + calls.append(url) + return f"body:{url}" + + monkeypatch.setattr("app.plugins.sdk.guarded_get", fake_guarded_get) + out = await sdk("demo").http_get("http://example.test/x") + assert out == "body:http://example.test/x" + assert calls == ["http://example.test/x"] + + +def test_sdk_exposes_shared_services() -> None: + s = sdk("demo") + assert s.name == "demo" + assert s.settings is not None + # blob store + outbox helpers are reachable through the SDK namespace. + assert hasattr(s.blob, "store") and hasattr(s.blob, "load") + assert hasattr(s.outbox, "upsert_draft") diff --git a/backend/tests/plugins/test_types.py b/backend/tests/plugins/test_types.py new file mode 100644 index 0000000..4b495b5 --- /dev/null +++ b/backend/tests/plugins/test_types.py @@ -0,0 +1,34 @@ +"""Frozen plugin contract validation (SPEC §11).""" + +from __future__ import annotations + +import pytest + +from app.plugins.sdk import Plugin, Slot, UiManifest + + +def test_slot_rejects_unknown_name() -> None: + Slot(name="message.actions", bundle="/b.js") # a known slot is fine + with pytest.raises(ValueError, match="unknown slot"): + Slot(name="mail.everywhere", bundle="/b.js") + + +def test_plugin_name_must_be_identifier_like() -> None: + Plugin(name="filters", version="1.0") + Plugin(name="my_plugin2", version="1.0") + for bad in ("Filters", "2cool", "has-dash", "with space", ""): + with pytest.raises(ValueError, match="invalid plugin name"): + Plugin(name=bad, version="1.0") + + +def test_plugin_defaults_are_independent() -> None: + a = Plugin(name="a", version="1.0") + b = Plugin(name="b", version="1.0") + assert a.hooks == {} and a.jobs == [] and a.ui_manifest is None + a.hooks["x"] = lambda _p: None # type: ignore[assignment,return-value] + assert b.hooks == {} # default_factory gives each instance its own dict + + +def test_ui_manifest_defaults() -> None: + m = UiManifest() + assert m.slots == [] and m.pages == [] and m.capabilities == {} diff --git a/backend/tests/plugins/test_ui_manifest_api.py b/backend/tests/plugins/test_ui_manifest_api.py new file mode 100644 index 0000000..42c6599 --- /dev/null +++ b/backend/tests/plugins/test_ui_manifest_api.py @@ -0,0 +1,73 @@ +"""GET /api/plugins/ui-manifest returns the aggregated manifest (SPEC §11.3).""" + +from __future__ import annotations + +import uuid + +import httpx +from httpx import ASGITransport +from sqlalchemy import delete +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine +from sqlalchemy.pool import NullPool + +from app.config import get_settings +from app.identity.passwords import hash_password +from app.identity.sessions import create_session +from app.main import create_app +from app.models import User +from app.plugins import loader +from app.plugins.sdk import Plugin, Slot, UiManifest + + +async def test_ui_manifest_endpoint() -> None: + engine = create_async_engine(get_settings().database_url, poolclass=NullPool) + sm = async_sessionmaker(engine, expire_on_commit=False) + session = sm() + user = User(handle=f"ui{uuid.uuid4().hex[:8]}@test.local", password_hash=hash_password("pw")) + session.add(user) + await session.commit() + + loader.set_plugins_for_test( + [ + Plugin( + name="labels", + version="1.0", + ui_manifest=UiManifest(slots=[Slot(name="message.actions", bundle="/labels.js")]), + ) + ] + ) + + token = await create_session(user.id, "pytest") + transport = ASGITransport(app=create_app()) + client = httpx.AsyncClient( + transport=transport, base_url="http://testserver", cookies={"session": token} + ) + try: + # Unauthenticated request is rejected. + async with httpx.AsyncClient( + transport=transport, base_url="http://testserver" + ) as anon: + assert (await anon.get("/api/plugins/ui-manifest")).status_code == 401 + + resp = await client.get("/api/plugins/ui-manifest") + assert resp.status_code == 200 + body = resp.json() + assert body == { + "plugins": [ + { + "name": "labels", + "version": "1.0", + "builtin": False, + "slots": [{"name": "message.actions", "bundle": "/labels.js"}], + "pages": [], + "capabilities": {}, + } + ] + } + finally: + await client.aclose() + async with sm() as cleanup: + await cleanup.execute(delete(User).where(User.id == user.id)) + await cleanup.commit() + await session.close() + await engine.dispose() diff --git a/backend/workers/run.py b/backend/workers/run.py index 9d4a4d5..7c84cfd 100644 --- a/backend/workers/run.py +++ b/backend/workers/run.py @@ -33,7 +33,13 @@ logger = logging.getLogger(__name__) _SHUTDOWN_DRAIN_S = 10.0 # §7: release leases + drain within this budget -def _worker(functions: list[Any], queue_name: str, max_jobs: int, job_timeout: int) -> Worker: +def _worker( + functions: list[Any], + queue_name: str, + max_jobs: int, + job_timeout: int, + cron_jobs: list[Any] | None = None, +) -> Worker: return Worker( functions=functions, queue_name=queue_name, @@ -41,6 +47,7 @@ def _worker(functions: list[Any], queue_name: str, max_jobs: int, job_timeout: i max_jobs=max_jobs, job_timeout=job_timeout, handle_signals=False, # the runner owns signal handling (graceful shutdown) + cron_jobs=cron_jobs or [], ) @@ -49,8 +56,19 @@ def build_arq_workers(role: str) -> list[Worker]: s = get_settings() if role not in ("jobs", "all"): return [] + # Plugin jobs (SPEC §11.2) join the interactive queue; cron-scheduled ones run + # on that worker too. With no plugins installed these are empty (inert). + from app.plugins.jobs import plugin_arq_functions, plugin_cron_jobs + + interactive: list[Any] = [*INTERACTIVE_FUNCTIONS, *plugin_arq_functions()] return [ - _worker(INTERACTIVE_FUNCTIONS, QUEUE_INTERACTIVE, s.interactive_concurrency, 120), + _worker( + interactive, + QUEUE_INTERACTIVE, + s.interactive_concurrency, + 120, + cron_jobs=plugin_cron_jobs(), + ), _worker(BULK_FUNCTIONS, QUEUE_BULK, s.bulk_concurrency, 3600), ] @@ -59,6 +77,13 @@ async def _run() -> None: logging.basicConfig(level=get_settings().log_level) role = get_settings().worker_role + # Subscribe plugin hooks so events published during sync (message.new, …) reach + # them in this process too (SPEC §11; the API process registers separately). + from app.events import get_event_bus + from app.plugins import loader + + loader.register_hooks(get_event_bus()) + coordinator = SyncCoordinator() if role in ("sync", "all") else None if coordinator is not None: await coordinator.start()