"""Antispam HTTP surface + RLS isolation (SPEC §C.6, acceptance 1 and 4). Exercises the real vertical: entry-point discovery -> router at /api/plugins/antispam -> SDK auth -> RLS-scoped session -> the plg_antispam_* tables. Requires the plugin installed and `install antispam` run. """ from __future__ import annotations import importlib import types import uuid from collections.abc import AsyncIterator import httpx import pytest import pytest_asyncio from httpx import ASGITransport from sqlalchemy import delete, text from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from sqlalchemy.pool import NullPool # NB: `from webmail_plugin_antispam import router` would bind the APIRouter # object, not the module — __init__ re-exports it. Patch the engine module, # which router.py resolves by attribute at call time. from webmail_plugin_antispam import rspamd as rspamd_mod 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 BASE = "/api/plugins/antispam" class _SdkProxy: """The router's SDK handle with a few calls swapped out. The SDK object is a frozen dataclass, and `.messages` / `.accounts` hand out a fresh object on every access, so patching attributes on them does not stick.""" def __init__(self, real: object, **overrides: object) -> None: self._real = real self._overrides = overrides def __getattr__(self, name: str) -> object: if name in self._overrides: return self._overrides[name] return getattr(self._real, name) def _router_module() -> object: return importlib.import_module("webmail_plugin_antispam.router") async def _raw(*_a: object, **_k: object) -> bytes: return b"Subject: x\r\n\r\nbody" async def _mk_user(sm: async_sessionmaker) -> uuid.UUID: # type: ignore[type-arg] async with sm() as s: user = User( handle=f"as{uuid.uuid4().hex[:8]}@test.local", password_hash=hash_password("pw") ) s.add(user) await s.commit() return user.id class _Ctx: user_a: uuid.UUID user_b: uuid.UUID client_a: httpx.AsyncClient client_b: httpx.AsyncClient @pytest_asyncio.fixture async def ctx() -> AsyncIterator[_Ctx]: engine = create_async_engine(get_settings().database_url, poolclass=NullPool) sm = async_sessionmaker(engine, expire_on_commit=False) c = _Ctx() c.user_a = await _mk_user(sm) c.user_b = await _mk_user(sm) transport = ASGITransport(app=create_app()) async def client(uid: uuid.UUID) -> httpx.AsyncClient: token = await create_session(uid, "pytest") return httpx.AsyncClient( transport=transport, base_url="http://testserver", cookies={"session": token, "csrf": "csrf"}, headers={"X-CSRF": "csrf"}, ) c.client_a = await client(c.user_a) c.client_b = await client(c.user_b) try: yield c finally: await c.client_a.aclose() await c.client_b.aclose() async with sm() as s: for table in ("plg_antispam_rules", "plg_antispam_settings", "plg_antispam_verdicts"): await s.execute( text(f"DELETE FROM {table} WHERE user_id IN (:a, :b)"), {"a": str(c.user_a), "b": str(c.user_b)}, ) await s.execute(delete(User).where(User.id.in_([c.user_a, c.user_b]))) await s.commit() await engine.dispose() async def test_settings_defaults_then_merge(ctx: _Ctx) -> None: """A user who has never opened the page still gets the §C.3 defaults.""" s = (await ctx.client_a.get(f"{BASE}/settings")).json() assert (s["move_threshold"], s["banner_threshold"]) == (12.0, 6.0) assert s["network_checks"] is False # privacy default (§C.2) assert s["enabled_accounts"] == [] # empty = every account # PATCH merges: the field we leave out keeps its value. s = (await ctx.client_a.patch(f"{BASE}/settings", json={"move_threshold": 15.0})).json() assert s["move_threshold"] == 15.0 assert s["banner_threshold"] == 6.0 async def test_banner_threshold_cannot_exceed_move(ctx: _Ctx) -> None: """A banner above the move score would never be seen — it is capped, not refused.""" s = ( await ctx.client_a.patch( f"{BASE}/settings", json={"move_threshold": 10.0, "banner_threshold": 20.0} ) ).json() assert s["banner_threshold"] == 10.0 async def test_rules_crud_and_validation(ctx: _Ctx) -> None: created = await ctx.client_a.post( f"{BASE}/rules", json={"kind": "deny", "pattern": "@Bad.TEST"} ) assert created.status_code == 200 assert created.json()["pattern"] == "@bad.test" # normalised listed = (await ctx.client_a.get(f"{BASE}/rules")).json() assert [r["pattern"] for r in listed] == ["@bad.test"] # Re-adding the same pattern is idempotent, not a duplicate. await ctx.client_a.post(f"{BASE}/rules", json={"kind": "deny", "pattern": "@bad.test"}) assert len((await ctx.client_a.get(f"{BASE}/rules")).json()) == 1 assert ( await ctx.client_a.post(f"{BASE}/rules", json={"kind": "deny", "pattern": "nonsense"}) ).status_code == 400 assert ( await ctx.client_a.post(f"{BASE}/rules", json={"kind": "maybe", "pattern": "a@b.test"}) ).status_code == 422 await ctx.client_a.delete(f"{BASE}/rules/{listed[0]['id']}") assert (await ctx.client_a.get(f"{BASE}/rules")).json() == [] async def test_rls_isolation_between_users(ctx: _Ctx) -> None: """Acceptance 4's per-user isolation, at the storage layer.""" await ctx.client_a.post(f"{BASE}/rules", json={"kind": "allow", "pattern": "a@only.test"}) await ctx.client_a.patch(f"{BASE}/settings", json={"move_threshold": 9.0}) assert (await ctx.client_b.get(f"{BASE}/rules")).json() == [] assert (await ctx.client_b.get(f"{BASE}/settings")).json()["move_threshold"] == 12.0 # B cannot delete A's rule even knowing its id. a_rule = (await ctx.client_a.get(f"{BASE}/rules")).json()[0] await ctx.client_b.delete(f"{BASE}/rules/{a_rule['id']}") assert len((await ctx.client_a.get(f"{BASE}/rules")).json()) == 1 async def test_context_for_an_unknown_message(ctx: _Ctx) -> None: """The slots ask about every message they mount on, including ones we have never classified — that must answer, not 404.""" body = (await ctx.client_a.get(f"{BASE}/context/{uuid.uuid4()}")).json() assert body["verdict"] is None assert body["special_use"] is None # not this user's message: no folder to report async def test_stats_work_without_an_engine(ctx: _Ctx, monkeypatch: pytest.MonkeyPatch) -> None: """The stats page must render on a box where Rspamd is not running. Simulated rather than inferred from the environment: the dev stack now runs an engine, so "no engine" has to be arranged deliberately or this test only passes by accident. """ async def _no_engine(**_kw: object) -> None: return None monkeypatch.setattr(rspamd_mod, "stats", _no_engine) body = (await ctx.client_a.get(f"{BASE}/stats")).json() assert body["counts"] == {} assert body["top_symbols"] == [] assert body["bayes"] is None # unreachable engine, not an error async def test_stats_include_engine_state_when_it_is_up(ctx: _Ctx) -> None: """And when an engine IS present, its numbers come through.""" body = (await ctx.client_a.get(f"{BASE}/stats")).json() assert body["bayes"] is None or isinstance(body["bayes"], dict) async def test_ui_bundles_are_served_and_gated(ctx: _Ctx) -> None: for name in ("settings.js", "banner.js", "menu.js"): r = await ctx.client_a.get(f"{BASE}/ui/{name}") assert r.status_code == 200 assert "javascript" in r.headers["content-type"] assert (await ctx.client_a.get(f"{BASE}/ui/../hook.py")).status_code == 404 async def test_learn_records_even_without_a_prior_verdict( ctx: _Ctx, monkeypatch: pytest.MonkeyPatch ) -> None: """Mail the provider filed as spam has no verdict row of ours to update. The write must therefore upsert — and it must happen before the move, or the passive-learning hook re-teaches the lesson the move announces. """ router_mod = _router_module() async def _accepted(_raw: bytes, **_kw: object) -> bool: return True monkeypatch.setattr( router_mod, "_SDK", _SdkProxy(router_mod._SDK, messages=_SdkProxy(router_mod._SDK.messages, raw=_raw)), ) monkeypatch.setattr(rspamd_mod, "learn", _accepted) message_id = uuid.uuid4() r = await ctx.client_a.post( f"{BASE}/learn", json={"message_id": str(message_id), "verdict": "spam"} ) assert r.status_code == 200 assert r.json()["taught"] is True assert r.json()["moved"] is False # not this user's message: nothing to move body = (await ctx.client_a.get(f"{BASE}/context/{message_id}")).json() assert body["verdict"]["learned"] == "spam" async def test_a_refused_lesson_is_not_recorded(ctx: _Ctx, monkeypatch: pytest.MonkeyPatch) -> None: """The engine said no (a 401 from a mismatched controller password, say). Recording it anyway made the menu say "Marked as spam" for a lesson never taught, and its dedup then blocked every retry. """ router_mod = _router_module() async def _refused(_raw: bytes, **_kw: object) -> bool: return False monkeypatch.setattr( router_mod, "_SDK", _SdkProxy(router_mod._SDK, messages=_SdkProxy(router_mod._SDK.messages, raw=_raw)), ) monkeypatch.setattr(rspamd_mod, "learn", _refused) message_id = uuid.uuid4() r = await ctx.client_a.post( f"{BASE}/learn", json={"message_id": str(message_id), "verdict": "spam"} ) assert r.status_code == 200 assert r.json()["taught"] is False body = (await ctx.client_a.get(f"{BASE}/context/{message_id}")).json() assert body["verdict"] is None async def test_bootstrap_now_queues_only_this_users_accounts( ctx: _Ctx, monkeypatch: pytest.MonkeyPatch ) -> None: router_mod = _router_module() mine, theirs = uuid.uuid4(), uuid.uuid4() queued: list[dict[str, object]] = [] async def _live() -> list[tuple[uuid.UUID, uuid.UUID]]: return [(ctx.user_a, mine), (ctx.user_b, theirs)] async def _enqueue(_func: object, **kwargs: object) -> None: queued.append(kwargs) monkeypatch.setattr( router_mod, "_SDK", _SdkProxy(router_mod._SDK, accounts=types.SimpleNamespace(live=_live), enqueue=_enqueue), ) r = await ctx.client_a.post(f"{BASE}/bootstrap") assert r.json() == {"queued": 1} assert queued == [{"user_id": str(ctx.user_a), "account_id": str(mine), "force": True}]