Antispam plugin (M7 phase C1): classify, file, and stay out of the way

Second-pass spam filtering per SPEC Appendix C. The engine is Rspamd in its own
container; this plugin is glue and policy, never a spam engine of its own.

Classification (§C.4) hooks message.new, acts only on inbox arrivals of enabled
accounts, and runs deny list -> allow list -> engine -> thresholds. Every
failure path ends in "deliver normally, unclassified": an unreachable engine, a
missing blob, an account with no Junk folder, a failed verdict write. Nothing
here may delay or lose mail, so no service declares depends_on: rspamd.

Two containers, started with everything else rather than behind a profile:
rspamd, and a redis of its own. The separate redis is not incidental — the app's
redis holds sync leases and arq queues under noeviction, and an ever-growing
Bayes corpus must never be able to fail a job enqueue. This one takes RDB
snapshots and evicts, because its data is derived and retrainable.

Network checks (RBL/SURBL/fuzzy) are off: they leak message-derived hashes and
URLs to third parties, which a self-hosted server's users did not ask for. The
engine makes no outbound connections in the default configuration.

The controller password is generated into .env like MASTER_KEY. It is internal;
humans will reach Rspamd through the admin panel, authenticated as the ordinary
administrator, rather than typing it.

local.d/groups.conf zeroes six SMTP-time symbols — DATE_IN_PAST,
PREVIOUSLY_DELIVERED, HFILTER_HOSTNAME_UNKNOWN, RCVD_NO_TLS_LAST, ARC_NA,
DMARC_NA — and greylisting is disabled. Measured against 398 real delivered
messages, those checks were firing on 44-71% of ordinary mail and pushing 11.3%
of it over the banner threshold; scoring mail at rest for facts true of every
delivered message is simply the wrong question. After: median 3.09 -> 0.70,
p95 7.00 -> 3.40, banner rate 11.3% -> 1.0%. The spec's 12.0/6.0 thresholds were
never the problem and are unchanged. Symbols are zeroed rather than removed so
verdicts stay explainable, and they matter again at the M6 edge (§C.7).

Still to come in C1: the §C.6 router and UI slots. C2 adds the learning loop —
until Bayes is trained, genuine scams and legitimate bulk mail overlap in the
5-8 band and no threshold separates them.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018SqJYt7jD7oUEkoY68sBZo
This commit is contained in:
2026-09-16 22:14:43 +02:00
co-authored by Claude Opus 5
parent 428ec58c71
commit 95ca1b49f6
21 changed files with 810 additions and 3 deletions
@@ -0,0 +1,21 @@
[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"
[project]
name = "webmail-plugin-antispam"
version = "0.1.0"
description = "Anti-spam — the third heavy plugin (SPEC Appendix C). Second-pass classification via Rspamd, allow/deny lists, per-user Bayes."
requires-python = ">=3.11"
# The engine is a container, not a library: the plugin is glue + policy (§C.2).
# httpx is already a core dependency, used here to reach the Rspamd container.
dependencies = []
[project.entry-points."webmail.plugins"]
antispam = "webmail_plugin_antispam:plugin"
[tool.setuptools]
packages = ["webmail_plugin_antispam"]
[tool.setuptools.package-data]
webmail_plugin_antispam = ["ui/*.js", "migrations/versions/*.py"]
@@ -0,0 +1,31 @@
"""``antispam`` — second-pass spam classification (SPEC Appendix C, M7).
For synced accounts the provider has usually filtered spam already, so this is a
*second pass*: it catches what a weak provider missed, gives one consistent
experience across accounts, and trains a per-user Bayes that belongs to the user
rather than to their mail host.
Phases: **C1** (here) scores new inbox mail, moves high scorers to Junk through
the core move machinery so the verdict reaches the IMAP server, banners the
uncertain middle, and honours per-user allow/deny lists. C2 adds the learning
loop; C3 moves the same engine to the SMTP edge when M6 lands.
The engine is Rspamd in its own container (§C.2) — this package is glue and
policy, never a spam engine of its own.
"""
from __future__ import annotations
from pathlib import Path
from app.plugins.sdk import Plugin
from webmail_plugin_antispam.hook import on_message_new
_HERE = Path(__file__).parent
plugin = Plugin(
name="antispam",
version="0.1.0",
hooks={"message.new": on_message_new},
migrations=str(_HERE / "migrations" / "versions"),
)
@@ -0,0 +1,118 @@
"""``message.new`` hook — the C1 classification flow (SPEC §C.4).
Runs in the sync worker wherever ``message.new`` is published. The invariant that
shapes every branch: **spam filtering must never delay or lose mail**. Every
failure path here ends in "deliver normally, unclassified" — an unreachable
engine, a missing blob, an account with no Junk folder, a database error while
recording the verdict.
"""
from __future__ import annotations
import json
import logging
import uuid
from typing import Any
from sqlalchemy import text
from app.plugins.sdk import sdk
from webmail_plugin_antispam import policy, rspamd
logger = logging.getLogger("plg.antispam")
_SDK = sdk("antispam")
async def _record(
user_id: uuid.UUID,
message_id: uuid.UUID,
*,
score: float,
symbols: list[str],
action: str,
) -> None:
"""Write the verdict row. A failure here must not affect delivery."""
try:
async with _SDK.session(user_id) as s:
await s.execute(
text(
"INSERT INTO plg_antispam_verdicts "
" (message_id, user_id, score, symbols, action) "
"VALUES (:m, :u, :sc, CAST(:sy AS jsonb), :a) "
"ON CONFLICT (message_id) DO NOTHING"
),
{
"m": str(message_id),
"u": str(user_id),
"sc": score,
"sy": json.dumps(symbols),
"a": action,
},
)
await s.commit()
except Exception as exc: # noqa: BLE001
logger.warning("could not record antispam verdict for %s: %s", message_id, exc)
async def on_message_new(payload: dict[str, Any]) -> None:
uid_raw = payload.get("user_id")
mid_raw = payload.get("message_id")
if not isinstance(uid_raw, str) or not isinstance(mid_raw, str):
return
user_id, message_id = uuid.UUID(uid_raw), uuid.UUID(mid_raw)
env = await _SDK.messages.envelope(user_id, message_id)
if env is None:
return
# Only newly-arrived inbox mail (§C.4). Never Sent/Drafts/Junk/Trash: scoring
# a message the user just sent, or one the provider already filed as spam,
# would be noise at best and a misfiled reply at worst.
if env.get("mailbox_special_use") != "inbox":
return
settings = await policy.load_settings(user_id)
account_id = str(env.get("account_id") or "")
if not settings.covers(account_id):
return
rules = await policy.load_rules(user_id)
sender = str(env.get("from_email") or "")
# A deny-list hit needs no engine at all (acceptance 5), so check rules first
# and only pay for scoring when the lists have nothing to say.
action = policy.decide(sender=sender, rules=rules, score=None, settings=settings)
score, symbols = 0.0, []
if action == "none": # neither list matched — ask the engine
raw = await _SDK.messages.raw(user_id, message_id)
if raw is not None:
verdict = await rspamd.check(
raw, user_key=str(user_id), network_checks=settings.network_checks
)
if verdict is not None:
score, symbols = verdict.score, verdict.symbols
action = policy.decide(
sender=sender, rules=rules, score=verdict.score, settings=settings
)
if action in ("moved", "deny_listed"):
moved = await _SDK.actions.move_to_special(user_id, message_id, "junk")
if not moved:
# No Junk folder on this account (or it is already there). Downgrade
# to a banner rather than dropping the verdict: the user still sees
# the warning, and nothing is lost.
action = "banner" if action == "moved" else "deny_listed"
await _record(user_id, message_id, score=score, symbols=symbols, action=action)
await _SDK.publish(
"plugin.antispam.classified",
{
"user_id": str(user_id),
"message_id": str(message_id),
"score": score,
"action": action,
},
)
@@ -0,0 +1,87 @@
"""antispam plugin: settings, allow/deny rules, verdicts — with RLS (SPEC §C.3)
Revision ID: antispam0001
Revises:
Create Date: 2026-09-16
"""
from __future__ import annotations
from collections.abc import Sequence
from alembic import op
from app.plugins.rls import enable_rls
revision: str = "antispam0001"
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_antispam_settings (
user_id uuid PRIMARY KEY,
-- empty array = every account; §C.3 keeps the "all" case cheap
enabled_accounts uuid[] NOT NULL DEFAULT '{}',
move_threshold real NOT NULL DEFAULT 12.0, -- score >= : move to Junk
banner_threshold real NOT NULL DEFAULT 6.0, -- score >= : warn banner
-- Off by default on purpose (§C.2): RBL/SURBL/fuzzy lookups leak
-- message-derived hashes and URLs to third parties.
network_checks boolean NOT NULL DEFAULT false,
learn_from_provider_junk boolean NOT NULL DEFAULT true
)
"""
)
op.execute(
"""
CREATE TABLE plg_antispam_rules (
id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
user_id uuid NOT NULL,
kind text NOT NULL, -- allow|deny
pattern citext NOT NULL, -- an address, or @domain for a whole domain
created_at timestamptz NOT NULL DEFAULT now(),
UNIQUE (user_id, kind, pattern),
CONSTRAINT ck_plg_antispam_rules_kind CHECK (kind IN ('allow', 'deny'))
)
"""
)
op.execute(
"""
CREATE TABLE plg_antispam_verdicts (
message_id uuid PRIMARY KEY,
user_id uuid NOT NULL,
score real NOT NULL,
symbols jsonb NOT NULL DEFAULT '[]', -- rspamd symbols, for explainability
action text NOT NULL, -- none|banner|moved|allow_listed|deny_listed
learned text, -- spam|ham|NULL (C2 learning loop)
created_at timestamptz NOT NULL DEFAULT now(),
CONSTRAINT ck_plg_antispam_verdicts_action CHECK (
action IN ('none', 'banner', 'moved', 'allow_listed', 'deny_listed')
),
CONSTRAINT ck_plg_antispam_verdicts_learned CHECK (
learned IS NULL OR learned IN ('spam', 'ham')
)
)
"""
)
# The stats page counts by action, and the banner slot looks up one message.
op.execute(
"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)"
)
enable_rls("plg_antispam_settings")
enable_rls("plg_antispam_rules")
enable_rls("plg_antispam_verdicts")
def downgrade() -> None:
op.execute("DROP TABLE IF EXISTS plg_antispam_verdicts")
op.execute("DROP TABLE IF EXISTS plg_antispam_rules")
op.execute("DROP TABLE IF EXISTS plg_antispam_settings")
@@ -0,0 +1,122 @@
"""Settings and allow/deny rules — the policy half of classification (SPEC §C.3).
Kept apart from the hook so the decision logic is testable without a database or
an engine: :func:`decide` is a pure function of (rules, score, thresholds).
"""
from __future__ import annotations
import uuid
from dataclasses import dataclass
from typing import Any
from sqlalchemy import text
from app.plugins.sdk import sdk
_SDK = sdk("antispam")
# §C.3 defaults, used when a user has never opened the settings page.
DEFAULTS = {
"enabled_accounts": [], # empty = every account
"move_threshold": 12.0,
"banner_threshold": 6.0,
"network_checks": False,
"learn_from_provider_junk": True,
}
@dataclass(frozen=True)
class Settings:
enabled_accounts: list[str]
move_threshold: float
banner_threshold: float
network_checks: bool
learn_from_provider_junk: bool
def covers(self, account_id: str) -> bool:
"""Empty list means every account — the common case stays cheap."""
return not self.enabled_accounts or account_id in self.enabled_accounts
async def load_settings(user_id: uuid.UUID) -> Settings:
async with _SDK.session(user_id) as s:
row = (
await s.execute(
text(
"SELECT enabled_accounts, move_threshold, banner_threshold, "
"network_checks, learn_from_provider_junk "
"FROM plg_antispam_settings WHERE user_id = :u"
),
{"u": str(user_id)},
)
).first()
if row is None:
d: Any = DEFAULTS
return Settings(
enabled_accounts=[],
move_threshold=d["move_threshold"],
banner_threshold=d["banner_threshold"],
network_checks=d["network_checks"],
learn_from_provider_junk=d["learn_from_provider_junk"],
)
return Settings(
enabled_accounts=[str(a) for a in (row[0] or [])],
move_threshold=float(row[1]),
banner_threshold=float(row[2]),
network_checks=bool(row[3]),
learn_from_provider_junk=bool(row[4]),
)
async def load_rules(user_id: uuid.UUID) -> dict[str, set[str]]:
"""``{"allow": {...}, "deny": {...}}`` of lower-cased patterns."""
async with _SDK.session(user_id) as s:
rows = (
await s.execute(
text("SELECT kind, pattern FROM plg_antispam_rules WHERE user_id = :u"),
{"u": str(user_id)},
)
).all()
out: dict[str, set[str]] = {"allow": set(), "deny": set()}
for kind, pattern in rows:
if kind in out:
out[kind].add(str(pattern).strip().lower())
return out
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()
if not addr:
return False
if addr in patterns:
return True
_, _, domain = addr.partition("@")
return bool(domain) and f"@{domain}" in patterns
def decide(
*,
sender: str,
rules: dict[str, set[str]],
score: float | None,
settings: Settings,
) -> 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``.
"""
if matches(rules["deny"], sender):
return "deny_listed"
if matches(rules["allow"], sender):
return "allow_listed"
if score is None:
return "none"
if score >= settings.move_threshold:
return "moved"
if score >= settings.banner_threshold:
return "banner"
return "none"
@@ -0,0 +1,146 @@
"""Rspamd HTTP client (SPEC §C.2).
The engine is a container, not a library — this module is the whole of our
coupling to it. Two rules shape everything here:
* **Fail open.** Rspamd being down, slow or wrong must never delay or lose mail
(§C.4). Every entry point returns ``None`` instead of raising, and the caller
treats ``None`` as "deliver normally, unclassified".
* **Not through the SDK's HTTP.** ``sdk.http`` applies the §10.6 SSRF guard,
which blocks private addresses — correct for user-supplied CalDAV URLs, wrong
for an operator-configured service on the compose network. The URL here comes
from the environment, never from a user, so the guard protects nothing and
would only forbid the one address that works.
"""
from __future__ import annotations
import logging
import os
from dataclasses import dataclass, field
from typing import Any
import httpx
logger = logging.getLogger("plg.antispam")
# The compose service name; operators override for an external engine.
# Rspamd splits its HTTP over two workers and they are NOT interchangeable:
# :11333 scan worker -> /checkv2, unauthenticated
# :11334 controller -> /learnspam, /learnham, /stat, password-gated
# Pointing learning at the scan port silently fails, so both are named here.
RSPAMD_URL = os.environ.get("RSPAMD_URL", "http://rspamd:11333")
RSPAMD_CONTROLLER_URL = os.environ.get("RSPAMD_CONTROLLER_URL", "http://rspamd:11334")
# Internal credential (generated into .env by `webmail.sh init`). Humans never
# present this: the admin panel authenticates them as the webmail administrator
# and talks to the controller on their behalf.
CONTROLLER_PASSWORD = os.environ.get("RSPAMD_CONTROLLER_PASSWORD", "")
# Generous enough for a large message, short enough that a hung engine cannot
# stall the sync worker: classification is best-effort, delivery is not.
TIMEOUT = float(os.environ.get("RSPAMD_TIMEOUT", "10"))
# Settings profiles defined in docker/rspamd/local.d/settings.conf. The privacy
# default (§C.2) is enforced in that file, not here: "no network" is the profile
# that omits the RBL/SURBL/fuzzy modules entirely, so a bug in this client
# cannot silently start leaking URLs and hashes to third parties.
SETTINGS_OFFLINE = "webmail_offline"
SETTINGS_NETWORK = "webmail_network"
@dataclass(frozen=True)
class Verdict:
"""What the engine thought. ``symbols`` is kept for explainability (§C.3)."""
score: float
required_score: float
action: str
symbols: list[str] = field(default_factory=list)
def _parse(data: dict[str, Any]) -> Verdict | None:
try:
score = float(data["score"])
except (KeyError, TypeError, ValueError):
return None
raw_symbols = data.get("symbols")
symbols = sorted(raw_symbols) if isinstance(raw_symbols, dict) else []
try:
required = float(data.get("required_score", 0.0))
except (TypeError, ValueError):
required = 0.0
return Verdict(
score=score,
required_score=required,
action=str(data.get("action", "no action")),
symbols=symbols,
)
def _headers(user_key: str, network_checks: bool) -> dict[str, str]:
# Deliver-To is what Rspamd keys per-user settings and per-user Bayes on, so
# one user's training never moves another user's scores (§C.5, acceptance 4).
return {
"Deliver-To": user_key,
"Settings-ID": SETTINGS_NETWORK if network_checks else SETTINGS_OFFLINE,
}
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:
resp = await client.post(
f"{RSPAMD_URL}/checkv2",
content=raw,
headers=_headers(user_key, network_checks),
)
resp.raise_for_status()
return _parse(resp.json())
except Exception as exc: # noqa: BLE001 - fail open, whatever went wrong
logger.warning("rspamd check unavailable (%s); delivering unclassified", exc)
return None
def _controller_headers(user_key: str) -> dict[str, str]:
headers = {"Deliver-To": user_key}
if CONTROLLER_PASSWORD:
headers["Password"] = CONTROLLER_PASSWORD
return headers
async def learn(raw: bytes, *, user_key: str, spam: bool) -> bool:
"""Teach this user's Bayes. Returns False if the engine was unreachable.
Goes to the CONTROLLER with the password, unlike :func:`check`. Used by the
C2 learning loop.
"""
endpoint = "/learnspam" if spam else "/learnham"
try:
async with httpx.AsyncClient(timeout=TIMEOUT) as client:
resp = await client.post(
f"{RSPAMD_CONTROLLER_URL}{endpoint}",
content=raw,
headers=_controller_headers(user_key),
)
resp.raise_for_status()
return True
except Exception as exc: # noqa: BLE001
logger.warning("rspamd learn failed (%s)", exc)
return False
async def stats(*, user_key: str) -> dict[str, Any] | None:
"""Controller ``/stat`` — feeds the §C.6 stats page. None when unreachable."""
try:
async with httpx.AsyncClient(timeout=TIMEOUT) as client:
resp = await client.get(
f"{RSPAMD_CONTROLLER_URL}/stat", headers=_controller_headers(user_key)
)
resp.raise_for_status()
data = resp.json()
return data if isinstance(data, dict) else None
except Exception as exc: # noqa: BLE001
logger.warning("rspamd stat unavailable (%s)", exc)
return None