lukas/webmail-client
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
d0caf76b03e5865e25535490224269765bea398c
webmail-client/backend/app/models.py
T
lukasandClaude Opus 5.5 fda4cc6f16 Installable app, push notifications about new mail, Install app entry 
The webmail is now a PWA: web app manifest, icons (any + maskable, a status-bar
badge), theme-coloured title bar following the chosen theme, and an app-icon
badge with the unread Inbox count. The service worker is always installed;
offline access is on only when it is registered as /sw.js?offline=1, so
turning it on or off is an update that applies at once.

Notifications (Settings → Accounts → Notifications, per device) use Web Push.
Encryption (RFC 8291) and VAPID signing (RFC 8292) are implemented in
app/push/webpush.py — pywebpush needs a newer cryptography than we pin — and
checked against the RFC's worked example. The server makes its key pair on
first use (private half encrypted with MASTER_KEY). New mail found by the live
sync in an Inbox enqueues a job 5 s later that tells the user's devices about
messages still unread there, under 2 hours old and not from the user; more
than three become one "N new messages". A subscription belongs to the session
it was made under: signed out or revoked, nothing more is sent there. Endpoints
are limited to the known push services (no SSRF). Nothing shows while the app
is in front (except on Apple, which requires it). Tested with real pushes
through Google's service in Chromium and Opera.

The mobile drawer gets "Install app": Chrome's own install dialog where the
browser offers one, else short steps for iPhone/iPad, Opera and Firefox, with
"Don't show this again". docs/nas-install.md §5.3 covers installing,
notifications, Public URL and a later Android app (TWA).

Migration 0022: push_subscriptions, app_settings.vapid_*.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-08 18:52:26 +02:00

424 lines
19 KiB
Python
Raw Blame History

"""SQLAlchemy ORM models mirroring the authoritative DDL in SPEC §4.
The Alembic initial migration is generated to match these models exactly.
Invariant I-7: every table carrying user data has an indexed ``user_id``.
"""
from __future__ import annotations
import datetime as dt
import uuid
from typing import Any
from sqlalchemy import (
BigInteger,
CheckConstraint,
ForeignKey,
Index,
Integer,
Text,
UniqueConstraint,
text,
)
from sqlalchemy.dialects.postgresql import (
ARRAY,
BYTEA,
CITEXT,
JSONB,
TIMESTAMP,
TSVECTOR,
UUID,
)
from sqlalchemy.orm import Mapped, mapped_column
from app.db import Base
def _uuid_pk() -> Mapped[uuid.UUID]:
return mapped_column(
UUID(as_uuid=True),
primary_key=True,
server_default=text("gen_random_uuid()"),
)
def _now() -> Mapped[dt.datetime]:
return mapped_column(TIMESTAMP(timezone=True), nullable=False, server_default=text("now()"))
class AppSettings(Base):
"""Instance-wide settings, exactly one row (id = 1).
Distinct from :func:`app.config.get_settings`, which is deployment environment
read at boot; these are options an administrator changes at runtime from the
admin panel. Locking this row serialises the registration decision below.
"""
__tablename__ = "app_settings"
__table_args__ = (
CheckConstraint("id = 1", name="ck_app_settings_singleton"),
CheckConstraint("default_quota_bytes > 0", name="ck_app_settings_quota_positive"),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, server_default=text("1"))
# Whether POST /auth/register accepts new users. Closed automatically once the
# first account (the administrator) is created; an admin reopens it later.
registration_open: Mapped[bool] = mapped_column(nullable=False, server_default=text("false"))
# Quota handed to the NEXT user created. Existing users keep whatever
# users.quota_bytes already says — changing this never resizes a mailbox.
default_quota_bytes: Mapped[int] = mapped_column(
BigInteger, nullable=False, server_default=text("5368709120")
)
# Outgoing system mail (password-reset links): the SMTP account the server
# sends from. The password is vault-encrypted (like Account.secret_ct).
smtp_host: Mapped[str] = mapped_column(Text, nullable=False, server_default=text("''"))
smtp_port: Mapped[int] = mapped_column(Integer, nullable=False, server_default=text("587"))
smtp_security: Mapped[str] = mapped_column(
Text, nullable=False, server_default=text("'starttls'")
) # starttls | ssl | none
smtp_username: Mapped[str] = mapped_column(Text, nullable=False, server_default=text("''"))
smtp_secret_ct: Mapped[bytes | None] = mapped_column(BYTEA)
smtp_from: Mapped[str] = mapped_column(Text, nullable=False, server_default=text("''"))
# Where users reach the app, e.g. https://mail.example.com — reset links are
# built from this, never from the request's Host header.
public_url: Mapped[str] = mapped_column(Text, nullable=False, server_default=text("''"))
# The key pair push notifications are signed with (VAPID, RFC 8292): made on
# first use; the private half vault-encrypted, the public half as browsers
# take it (base64url, uncompressed P-256 point).
vapid_public: Mapped[str] = mapped_column(Text, nullable=False, server_default=text("''"))
vapid_private_ct: Mapped[bytes | None] = mapped_column(BYTEA)
updated_at: Mapped[dt.datetime] = _now()
class User(Base):
__tablename__ = "users"
id: Mapped[uuid.UUID] = _uuid_pk()
handle: Mapped[str] = mapped_column(CITEXT, unique=True, nullable=False)
password_hash: Mapped[str] = mapped_column(Text, nullable=False)
totp_secret: Mapped[bytes | None] = mapped_column(BYTEA)
recovery_email: Mapped[str | None] = mapped_column(CITEXT)
settings: Mapped[dict[str, object]] = mapped_column(
JSONB, nullable=False, server_default=text("'{}'")
)
# Storage quota in bytes; per-user so an admin can override it later.
# Default 5 GiB (5 * 1024^3).
quota_bytes: Mapped[int] = mapped_column(
BigInteger, nullable=False, server_default=text("5368709120")
)
# The first account created on a fresh install bootstraps as administrator;
# the admin panel can grant it to others later.
is_admin: Mapped[bool] = mapped_column(nullable=False, server_default=text("false"))
# Set while an administrator has blocked the account (no sign-in, sessions
# refused); NULL when not blocked.
blocked_at: Mapped[dt.datetime | None] = mapped_column(TIMESTAMP(timezone=True))
created_at: Mapped[dt.datetime] = _now()
class Account(Base):
__tablename__ = "accounts"
__table_args__ = (UniqueConstraint("user_id", "email", name="uq_accounts_user_email"),)
id: Mapped[uuid.UUID] = _uuid_pk()
user_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
email: Mapped[str] = mapped_column(CITEXT, nullable=False)
display_name: Mapped[str] = mapped_column(Text, nullable=False, server_default=text("''"))
color: Mapped[str] = mapped_column(Text, nullable=False, server_default=text("'#888888'"))
# Per-account HTML signature, inserted into new messages (and via compose menu).
signature: Mapped[str] = mapped_column(Text, nullable=False, server_default=text("''"))
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)
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("''"))
# How much history to import (SPEC §12 M2): all | 3months | new. Drives the
# initial backfill; backfill_total/backfill_done feed the progress UX.
sync_window: Mapped[str] = mapped_column(Text, nullable=False, server_default=text("'all'"))
# Whether the Spam/Junk folder is synced (opt-in; SPEC §6.3 folder selection).
sync_spam: Mapped[bool] = mapped_column(nullable=False, server_default=text("false"))
backfill_total: Mapped[int] = mapped_column(Integer, nullable=False, server_default=text("0"))
backfill_done: Mapped[int] = mapped_column(Integer, nullable=False, server_default=text("0"))
last_sync_at: Mapped[dt.datetime | None] = mapped_column(TIMESTAMP(timezone=True))
# The server has not answered for a while ("unreachable"); cleared on success.
sync_error: Mapped[str | None] = mapped_column(Text)
sync_error_since: Mapped[dt.datetime | None] = mapped_column(TIMESTAMP(timezone=True))
created_at: Mapped[dt.datetime] = _now()
class Mailbox(Base):
__tablename__ = "mailboxes"
__table_args__ = (UniqueConstraint("account_id", "path", name="uq_mailboxes_account_path"),)
id: Mapped[uuid.UUID] = _uuid_pk()
account_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), ForeignKey("accounts.id", ondelete="CASCADE"), nullable=False
)
user_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), nullable=False
)
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"))
class Thread(Base):
__tablename__ = "threads"
__table_args__ = (Index("ix_threads_user_last", "user_id", text("last_message_at DESC")),)
id: Mapped[uuid.UUID] = _uuid_pk()
user_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), nullable=False
)
subject_norm: Mapped[str] = mapped_column(Text, nullable=False)
snippet: Mapped[str] = mapped_column(Text, nullable=False, server_default=text("''"))
participants: Mapped[list[dict[str, str]]] = mapped_column(
JSONB, nullable=False, server_default=text("'[]'")
)
account_ids: Mapped[list[uuid.UUID]] = mapped_column(
ARRAY(UUID(as_uuid=True)), nullable=False, server_default=text("'{}'")
)
message_count: Mapped[int] = mapped_column(Integer, nullable=False, server_default=text("0"))
unread_count: Mapped[int] = mapped_column(Integer, nullable=False, server_default=text("0"))
has_attachments: Mapped[bool] = mapped_column(nullable=False, server_default=text("false"))
pinned: Mapped[bool] = mapped_column(nullable=False, server_default=text("false"))
last_message_at: Mapped[dt.datetime] = mapped_column(TIMESTAMP(timezone=True), nullable=False)
class Message(Base):
__tablename__ = "messages"
__table_args__ = (
UniqueConstraint("mailbox_id", "uid", name="uq_messages_mailbox_uid"),
Index("ix_messages_user_thread", "user_id", "thread_id"),
Index("ix_messages_account_msgid", "account_id", "message_id_hdr"),
Index("ix_messages_user_msgid", "user_id", "message_id_hdr"),
Index("ix_messages_body_indexed", "body_indexed", postgresql_using="gin"),
)
id: Mapped[uuid.UUID] = _uuid_pk()
user_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), nullable=False
)
account_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), ForeignKey("accounts.id", ondelete="CASCADE"), nullable=False
)
mailbox_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), ForeignKey("mailboxes.id", ondelete="CASCADE"), nullable=False
)
# The folder this message was in before being moved to Archive/Trash/Junk, so
# Restore can return it there (e.g. Sent) instead of the Inbox. NULL otherwise.
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)
thread_id: Mapped[uuid.UUID | None] = mapped_column(
UUID(as_uuid=True), ForeignKey("threads.id")
)
message_id_hdr: Mapped[str | None] = mapped_column(Text)
# The real header Message-ID as delivered (kept even when message_id_hdr is
# synthesized for a cross-account duplicate, §7.1). Identifies one delivery, so
# the same email fanned out to several accounts collapses to one row in a thread.
delivery_id: Mapped[str | None] = mapped_column(Text)
in_reply_to: Mapped[str | None] = mapped_column(Text)
references_hdr: Mapped[list[str]] = mapped_column(
ARRAY(Text), nullable=False, server_default=text("'{}'")
)
subject: Mapped[str] = mapped_column(Text, nullable=False, server_default=text("''"))
from_addr: Mapped[list[dict[str, str]]] = mapped_column(
JSONB, nullable=False, server_default=text("'[]'")
)
to_addr: Mapped[list[dict[str, str]]] = mapped_column(
JSONB, nullable=False, server_default=text("'[]'")
)
cc_addr: Mapped[list[dict[str, str]]] = mapped_column(
JSONB, nullable=False, server_default=text("'[]'")
)
bcc_addr: Mapped[list[dict[str, str]]] = mapped_column(
JSONB, nullable=False, server_default=text("'[]'")
)
date_hdr: Mapped[dt.datetime | None] = mapped_column(TIMESTAMP(timezone=True))
internal_date: Mapped[dt.datetime] = mapped_column(TIMESTAMP(timezone=True), nullable=False)
flags: Mapped[list[str]] = mapped_column(
ARRAY(Text), nullable=False, server_default=text("'{}'")
)
size_bytes: Mapped[int] = mapped_column(Integer, nullable=False, server_default=text("0"))
has_attachments: Mapped[bool] = mapped_column(nullable=False, server_default=text("false"))
has_remote_images: Mapped[bool] = mapped_column(nullable=False, server_default=text("false"))
snippet: Mapped[str] = mapped_column(Text, nullable=False, server_default=text("''"))
blob_sha256: Mapped[str] = mapped_column(Text, nullable=False)
body_indexed: Mapped[str | None] = mapped_column(TSVECTOR)
class BodyPart(Base):
__tablename__ = "body_parts"
id: Mapped[uuid.UUID] = _uuid_pk()
message_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), ForeignKey("messages.id", ondelete="CASCADE"), nullable=False
)
kind: Mapped[str] = mapped_column(Text, nullable=False) # text | html_sanitized
content: Mapped[str] = mapped_column(Text, nullable=False)
# SPEC §10.4: re-sanitize on render if the sanitizer version changed.
sanitizer_v: Mapped[int] = mapped_column(Integer, nullable=False, server_default=text("0"))
class Attachment(Base):
__tablename__ = "attachments"
id: Mapped[uuid.UUID] = _uuid_pk()
message_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), ForeignKey("messages.id", ondelete="CASCADE"), nullable=False
)
user_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), nullable=False)
filename: Mapped[str] = mapped_column(Text, nullable=False)
mime_type: Mapped[str] = mapped_column(Text, nullable=False)
size_bytes: Mapped[int] = mapped_column(Integer, nullable=False)
content_id: Mapped[str | None] = mapped_column(Text)
blob_sha256: Mapped[str] = mapped_column(Text, nullable=False)
class ComposeAttachment(Base):
"""A file uploaded for an outgoing message, referenced by a draft's payload."""
__tablename__ = "compose_attachments"
id: Mapped[uuid.UUID] = _uuid_pk()
user_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), nullable=False
)
filename: Mapped[str] = mapped_column(Text, nullable=False)
mime_type: Mapped[str] = mapped_column(Text, nullable=False)
size_bytes: Mapped[int] = mapped_column(Integer, nullable=False)
blob_sha256: Mapped[str] = mapped_column(Text, nullable=False)
created_at: Mapped[dt.datetime] = _now()
class Draft(Base):
__tablename__ = "drafts"
id: Mapped[uuid.UUID] = _uuid_pk()
user_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), nullable=False
)
account_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), ForeignKey("accounts.id", ondelete="CASCADE"), nullable=False
)
thread_id: Mapped[uuid.UUID | None] = mapped_column(
UUID(as_uuid=True), ForeignKey("threads.id")
)
reply_to_message_id: Mapped[uuid.UUID | None] = mapped_column(
UUID(as_uuid=True), ForeignKey("messages.id")
)
kind: Mapped[str] = mapped_column(Text, nullable=False, server_default=text("'reply'"))
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)
updated_at: Mapped[dt.datetime] = _now()
class Outbox(Base):
__tablename__ = "outbox"
id: Mapped[uuid.UUID] = _uuid_pk()
user_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), nullable=False)
draft_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), nullable=False)
state: Mapped[str] = mapped_column(Text, nullable=False, server_default=text("'undo_window'"))
send_after: Mapped[dt.datetime] = mapped_column(TIMESTAMP(timezone=True), nullable=False)
error: Mapped[str | None] = mapped_column(Text)
# Retries after a send the server could not take (0021): state "retrying".
attempts: Mapped[int] = mapped_column(Integer, nullable=False, server_default=text("0"))
next_attempt_at: Mapped[dt.datetime | None] = mapped_column(TIMESTAMP(timezone=True))
retry_until: Mapped[dt.datetime | None] = mapped_column(TIMESTAMP(timezone=True))
class SavedSearch(Base):
__tablename__ = "saved_searches"
__table_args__ = (UniqueConstraint("user_id", "name", name="uq_saved_searches_user_name"),)
id: Mapped[uuid.UUID] = _uuid_pk()
user_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), nullable=False
)
name: Mapped[str] = mapped_column(Text, nullable=False)
query: Mapped[str] = mapped_column(Text, nullable=False)
pinned: Mapped[bool] = mapped_column(nullable=False, server_default=text("true"))
class PushSubscription(Base):
"""A browser or installed app to tell about new mail (Web Push)."""
__tablename__ = "push_subscriptions"
id: Mapped[uuid.UUID] = _uuid_pk()
user_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
# The push service URL the browser handed out; posting to it reaches the device.
endpoint: Mapped[str] = mapped_column(Text, nullable=False, unique=True)
# The browser's keys the message is encrypted to (RFC 8291), base64url.
p256dh: Mapped[str] = mapped_column(Text, nullable=False)
auth: Mapped[str] = mapped_column(Text, nullable=False)
# The sign-in it was made under: nothing is sent once that session is gone.
session_sid: Mapped[str] = mapped_column(Text, nullable=False)
user_agent: Mapped[str] = mapped_column(Text, nullable=False, server_default=text("''"))
created_at: Mapped[dt.datetime] = _now()
last_used_at: Mapped[dt.datetime | None] = mapped_column(TIMESTAMP(timezone=True))
class PluginKv(Base):
"""Namespaced key/value store for plugins (SPEC §11 ``plugin_kv``).
Core-owned infrastructure (not a ``plg_``-prefixed plugin table): every plugin
gets a slice keyed by (plugin, user_id, key). ``value`` is arbitrary JSON.
Invariant I-7 (user-data rows carry ``user_id``) holds here too."""
__tablename__ = "plugin_kv"
plugin: Mapped[str] = mapped_column(Text, primary_key=True)
user_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), primary_key=True
)
key: Mapped[str] = mapped_column(Text, primary_key=True)
value: Mapped[Any] = mapped_column(JSONB, nullable=False)
class KnownRecipient(Base):
__tablename__ = "known_recipients"
__table_args__ = (
Index(
"ix_known_recipients_user_email",
"user_id",
text("email text_pattern_ops"),
),
)
user_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey("users.id", ondelete="CASCADE"),
primary_key=True,
)
email: Mapped[str] = mapped_column(CITEXT, primary_key=True)
name: Mapped[str] = mapped_column(Text, nullable=False, server_default=text("''"))
times_used: Mapped[int] = mapped_column(Integer, nullable=False, server_default=text("0"))
last_used_at: Mapped[dt.datetime] = mapped_column(TIMESTAMP(timezone=True), nullable=False)
__all__ = [
"Account",
"Attachment",
"BodyPart",
"Draft",
"KnownRecipient",
"Mailbox",
"Message",
"Outbox",
"SavedSearch",
"Thread",
"User",
]
Reference in New Issue View Git Blame Copy Permalink