diff --git a/backend/app/accounts/schemas.py b/backend/app/accounts/schemas.py index 24bdce2..ea3a7f0 100644 --- a/backend/app/accounts/schemas.py +++ b/backend/app/accounts/schemas.py @@ -102,3 +102,4 @@ class AccountDetailOut(BaseModel): backfill_done: int imap_config: dict[str, object] smtp_config: dict[str, object] + kind: str = "external" diff --git a/backend/app/api/accounts.py b/backend/app/api/accounts.py index 1bea548..c3d3e49 100644 --- a/backend/app/api/accounts.py +++ b/backend/app/api/accounts.py @@ -104,6 +104,7 @@ async def list_accounts(user: CurrentUser, db: DbSession) -> list[AccountOut]: sync_error=a.sync_error, sync_error_since=a.sync_error_since.isoformat() if a.sync_error_since else None, state_detail=a.state_detail, + kind=a.kind, ) for a in accounts ] @@ -225,6 +226,8 @@ async def update_account( # re-encrypt the secret and re-sync. The password must be supplied (§10.2: the # stored secret is only decrypted in the worker, never here). conn_change = body.imap is not None or body.smtp is not None or body.password is not None + if conn_change and account.kind == "native": + raise_api_error(400, "native_account", "the settings of your own mailbox cannot be changed") if conn_change: if not body.password: raise_api_error( @@ -267,10 +270,20 @@ async def update_account( @router.delete("/{account_id}", dependencies=[Depends(require_csrf)]) async def remove_account(account_id: uuid.UUID, user: CurrentUser, db: DbSession) -> dict[str, str]: + kind = await db.scalar( + select(Account.kind).where(Account.id == account_id, Account.user_id == user.id) + ) + if kind == "native": + raise_api_error(409, "native_account", "your own mailbox cannot be removed") ok = await delete_account(db, user_id=user.id, account_id=account_id) if not ok: raise_api_error(404, "not_found", "account not found") await db.commit() + # Removing an account for the user's own address (the mail at its old + # provider) makes room for their hosted mailbox's (App. D §D.4). + from app.hosted import native + + await native.ensure_and_sync(db, user) return {"status": "deleted"} @@ -288,4 +301,5 @@ def _detail(account: Account) -> AccountDetailOut: backfill_done=account.backfill_done, imap_config=account.imap_config, smtp_config=account.smtp_config, + kind=account.kind, ) diff --git a/backend/app/api/admin.py b/backend/app/api/admin.py index 7e143ac..2535eb9 100644 --- a/backend/app/api/admin.py +++ b/backend/app/api/admin.py @@ -19,6 +19,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.api.errors import raise_api_error from app.db import get_session +from app.hosted.admin import MailServerError from app.identity.deps import CurrentAdmin, require_csrf from app.identity.mailer import LOGO_HTML, MailerError, mailer_config, send_mail from app.identity.schemas import ( @@ -183,7 +184,12 @@ async def patch_user( if body.quota_bytes is not None: user.quota_bytes = body.quota_bytes if body.blocked is not None: - await set_blocked(db, user, body.blocked) + try: + await set_blocked(db, user, body.blocked) + except MailServerError as exc: + raise_api_error( + 502, "mail_server_unavailable", f"the mail server did not answer: {exc}" + ) await db.commit() used = await db.scalar( select(func.coalesce(func.sum(Message.size_bytes), 0)).where(Message.user_id == user.id) @@ -200,5 +206,8 @@ async def patch_user( async def remove_user(user_id: uuid.UUID, admin: CurrentAdmin, db: DbSession) -> dict[str, str]: """Delete a user and all of their data. Cannot be undone.""" user = await _target(db, admin, user_id) - await delete_user(db, user) + try: + await delete_user(db, user) + except MailServerError as exc: + raise_api_error(502, "mail_server_unavailable", f"the mail server did not answer: {exc}") return {"status": "deleted"} diff --git a/backend/app/api/compose.py b/backend/app/api/compose.py index 71c4f81..ae983dd 100644 --- a/backend/app/api/compose.py +++ b/backend/app/api/compose.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import contextlib import uuid from datetime import UTC, datetime, timedelta from typing import Annotated @@ -17,7 +18,7 @@ from app.config import get_settings from app.db import get_session from app.identity.deps import CurrentUser, require_csrf from app.indexer import attachments as att_store -from app.indexer import blobstore +from app.indexer import blobstore, rawstore from app.models import Attachment, BodyPart, ComposeAttachment, Draft, Mailbox, Message, Outbox from app.outbox.jobs import RETRY_WINDOW from app.outbox.schemas import ( @@ -208,6 +209,9 @@ async def adopt_remote_draft(message_id: uuid.UUID, user: CurrentUser, db: DbSes ) ) attachment_ids: list[str] = [] + if att_rows: + with contextlib.suppress(FileNotFoundError): # fetched if a hosted mailbox's + await rawstore.ensure(msg) for a in att_rows: # Kept only inside the message: give the compose attachment a blob of its own. if not await asyncio.to_thread( diff --git a/backend/app/api/messages.py b/backend/app/api/messages.py index 55c8916..a8c863b 100644 --- a/backend/app/api/messages.py +++ b/backend/app/api/messages.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import contextlib import html as html_lib import io import uuid @@ -21,7 +22,7 @@ from app.api.settings import add_allowed_image_sender, allowed_image_senders from app.db import get_session from app.identity.deps import CurrentUser, require_csrf from app.indexer import attachments as att_store -from app.indexer import blobstore, thumbs +from app.indexer import rawstore, thumbs from app.indexer.mime import parse_message from app.indexer.sanitize import SANITIZER_VERSION, sanitize_html from app.indexer.thumbs import thumb_type @@ -128,7 +129,7 @@ async def message_html( # If the raw blob is unavailable, fall back to whatever is already stored. if allow_remote or stale: try: - raw = blobstore.load(user.id, message.blob_sha256) + raw = await rawstore.ensure_bytes(message) except FileNotFoundError: raw = None if raw is not None: @@ -191,8 +192,16 @@ def _content_disposition(disposition: str, filename: str) -> str: return f"{disposition}; filename=\"{fallback}\"; filename*=UTF-8''{quote(filename)}" +async def _raw_ready(msg: Message) -> None: + """Have the message's raw source here (a hosted mailbox's is fetched from + the mail server when not cached); a miss is left to the reader to report.""" + with contextlib.suppress(FileNotFoundError): + await rawstore.ensure(msg) + + async def _attachment_bytes(msg: Message, att: Attachment) -> bytes: """An attachment's bytes, taken out of its message (404 if they cannot be).""" + await _raw_ready(msg) try: return await asyncio.to_thread( att_store.load, msg.user_id, msg.blob_sha256, att.blob_sha256, att.part_index @@ -236,6 +245,7 @@ async def forward_attachments( of its own (shared, by hash, with any other use of the same file). Returns them as the composer lists its attachments.""" msg = await _owned_message(db, user.id, message_id) + await _raw_ready(msg) rows = [ ComposeAttachment( user_id=user.id, @@ -315,6 +325,8 @@ async def attachment_thumbnail( kind = thumb_type(att.mime_type, att.filename) if att is not None else None if att is None or kind is None: raise_api_error(404, "not_found", "no thumbnail") + if not thumbs.made(msg.user_id, att.blob_sha256): + await _raw_ready(msg) data = await asyncio.to_thread( thumbs.thumbnail, msg.user_id, @@ -350,6 +362,8 @@ async def message_attachment( ) if att is None: raise_api_error(404, "not_found", "attachment not found") + if not att_store.has_file(msg.user_id, att.blob_sha256): + await _raw_ready(msg) try: # Taken out of the message into a short-lived file (kept a day after # its last use), so it can be streamed with HTTP Range like before. diff --git a/backend/app/api/schemas.py b/backend/app/api/schemas.py index 827e861..68a0f38 100644 --- a/backend/app/api/schemas.py +++ b/backend/app/api/schemas.py @@ -102,6 +102,8 @@ class AccountOut(BaseModel): sync_error_since: str | None = None # Why the state is what it is (e.g. the server's reply to a refused sign-in). state_detail: str = "" + # "native": the user's own hosted mailbox — cannot be removed (App. D §D.4). + kind: str = "external" class SearchThreadHit(BaseModel): diff --git a/backend/app/config.py b/backend/app/config.py index a8654c5..4105a0c 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -74,6 +74,33 @@ class Settings(BaseSettings): # Auto-purge Trash/Spam messages older than this many days (0 disables). trash_retention_days: int = Field(default=30, alias="TRASH_RETENTION_DAYS") + # --- Hosted mailboxes (SPEC Appendix D, app/hosted/) ----------------------- + # Off: no mail server, byte-identical to before (every account external). + hosted_mailboxes: bool = Field(default=False, alias="HOSTED_MAILBOXES") + # The mail server's management API (JMAP), signed in to as its recovery admin. + stalwart_url: str = Field(default="http://stalwart:8080", alias="STALWART_URL") + stalwart_admin_secret: str = Field(default="", alias="STALWART_ADMIN_SECRET") + # The webmail worker's own sign-in, which may act for any mailbox + # (``user%worker``): no user's password is ever stored (§D.3). + stalwart_master_secret: str = Field(default="", alias="STALWART_MASTER_SECRET") + # The mail server's role on Postgres (its database, and the directory view). + stalwart_db_password: str = Field(default="", alias="STALWART_DB_PASSWORD") + stalwart_db_name: str = Field(default="stalwart", alias="STALWART_DB_NAME") + # Where the mail server reaches Postgres (its own network view of it). + stalwart_pg_host: str = Field(default="postgres", alias="STALWART_PG_HOST") + # IMAP/SMTP as the webmail reaches them (TLS, certificate not checked: the + # traffic stays on the Docker network or loopback). + stalwart_imap_host: str = Field(default="stalwart", alias="STALWART_IMAP_HOST") + stalwart_imap_port: int = Field(default=993, alias="STALWART_IMAP_PORT") + stalwart_smtp_host: str = Field(default="stalwart", alias="STALWART_SMTP_HOST") + stalwart_smtp_port: int = Field(default=465, alias="STALWART_SMTP_PORT") + # The mail server's public name (SMTP greeting, reverse DNS, what mail apps + # are told to connect to). Default: mail.APP_DOMAIN. + mail_hostname: str = Field(default="", alias="MAIL_HOSTNAME") + # Raw messages of hosted mailboxes are kept only in a cache (§D.6 item 2); + # beyond this per user, the least recently used go first. + raw_cache_mb: int = Field(default=256, alias="RAW_CACHE_MB") + @lru_cache(maxsize=1) def get_settings() -> Settings: diff --git a/backend/app/hosted/__init__.py b/backend/app/hosted/__init__.py new file mode 100644 index 0000000..82aed81 --- /dev/null +++ b/backend/app/hosted/__init__.py @@ -0,0 +1,34 @@ +"""Hosted mailboxes (SPEC Appendix D): every user's own mailbox on the Stalwart +mail server beside the app. + +* :mod:`app.hosted.admin` — the mail server's management API (JMAP); +* :mod:`app.hosted.provision` — idempotent setup: its database and role, the + directory over the ``mail_directory`` view, the worker's sign-in; +* :mod:`app.hosted.native` — the permanent account in the webmail, and how the + worker signs in to it without any stored password; +* :mod:`app.hosted.rawcache` — raw messages kept once, on the mail server: the + webmail keeps only a bounded cache of them (§D.6 item 2). + +All of it is inert while ``HOSTED_MAILBOXES`` is off. +""" + +from __future__ import annotations + +from app.config import get_settings + +# The webmail worker's sign-in on the mail server: ``user@domain%WORKER`` acts +# for that user (Stalwart's master-user login, given the impersonate role only). +WORKER_LOCALPART = "webmail-worker" + + +def enabled() -> bool: + return get_settings().hosted_mailboxes + + +def worker_address() -> str: + return f"{WORKER_LOCALPART}@{get_settings().app_domain}" + + +def mail_hostname() -> str: + settings = get_settings() + return settings.mail_hostname or f"mail.{settings.app_domain}" diff --git a/backend/app/hosted/admin.py b/backend/app/hosted/admin.py new file mode 100644 index 0000000..712f8fe --- /dev/null +++ b/backend/app/hosted/admin.py @@ -0,0 +1,177 @@ +"""The mail server's management API (Stalwart 0.16, JMAP ``x:`` methods). + +Signed in to as the recovery admin (``STALWART_RECOVERY_ADMIN`` in the mail +server's environment, ``STALWART_ADMIN_SECRET`` here). Objects are Stalwart's +registry: ``x:Account``, ``x:Domain``, ``x:Directory``, ``x:Role``, +``x:AppPassword``… with the usual JMAP ``/get``, ``/query``, ``/set``. +""" + +from __future__ import annotations + +import base64 +from typing import Any + +import httpx + +from app.config import get_settings +from app.hosted import worker_address + +ADMIN_USER = "admin" +_USING = ["urn:ietf:params:jmap:core", "urn:stalwart:jmap"] +_TIMEOUT = 15.0 + + +class MailServerError(Exception): + """The mail server refused a call, or could not be reached.""" + + +def _basic(user: str, secret: str) -> str: + return "Basic " + base64.b64encode(f"{user}:{secret}".encode()).decode() + + +def _admin_auth() -> str: + return _basic(ADMIN_USER, get_settings().stalwart_admin_secret) + + +async def call( + *method_calls: tuple[str, dict[str, Any]], auth: str | None = None +) -> list[dict[str, Any]]: + """Run JMAP method calls in one request; return each call's arguments. + + Raises :class:`MailServerError` on a method error or a refused ``set``.""" + body = { + "using": _USING, + "methodCalls": [[name, args, str(i)] for i, (name, args) in enumerate(method_calls)], + } + try: + async with httpx.AsyncClient(timeout=_TIMEOUT) as client: + resp = await client.post( + f"{get_settings().stalwart_url}/jmap", + json=body, + headers={"Authorization": auth or _admin_auth()}, + ) + except httpx.HTTPError as exc: + raise MailServerError(f"mail server unreachable: {exc}") from exc + if resp.status_code != 200: + raise MailServerError(f"mail server answered {resp.status_code}: {resp.text[:300]}") + out: list[dict[str, Any]] = [] + for name, args, _ in resp.json()["methodResponses"]: + if name == "error": + raise MailServerError(f"{args.get('type')}: {args.get('description', '')}") + for kind in ("notCreated", "notUpdated", "notDestroyed"): + if args.get(kind): + raise MailServerError(f"{name} {kind}: {args[kind]}") + out.append(args) + return out + + +async def get_all(obj: str, **args: Any) -> list[dict[str, Any]]: + """Every object of a type (registry objects are few).""" + (res,) = await call((f"x:{obj}/get", {"ids": None, **args})) + return list(res["list"]) + + +async def get_one(obj: str, object_id: str = "singleton") -> dict[str, Any]: + (res,) = await call((f"x:{obj}/get", {"ids": [object_id]})) + if not res["list"]: + raise MailServerError(f"{obj} {object_id} not found") + return dict(res["list"][0]) + + +async def create(obj: str, value: dict[str, Any], **args: Any) -> str: + (res,) = await call((f"x:{obj}/set", {"create": {"n": value}, **args})) + return str(res["created"]["n"]["id"]) + + +async def update(obj: str, object_id: str, patch: dict[str, Any], **args: Any) -> None: + await call((f"x:{obj}/set", {"update": {object_id: patch}, **args})) + + +async def destroy(obj: str, ids: list[str], **args: Any) -> None: + if ids: + await call((f"x:{obj}/set", {"destroy": ids, **args})) + + +# ---- accounts ---------------------------------------------------------------------- + + +async def account_id(email: str) -> str | None: + """The mail server's id of a mailbox, if it knows it yet. + + Accounts from the directory reach the server's registry the first time + they are used (a sign-in, a delivery): see :func:`ensure_account`.""" + local = email.split("@", 1)[0].lower() + (res,) = await call(("x:Account/query", {"filter": {"name": local}})) + ids = list(res.get("ids") or []) + if not ids: + return None + for acc in (await call(("x:Account/get", {"ids": ids, "properties": ["emailAddress"]})))[0][ + "list" + ]: + if str(acc.get("emailAddress", "")).lower() == email.lower(): + return str(acc["id"]) + return None + + +async def ensure_account(email: str) -> str: + """The mailbox's id, bringing it into the server's registry if new: a + sign-in as the worker acting for it does that (§D.3).""" + found = await account_id(email) + if found is not None: + return found + auth = _basic(f"{email}%{worker_address()}", get_settings().stalwart_master_secret) + try: + async with httpx.AsyncClient(timeout=_TIMEOUT) as client: + resp = await client.get( + f"{get_settings().stalwart_url}/jmap/session", headers={"Authorization": auth} + ) + except httpx.HTTPError as exc: + raise MailServerError(f"mail server unreachable: {exc}") from exc + if resp.status_code != 200: + raise MailServerError(f"cannot open the mailbox {email}: {resp.status_code}") + found = await account_id(email) + if found is None: + raise MailServerError(f"mailbox {email} not found after sign-in") + return found + + +async def set_blocked(email: str, blocked: bool) -> None: + """A blocked user's mailbox cannot be signed in to at all — app passwords + and the worker included. Mail to it is still accepted.""" + acc = await account_id(email) + if acc is None: + return # never used: the directory alone keeps it shut + permissions: dict[str, Any] = ( + {"@type": "Merge", "enabledPermissions": {}, "disabledPermissions": {"authenticate": True}} + if blocked + else {"@type": "Inherit"} + ) + await update("Account", acc, {"permissions": permissions}) + + +async def delete_account(email: str) -> None: + """Remove the mailbox and all its mail from the mail server.""" + acc = await account_id(email) + if acc is not None: + await destroy("Account", [acc]) + + +# ---- app passwords (§D.5) ------------------------------------------------------------- + + +async def create_app_password(email: str, description: str) -> tuple[str, str]: + """A new app password for the mailbox: (server id, the password itself).""" + acc = await ensure_account(email) + (res,) = await call( + ("x:AppPassword/set", {"accountId": acc, "create": {"n": {"description": description}}}) + ) + created = res["created"]["n"] + return str(created["id"]), str(created["secret"]) + + +async def delete_app_passwords(email: str, server_ids: list[str]) -> None: + acc = await account_id(email) + if acc is not None and server_ids: + (res,) = await call(("x:AppPassword/get", {"accountId": acc, "ids": server_ids})) + present = [str(p["id"]) for p in res["list"]] + await destroy("AppPassword", present, accountId=acc) diff --git a/backend/app/hosted/native.py b/backend/app/hosted/native.py new file mode 100644 index 0000000..af0ff8f --- /dev/null +++ b/backend/app/hosted/native.py @@ -0,0 +1,154 @@ +"""The user's own mailbox as a webmail account (SPEC App. D §D.4). + +Every user has one ``kind = 'native'`` account: their handle's mailbox on the +hosted mail server. It is created with the user (and, for users from before, +by :func:`ensure_all`, run by provisioning), cannot be removed, and its server +settings are not the user's to change. + +No password of the user's is stored for it: the worker signs in as itself, +acting for the user (``user@domain%webmail-worker``, §D.3), over the Docker +network (or loopback in dev) — TLS, with the certificate left unchecked since +the traffic never leaves the host. +""" + +from __future__ import annotations + +import logging +import ssl +import uuid + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.config import get_settings +from app.hosted import enabled, worker_address +from app.models import Account, User +from app.providers.imap_smtp import ProviderConfig +from app.security.crypto import encrypt_secret + +logger = logging.getLogger(__name__) + +KIND = "native" + + +def is_native(account: Account) -> bool: + return account.kind == KIND + + +def _server(host: str, port: int) -> dict[str, object]: + return {"host": host, "port": port, "tls": "ssl", "auth": "password"} + + +def internal_tls() -> ssl.SSLContext: + """TLS to the mail server on the internal network: its certificate is for + its public name (or self-signed), not the container's.""" + ctx = ssl.create_default_context() + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + return ctx + + +def native_provider_config(email: str) -> ProviderConfig: + """IMAP/SMTP to the user's mailbox, signed in as the worker acting for them. + + Read from the settings, not the account row: the mail server's address is + the deployment's, and may change.""" + settings = get_settings() + return ProviderConfig( + imap_host=settings.stalwart_imap_host, + imap_port=settings.stalwart_imap_port, + imap_tls="ssl", + smtp_host=settings.stalwart_smtp_host, + smtp_port=settings.stalwart_smtp_port, + smtp_tls="ssl", + username=f"{email}%{worker_address()}", + password=settings.stalwart_master_secret, + tls_context=internal_tls(), + internal=True, + ) + + +async def native_account(db: AsyncSession, user_id: uuid.UUID) -> Account | None: + account: Account | None = await db.scalar( + select(Account).where(Account.user_id == user_id, Account.kind == KIND) + ) + return account + + +async def ensure(db: AsyncSession, user: User) -> Account | None: + """The user's native account, created (and its first sync queued by the + caller, see :func:`ensure_and_sync`) if missing. None when hosted + mailboxes are off. Flushes, does not commit.""" + if not enabled(): + return None + existing = await native_account(db, user.id) + if existing is not None: + return existing + # An account for the same address added by hand: the same address at + # another provider (the mail before it moved here — e.g. while the domain + # still points elsewhere). It stays as it is; the mailbox account is added + # once it is removed (app/api/accounts.py). + same = await db.scalar( + select(Account.id).where(Account.user_id == user.id, Account.email == user.handle) + ) + if same is not None: + logger.info("hosted: %s already has an account for its address; not added", user.handle) + return None + settings = get_settings() + imap = _server(settings.stalwart_imap_host, settings.stalwart_imap_port) + smtp = _server(settings.stalwart_smtp_host, settings.stalwart_smtp_port) + account = Account( + user_id=user.id, + email=user.handle, + kind=KIND, + imap_config=imap, + smtp_config=smtp, + secret_ct=encrypt_secret(b""), # nothing: the worker signs in as itself + discovery_src="hosted", + state="backfilling", + sync_window="all", + # The mail server's own spam filter files into Junk: show it. + sync_spam=True, + ) + db.add(account) + await db.flush() + return account + + +async def ensure_and_sync(db: AsyncSession, user: User) -> Account | None: + """:func:`ensure`, committed, and the account's first sync queued if new.""" + before = await native_account(db, user.id) + account = await ensure(db, user) + await db.commit() + if account is not None and before is None: + from app.queue import enqueue_job + + await enqueue_job("sync_bootstrap_account", str(account.id)) + return account + + +async def ensure_all() -> int: + """Give every user without one their native account (users from before + hosted mailboxes). Returns how many were created.""" + from app.db import session_scope + + created = 0 + async with session_scope() as db: + users = list( + await db.scalars( + select(User).where( + ~select(Account.id) + .where(Account.user_id == User.id, Account.kind == KIND) + .exists() + ) + ) + ) + for user in users: + async with session_scope() as db: + fresh = await db.get(User, user.id) + if fresh is None: + continue + await ensure_and_sync(db, fresh) + created += 1 + logger.info("hosted: mailbox account for %s", fresh.handle) + return created diff --git a/backend/app/hosted/provision.py b/backend/app/hosted/provision.py new file mode 100644 index 0000000..ba04afc --- /dev/null +++ b/backend/app/hosted/provision.py @@ -0,0 +1,323 @@ +"""Set up the mail server for hosted mailboxes (SPEC App. D §D.2/§D.3). Idempotent. + +Run after the migrations (the ``migrate`` service in production; in dev +``python -m app.hosted.provision``). Two halves: + +1. **Postgres** (as the app's own, privileged database user): the mail + server's role ``stalwart`` (password ``STALWART_DB_PASSWORD``), which owns its + database and may read the ``mail_directory`` view of the app's database and + nothing else there; and the worker's sign-in in ``mail_service_accounts``. + The mail server cannot start before this has run once. +2. **The mail server** (its management API, as the recovery admin): the mail + domain and host name, mail kept on disk (not in Postgres), the SQL directory + over ``mail_directory`` as the only source of accounts, the worker's + impersonate-only role, the submission port for mail apps. +""" + +from __future__ import annotations + +import asyncio +import logging +import re +from typing import Any +from urllib.parse import urlparse + +import asyncpg # type: ignore[import-untyped] + +from app.config import get_settings +from app.hosted import admin, mail_hostname, worker_address +from app.identity.passwords import hash_password, verify_password + +logger = logging.getLogger(__name__) + +DB_ROLE = "stalwart" +DIRECTORY_NAME = "Ultimo webmail users" +WORKER_ROLE_NAME = "Ultimo webmail worker" +BLOB_PATH = "/var/lib/stalwart/blobs" + +# The directory's queries ($1: the address signed in with / mail is for). A +# blocked user cannot sign in, but still receives mail. +QUERY_LOGIN = ( + "SELECT name, secret, type, description FROM mail_directory WHERE name = $1 AND active" +) +QUERY_RECIPIENT = ( + "SELECT name, NULL::text AS secret, type, description FROM mail_directory WHERE name = $1" +) + + +def _pg_dsn() -> str: + return get_settings().database_url.replace("postgresql+asyncpg://", "postgresql://", 1) + + +def _app_db_name() -> str: + return urlparse(_pg_dsn()).path.lstrip("/") + + +def _quote_literal(value: str) -> str: + return "'" + value.replace("'", "''") + "'" + + +async def provision_database() -> None: + settings = get_settings() + if not settings.stalwart_db_password or not settings.stalwart_master_secret: + raise RuntimeError("STALWART_DB_PASSWORD and STALWART_MASTER_SECRET must be set") + conn = await asyncpg.connect(_pg_dsn()) + try: + password = _quote_literal(settings.stalwart_db_password) + if await conn.fetchval("SELECT 1 FROM pg_roles WHERE rolname = $1", DB_ROLE): + await conn.execute(f"ALTER ROLE {DB_ROLE} WITH LOGIN PASSWORD {password}") + else: + await conn.execute(f"CREATE ROLE {DB_ROLE} WITH LOGIN PASSWORD {password}") + db = settings.stalwart_db_name + if not await conn.fetchval("SELECT 1 FROM pg_database WHERE datname = $1", db): + await conn.execute(f'CREATE DATABASE "{db}" OWNER {DB_ROLE}') + logger.info("hosted: created database %s", db) + await conn.execute(f"GRANT SELECT ON mail_directory TO {DB_ROLE}") + + # The worker's sign-in: re-hashed only when the secret changed. + name = worker_address() + current = await conn.fetchval( + "SELECT secret FROM mail_service_accounts WHERE name = $1", name + ) + if current is None or not verify_password(current, settings.stalwart_master_secret): + await conn.execute( + """ + INSERT INTO mail_service_accounts (name, secret, description) + VALUES ($1, $2, 'Webmail worker') + ON CONFLICT (name) DO UPDATE SET secret = EXCLUDED.secret + """, + name, + hash_password(settings.stalwart_master_secret), + ) + finally: + await conn.close() + + +# ---- the mail server ------------------------------------------------------------------ + + +async def _wait_for_server(timeout_s: float = 120.0) -> None: + deadline = asyncio.get_running_loop().time() + timeout_s + while True: + try: + await admin.get_one("SystemSettings") + return + except admin.MailServerError: + if asyncio.get_running_loop().time() > deadline: + raise + await asyncio.sleep(2) + + +def _by(items: list[dict[str, Any]], key: str, value: str) -> dict[str, Any] | None: + return next((i for i in items if i.get(key) == value), None) + + +async def _domain() -> str: + name = get_settings().app_domain.lower() + found = _by(await admin.get_all("Domain"), "name", name) + if found is not None: + return str(found["id"]) + domain_id = await admin.create("Domain", {"name": name}) + logger.info("hosted: created mail domain %s", name) + return domain_id + + +async def _directory() -> str: + settings = get_settings() + dsn = urlparse(_pg_dsn()) + value: dict[str, Any] = { + "@type": "Sql", + "description": DIRECTORY_NAME, + "columnEmail": "name", + "columnSecret": "secret", + "columnClass": "type", + "columnDescription": "description", + "queryLogin": QUERY_LOGIN, + "queryRecipient": QUERY_RECIPIENT, + "queryMemberOf": None, + "queryEmailAliases": None, + "store": { + "@type": "PostgreSql", + "host": settings.stalwart_pg_host, + "port": 5432 if settings.stalwart_pg_host == "postgres" else (dsn.port or 5432), + "database": _app_db_name(), + "authUsername": DB_ROLE, + "authSecret": {"@type": "EnvironmentVariable", "variableName": "STALWART_DB_PASSWORD"}, + "useTls": False, + "allowInvalidCerts": False, + "poolMaxConnections": 5, + "poolRecyclingMethod": "fast", + "readReplicas": {}, + "timeout": 15000, + "options": None, + }, + } + found = _by(await admin.get_all("Directory"), "description", DIRECTORY_NAME) + if found is None: + directory_id = await admin.create("Directory", value) + logger.info("hosted: created the SQL directory") + else: + directory_id = str(found["id"]) + await admin.update("Directory", directory_id, value) + await admin.update("Authentication", "singleton", {"directoryId": directory_id}) + return directory_id + + +async def _worker_role() -> str: + perms = {"impersonate": True, "authenticate": True} + found = _by(await admin.get_all("Role"), "description", WORKER_ROLE_NAME) + if found is not None: + role_id = str(found["id"]) + await admin.update("Role", role_id, {"enabledPermissions": perms}) + return role_id + return await admin.create( + "Role", {"description": WORKER_ROLE_NAME, "enabledPermissions": perms} + ) + + +async def _worker_account(role_id: str) -> None: + """The worker may act for any mailbox and do nothing as itself.""" + # Signing in brings it from the directory into the server's registry. With + # its role already set that sign-in fails (it may not open a session as + # itself) — after the account exists, which is all this needs. + worker = worker_address() + acc = await admin.account_id(worker) + if acc is None: + import httpx + + async with httpx.AsyncClient(timeout=15) as client: + await client.get( + f"{get_settings().stalwart_url}/jmap/session", + auth=(worker, get_settings().stalwart_master_secret), + ) + acc = await admin.account_id(worker) + if acc is None: + raise admin.MailServerError("the worker account did not appear on the mail server") + await admin.update("Account", acc, {"roles": {"@type": "Custom", "roleIds": {role_id: True}}}) + + +async def _reload() -> None: + """Apply the settings now. Stalwart 0.16.19 ships a few spam-list rules its + own validator rejects (``bit_and`` in SURBL tags); they never load, but + they make every reload fail — so they are removed, which changes nothing.""" + for _ in range(10): + try: + await admin.create("Action", {"@type": "ReloadSettings"}) + return + except admin.MailServerError as exc: + match = re.search(r"'object': '(SpamDnsblServer)', 'id': '(\w+)'", str(exc)) + if match is None: + raise + logger.info("hosted: removing the invalid built-in rule %s", match.group(2)) + await admin.destroy(match.group(1), [match.group(2)]) + raise admin.MailServerError("settings reload kept failing") + + +async def _logging() -> None: + """Log to the console (``docker logs``), not to files inside the container.""" + for tracer in await admin.get_all("Tracer"): + if tracer.get("@type") != "Stdout": + await admin.update( + "Tracer", + str(tracer["id"]), + { + "@type": "Stdout", + "ansi": False, + "buffered": False, + "enable": True, + "level": "info", + "events": {}, + "eventsPolicy": "exclude", + }, + ) + + +async def _blob_store() -> bool: + """Mail on disk (``DATA_ROOT/stalwart``), not inside Postgres: database + backups stay small. Takes a restart; True when one is needed.""" + current = await admin.get_one("BlobStore") + if current.get("@type") == "FileSystem" and current.get("path") == BLOB_PATH: + return False + await admin.update( + "BlobStore", "singleton", {"@type": "FileSystem", "path": BLOB_PATH, "depth": 2} + ) + logger.info("hosted: mail is now kept on disk (takes effect after a restart)") + return True + + +async def _listeners() -> None: + """Add submission with STARTTLS (587) beside the defaults: smtp 25, submissions + 465, imaps 993 (and http 8080 for this API).""" + listeners = await admin.get_all("NetworkListener") + if _by(listeners, "name", "submission") is None: + await admin.create( + "NetworkListener", + {"name": "submission", "bind": {"[::]:587": True}, "protocol": "smtp", "useTls": True}, + ) + + +# Folder names for new mailboxes: the plain ones mail apps expect, not +# Stalwart's Outlook-style defaults ("Deleted Items", "Junk Mail", "Sent +# Items"). Archive is made too: the webmail archives into it. +FOLDERS = { + "trash": "Trash", + "junk": "Junk", + "drafts": "Drafts", + "sent": "Sent", + "archive": "Archive", +} + + +async def _folders() -> None: + folders = { + use: {"name": name, "create": True, "subscribe": True, "aliases": {}} + for use, name in FOLDERS.items() + } + await admin.update("Email", "singleton", {"defaultFolders": folders}) + + +async def provision_server() -> bool: + """Set the mail server up; True when it must be restarted to finish.""" + await _wait_for_server() + await _logging() + restart = await _blob_store() + domain_id = await _domain() + await admin.update( + "SystemSettings", + "singleton", + {"defaultDomainId": domain_id, "defaultHostname": mail_hostname()}, + ) + await _listeners() + await _folders() + await _directory() + role_id = await _worker_role() + await _reload() + await _worker_account(role_id) + return restart + + +async def provision() -> bool: + from app.hosted import native + + await provision_database() + restart = await provision_server() + created = await native.ensure_all() + if created: + logger.info("hosted: added the mailbox account of %d existing user(s)", created) + return restart + + +def main() -> None: + logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") + logging.getLogger("httpx").setLevel(logging.WARNING) + if not get_settings().hosted_mailboxes: + logger.info("hosted mailboxes are off (HOSTED_MAILBOXES): nothing to set up") + return + if asyncio.run(provision()): + logger.warning("hosted mailboxes: RESTART the mail server (stalwart) to finish") + else: + logger.info("hosted mailboxes: mail server ready") + + +if __name__ == "__main__": + main() diff --git a/backend/app/identity/app_passwords.py b/backend/app/identity/app_passwords.py index 3a84071..51f1131 100644 --- a/backend/app/identity/app_passwords.py +++ b/backend/app/identity/app_passwords.py @@ -1,14 +1,18 @@ """App passwords: one generated password per mail app (SPEC App. D §D.5). A mail app (Thunderbird, a phone's Mail app) signs in to the hosted mailbox -over IMAP/SMTP, where no 2FA code can be asked for. It gets a password of its -own — 16 random letters, shown once — that can be revoked without touching the -main password or the other apps. The webmail itself never accepts one. +over IMAP/SMTP, and a calendar or contacts app to the DAV server — never with +the account password (no 2FA code can be asked for there), always with one of +these. Each gets a password of its own, shown once, that can be +revoked without touching the main password or the other apps. The webmail +itself never accepts one. -Stored as argon2id hashes (identity/passwords.py), the form the mail server's -SQL directory reads (§D.3). The hash is of the bare 16 letters: the mail -server compares exactly what an app sends, so the password is copied and -entered without the spaces it is shown with. +With hosted mailboxes on, the mail server makes the password (``app_`` and 36 +more letters and digits: its format names the credential) and checks it for +IMAP/SMTP itself; its argon2id hash is kept here too, for the DAV server, with +the server's id to revoke it there. Without, it is 16 random letters, for DAV +only. Either way the hash is of the bare password: apps send exactly that, so +it is copied and entered without the spaces it is shown with. """ from __future__ import annotations @@ -20,10 +24,12 @@ import uuid from sqlalchemy import delete, func, select from sqlalchemy.ext.asyncio import AsyncSession +from app import hosted from app.identity.passwords import hash_password, verify_password -from app.models import UserAppPassword +from app.models import User, UserAppPassword LENGTH = 16 # 26^16 ≈ 2^75 +SERVER_LENGTH = 40 # the mail server's: "app_" + base32 of a 4-byte id and 18 random bytes MAX_PER_USER = 25 NAME_MAX = 60 _ALPHABET = string.ascii_lowercase @@ -66,16 +72,45 @@ async def create(db: AsyncSession, user_id: uuid.UUID, name: str) -> tuple[UserA ) if (count or 0) >= MAX_PER_USER: raise TooMany - password = generate() + name = name.strip()[:NAME_MAX] + server_id: str | None = None + if hosted.enabled(): + from app.hosted import admin as mail_admin + + handle = await db.scalar(select(User.handle).where(User.id == user_id)) + server_id, password = await mail_admin.create_app_password(str(handle), name) + else: + password = generate() row = UserAppPassword( - user_id=user_id, name=name.strip()[:NAME_MAX], password_hash=hash_password(password) + user_id=user_id, name=name, password_hash=hash_password(password), server_id=server_id ) db.add(row) await db.commit() return row, password +async def _revoke_on_server( + db: AsyncSession, user_id: uuid.UUID, rows: list[UserAppPassword] +) -> None: + """Revoke on the mail server too (first: if it cannot be reached, the + password keeps working there, so nothing is revoked here either).""" + server_ids = [r.server_id for r in rows if r.server_id] + if server_ids and hosted.enabled(): + from app.hosted import admin as mail_admin + + handle = await db.scalar(select(User.handle).where(User.id == user_id)) + await mail_admin.delete_app_passwords(str(handle), server_ids) + + async def revoke(db: AsyncSession, user_id: uuid.UUID, app_password_id: uuid.UUID) -> bool: + rows = list( + await db.scalars( + select(UserAppPassword).where( + UserAppPassword.id == app_password_id, UserAppPassword.user_id == user_id + ) + ) + ) + await _revoke_on_server(db, user_id, rows) result = await db.execute( delete(UserAppPassword).where( UserAppPassword.id == app_password_id, UserAppPassword.user_id == user_id @@ -88,6 +123,7 @@ async def revoke(db: AsyncSession, user_id: uuid.UUID, app_password_id: uuid.UUI async def revoke_all(db: AsyncSession, user_id: uuid.UUID) -> None: """Every app signs in again (the main password was reset).""" + await _revoke_on_server(db, user_id, await list_for(db, user_id)) await db.execute(delete(UserAppPassword).where(UserAppPassword.user_id == user_id)) await db.commit() await _changed(user_id) @@ -96,7 +132,7 @@ async def revoke_all(db: AsyncSession, user_id: uuid.UUID) -> None: async def verify(db: AsyncSession, user_id: uuid.UUID, password: str) -> UserAppPassword | None: """The app password ``password`` matches, if any — for the mail server's sign-in and for tests; never for the webmail's own sign-in.""" - if len(password) != LENGTH: + if len(password) not in (LENGTH, SERVER_LENGTH): return None for row in await list_for(db, user_id): if verify_password(row.password_hash, password): diff --git a/backend/app/identity/routes.py b/backend/app/identity/routes.py index 1cbbeee..792f6b8 100644 --- a/backend/app/identity/routes.py +++ b/backend/app/identity/routes.py @@ -6,16 +6,18 @@ from __future__ import annotations import logging import re import uuid -from typing import Annotated +from typing import Annotated, NoReturn from fastapi import APIRouter, BackgroundTasks, Depends, Request, Response from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession +from app import hosted from app.api.errors import raise_api_error from app.config import get_settings from app.db import get_session from app.events import get_event_bus +from app.hosted.admin import MailServerError from app.identity import app_passwords, trusted_devices from app.identity import totp as totp_mod from app.identity.cookies import ( @@ -105,6 +107,7 @@ def _user_out(user: User) -> UserOut: settings=user.settings, totp_enabled=user.totp_secret is not None, is_admin=user.is_admin, + mail_host=hosted.mail_hostname() if hosted.enabled() else None, ) @@ -132,6 +135,12 @@ async def register( except RegistrationError as exc: raise_api_error(_REGISTRATION_ERROR_STATUS.get(exc.code, 400), exc.code, exc.message) await db.commit() + # The user's own mailbox (App. D §D.4): it exists on the mail server as soon + # as the user does (the directory reads the users table); this adds it to + # the webmail. + from app.hosted import native + + await native.ensure_and_sync(db, user) await _establish_session(request, response, user) return _user_out(user) @@ -316,7 +325,12 @@ async def password_reset(body: PasswordResetIn, db: DbSession) -> MessageOut: await db.commit() await destroy_all_sessions(user.id) # force re-login everywhere after reset # Mail apps sign in again too: the reset may be the owner taking the account back. - await app_passwords.revoke_all(db, user.id) + try: + await app_passwords.revoke_all(db, user.id) + except MailServerError: + # The new password is set; the app passwords stay listed (and working) + # until the mail server answers, so the user can still revoke them. + logger.exception("password reset: could not revoke app passwords on the mail server") # Whoever reset the password may be the owner taking the account back: no # device keeps skipping the code on the old trust. await trusted_devices.revoke_all(user.id) @@ -326,6 +340,12 @@ async def password_reset(body: PasswordResetIn, db: DbSession) -> MessageOut: # --- App passwords (SPEC App. D §D.5) ----------------------------------------- +def _mail_server_down() -> NoReturn: + raise_api_error( + 502, "mail_server_unavailable", "the mail server is not answering; try again shortly" + ) + + def _app_password_out(row: object) -> AppPasswordOut: from app.models import UserAppPassword @@ -361,6 +381,8 @@ async def create_app_password( raise_api_error( 409, "too_many", f"you can have up to {app_passwords.MAX_PER_USER} app passwords" ) + except MailServerError: + _mail_server_down() return AppPasswordCreatedOut(**_app_password_out(row).model_dump(), app_password=password) @@ -372,7 +394,11 @@ async def create_app_password( async def revoke_app_password( app_password_id: uuid.UUID, user: CurrentUser, db: DbSession ) -> MessageOut: - if not await app_passwords.revoke(db, user.id, app_password_id): + try: + revoked = await app_passwords.revoke(db, user.id, app_password_id) + except MailServerError: + _mail_server_down() + if not revoked: raise_api_error(404, "not_found", "app password not found") return MessageOut() diff --git a/backend/app/identity/schemas.py b/backend/app/identity/schemas.py index 7af1bb6..9fc9763 100644 --- a/backend/app/identity/schemas.py +++ b/backend/app/identity/schemas.py @@ -93,6 +93,9 @@ class UserOut(BaseModel): settings: dict[str, object] totp_enabled: bool = False is_admin: bool = False + # The hosted mail server's name for mail apps (IMAP 993, SMTP 465/587), or + # None when there are no hosted mailboxes (SPEC App. D). + mail_host: str | None = None class RegistrationStatusOut(BaseModel): diff --git a/backend/app/identity/user_admin.py b/backend/app/identity/user_admin.py index f832958..19c18fd 100644 --- a/backend/app/identity/user_admin.py +++ b/backend/app/identity/user_admin.py @@ -67,7 +67,16 @@ async def search_users(db: AsyncSession, q: str) -> list[tuple[User, int, list[s async def set_blocked(db: AsyncSession, user: User, blocked: bool) -> None: - """Block or unblock. Blocking also signs the user out everywhere.""" + """Block or unblock. Blocking also signs the user out everywhere, and shuts + their hosted mailbox to every sign-in (mail apps included); it still + receives mail. The mail server goes first: if it cannot be told, nothing + changes.""" + from app import hosted + + if hosted.enabled(): + from app.hosted import admin as mail_admin + + await mail_admin.set_blocked(user.handle, blocked) if blocked and user.blocked_at is None: user.blocked_at = dt.datetime.now(dt.UTC) await db.flush() @@ -97,8 +106,18 @@ async def _plugin_tables(db: AsyncSession) -> list[str]: async def delete_user(db: AsyncSession, user: User) -> None: - """Delete ``user`` and everything they own. Commits.""" + """Delete ``user`` and everything they own. Commits. + + Their hosted mailbox goes first, with all its mail: if the mail server + cannot be reached nothing is deleted — a mailbox left behind would be + handed, mail and all, to whoever registers the address next.""" + from app import hosted + user_id = user.id + if hosted.enabled(): + from app.hosted import admin as mail_admin + + await mail_admin.delete_account(user.handle) # Mail accounts first, one by one: this is what stops their sync runners and # tidies threads, exactly as when the user removes an account themselves. diff --git a/backend/app/indexer/attachments.py b/backend/app/indexer/attachments.py index ae3f812..11573c5 100644 --- a/backend/app/indexer/attachments.py +++ b/backend/app/indexer/attachments.py @@ -10,7 +10,9 @@ decoded on its own). It is taken out of the raw message when wanted: * :func:`dedupe_existing` — the one-time pass removing the separate copies stored before, each only after checking it can be taken out of its message. -Blocking (parsing, disk): call these in a thread. +Blocking (parsing, disk): call these in a thread — after +:func:`app.indexer.rawstore.ensure` for the message, which fetches a hosted +mailbox's message from its mail server when it is not cached. """ from __future__ import annotations @@ -23,7 +25,7 @@ import uuid from collections import defaultdict from pathlib import Path -from app.indexer import blobstore +from app.indexer import blobstore, rawstore from app.indexer.mime import load_message, part_payload logger = logging.getLogger(__name__) @@ -45,8 +47,8 @@ def load(user_id: uuid.UUID, message_sha: str, sha: str, part_index: int | None) legacy = _legacy(user_id, sha) if legacy is not None: return legacy.read_bytes() - raw = blobstore.blob_path(user_id, message_sha) - if part_index == NOT_IN_MESSAGE or not raw.exists(): + raw = rawstore.locate(user_id, message_sha) + if part_index == NOT_IN_MESSAGE or raw is None: raise FileNotFoundError(sha) payload = part_payload(load_message(path=raw), part_index, sha) if payload is None: @@ -82,6 +84,11 @@ def prune_cache(cache: Path) -> None: path.unlink(missing_ok=True) +def has_file(user_id: uuid.UUID, sha: str) -> bool: + """Whether :func:`file` can answer without reading the message.""" + return _legacy(user_id, sha) is not None or (_cache_dir(user_id) / sha).exists() + + def file(user_id: uuid.UUID, message_sha: str, sha: str, part_index: int | None) -> Path: """A file holding the attachment, to stream from. Raises FileNotFoundError.""" legacy = _legacy(user_id, sha) @@ -167,9 +174,9 @@ async def dedupe_existing(batch: int = 200, *, user_id: uuid.UUID | None = None) located: dict[uuid.UUID, int] = {} for (owner, msg_sha), atts in by_message.items(): - raw = blobstore.blob_path(owner, msg_sha) + raw = rawstore.locate(owner, msg_sha) try: - indexes = await asyncio.to_thread(_part_indexes, raw) if raw.exists() else {} + indexes = await asyncio.to_thread(_part_indexes, raw) if raw is not None else {} except Exception: # noqa: BLE001 - unreadable message: keep the copies logger.warning("attachments: cannot parse %s", raw, exc_info=True) indexes = {} diff --git a/backend/app/indexer/gc.py b/backend/app/indexer/gc.py index f8f2b6e..faaaf17 100644 --- a/backend/app/indexer/gc.py +++ b/backend/app/indexer/gc.py @@ -11,7 +11,12 @@ that for every user, whatever the way the rows went: a sent message (its draft is gone) or a composer closed without saving; * message/attachment files no row refers to; * attachment previews (thumbnails) of attachments that are gone; -* streaming copies past their keep time (app/indexer/attachments.py). +* streaming copies past their keep time (app/indexer/attachments.py); +* cached raw messages of a hosted mailbox beyond the cache's room + (app/indexer/rawstore.py). + +It also recounts threads whose unread/message counters disagree with their +messages, and drops threads left empty (app/threading_/threader.py). A file younger than an hour is left alone: it may belong to work in progress (a message being stored, a sent message waiting for its Sent copy). @@ -31,6 +36,7 @@ from sqlalchemy import delete, or_, select from app.config import get_settings from app.indexer import attachments as att_store +from app.indexer import rawstore logger = logging.getLogger(__name__) @@ -94,9 +100,9 @@ def _old(path: Path, now: float) -> bool: async def sweep_user(user_id: uuid.UUID) -> dict[str, int]: """Clean one user's stored files; returns what was removed.""" from app.db import session_scope - from app.models import Attachment, ComposeAttachment, Draft, Message + from app.models import Account, Attachment, ComposeAttachment, Draft, Message - stats = {"uploads": 0, "files": 0, "bytes": 0, "thumbs": 0} + stats = {"uploads": 0, "files": 0, "bytes": 0, "thumbs": 0, "threads_repaired": 0} async with session_scope() as db: # Uploads no draft refers to, once a week old. used = _draft_upload_ids( @@ -116,8 +122,14 @@ async def sweep_user(user_id: uuid.UUID) -> dict[str, int]: async with session_scope() as db: # Every file still referred to. Attachments count only where their file # is their only copy (not found in the message, or not yet looked for). + # A hosted mailbox's messages are kept by its mail server: their raw + # file here (left by a Sent copy filed late) is not needed. referenced = set( - await db.scalars(select(Message.blob_sha256).where(Message.user_id == user_id)) + await db.scalars( + select(Message.blob_sha256) + .join(Account, Account.id == Message.account_id) + .where(Message.user_id == user_id, Account.kind != "native") + ) ) referenced |= set( await db.scalars( @@ -140,6 +152,12 @@ async def sweep_user(user_id: uuid.UUID) -> dict[str, int]: await db.scalars(select(Attachment.blob_sha256).where(Attachment.user_id == user_id)) ) + # Counters out of step with their messages (see repair_threads). + from app.threading_.threader import repair_threads + + async with session_scope() as db: + stats["threads_repaired"] = await repair_threads(db, user_id) + root = Path(get_settings().blob_root) / str(user_id) if not root.is_dir(): return stats @@ -168,6 +186,7 @@ async def sweep_user(user_id: uuid.UUID) -> dict[str, int]: cache = root / "cache" if cache.is_dir(): att_store.prune_cache(cache) + rawstore.prune(user_id) return stats @@ -178,7 +197,7 @@ async def sweep(user_id: uuid.UUID | None = None) -> dict[str, int]: async with session_scope() as db: users = [user_id] if user_id is not None else list(await db.scalars(select(User.id))) - total = {"users": 0, "uploads": 0, "files": 0, "bytes": 0, "thumbs": 0} + total = {"users": 0, "uploads": 0, "files": 0, "bytes": 0, "thumbs": 0, "threads_repaired": 0} for uid in users: try: stats = await sweep_user(uid) diff --git a/backend/app/indexer/indexer.py b/backend/app/indexer/indexer.py index 9283d23..3783959 100644 --- a/backend/app/indexer/indexer.py +++ b/backend/app/indexer/indexer.py @@ -19,10 +19,10 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.concurrency import loop_semaphore from app.config import get_settings -from app.indexer import blobstore +from app.indexer import blobstore, rawstore from app.indexer.mime import ParsedEmail, parse_message, parse_message_file from app.indexer.sanitize import sanitize_html -from app.models import Attachment, BodyPart, KnownRecipient, Message +from app.models import Account, Attachment, BodyPart, KnownRecipient, Message from app.providers.base import RemoteMessage from app.search.base import SearchDoc from app.threading_.normalize import synthesize_message_id @@ -118,6 +118,13 @@ async def _resolve_message_id( return parsed_id +async def _keeps_raw(session: AsyncSession, account_id: uuid.UUID) -> bool: + """Whether the raw message is kept for good: not for a hosted mailbox, whose + mail server holds it (it is only cached, App. D §D.6 item 2).""" + kind = await session.scalar(select(Account.kind).where(Account.id == account_id)) + return kind != "native" + + async def _parse_bounded(user_id: uuid.UUID, sha: str, raw: bytes) -> ParsedEmail: """Parse under the PARSE_CONCURRENCY cap; stream large messages from disk (§7). @@ -128,8 +135,9 @@ async def _parse_bounded(user_id: uuid.UUID, sha: str, raw: bytes) -> ParsedEmai settings = get_settings() threshold = settings.stream_parse_mb * 1024 * 1024 async with loop_semaphore("mime_parse", "parse", settings.parse_concurrency): - if len(raw) > threshold: - return parse_message_file(blobstore.blob_path(user_id, sha)) + path = rawstore.locate(user_id, sha) if len(raw) > threshold else None + if path is not None: + return parse_message_file(path) return parse_message(raw) @@ -154,7 +162,10 @@ async def persist_message( if existing is not None: return None - sha = blobstore.store(user_id, remote.raw) + if await _keeps_raw(session, account_id): + sha = blobstore.store(user_id, remote.raw) + else: + sha = rawstore.put_cached(user_id, remote.raw) parsed = await _parse_bounded(user_id, sha, remote.raw) message_id_hdr = await _resolve_message_id( diff --git a/backend/app/indexer/rawstore.py b/backend/app/indexer/rawstore.py new file mode 100644 index 0000000..106a226 --- /dev/null +++ b/backend/app/indexer/rawstore.py @@ -0,0 +1,170 @@ +"""Where a message's raw source is (SPEC App. D §D.6 item 2). + +An external account's messages are kept whole in the blob store, as always. +A hosted mailbox's are not: the mail server beside the app holds them, so the +webmail keeps only what it reads them for (headers, the sanitized body, the +snippet, the search index) and a **cache** of raw messages — the newest ones +on arrival, the ones opened since — bounded per user (``RAW_CACHE_MB``, +least recently used out first). Anything else is fetched from the mail server +when needed: downloading an attachment, "show original", forwarding. + +Layout: ``${BLOB_ROOT}/{user_id}/rawcache/{sha256}``, the same content hash +``messages.blob_sha256`` names. + +* :func:`locate` — the file, if it is here (blob store or cache); cheap, sync; +* :func:`ensure` — the file, fetched from the mail server if need be (async): + call it before handing the work to a thread that reads the file. +""" + +from __future__ import annotations + +import contextlib +import logging +import time +import uuid +from pathlib import Path + +from app.config import get_settings +from app.indexer import blobstore + +logger = logging.getLogger(__name__) + +CACHE_DIR = "rawcache" +KEEP_S = 30 * 24 * 3600 # unused this long: dropped whatever the room +PRUNE_EVERY = 50 +_writes: dict[uuid.UUID, int] = {} + + +def cache_dir(user_id: uuid.UUID) -> Path: + return blobstore.blob_path(user_id, "00").parent.parent / CACHE_DIR + + +def cache_path(user_id: uuid.UUID, sha: str) -> Path: + return cache_dir(user_id) / sha + + +def locate(user_id: uuid.UUID, sha: str) -> Path | None: + """The message's raw file, if it is kept or cached here.""" + kept = blobstore.blob_path(user_id, sha) + if kept.exists(): + return kept + cached = cache_path(user_id, sha) + if cached.exists(): + with contextlib.suppress(OSError): + cached.touch() # used again: kept longer + return cached + return None + + +def load(user_id: uuid.UUID, sha: str) -> bytes: + """The raw bytes if they are here. Raises FileNotFoundError.""" + path = locate(user_id, sha) + if path is None: + raise FileNotFoundError(sha) + return path.read_bytes() + + +def put_cached(user_id: uuid.UUID, data: bytes, sha: str | None = None) -> str: + """Cache a raw message (``sha`` names it; its content hash by default).""" + sha = sha or blobstore.sha256_hex(data) + if blobstore.exists(user_id, sha): + return sha # kept for good already (another account has the same message) + path = cache_path(user_id, sha) + if not path.exists(): + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(f".{uuid.uuid4().hex}.tmp") + tmp.write_bytes(data) + tmp.replace(path) + else: + path.touch() + # Pruning lists the whole cache: done every PRUNE_EVERY writes, not each + # (a backfill writes thousands); meanwhile it overshoots by that much at most. + writes = _writes.get(user_id, 0) + 1 + _writes[user_id] = writes % PRUNE_EVERY + if writes >= PRUNE_EVERY: + with contextlib.suppress(OSError): # a file removed meanwhile by another request + prune(user_id) + return sha + + +def prune(user_id: uuid.UUID, *, max_bytes: int | None = None) -> int: + """Drop cached messages unused for KEEP_S, and the least recently used + beyond the cap. Returns bytes freed.""" + cache = cache_dir(user_id) + if not cache.is_dir(): + return 0 + cap = max_bytes if max_bytes is not None else get_settings().raw_cache_mb * 1024 * 1024 + files = [(p, p.stat()) for p in cache.iterdir() if p.is_file() and not p.name.endswith(".tmp")] + now = time.time() + total = freed = 0 + for path, st in sorted(files, key=lambda f: f[1].st_mtime, reverse=True): + total += st.st_size + if now - st.st_mtime > KEEP_S or total > cap: + path.unlink(missing_ok=True) + freed += st.st_size + return freed + + +def drop(user_id: uuid.UUID, sha: str) -> None: + cache_path(user_id, sha).unlink(missing_ok=True) + + +async def ensure(message: object) -> Path: + """The raw file of ``message`` (a :class:`~app.models.Message`), fetched + from the hosted mail server into the cache when it is not here. Raises + FileNotFoundError when it cannot be had (an external account's message + missing from the blob store; a native one gone from the server).""" + from app.models import Message + + assert isinstance(message, Message) + found = locate(message.user_id, message.blob_sha256) + if found is not None: + return found + raw = await _fetch_native(message) + if raw is None: + raise FileNotFoundError(message.blob_sha256) + got = blobstore.sha256_hex(raw) + if got != message.blob_sha256: + # The server's copy differs from what was indexed (it should not): + # cached under the indexed name, so the message can still be read. + logger.warning("raw of message %s changed on the mail server", message.id) + put_cached(message.user_id, raw, message.blob_sha256) + return cache_path(message.user_id, message.blob_sha256) + + +async def ensure_bytes(message: object) -> bytes: + path = await ensure(message) + import asyncio + + return await asyncio.to_thread(path.read_bytes) + + +async def _fetch_native(message: object) -> bytes | None: + from sqlalchemy import select + + from app.db import session_scope + from app.models import Account, Mailbox, Message + + assert isinstance(message, Message) + async with session_scope() as db: + row = ( + await db.execute( + select(Account.kind, Account.email, Mailbox.path) + .join(Mailbox, Mailbox.account_id == Account.id) + .where(Account.id == message.account_id, Mailbox.id == message.mailbox_id) + ) + ).first() + if row is None or row.kind != "native": + return None + from app.hosted.native import native_provider_config + from app.providers.base import ProviderError + from app.providers.imap_smtp import ImapSmtpProvider + + provider = ImapSmtpProvider(native_provider_config(row.email)) + try: + return await provider.fetch_raw(row.path, message.uid) + except ProviderError as exc: + logger.warning("cannot fetch message %s from the mail server: %s", message.id, exc) + return None + finally: + await provider.close() diff --git a/backend/app/indexer/thumbs.py b/backend/app/indexer/thumbs.py index e4db539..e8b1dee 100644 --- a/backend/app/indexer/thumbs.py +++ b/backend/app/indexer/thumbs.py @@ -87,6 +87,11 @@ def _render(data: bytes, mime_type: str) -> bytes | None: return out.getvalue() +def made(user_id: uuid.UUID, sha: str) -> bool: + """Whether the attachment's thumbnail (or its absence) is settled already.""" + return _path(user_id, sha, ".webp").exists() or _path(user_id, sha, ".none").exists() + + def thumbnail( user_id: uuid.UUID, sha: str, diff --git a/backend/app/models.py b/backend/app/models.py index 3c4f604..d833492 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -131,6 +131,10 @@ class Account(Base): imap_config: Mapped[dict[str, object]] = mapped_column(JSONB, nullable=False) smtp_config: Mapped[dict[str, object]] = mapped_column(JSONB, nullable=False) secret_ct: Mapped[bytes] = mapped_column(BYTEA, nullable=False) + # external: any IMAP/SMTP server (the user's password in secret_ct); + # native: the user's own mailbox on the hosted mail server — permanent, no + # stored password (SPEC App. D §D.4; app/hosted/). + kind: Mapped[str] = mapped_column(Text, nullable=False, server_default=text("'external'")) discovery_src: Mapped[str] = mapped_column(Text, nullable=False) state: Mapped[str] = mapped_column(Text, nullable=False, server_default=text("'backfilling'")) state_detail: Mapped[str] = mapped_column(Text, nullable=False, server_default=text("''")) @@ -161,9 +165,12 @@ class Mailbox(Base): ) path: Mapped[str] = mapped_column(Text, nullable=False) special_use: Mapped[str | None] = mapped_column(Text) - uidvalidity: Mapped[int] = mapped_column(Integer, nullable=False, server_default=text("0")) - highestmodseq: Mapped[int | None] = mapped_column(Integer) - last_seen_uid: Mapped[int] = mapped_column(Integer, nullable=False, server_default=text("0")) + # IMAP ids are unsigned 32-bit (MODSEQ 63-bit): bigint, not integer. + uidvalidity: Mapped[int] = mapped_column(BigInteger, nullable=False, server_default=text("0")) + highestmodseq: Mapped[int | None] = mapped_column(BigInteger) + last_seen_uid: Mapped[int] = mapped_column( + BigInteger, nullable=False, server_default=text("0") + ) class Thread(Base): @@ -214,7 +221,7 @@ class Message(Base): restore_mailbox_id: Mapped[uuid.UUID | None] = mapped_column( UUID(as_uuid=True), ForeignKey("mailboxes.id", ondelete="SET NULL") ) - uid: Mapped[int] = mapped_column(Integer, nullable=False) + uid: Mapped[int] = mapped_column(BigInteger, nullable=False) thread_id: Mapped[uuid.UUID | None] = mapped_column( UUID(as_uuid=True), ForeignKey("threads.id") ) @@ -321,7 +328,7 @@ class Draft(Base): payload: Mapped[dict[str, object]] = mapped_column(JSONB, nullable=False) # UID of the mirrored copy in the account's IMAP Drafts folder (so the draft # is visible/continuable on other mail clients). Replaced on each edit. - imap_uid: Mapped[int | None] = mapped_column(Integer) + imap_uid: Mapped[int | None] = mapped_column(BigInteger) updated_at: Mapped[dt.datetime] = _now() @@ -367,6 +374,8 @@ class UserAppPassword(Base): created_at: Mapped[dt.datetime] = _now() # Set by the mail server's sign-in, once there is one. last_used_at: Mapped[dt.datetime | None] = mapped_column(TIMESTAMP(timezone=True)) + # The hosted mail server's id of the same password (§D.5), when it has one. + server_id: Mapped[str | None] = mapped_column(Text) class PushSubscription(Base): diff --git a/backend/app/outbox/inline_images.py b/backend/app/outbox/inline_images.py index 56504ae..b14c2c6 100644 --- a/backend/app/outbox/inline_images.py +++ b/backend/app/outbox/inline_images.py @@ -22,6 +22,7 @@ from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.indexer import attachments as att_store +from app.indexer import rawstore from app.indexer.sanitize import INTERNAL_CID from app.models import Attachment, Message @@ -64,7 +65,7 @@ async def embed( message_id, content_id = target found = ( await session.execute( - select(Attachment, Message.blob_sha256) + select(Attachment, Message) .join(Message, Message.id == Attachment.message_id) .where( Attachment.message_id == message_id, @@ -76,10 +77,11 @@ async def embed( ).first() part: InlinePart | None = None if found is not None: - att, message_sha = found + att, message = found try: + await rawstore.ensure(message) # a hosted mailbox's: fetched if need be data = await asyncio.to_thread( - att_store.load, user_id, message_sha, att.blob_sha256, att.part_index + att_store.load, user_id, message.blob_sha256, att.blob_sha256, att.part_index ) cid = make_msgid(domain=domain)[1:-1] part = InlinePart(cid, att.mime_type, att.filename, data) diff --git a/backend/app/plugins/actions.py b/backend/app/plugins/actions.py index cf3e2ff..ec4c3ad 100644 --- a/backend/app/plugins/actions.py +++ b/backend/app/plugins/actions.py @@ -23,7 +23,7 @@ from sqlalchemy import select from app.db import session_scope from app.events import get_event_bus -from app.indexer import blobstore +from app.indexer import rawstore from app.models import Account, Mailbox, Message, Thread, User @@ -39,9 +39,8 @@ class PluginMessages: ) if msg is None or not msg.blob_sha256: return None - sha = msg.blob_sha256 try: - raw = blobstore.load(user_id, sha) + raw = await rawstore.ensure_bytes(msg) except FileNotFoundError: return None return message_from_bytes(raw).get(name) @@ -60,9 +59,8 @@ class PluginMessages: ) if msg is None or not msg.blob_sha256: return None - sha = msg.blob_sha256 try: - return blobstore.load(user_id, sha) + return await rawstore.ensure_bytes(msg) except FileNotFoundError: return None diff --git a/backend/app/providers/imap_smtp/provider.py b/backend/app/providers/imap_smtp/provider.py index 525e95f..30a6ba1 100644 --- a/backend/app/providers/imap_smtp/provider.py +++ b/backend/app/providers/imap_smtp/provider.py @@ -132,6 +132,12 @@ class ProviderConfig: smtp_tls: str # "ssl" | "starttls" | "plain" username: str password: str # plaintext; caller decrypts per §10.2 + # TLS settings other than the default (certificate checked): the hosted + # mail server on the internal network (app/hosted/native.py). + tls_context: ssl.SSLContext | None = None + # The hosted mail server: every user's mailbox is on that one host, so the + # per-host connection cap (which spares outside providers) does not apply. + internal: bool = False class ImapSmtpProvider(MailProvider): @@ -169,8 +175,11 @@ class ImapSmtpProvider(MailProvider): account = loop_semaphore( "imap_account", f"{cfg.username}@{cfg.imap_host}", _PER_ACCOUNT_CONN_CAP ) - host = loop_semaphore("imap_host", cfg.imap_host, get_settings().per_host_conn_cap) await account.acquire() # consistent order (account, then host) avoids deadlock + if cfg.internal: + self._conn_slots = [account] + return + host = loop_semaphore("imap_host", cfg.imap_host, get_settings().per_host_conn_cap) try: await host.acquire() except BaseException: @@ -193,7 +202,10 @@ class ImapSmtpProvider(MailProvider): imap: aioimaplib.IMAP4 if cfg.imap_tls == "ssl": imap = _IMAP4_SSL( - host=cfg.imap_host, port=cfg.imap_port, timeout=_CONNECT_TIMEOUT + host=cfg.imap_host, + port=cfg.imap_port, + timeout=_CONNECT_TIMEOUT, + ssl_context=cfg.tls_context, ) else: imap = _IMAP4(host=cfg.imap_host, port=cfg.imap_port, timeout=_CONNECT_TIMEOUT) @@ -356,6 +368,15 @@ class ImapSmtpProvider(MailProvider): resp = await imap.uid_search("ALL") return set(parse_search_uids(resp.lines)) + async def fetch_raw(self, mailbox: str, uid: int) -> bytes | None: + """One message's raw bytes, or None if it is not (or no longer) there.""" + await self._select(mailbox) + imap = await self._connect() + for message in await self._fetch_messages(imap, [uid]): + if message.uid == uid: + return message.raw + return None + async def _fetch_messages(self, imap: aioimaplib.IMAP4, uids: list[int]) -> list[RemoteMessage]: if not uids: return [] @@ -474,12 +495,12 @@ class ImapSmtpProvider(MailProvider): """Create a mailbox on the server (idempotent — ignores 'already exists').""" imap = await self._connect() with contextlib.suppress(Exception): - await imap.create(path) + await imap.create(self._wire(path)) async def move(self, uid: int, src: str, dst: str) -> int: await self._select(src) imap = await self._connect() - resp = await imap.uid("move", str(uid), dst) + resp = await imap.uid("move", str(uid), self._wire(dst)) if resp.result != "OK": raise ProviderConnectError(f"MOVE {uid} -> {dst} failed") for line in resp.lines: @@ -521,7 +542,11 @@ class ImapSmtpProvider(MailProvider): cfg = self._cfg use_tls = cfg.smtp_tls == "ssl" start_tls = cfg.smtp_tls == "starttls" - host_slot = loop_semaphore("smtp_host", cfg.smtp_host, get_settings().per_host_conn_cap) + host_slot = ( + contextlib.nullcontext() + if cfg.internal + else loop_semaphore("smtp_host", cfg.smtp_host, get_settings().per_host_conn_cap) + ) try: async with host_slot: # bound simultaneous sockets to the SMTP host (§4) await aiosmtplib.send( @@ -534,6 +559,7 @@ class ImapSmtpProvider(MailProvider): password=cfg.password, use_tls=use_tls, start_tls=start_tls, + tls_context=cfg.tls_context, timeout=_CONNECT_TIMEOUT, ) except aiosmtplib.SMTPException as exc: diff --git a/backend/app/sync/actions.py b/backend/app/sync/actions.py index a729cb9..91289be 100644 --- a/backend/app/sync/actions.py +++ b/backend/app/sync/actions.py @@ -77,7 +77,7 @@ async def apply_flags_to_messages( if message.thread_id is not None: thread_ids.add(message.thread_id) await session.flush() - for tid in thread_ids: + for tid in sorted(thread_ids): # one lock order across transactions await denormalize_thread(session, tid) return thread_ids @@ -504,7 +504,7 @@ async def empty_folder_job(ctx: dict[str, Any], user_id: str, special_use: str) await session.flush() # Threads with no messages left get dropped (drafts un-linked first); the # rest are re-denormalized so their counts/snippet reflect the removal. - for tid in thread_ids: + for tid in sorted(thread_ids): remaining = await session.scalar( select(func.count()).select_from(Message).where(Message.thread_id == tid) ) @@ -584,7 +584,7 @@ async def autopurge_trash_spam(ctx: dict[str, Any]) -> None: ) await session.execute(delete(Message).where(Message.id.in_(msg_ids))) await session.flush() - for tid in thread_ids: + for tid in sorted(thread_ids): remaining = await session.scalar( select(func.count()).select_from(Message).where(Message.thread_id == tid) ) diff --git a/backend/app/sync/engine.py b/backend/app/sync/engine.py index 2065831..5bde1b0 100644 --- a/backend/app/sync/engine.py +++ b/backend/app/sync/engine.py @@ -354,7 +354,18 @@ async def apply_delta( if delta.new_uidvalidity is not None: # UIDVALIDITY changed: wipe this mailbox's rows and re-backfill (SPEC §6.3). async with session_scope() as session: + # Their threads are recounted: a message not brought back by the + # re-import (gone meanwhile) must not stay counted. + wiped_threads: set[uuid.UUID] = { + tid + for tid in await session.scalars( + select(Message.thread_id).where(Message.mailbox_id == mailbox.id) + ) + if tid is not None + } await session.execute(delete(Message).where(Message.mailbox_id == mailbox.id)) + for tid in sorted(wiped_threads): + await denormalize_thread(session, tid) await session.execute( update(Mailbox) .where(Mailbox.id == mailbox.id) @@ -428,7 +439,7 @@ async def apply_delta( ) # After the deletes, so a thread left with no messages is removed rather # than recounted (denormalize_thread handles that case). - for tid in touched_threads: + for tid in sorted(touched_threads): # one lock order (denormalize_thread) await denormalize_thread(session, tid) updates = {"last_seen_uid": max_uid} diff --git a/backend/app/sync/provider_factory.py b/backend/app/sync/provider_factory.py index ccb48fd..a859bbe 100644 --- a/backend/app/sync/provider_factory.py +++ b/backend/app/sync/provider_factory.py @@ -3,6 +3,9 @@ This module runs only inside the worker/sync context; ``decrypt_secret`` (the credential vault) is invoked here to obtain the plaintext mailbox password. It must never be imported by API request code. + +A native account (the user's hosted mailbox, App. D §D.4) has no stored +password: the worker signs in as itself, acting for the user. """ from __future__ import annotations @@ -14,6 +17,10 @@ from app.security.crypto import decrypt_secret def build_provider(account: Account) -> ImapSmtpProvider: """Construct an :class:`ImapSmtpProvider` for ``account`` (password auth, M0).""" + if account.kind == "native": + from app.hosted.native import native_provider_config + + return ImapSmtpProvider(native_provider_config(account.email)) imap = account.imap_config smtp = account.smtp_config password = decrypt_secret(account.secret_ct).decode("utf-8") diff --git a/backend/app/threading_/threader.py b/backend/app/threading_/threader.py index 0509090..3571148 100644 --- a/backend/app/threading_/threader.py +++ b/backend/app/threading_/threader.py @@ -11,12 +11,12 @@ from __future__ import annotations import uuid from datetime import timedelta -from sqlalchemy import ColumnElement, delete, literal, select, update +from sqlalchemy import ColumnElement, delete, exists, func, literal, select, update from sqlalchemy import any_ as sa_any from sqlalchemy import or_ as sa_or from sqlalchemy.ext.asyncio import AsyncSession -from app.models import Message, Thread +from app.models import Draft, Message, Thread from app.threading_.normalize import is_synthetic, normalize_subject _SUBJECT_FALLBACK_WINDOW = timedelta(days=7) @@ -172,18 +172,38 @@ async def _create_thread(session: AsyncSession, message: Message) -> uuid.UUID: async def denormalize_thread(session: AsyncSession, thread_id: uuid.UUID) -> None: - """Recompute a thread's denormalized fields from its messages (SPEC §7.2.6).""" + """Recompute a thread's denormalized fields from its messages (SPEC §7.2.6). + + The thread row is locked *before* its messages are read: two transactions + recounting one thread at once (read in the webmail while a sync brings the + same flag from a phone) then take turns, and the second reads what the + first committed — instead of both reading first and the slower one writing + an old count (a message shown read, its thread still "unread"). The + messages are reloaded from the database, not taken from the session's + copies, which a bulk UPDATE may have left behind.""" + thread = await session.scalar( + select(Thread) + .where(Thread.id == thread_id) + .with_for_update() + .execution_options(populate_existing=True) + ) + if thread is None: + return messages = list( await session.scalars( select(Message) .where(Message.thread_id == thread_id) .order_by(Message.internal_date.desc()) + .execution_options(populate_existing=True) ) ) - thread = await session.get(Thread, thread_id) - if thread is None: - return if not messages: + # A reply still being written keeps its (now empty) thread: the draft + # points at it. + if await session.scalar(select(exists().where(Draft.thread_id == thread_id))): + thread.message_count = 0 + thread.unread_count = 0 + return await session.execute(delete(Thread).where(Thread.id == thread_id)) return @@ -241,3 +261,38 @@ async def thread_message(session: AsyncSession, message: Message) -> uuid.UUID: await denormalize_thread(session, thread_id) await session.flush() return thread_id + + +async def repair_threads(session: AsyncSession, user_id: uuid.UUID) -> int: + """Recount the user's threads whose stored counters disagree with their + messages (and drop threads left without any). A safety net run nightly + (app/indexer/gc.py): every write path recounts its threads, but a missed + one would otherwise show a read message as unread until it changed again. + Returns how many were repaired.""" + seen = Message.flags.any("\\Seen") # type: ignore[arg-type] + actual = ( + select( + Message.thread_id.label("tid"), + func.count().label("n"), + func.count().filter(~seen).label("unread"), + ) + .where(Message.user_id == user_id, Message.thread_id.is_not(None)) + .group_by(Message.thread_id) + .subquery() + ) + stale = list( + await session.scalars( + select(Thread.id) + .outerjoin(actual, actual.c.tid == Thread.id) + .where( + Thread.user_id == user_id, + (actual.c.tid.is_(None) & (Thread.message_count != 0)) + | (actual.c.tid.is_(None) & ~exists().where(Draft.thread_id == Thread.id)) + | (actual.c.n != Thread.message_count) + | (actual.c.unread != Thread.unread_count), + ) + ) + ) + for tid in sorted(stale): + await denormalize_thread(session, tid) + return len(stale) diff --git a/backend/migrations/versions/0025_hosted_mailboxes.py b/backend/migrations/versions/0025_hosted_mailboxes.py new file mode 100644 index 0000000..12b67c0 --- /dev/null +++ b/backend/migrations/versions/0025_hosted_mailboxes.py @@ -0,0 +1,75 @@ +"""hosted mailboxes: native accounts and the mail server's directory (SPEC App. D) + +* ``accounts.kind`` — ``external`` (any IMAP/SMTP server, the user's own + password in the vault) or ``native`` (the user's mailbox on the hosted mail + server: permanent, signed in to without any stored password, §D.4). +* ``mail_directory`` — the view the mail server's SQL directory reads (§D.3): + one row per user, its address — and no password: the webmail password never + opens the mailbox in a mail app, only app passwords do (the mail server's + own, §D.5) — and ``active`` false for a blocked user. Plus the service accounts of + ``mail_service_accounts``: the webmail's own worker, which may act for any + mailbox and nothing else. +* ``user_app_passwords.server_id`` — the mail server's id of the same app + password (§D.5), to revoke it there too. + +The mail server reads the view as its own role, given SELECT on it alone by +``app.hosted.provision`` (the role is created there, with its password). + +Revision ID: 0025 +Revises: 0024 +Create Date: 2026-10-11 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +from alembic import op + +revision: str = "0025" +down_revision: str | None = "0024" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.execute("ALTER TABLE accounts ADD COLUMN kind text NOT NULL DEFAULT 'external'") + op.execute( + "ALTER TABLE accounts ADD CONSTRAINT ck_accounts_kind " + "CHECK (kind IN ('external', 'native'))" + ) + # One mailbox per user. + op.execute("CREATE UNIQUE INDEX uq_accounts_native ON accounts (user_id) WHERE kind = 'native'") + op.execute("ALTER TABLE user_app_passwords ADD COLUMN server_id text") + op.execute( + """ + CREATE TABLE mail_service_accounts ( + name citext PRIMARY KEY, + secret text NOT NULL, + description text NOT NULL DEFAULT '' + ) + """ + ) + op.execute( + """ + CREATE VIEW mail_directory AS + SELECT u.handle AS name, + NULL::text AS secret, + 'individual'::text AS type, + u.handle::text AS description, + u.blocked_at IS NULL AS active + FROM users u + UNION ALL + SELECT s.name, s.secret, 'individual'::text, s.description, true + FROM mail_service_accounts s + """ + ) + + +def downgrade() -> None: + op.execute("DROP VIEW IF EXISTS mail_directory") + op.execute("DROP TABLE IF EXISTS mail_service_accounts") + op.execute("ALTER TABLE user_app_passwords DROP COLUMN IF EXISTS server_id") + op.execute("DROP INDEX IF EXISTS uq_accounts_native") + op.execute("ALTER TABLE accounts DROP CONSTRAINT IF EXISTS ck_accounts_kind") + op.execute("ALTER TABLE accounts DROP COLUMN IF EXISTS kind") diff --git a/backend/migrations/versions/0026_imap_ids_bigint.py b/backend/migrations/versions/0026_imap_ids_bigint.py new file mode 100644 index 0000000..3fea43b --- /dev/null +++ b/backend/migrations/versions/0026_imap_ids_bigint.py @@ -0,0 +1,40 @@ +"""IMAP ids as 64-bit integers + +UIDVALIDITY and UIDs are unsigned 32-bit numbers (RFC 9051), MODSEQ up to 63 +bits (RFC 7162); Postgres ``integer`` is signed 32-bit. Servers that hand out +values above 2^31 — the hosted mail server's UIDVALIDITY, for one — could not +be stored ("value out of int32 range") and their folders never synced. + +Revision ID: 0026 +Revises: 0025 +Create Date: 2026-10-11 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +from alembic import op + +revision: str = "0026" +down_revision: str | None = "0025" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +_COLUMNS = ( + ("mailboxes", "uidvalidity"), + ("mailboxes", "highestmodseq"), + ("mailboxes", "last_seen_uid"), + ("messages", "uid"), + ("drafts", "imap_uid"), +) + + +def upgrade() -> None: + for table, column in _COLUMNS: + op.execute(f"ALTER TABLE {table} ALTER COLUMN {column} TYPE bigint") + + +def downgrade() -> None: + for table, column in _COLUMNS: + op.execute(f"ALTER TABLE {table} ALTER COLUMN {column} TYPE integer") diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 169724a..07c0c22 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -18,6 +18,19 @@ os.environ.setdefault( os.environ["REDIS_URL"] = "redis://127.0.0.1:6380/15" os.environ.setdefault("MASTER_KEY", base64.b64encode(b"\x00" * 32).decode()) os.environ.setdefault("BLOB_ROOT", "./var/blobs-test") +# Hosted mailboxes off: tests that need the mail server turn them on themselves +# (tests/hosted/), against the dev stack's Stalwart container. +os.environ["HOSTED_MAILBOXES"] = "false" +for _key, _value in { + "STALWART_URL": "http://127.0.0.1:8180", + "STALWART_ADMIN_SECRET": "stalwart_admin_dev", + "STALWART_MASTER_SECRET": "stalwart_master_dev", + "STALWART_IMAP_HOST": "127.0.0.1", + "STALWART_IMAP_PORT": "1993", + "STALWART_SMTP_HOST": "127.0.0.1", + "STALWART_SMTP_PORT": "1465", +}.items(): + os.environ.setdefault(_key, _value) from collections.abc import AsyncIterator # noqa: E402 diff --git a/backend/tests/hosted/test_hosted_mailboxes.py b/backend/tests/hosted/test_hosted_mailboxes.py new file mode 100644 index 0000000..08c4d37 --- /dev/null +++ b/backend/tests/hosted/test_hosted_mailboxes.py @@ -0,0 +1,269 @@ +"""Hosted mailboxes (SPEC App. D) against the dev stack's Stalwart container. + +Sign-in through the directory (password, 2FA, block), app passwords made by the +mail server, the permanent native account, and raw messages kept only by the +mail server (fetched back on demand). Skipped when the container is not up. +""" + +from __future__ import annotations + +import asyncio +import imaplib +import smtplib +import ssl +import uuid +from collections.abc import AsyncIterator +from dataclasses import dataclass +from email.message import EmailMessage + +import httpx +import pytest +import pytest_asyncio +from httpx import ASGITransport +from sqlalchemy import delete, select + +from app.config import get_settings +from app.db import session_scope +from app.hosted import admin, native +from app.identity import app_passwords +from app.identity.passwords import hash_password +from app.identity.sessions import create_session +from app.identity.user_admin import delete_user, set_blocked +from app.indexer import blobstore, rawstore +from app.main import create_app +from app.models import Account, Attachment, Mailbox, Message, User + +PASSWORD = "Correct-Horse-7" + + +async def _server_domain() -> str | None: + try: + settings = await admin.get_one("SystemSettings") + domain = await admin.get_one("Domain", str(settings["defaultDomainId"])) + except admin.MailServerError: + return None + return str(domain["name"]) + + +@dataclass +class Ctx: + user: User + web: httpx.AsyncClient + + +@pytest_asyncio.fixture +async def ctx(monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[Ctx]: + domain = await _server_domain() + if domain is None: + pytest.skip("the dev mail server (docker compose service stalwart) is not running") + settings = get_settings() + monkeypatch.setattr(settings, "hosted_mailboxes", True) + monkeypatch.setattr(settings, "app_domain", domain) + handle = f"hosted{uuid.uuid4().hex[:8]}@{domain}" + async with session_scope() as db: + user = User(handle=handle, password_hash=hash_password(PASSWORD)) + db.add(user) + token = await create_session(user.id, "pytest") + web = httpx.AsyncClient( + transport=ASGITransport(app=create_app()), + base_url="http://testserver", + cookies={"session": token, "csrf": "c"}, + headers={"X-CSRF": "c"}, + ) + try: + yield Ctx(user, web) + finally: + await web.aclose() + await admin.delete_account(handle) + async with session_scope() as db: + await db.execute(delete(User).where(User.id == user.id)) + + +def _imap_login(user: str, password: str) -> bool: + tls = ssl.create_default_context() + tls.check_hostname = False + tls.verify_mode = ssl.CERT_NONE + settings = get_settings() + try: + # A blocked account's sign-in is not answered: the server drops the + # connection after a minute. Not answering within seconds is a refusal. + conn = imaplib.IMAP4_SSL( + settings.stalwart_imap_host, settings.stalwart_imap_port, ssl_context=tls, timeout=5 + ) + except OSError: + return False + try: + conn.login(user, password) + conn.logout() + return True + except (imaplib.IMAP4.error, OSError): + return False + + +async def _can_sign_in(user: str, password: str) -> bool: + return await asyncio.to_thread(_imap_login, user, password) + + +def _deliver(to: str, subject: str, attachment: bytes) -> None: + msg = EmailMessage() + msg["From"] = "Anna Test " + msg["To"] = to + msg["Subject"] = subject + msg.set_content("Hello,\n\nthe report is attached.\n") + msg.add_attachment(attachment, maintype="application", subtype="octet-stream", filename="r.bin") + with smtplib.SMTP(get_settings().stalwart_imap_host, 1025, timeout=20) as smtp: + smtp.send_message(msg) + + +async def test_the_webmail_password_never_opens_the_mailbox(ctx: Ctx) -> None: + """Mail apps sign in with app passwords only — 2FA or not.""" + assert not await _can_sign_in(ctx.user.handle, PASSWORD) + assert not await _can_sign_in(ctx.user.handle, "wrong") + + +async def test_app_passwords_work_for_mail_apps_and_dav_and_are_revoked_everywhere( + ctx: Ctx, +) -> None: + handle = ctx.user.handle + async with session_scope() as db: + row, password = await app_passwords.create(db, ctx.user.id, "Thunderbird") + assert password.startswith("app_") and len(password) == app_passwords.SERVER_LENGTH + assert row.server_id + assert await _can_sign_in(handle, password) + async with session_scope() as db: + assert await app_passwords.verify(db, ctx.user.id, password) is not None # DAV + + # 2FA turned on changes nothing for it. + async with session_scope() as db: + user = await db.get(User, ctx.user.id) + assert user is not None + user.totp_secret = b"x" + assert await _can_sign_in(handle, password) + + # Blocked: nothing opens the mailbox; unblocked: it does again. + async with session_scope() as db: + user = await db.get(User, ctx.user.id) + assert user is not None + await set_blocked(db, user, True) + assert not await _can_sign_in(handle, password) + async with session_scope() as db: + user = await db.get(User, ctx.user.id) + assert user is not None + await set_blocked(db, user, False) + assert await _can_sign_in(handle, password) + + async with session_scope() as db: + assert await app_passwords.revoke(db, ctx.user.id, row.id) + assert not await _can_sign_in(handle, password) + + +async def test_the_native_account_keeps_mail_once_and_cannot_be_removed(ctx: Ctx) -> None: + from app.sync.engine import bootstrap_and_backfill + + async with session_scope() as db: + user = await db.get(User, ctx.user.id) + assert user is not None + account = await native.ensure(db, user) + assert account is not None + account_id = account.id + payload = uuid.uuid4().bytes * 5000 + await asyncio.to_thread(_deliver, ctx.user.handle, "Report", payload) + + # Synced like any IMAP account (delivery is asynchronous: retry briefly). + message = None + for _ in range(20): + await bootstrap_and_backfill(account_id) + async with session_scope() as db: + message = await db.scalar( + select(Message).where(Message.account_id == account_id, Message.subject == "Report") + ) + if message is not None: + break + await asyncio.sleep(0.5) + assert message is not None + + # A new mailbox has the plain folder names (not "Deleted Items", …). + async with session_scope() as db: + folders = dict( + ( + await db.execute( + select(Mailbox.special_use, Mailbox.path).where( + Mailbox.account_id == account_id + ) + ) + ).all() + ) + assert folders["trash"] == "Trash" and folders["sent"] == "Sent" + assert folders["junk"] == "Junk" and folders["archive"] == "Archive" + + # The raw message is not kept for good — only cached. + sha = message.blob_sha256 + assert not blobstore.exists(ctx.user.id, sha) + assert rawstore.cache_path(ctx.user.id, sha).exists() + + # Gone from the cache, the attachment is fetched back from the mail server. + rawstore.drop(ctx.user.id, sha) + async with session_scope() as db: + att = await db.scalar(select(Attachment).where(Attachment.message_id == message.id)) + assert att is not None + got = await ctx.web.get(f"/api/messages/{message.id}/attachments/{att.id}?download=1") + assert got.status_code == 200 and got.content == payload + assert rawstore.cache_path(ctx.user.id, sha).exists() + + # Permanent: not removable, its server settings not changeable. + listed = (await ctx.web.get("/api/accounts")).json() + assert [a["kind"] for a in listed] == ["native"] + assert (await ctx.web.delete(f"/api/accounts/{account_id}")).status_code == 409 + patched = await ctx.web.patch(f"/api/accounts/{account_id}", json={"password": "x"}) + assert patched.status_code == 400 and patched.json()["error"]["code"] == "native_account" + + +async def test_deleting_the_user_deletes_the_mailbox(ctx: Ctx) -> None: + await admin.ensure_account(ctx.user.handle) # brings it into the server + assert await admin.account_id(ctx.user.handle) is not None + async with session_scope() as db: + user = await db.get(User, ctx.user.id) + assert user is not None + await delete_user(db, user) + assert await admin.account_id(ctx.user.handle) is None + async with session_scope() as db: + assert await db.scalar(select(Account.id).where(Account.user_id == ctx.user.id)) is None + + +async def test_the_cache_keeps_the_most_recently_used(ctx: Ctx) -> None: + import os + import time + + uid = ctx.user.id + old = rawstore.put_cached(uid, b"a" * 1000) + new = rawstore.put_cached(uid, b"b" * 1000) + past = time.time() - 60 + os.utime(rawstore.cache_path(uid, old), (past, past)) + rawstore.prune(uid, max_bytes=1500) + assert not rawstore.cache_path(uid, old).exists() + assert rawstore.cache_path(uid, new).exists() + stale = time.time() - rawstore.KEEP_S - 10 + os.utime(rawstore.cache_path(uid, new), (stale, stale)) + rawstore.prune(uid) + assert not rawstore.cache_path(uid, new).exists() + + +async def test_moves_reach_folders_with_spaces_in_their_names(ctx: Ctx) -> None: + """Older mailboxes (and many providers) say "Deleted Items": the name goes + over the wire quoted, or the move is refused and Trash never fills.""" + from app.providers.imap_smtp import ImapSmtpProvider + + await admin.ensure_account(ctx.user.handle) + msg = EmailMessage() + msg["From"] = "a@example.com" + msg["To"] = ctx.user.handle + msg["Subject"] = "Swipe me" + msg.set_content("x") + provider = ImapSmtpProvider(native.native_provider_config(ctx.user.handle)) + try: + await provider.create_mailbox("Deleted Items") + uid = await provider.append("INBOX", msg.as_bytes(), frozenset()) + moved = await provider.move(uid, "INBOX", "Deleted Items") + assert moved > 0 + finally: + await provider.close() diff --git a/backend/tests/indexer/test_streaming_parse.py b/backend/tests/indexer/test_streaming_parse.py index 5fc2dc7..f03175a 100644 --- a/backend/tests/indexer/test_streaming_parse.py +++ b/backend/tests/indexer/test_streaming_parse.py @@ -8,6 +8,7 @@ from __future__ import annotations import uuid from email.message import EmailMessage +from pathlib import Path from app.indexer import indexer from app.indexer.mime import parse_message, parse_message_file @@ -50,6 +51,8 @@ async def test_parse_bounded_streams_large_messages(monkeypatch) -> None: # typ monkeypatch.setattr(indexer, "parse_message_file", fake_file) monkeypatch.setattr(indexer, "parse_message", fake_bytes) + # The message's file, written before parsing (blob store or raw cache). + monkeypatch.setattr(indexer.rawstore, "locate", lambda user_id, sha: Path("/blob")) user_id = uuid.uuid4() small = await indexer._parse_bounded(user_id, "sha", b"x" * 16) diff --git a/backend/tests/plugins/test_antispam_api.py b/backend/tests/plugins/test_antispam_api.py index 274b612..0935683 100644 --- a/backend/tests/plugins/test_antispam_api.py +++ b/backend/tests/plugins/test_antispam_api.py @@ -115,7 +115,15 @@ async def test_settings_defaults_then_merge(ctx: _Ctx) -> None: 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 + assert s["enabled_accounts"] is None # every account + + # Every account off is stored as such (not read back as "all on"), and + # all_accounts goes back to every account. + s = (await ctx.client_a.patch(f"{BASE}/settings", json={"enabled_accounts": []})).json() + assert s["enabled_accounts"] == [] + assert (await ctx.client_a.get(f"{BASE}/settings")).json()["enabled_accounts"] == [] + s = (await ctx.client_a.patch(f"{BASE}/settings", json={"all_accounts": True})).json() + assert s["enabled_accounts"] is None # 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() diff --git a/backend/tests/plugins/test_antispam_corpus.py b/backend/tests/plugins/test_antispam_corpus.py index 2b05657..6cd1719 100644 --- a/backend/tests/plugins/test_antispam_corpus.py +++ b/backend/tests/plugins/test_antispam_corpus.py @@ -32,7 +32,7 @@ from webmail_plugin_antispam import policy, rspamd pytestmark = pytest.mark.asyncio SETTINGS = policy.Settings( - enabled_accounts=[], + enabled_accounts=None, move_threshold=12.0, banner_threshold=6.0, network_checks=False, diff --git a/backend/tests/plugins/test_antispam_hook_gating.py b/backend/tests/plugins/test_antispam_hook_gating.py index 85ced80..0fcf8e5 100644 --- a/backend/tests/plugins/test_antispam_hook_gating.py +++ b/backend/tests/plugins/test_antispam_hook_gating.py @@ -68,7 +68,7 @@ def wired(monkeypatch: pytest.MonkeyPatch): # type: ignore[no-untyped-def] async def _settings(_u: object) -> policy.Settings: return policy.Settings( - enabled_accounts=[], + enabled_accounts=None, move_threshold=12.0, banner_threshold=6.0, network_checks=False, diff --git a/backend/tests/plugins/test_antispam_learning.py b/backend/tests/plugins/test_antispam_learning.py index 9646c92..857d4c8 100644 --- a/backend/tests/plugins/test_antispam_learning.py +++ b/backend/tests/plugins/test_antispam_learning.py @@ -14,7 +14,7 @@ import pytest from webmail_plugin_antispam import hook, jobs, policy SETTINGS = policy.Settings( - enabled_accounts=[], + enabled_accounts=None, move_threshold=12.0, banner_threshold=6.0, network_checks=False, @@ -169,7 +169,7 @@ async def test_bootstrap_respects_the_setting(monkeypatch: pytest.MonkeyPatch) - async def _settings(_u: object) -> policy.Settings: return policy.Settings( - enabled_accounts=[], + enabled_accounts=None, move_threshold=12.0, banner_threshold=6.0, network_checks=False, diff --git a/backend/tests/plugins/test_antispam_policy.py b/backend/tests/plugins/test_antispam_policy.py index c2cb69a..0977d86 100644 --- a/backend/tests/plugins/test_antispam_policy.py +++ b/backend/tests/plugins/test_antispam_policy.py @@ -6,7 +6,7 @@ import pytest from webmail_plugin_antispam.policy import Settings, decide, matches SETTINGS = Settings( - enabled_accounts=[], + enabled_accounts=None, move_threshold=12.0, banner_threshold=6.0, network_checks=False, @@ -79,7 +79,7 @@ def test_pattern_matching(patterns: set[str], email: str, hit: bool) -> None: def test_per_account_enablement() -> None: - """Empty list means every account; a populated one is an allow-list.""" + """None means every account; a list is an allow-list — empty, none.""" everywhere = SETTINGS assert everywhere.covers("any-account-id") @@ -93,6 +93,15 @@ def test_per_account_enablement() -> None: assert scoped.covers("acc-1") assert not scoped.covers("acc-2") + nowhere = Settings( + enabled_accounts=[], + move_threshold=12.0, + banner_threshold=6.0, + network_checks=False, + learn_from_provider_junk=True, + ) + assert not nowhere.covers("acc-1") + # --- contacts are never filed as spam ---------------------------------------- diff --git a/backend/tests/sync/test_thread_repair.py b/backend/tests/sync/test_thread_repair.py new file mode 100644 index 0000000..2990e32 --- /dev/null +++ b/backend/tests/sync/test_thread_repair.py @@ -0,0 +1,104 @@ +"""Thread counters that drifted from their messages are repaired (nightly safety +net, app/threading_/threader.py repair_threads).""" + +from __future__ import annotations + +import uuid +from datetime import UTC, datetime +from email.message import EmailMessage + +from sqlalchemy import delete, select, update + +from app.db import session_scope +from app.identity.passwords import hash_password +from app.indexer.indexer import persist_message +from app.models import Account, Draft, Mailbox, Thread, User +from app.providers.base import RemoteMessage +from app.threading_.threader import repair_threads + + +def _raw(subject: str) -> bytes: + m = EmailMessage() + m["Subject"] = subject + m["From"] = "billing@example.com" + m["To"] = "me@example.com" + m["Message-ID"] = f"<{uuid.uuid4().hex}@example.com>" + m.set_content("receipt") + return m.as_bytes() + + +async def test_repair_recounts_stale_threads_and_drops_empty_ones() -> None: + async with session_scope() as db: + user = User( + handle=f"rep{uuid.uuid4().hex[:8]}@test.local", password_hash=hash_password("pw") + ) + db.add(user) + await db.flush() + account = Account( + user_id=user.id, + email="me@example.com", + imap_config={}, + smtp_config={}, + secret_ct=b"", + discovery_src="manual", + ) + db.add(account) + await db.flush() + inbox = Mailbox(account_id=account.id, user_id=user.id, path="INBOX", special_use="inbox") + db.add(inbox) + await db.flush() + result = await persist_message( + db, + user_id=user.id, + account_id=account.id, + mailbox_id=inbox.id, + mailbox_special_use="inbox", + remote=RemoteMessage( + uid=1, + raw=_raw("Your receipt"), + flags=frozenset({"\\Seen"}), + internal_date=datetime.now(UTC), + ), + ) + assert result is not None + tid = result.message.thread_id + assert tid is not None + # A read message whose thread still says "unread" (the bug), and a + # thread left behind with no messages at all. + await db.execute(update(Thread).where(Thread.id == tid).values(unread_count=1)) + empty = Thread( + user_id=user.id, + subject_norm="gone", + message_count=1, + unread_count=1, + last_message_at=datetime.now(UTC), + ) + db.add(empty) + # An emptied thread a reply draft still points at: kept, counts zeroed. + replied = Thread( + user_id=user.id, + subject_norm="replying", + message_count=1, + unread_count=0, + last_message_at=datetime.now(UTC), + ) + db.add(replied) + await db.flush() + db.add(Draft(user_id=user.id, account_id=account.id, thread_id=replied.id, payload={})) + await db.flush() + empty_id = empty.id + replied_id = replied.id + try: + async with session_scope() as db: + assert await repair_threads(db, user.id) == 3 + async with session_scope() as db: + thread = await db.get(Thread, tid) + assert thread is not None and thread.unread_count == 0 and thread.message_count == 1 + assert await db.get(Thread, empty_id) is None + kept = await db.get(Thread, replied_id) + assert kept is not None and kept.message_count == 0 + assert await repair_threads(db, user.id) == 0 # nothing left to repair + finally: + async with session_scope() as db: + await db.execute(delete(User).where(User.id == user.id)) # cascades + assert await db.scalar(select(User.id).where(User.id == user.id)) is None diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml index a9710c9..0e6573c 100644 --- a/docker/docker-compose.yml +++ b/docker/docker-compose.yml @@ -117,6 +117,29 @@ services: timeout: 3s retries: 10 + # Hosted mailboxes (SPEC Appendix D). Everything but where its database is + # (config.json) lives in that database and is set up by + # `python -m app.hosted.provision`, which also creates the database and its + # role — so the first start waits (restarting) until that has run once. + # Dev publishes the ports on loopback, off the standard numbers. + stalwart: + image: stalwartlabs/stalwart:v0.16.19 + environment: + STALWART_DB_PASSWORD: ${STALWART_DB_PASSWORD:-stalwart_dev} + STALWART_RECOVERY_ADMIN: admin:${STALWART_ADMIN_SECRET:-stalwart_admin_dev} + volumes: + - ./stalwart/config.json:/etc/stalwart/config.json:ro + - stalwartdata:/var/lib/stalwart + ports: + - "127.0.0.1:8180:8080" # management API (JMAP) + - "127.0.0.1:1025:25" # SMTP — mail from "outside" + - "127.0.0.1:1465:465" # submissions + - "127.0.0.1:1587:587" # submission (STARTTLS) + - "127.0.0.1:1993:993" # IMAPS + depends_on: + postgres: { condition: service_healthy } + restart: unless-stopped + # Test mail servers (integration tests + manual dev against a throwaway mailbox) # Spelling and grammar checking for the compose window (POST /api/spellcheck # forwards to it; the browser never talks to it directly). No n-gram data: @@ -152,3 +175,4 @@ volumes: blobs: rspamddata: rspamdredis: + stalwartdata: diff --git a/docker/stalwart/config.json b/docker/stalwart/config.json new file mode 100644 index 0000000..a11266d --- /dev/null +++ b/docker/stalwart/config.json @@ -0,0 +1,15 @@ +{ + "@type": "PostgreSql", + "host": "postgres", + "port": 5432, + "database": "stalwart", + "authUsername": "stalwart", + "authSecret": { "@type": "EnvironmentVariable", "variableName": "STALWART_DB_PASSWORD" }, + "useTls": false, + "allowInvalidCerts": false, + "poolMaxConnections": 10, + "poolRecyclingMethod": "fast", + "timeout": 15000, + "readReplicas": {}, + "options": null +} diff --git a/docs/mail-server-network.md b/docs/mail-server-network.md index ed864ea..f992d31 100644 --- a/docs/mail-server-network.md +++ b/docs/mail-server-network.md @@ -10,7 +10,7 @@ Notes for switching `ultimo.email` mail from home.pl to Ultimo's own mail server |---|---| | Hosting | **Everything on the QNAP** — receive and send directly (spec §D.1 option A). A relay (home.pl SMTP, SMTP2GO, Brevo) is optional, as a fallback only. | | Domain | Addresses `name@ultimo.email`; webmail stays at `https://ultimo.email`. | -| Mail apps | **Allowed**: Thunderbird and phone mail apps use the mailboxes over IMAP/SMTP, signing in with **app passwords** (required when 2FA is on). | +| Mail apps | **Allowed**: Thunderbird and phone mail apps use the mailboxes over IMAP/SMTP, signing in with **app passwords** only (never the webmail password). | | Mail host name | **`mail.ultimo.email`** (recommended; `ultimo.email` would also work, see below). | ## Current setup diff --git a/frontend/src/components/thread/MessageBody.tsx b/frontend/src/components/thread/MessageBody.tsx index 7fa8844..3de6cb1 100644 --- a/frontend/src/components/thread/MessageBody.tsx +++ b/frontend/src/components/thread/MessageBody.tsx @@ -3,7 +3,9 @@ import { useCallback, useEffect, useRef, useState } from "react"; import { useT } from "../../i18n"; import { api } from "../../lib/api"; +import { adaptForDarkMode, useDarkMessages, withDarkPage } from "../../lib/darkMessage"; import { fixInvisibleText } from "../../lib/readability"; +import { useTheme } from "../../lib/theme"; import type { MessageEnvelope } from "../../lib/types"; import { Attachments } from "./Attachments"; @@ -30,6 +32,12 @@ export function MessageBody({ const onMailtoRef = useRef(onMailto); onMailtoRef.current = onMailto; const [height, setHeight] = useState(80); + // Dark message mode (lib/darkMessage.ts): a dark page in the dark theme, + // unless switched off in the subject bar. + const darkMessages = useDarkMessages(); + const dark = useTheme().resolved === "dark" && darkMessages; + const darkRef = useRef(dark); + darkRef.current = dark; const { data, isLoading, isError, error } = useQuery({ queryKey: ["message-html", message.id, allowed], @@ -89,7 +97,8 @@ export function MessageBody({ btn.setAttribute( "style", "display:inline-block;margin:6px 0;padding:1px 9px;border-radius:8px;border:none;" + - "background:#e5e7eb;color:#4b5563;cursor:pointer;font:inherit;font-size:15px;line-height:1.5;", + (darkRef.current ? "background:#3f444a;color:#c9ccd1;" : "background:#e5e7eb;color:#4b5563;") + + "cursor:pointer;font:inherit;font-size:15px;line-height:1.5;", ); btn.addEventListener("click", () => { quote.style.display = quote.style.display === "none" ? "" : "none"; @@ -131,8 +140,13 @@ export function MessageBody({ // overflow removes any phantom scrollbar left by sub-pixel/line-box rounding // (horizontal stays scrollable for genuinely wide emails). doc.documentElement.style.overflowY = "hidden"; - // Before the quote is folded away: hidden text counts as hidden on purpose. - fixInvisibleText(doc); // white-on-white from a dark-mode sender (lib/readability.ts) + if (darkRef.current) { + // The sender's colours that would not read on the dark page. + adaptForDarkMode(doc); + } else { + // Before the quote is folded away: hidden text counts as hidden on purpose. + fixInvisibleText(doc); // white-on-white from a dark-mode sender (lib/readability.ts) + } injectQuoteToggle(doc); resize(); // Re-measure after the first paint settles (fonts/layout can shift the height @@ -175,7 +189,7 @@ export function MessageBody({ }; frame = requestAnimationFrame(check); return () => cancelAnimationFrame(frame); - }, [data, prepare]); + }, [data, dark, prepare]); useEffect(() => () => observerRef.current?.disconnect(), []); @@ -213,13 +227,15 @@ export function MessageBody({ )} {data !== undefined && !isLoading && (