lukasandClaude Opus 5
fe471dbf01
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
95 lines
3.3 KiB
Python
95 lines
3.3 KiB
Python
"""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
|