Antispam: mirror the engine in dev, and never file a contact as spam

Dev compose gains rspamd + redis-rspamd, matching production. The dev API and
worker run on the host rather than in the compose network, so the ports are
published on loopback and RSPAMD_URL points at 127.0.0.1 — documented in
.env.example.

Contacts now cap a verdict at "banner": mail from someone in the user's address
book is never moved to Junk. Capped rather than exempted, because the way this
fails in practice is a contact's account being compromised and spamming its own
address book — the warning still has value even when the move would be wrong.
An explicit deny rule still wins; that is the user overruling us deliberately.

The lookup reads the contacts plugin's table directly and forgives its absence:
no contacts plugin means no contacts to check, not an error. Addresses are
stored as the user typed them, so both sides are lower-cased rather than relying
on the jsonb containment index, and secondary addresses in the emails array
count as well as the primary. It runs only when the verdict would otherwise be a
move, so the common path pays nothing.

Also fixes a test that asserted stats degrade "without an engine" by relying on
no engine being present — true until this commit. It now simulates the absence.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018SqJYt7jD7oUEkoY68sBZo

.env.example

@@ -11,3 +11,10 @@ POLL_INTERVAL=120
 SESSION_TTL_DAYS=30
 MAX_ATTACHMENT_MB=25
 LOG_LEVEL=DEBUG
+
+# --- Anti-spam (SPEC Appendix C) ---------------------------------------------
+# `make up` starts rspamd + redis-rspamd. The API and worker run on the host in
+# dev, so they use the published loopback ports rather than the service name.
+RSPAMD_URL=http://127.0.0.1:11333
+RSPAMD_CONTROLLER_URL=http://127.0.0.1:11334
+RSPAMD_CONTROLLER_PASSWORD=devpass

backend/tests/plugins/test_antispam_api.py

@@ -11,12 +11,18 @@ 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
@@ -145,14 +151,31 @@ async def test_verdict_absent_for_unclassified_message(ctx: _Ctx) -> None:
     assert (await ctx.client_a.get(f"{BASE}/verdict/{uuid.uuid4()}")).json() is None
 
 
-async def test_stats_work_without_an_engine(ctx: _Ctx) -> None:
-    """The stats page must render on a box where Rspamd is not running."""
+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", "actions.js"):
         r = await ctx.client_a.get(f"{BASE}/ui/{name}")

backend/tests/plugins/test_antispam_contacts.py

@@ -0,0 +1,94 @@
+"""Mail from the address book is never filed as spam (cross-plugin, §C.4).
+
+Exercises the real lookup against the contacts plugin's own table, which is what
+the hook consults. Requires both plugins installed.
+"""
+
+from __future__ import annotations
+
+import uuid
+from collections.abc import AsyncIterator
+
+import pytest_asyncio
+from sqlalchemy import delete, text
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+from sqlalchemy.pool import NullPool
+from webmail_plugin_antispam.policy import sender_in_contacts
+
+from app.config import get_settings
+from app.identity.passwords import hash_password
+from app.models import User
+from app.plugins.sdk import sdk
+
+_SDK = sdk("antispam")
+
+
+@pytest_asyncio.fixture
+async def user_with_contact() -> AsyncIterator[uuid.UUID]:
+    engine = create_async_engine(get_settings().database_url, poolclass=NullPool)
+    sm = async_sessionmaker(engine, expire_on_commit=False)
+    async with sm() as s:
+        user = User(
+            handle=f"ac{uuid.uuid4().hex[:8]}@test.local", password_hash=hash_password("pw")
+        )
+        s.add(user)
+        await s.commit()
+        user_id = user.id
+
+    async with _SDK.session(user_id) as s:
+        book_id = (
+            await s.execute(
+                text(
+                    "INSERT INTO plg_contacts_books (user_id, name) VALUES (:u, 'Test') "
+                    "RETURNING id"
+                ),
+                {"u": str(user_id)},
+            )
+        ).scalar()
+        await s.execute(
+            text(
+                "INSERT INTO plg_contacts_contacts "
+                "  (user_id, book_id, vcard_uid, name_display, default_email, emails) "
+                "VALUES (:u, :b, :v, 'A Friend', 'Friend@Example.COM', "
+                "        CAST(:e AS jsonb))"
+            ),
+            {
+                "u": str(user_id),
+                "b": str(book_id),
+                "v": str(uuid.uuid4()),
+                "e": '[{"value": "Other.Address@Example.com", "type": "work"}]',
+            },
+        )
+        await s.commit()
+    try:
+        yield user_id
+    finally:
+        async with sm() as s:
+            await s.execute(
+                text("DELETE FROM plg_contacts_books WHERE user_id = :u"), {"u": str(user_id)}
+            )
+            await s.execute(delete(User).where(User.id == user_id))
+            await s.commit()
+        await engine.dispose()
+
+
+async def test_primary_address_matches_case_insensitively(user_with_contact: uuid.UUID) -> None:
+    """Contacts store what the user typed, so the comparison cannot be exact."""
+    for probe in ("friend@example.com", "Friend@Example.COM", "  FRIEND@EXAMPLE.com  "):
+        assert await sender_in_contacts(user_with_contact, probe) is True
+
+
+async def test_secondary_addresses_in_the_emails_array_match(
+    user_with_contact: uuid.UUID,
+) -> None:
+    assert await sender_in_contacts(user_with_contact, "other.address@example.com") is True
+
+
+async def test_a_stranger_does_not_match(user_with_contact: uuid.UUID) -> None:
+    assert await sender_in_contacts(user_with_contact, "nobody@example.com") is False
+    assert await sender_in_contacts(user_with_contact, "") is False
+
+
+async def test_another_users_contacts_do_not_count(user_with_contact: uuid.UUID) -> None:
+    """RLS: one user's address book must not shield another user's spam."""
+    assert await sender_in_contacts(uuid.uuid4(), "friend@example.com") is False

backend/tests/plugins/test_antispam_policy.py

@@ -92,3 +92,44 @@ def test_per_account_enablement() -> None:
     )
     assert scoped.covers("acc-1")
     assert not scoped.covers("acc-2")
+
+
+# --- contacts are never filed as spam ----------------------------------------
+
+
+def test_a_contact_is_warned_about_but_never_moved() -> None:
+    """Mail from someone in the address book must not vanish into Junk."""
+    assert _decide(score=99.0) == "moved"  # same score, not a contact
+    assert (
+        decide(
+            sender="friend@example.com",
+            rules=NO_RULES,
+            score=99.0,
+            settings=SETTINGS,
+            contact=True,
+        )
+        == "banner"
+    )
+
+
+def test_a_contact_below_the_move_threshold_is_unaffected() -> None:
+    for score, expected in ((0.0, "none"), (6.0, "banner"), (11.9, "banner")):
+        assert (
+            decide(
+                sender="friend@example.com",
+                rules=NO_RULES,
+                score=score,
+                settings=SETTINGS,
+                contact=True,
+            )
+            == expected
+        )
+
+
+def test_an_explicit_deny_still_beats_being_a_contact() -> None:
+    """Deny-listing someone you also have saved is a deliberate act — honour it."""
+    rules = {"allow": set(), "deny": {"ex@example.com"}}
+    assert (
+        decide(sender="ex@example.com", rules=rules, score=0.0, settings=SETTINGS, contact=True)
+        == "deny_listed"
+    )

docker/docker-compose.yml

@@ -66,6 +66,52 @@ services:
       postgres: { condition: service_healthy }
       redis: { condition: service_healthy }
 
+  # Anti-spam engine, mirroring the production stack (see ../docker-compose.yml).
+  # Dev publishes the ports on loopback so the plugin can be driven from the host
+  # (RSPAMD_URL=http://127.0.0.1:11333) as well as from a container.
+  rspamd:
+    image: rspamd/rspamd:3.9
+    # Root only long enough to write the controller password; rspamd then drops
+    # to _rspamd itself. /etc/rspamd is root-owned in the image.
+    user: root
+    entrypoint:
+      - /bin/sh
+      - -c
+      - |
+        mkdir -p /etc/rspamd/override.d
+        printf 'password = "%s";\nenable_password = "%s";\n' \
+          "$$RSPAMD_CONTROLLER_PASSWORD" "$$RSPAMD_CONTROLLER_PASSWORD" \
+          > /etc/rspamd/override.d/worker-controller.inc
+        exec /usr/bin/rspamd -f -u _rspamd -g _rspamd
+    environment:
+      RSPAMD_CONTROLLER_PASSWORD: ${RSPAMD_CONTROLLER_PASSWORD:-devpass}
+    volumes:
+      - ./rspamd/local.d:/etc/rspamd/local.d:ro
+      - rspamddata:/var/lib/rspamd
+    ports:
+      - "127.0.0.1:11333:11333"   # scan worker  (/checkv2)
+      - "127.0.0.1:11334:11334"   # controller   (/learnspam, /stat)
+    healthcheck:
+      test: ["CMD-SHELL", "wget -qO- http://127.0.0.1:11334/ping | grep -qi pong"]
+      interval: 10s
+      timeout: 5s
+      retries: 12
+      start_period: 15s
+    depends_on:
+      redis-rspamd: { condition: service_healthy }
+
+  # Bayes/fuzzy storage, separate from the app redis on purpose (SPEC §C.2).
+  redis-rspamd:
+    image: redis:7-alpine
+    command: [redis-server, --save, "300", "1", --maxmemory, 128mb, --maxmemory-policy, allkeys-lru]
+    volumes:
+      - rspamdredis:/data
+    healthcheck:
+      test: ["CMD", "redis-cli", "ping"]
+      interval: 5s
+      timeout: 3s
+      retries: 10
+
   # Test mail servers (integration tests + manual dev against a throwaway mailbox)
   greenmail:
     image: greenmail/standalone:2.0.1
@@ -82,3 +128,5 @@ volumes:
   pgdata:
   meilidata:
   blobs:
+  rspamddata:
+  rspamdredis:

plugins/webmail_plugin_antispam/webmail_plugin_antispam/hook.py

@@ -106,6 +106,19 @@ async def on_message_new(payload: dict[str, Any]) -> None:
                 action = policy.decide(
                     sender=sender, rules=rules, score=verdict.score, settings=settings
                 )
+                # Only ask the address book when the answer could change the
+                # outcome — i.e. when we are about to move the message. On the
+                # overwhelmingly common path (not spam) this costs nothing.
+                if action == "moved" and await policy.sender_in_contacts(
+                    user_id, sender
+                ):
+                    action = policy.decide(
+                        sender=sender,
+                        rules=rules,
+                        score=verdict.score,
+                        settings=settings,
+                        contact=True,
+                    )
 
     if action in ("moved", "deny_listed"):
         moved = await _SDK.actions.move_to_special(user_id, message_id, "junk")

plugins/webmail_plugin_antispam/webmail_plugin_antispam/migrations/versions/0001_init.py

@@ -72,7 +72,9 @@ def upgrade() -> None:
         "CREATE INDEX ix_plg_antispam_verdicts_user_action "
         "ON plg_antispam_verdicts (user_id, action)"
     )
-    op.execute("CREATE INDEX ix_plg_antispam_rules_user_kind ON plg_antispam_rules (user_id, kind)")
+    op.execute(
+        "CREATE INDEX ix_plg_antispam_rules_user_kind ON plg_antispam_rules (user_id, kind)"
+    )
 
     enable_rls("plg_antispam_settings")
     enable_rls("plg_antispam_rules")

plugins/webmail_plugin_antispam/webmail_plugin_antispam/policy.py

@@ -85,6 +85,41 @@ async def load_rules(user_id: uuid.UUID) -> dict[str, set[str]]:
     return out
 
 
+async def sender_in_contacts(user_id: uuid.UUID, email: str) -> bool:
+    """Is this sender in the user's address book?
+
+    Someone you have deliberately saved is not someone whose mail should vanish
+    into Junk — so a contact caps the verdict at a banner (see :func:`decide`).
+
+    Reads the contacts plugin's table directly. Cross-plugin, and deliberately
+    forgiving: if that plugin is not installed the table does not exist, which is
+    not an error here, just "no contacts to check". The addresses are stored as
+    the user typed them, so both sides are lower-cased rather than relying on the
+    jsonb containment index.
+    """
+    addr = (email or "").strip().lower()
+    if not addr:
+        return False
+    try:
+        async with _SDK.session(user_id) as s:
+            row = (
+                await s.execute(
+                    text(
+                        "SELECT 1 FROM plg_contacts_contacts "
+                        "WHERE user_id = :u AND ("
+                        "  lower(default_email) = :e "
+                        "  OR EXISTS (SELECT 1 FROM jsonb_array_elements(emails) AS x "
+                        "             WHERE lower(x->>'value') = :e)) "
+                        "LIMIT 1"
+                    ),
+                    {"u": str(user_id), "e": addr},
+                )
+            ).first()
+        return row is not None
+    except Exception:  # noqa: BLE001 - contacts plugin absent, or its schema moved
+        return False
+
+
 def matches(patterns: set[str], email: str) -> bool:
     """A pattern is either a whole address or ``@domain`` for everyone there."""
     addr = (email or "").strip().lower()
@@ -102,12 +137,19 @@ def decide(
     rules: dict[str, set[str]],
     score: float | None,
     settings: Settings,
+    contact: bool = False,
 ) -> str:
     """The §C.4 decision, in order. Returns the verdict action.
 
     Deny beats allow beats engine: an explicit user rule is a statement about a
     sender that no score should override. ``score is None`` means the engine
     could not be reached — fail open, deliver as ``none``.
+
+    ``contact`` (the sender is in the user's address book) caps the outcome at a
+    banner: their mail is never moved to Junk. Not silenced entirely, because the
+    way this goes wrong in practice is a contact's account being compromised and
+    spamming its own address book — the user still deserves the warning. An
+    explicit deny rule still wins: that is the user overruling us on purpose.
     """
     if matches(rules["deny"], sender):
         return "deny_listed"
@@ -116,7 +158,7 @@ def decide(
     if score is None:
         return "none"
     if score >= settings.move_threshold:
-        return "moved"
+        return "banner" if contact else "moved"
     if score >= settings.banner_threshold:
         return "banner"
     return "none"

plugins/webmail_plugin_antispam/webmail_plugin_antispam/router.py

@@ -49,17 +49,23 @@ async def get_settings(user_id: CurrentUserId) -> SettingsOut:
     )
 
 
-@router.patch("/settings", response_model=SettingsOut, dependencies=[Depends(require_csrf)])
+@router.patch(
+    "/settings", response_model=SettingsOut, dependencies=[Depends(require_csrf)]
+)
 async def patch_settings(body: SettingsIn, user_id: CurrentUserId) -> SettingsOut:
     current = await policy.load_settings(user_id)
     merged = {
         "enabled_accounts": (
-            current.enabled_accounts if body.enabled_accounts is None else body.enabled_accounts
+            current.enabled_accounts
+            if body.enabled_accounts is None
+            else body.enabled_accounts
         ),
         "move_threshold": body.move_threshold or current.move_threshold,
         "banner_threshold": body.banner_threshold or current.banner_threshold,
         "network_checks": (
-            current.network_checks if body.network_checks is None else body.network_checks
+            current.network_checks
+            if body.network_checks is None
+            else body.network_checks
         ),
         "learn_from_provider_junk": (
             current.learn_from_provider_junk
@@ -166,7 +172,9 @@ async def delete_rule(rule_id: uuid.UUID, user_id: CurrentUserId) -> dict[str, s
 
 
 @router.get("/verdict/{message_id}", response_model=VerdictOut | None)
-async def get_verdict(message_id: uuid.UUID, user_id: CurrentUserId) -> VerdictOut | None:
+async def get_verdict(
+    message_id: uuid.UUID, user_id: CurrentUserId
+) -> VerdictOut | None:
     """What the banner slot asks before deciding whether to render anything."""
     async with _SDK.session(user_id) as s:
         row = (
@@ -262,5 +270,7 @@ async def stats(user_id: CurrentUserId) -> StatsOut:
 @router.get("/ui/{filename}")
 async def bundle(filename: str) -> FileResponse:
     if filename not in _BUNDLES:
-        raise HTTPException(status_code=404, detail={"code": "not_found", "message": "no bundle"})
+        raise HTTPException(
+            status_code=404, detail={"code": "not_found", "message": "no bundle"}
+        )
     return FileResponse(_UI / filename, media_type="text/javascript")

plugins/webmail_plugin_antispam/webmail_plugin_antispam/rspamd.py

@@ -85,7 +85,9 @@ def _headers(user_key: str, network_checks: bool) -> dict[str, str]:
     }
 
 
-async def check(raw: bytes, *, user_key: str, network_checks: bool = False) -> Verdict | None:
+async def check(
+    raw: bytes, *, user_key: str, network_checks: bool = False
+) -> Verdict | None:
     """Score one raw RFC822 message. ``None`` means "could not classify"."""
     try:
         async with httpx.AsyncClient(timeout=TIMEOUT) as client: