M3 plugin platform: SDK + loader, RLS-isolated tables, migrations, jobs
Backend foundation for the plugin platform (SPEC §11), Phases 1-3.
Phase 1 — SDK + loader:
- app/plugins/types.py: frozen Plugin/UiManifest/Slot/Page/JobSpec contract.
- app/plugins/sdk.py: sdk(name) facade — the only app.* import allowed for plugins
(scoped session, plugin.{name}.* publish, blob, outbox, settings, SSRF http_get, KV, enqueue).
- app/plugins/loader.py: entry-point discovery (webmail.plugins), hook forwarding
(API + worker), router mounting at /api/plugins/{name}, ui-manifest aggregation.
- plugin_kv core table (migration 0013) + GET /api/plugins/ui-manifest.
Phase 2 — non-removable core plugins + DB-enforced privacy:
- builtin flag (calendar/contacts will be builtin=True; uninstall refuses them).
- RLS per-user isolation on plg_* tables: least-privilege webmail_plugin role
(migration 0014), plugin_session(user_id) sets LOCAL ROLE + app.current_user_id,
rls.enable_rls() (ENABLE+FORCE RLS + USING/WITH CHECK policy). Mail untouched.
- Per-plugin Alembic runner (shared env, dedicated version table) + install/uninstall
--drop-data CLI; skeleton fixture proves install -> run -> uninstall --drop-data
leaves zero residue (§12 M3 acceptance).
Phase 3 — plugin jobs (§11.2):
- JobSpec registered as arq function plg.{plugin}.{func} + cron (5-field parser);
singleton=True guarded by a per-(plugin,user) Redis lock; wired onto the shared
interactive worker.
Gates: ruff, mypy --strict (94 files + tests), full suite 326 passed. RLS isolation,
migration round-trip, and singleton lock covered by tests.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Rcj3TpydMTzTSmU7kunzf8
This commit is contained in:
Claude Opus 4.8
@@ -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)
|
||||
|
||||
@@ -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()
|
||||
@@ -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()
|
||||
|
||||
@@ -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__ = (
|
||||
|
||||
@@ -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``").
|
||||
"""
|
||||
@@ -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())
|
||||
@@ -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"}
|
||||
@@ -0,0 +1,58 @@
|
||||
"""Plugin management CLI (SPEC §11.1).
|
||||
|
||||
python -m app.plugins.cli list
|
||||
python -m app.plugins.cli install <name>
|
||||
python -m app.plugins.cli uninstall <name> [--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())
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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]
|
||||
@@ -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
|
||||
@@ -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))
|
||||
@@ -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)
|
||||
@@ -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>")``."""
|
||||
|
||||
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)
|
||||
@@ -0,0 +1,106 @@
|
||||
# INTERFACE VERSION: 1
|
||||
"""Frozen plugin contract types (SPEC §11).
|
||||
|
||||
A plugin is a Python package ``webmail_plugin_<name>`` 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)"
|
||||
)
|
||||
@@ -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")
|
||||
@@ -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
|
||||
$$;
|
||||
"""
|
||||
)
|
||||
@@ -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")
|
||||
@@ -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)
|
||||
@@ -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
|
||||
@@ -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"}
|
||||
@@ -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()
|
||||
@@ -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"},
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -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")
|
||||
@@ -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}"))
|
||||
@@ -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")
|
||||
@@ -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 == {}
|
||||
@@ -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()
|
||||
+27
-2
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user