feat!: remove direct messaging (chat)
E2EE direct messages, the chat dock, the Messages module, the message WebSocket and its realtime registry, and the messages / message_keys tables (migration 0097; the downgrade restores the empty schema). Prod held two welcome notes, no message between people and one key — never used (owner decision, E34). WebSocket now has no owner module: the lint bans it everywhere. Privacy policy updated: it no longer promises E2EE for a feature that is gone.
This commit is contained in:
@@ -824,96 +824,6 @@
|
||||
}
|
||||
}
|
||||
],
|
||||
"./app/routes/messages.py": [
|
||||
{
|
||||
"code": "reportCallIssue",
|
||||
"range": {
|
||||
"startColumn": 11,
|
||||
"endColumn": 43,
|
||||
"lineCount": 1
|
||||
}
|
||||
},
|
||||
{
|
||||
"code": "reportArgumentType",
|
||||
"range": {
|
||||
"startColumn": 23,
|
||||
"endColumn": 27,
|
||||
"lineCount": 1
|
||||
}
|
||||
},
|
||||
{
|
||||
"code": "reportCallIssue",
|
||||
"range": {
|
||||
"startColumn": 13,
|
||||
"endColumn": 5,
|
||||
"lineCount": 7
|
||||
}
|
||||
},
|
||||
{
|
||||
"code": "reportArgumentType",
|
||||
"range": {
|
||||
"startColumn": 8,
|
||||
"endColumn": 15,
|
||||
"lineCount": 5
|
||||
}
|
||||
},
|
||||
{
|
||||
"code": "reportReturnType",
|
||||
"range": {
|
||||
"startColumn": 15,
|
||||
"endColumn": 72,
|
||||
"lineCount": 1
|
||||
}
|
||||
},
|
||||
{
|
||||
"code": "reportCallIssue",
|
||||
"range": {
|
||||
"startColumn": 30,
|
||||
"endColumn": 48,
|
||||
"lineCount": 1
|
||||
}
|
||||
},
|
||||
{
|
||||
"code": "reportArgumentType",
|
||||
"range": {
|
||||
"startColumn": 41,
|
||||
"endColumn": 44,
|
||||
"lineCount": 1
|
||||
}
|
||||
},
|
||||
{
|
||||
"code": "reportOptionalMemberAccess",
|
||||
"range": {
|
||||
"startColumn": 36,
|
||||
"endColumn": 38,
|
||||
"lineCount": 1
|
||||
}
|
||||
},
|
||||
{
|
||||
"code": "reportOptionalMemberAccess",
|
||||
"range": {
|
||||
"startColumn": 71,
|
||||
"endColumn": 73,
|
||||
"lineCount": 1
|
||||
}
|
||||
},
|
||||
{
|
||||
"code": "reportArgumentType",
|
||||
"range": {
|
||||
"startColumn": 30,
|
||||
"endColumn": 39,
|
||||
"lineCount": 1
|
||||
}
|
||||
},
|
||||
{
|
||||
"code": "reportOptionalMemberAccess",
|
||||
"range": {
|
||||
"startColumn": 27,
|
||||
"endColumn": 29,
|
||||
"lineCount": 1
|
||||
}
|
||||
}
|
||||
],
|
||||
"./app/routes/playlists.py": [
|
||||
{
|
||||
"code": "reportOptionalOperand",
|
||||
|
||||
@@ -0,0 +1,69 @@
|
||||
"""messages: drop direct messaging (the owner's decision, 2026-10-04)
|
||||
|
||||
Chat — end-to-end encrypted direct messages plus the server's welcome note — was never used: prod held
|
||||
two welcome notes, no message between people, and one key. Its tables go with its code. The rows are
|
||||
not migrated anywhere; the nightly backup holds them.
|
||||
|
||||
The downgrade restores the schema as it stood (0026 + 0027), empty: a message can't be brought back
|
||||
from a migration, and the E2EE key blobs were only ever readable by their owner's browser anyway.
|
||||
"""
|
||||
|
||||
import sqlalchemy as sa
|
||||
from alembic import op
|
||||
|
||||
revision: str = "0097_drop_messages"
|
||||
down_revision: str | None = "0096_videos_sort_indexes"
|
||||
branch_labels: str | None = None
|
||||
depends_on: str | None = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.drop_table("message_keys")
|
||||
op.drop_index("ix_messages_created_at", table_name="messages")
|
||||
op.drop_index("ix_messages_recipient_id", table_name="messages")
|
||||
op.drop_index("ix_messages_sender_id", table_name="messages")
|
||||
op.drop_table("messages")
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.create_table(
|
||||
"messages",
|
||||
sa.Column("id", sa.Integer(), primary_key=True),
|
||||
sa.Column(
|
||||
"sender_id", sa.Integer(), sa.ForeignKey("users.id", ondelete="CASCADE"), nullable=True
|
||||
),
|
||||
sa.Column(
|
||||
"recipient_id",
|
||||
sa.Integer(),
|
||||
sa.ForeignKey("users.id", ondelete="CASCADE"),
|
||||
nullable=False,
|
||||
),
|
||||
sa.Column("body", sa.Text(), nullable=True),
|
||||
sa.Column("read_at", sa.DateTime(timezone=True), nullable=True),
|
||||
sa.Column(
|
||||
"created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False
|
||||
),
|
||||
sa.Column("kind", sa.String(length=16), nullable=False, server_default="user"),
|
||||
sa.Column("ciphertext", sa.Text(), nullable=True),
|
||||
sa.Column("iv", sa.String(length=32), nullable=True),
|
||||
)
|
||||
op.create_index("ix_messages_sender_id", "messages", ["sender_id"])
|
||||
op.create_index("ix_messages_recipient_id", "messages", ["recipient_id"])
|
||||
op.create_index("ix_messages_created_at", "messages", ["created_at"])
|
||||
op.create_table(
|
||||
"message_keys",
|
||||
sa.Column(
|
||||
"user_id",
|
||||
sa.Integer(),
|
||||
sa.ForeignKey("users.id", ondelete="CASCADE"),
|
||||
primary_key=True,
|
||||
),
|
||||
sa.Column("public_key", sa.Text(), nullable=False),
|
||||
sa.Column("wrapped_private_key", sa.Text(), nullable=False),
|
||||
sa.Column("salt", sa.String(length=64), nullable=False),
|
||||
sa.Column("wrap_iv", sa.String(length=32), nullable=False),
|
||||
sa.Column("key_check", sa.Text(), nullable=False),
|
||||
sa.Column(
|
||||
"created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False
|
||||
),
|
||||
)
|
||||
@@ -21,7 +21,7 @@ Three layers, by *who* runs the code — not by feature:
|
||||
`issues/bundle.py` is the one definition of a bug report's files, for the zip export and the tracker.
|
||||
|
||||
3. **Flat `app/*.py` core modules — cross-cutting primitives.** `auth`, `sysconfig`, `quota`, `audit`,
|
||||
`notifications`, `userscope`, `realtime`, `ratelimit`, `progress`, `logs` (the one logging
|
||||
`notifications`, `userscope`, `ratelimit`, `progress`, `logs` (the one logging
|
||||
configuration), `logsink` (its `log_events` table), `diagnostics` (a user's consented log
|
||||
level), `debugsession` (a recording that tags its user's lines, E16) and `collation` (A→Z in the
|
||||
reader's alphabet, E30), … are single-file primitives many layers import. They depend on nothing above them (a core
|
||||
|
||||
+3
-4
@@ -278,8 +278,7 @@ ACTIVE_ACCOUNT_HEADER = "x-siftlode-account"
|
||||
|
||||
def resolved_user_id(conn: HTTPConnection, *, strict: bool = True) -> tuple[int | None, bool]:
|
||||
"""Which account this connection acts as, and whether that's the session's default account.
|
||||
Takes any HTTPConnection — an HTTP Request OR a WebSocket — so the WS push channel shares this
|
||||
exact per-tab resolution instead of re-implementing it.
|
||||
Takes an HTTPConnection rather than a Request: it reads only the session and the headers.
|
||||
|
||||
The signed cookie holds the *wallet* (`account_ids`, every account signed into this browser)
|
||||
plus a default `user_id`. A caller may name its account with the X-Siftlode-Account header (or
|
||||
@@ -297,8 +296,8 @@ def resolved_user_id(conn: HTTPConnection, *, strict: bool = True) -> tuple[int
|
||||
"""
|
||||
default_id = conn.session.get("user_id")
|
||||
wallet = conn.session.get("account_ids") or []
|
||||
# Header for normal XHR; ?account= for contexts that can't set headers (WebSocket, and a
|
||||
# plain <a> file download). Both are wallet-gated below.
|
||||
# Header for normal XHR; ?account= for a context that can't set headers (a plain <a> file
|
||||
# download). Both are wallet-gated below.
|
||||
hdr = conn.headers.get(ACTIVE_ACCOUNT_HEADER) or conn.query_params.get("account")
|
||||
if hdr:
|
||||
try:
|
||||
|
||||
@@ -688,7 +688,6 @@ _POLLED_ROUTES = frozenset(
|
||||
{
|
||||
"/api/me",
|
||||
"/api/me/notifications/unread_count",
|
||||
"/api/messages/unread_count",
|
||||
"/api/downloads",
|
||||
"/api/downloads/index",
|
||||
"/api/downloads/usage",
|
||||
|
||||
@@ -44,7 +44,6 @@ from app.routes import ( # noqa: E402 — logging is configured first
|
||||
feed,
|
||||
health,
|
||||
me,
|
||||
messages,
|
||||
notifications,
|
||||
playlists,
|
||||
plex as plex_routes,
|
||||
@@ -311,7 +310,6 @@ app.include_router(me.router)
|
||||
app.include_router(diagnostics_routes.router)
|
||||
app.include_router(debug_sessions_routes.router)
|
||||
app.include_router(notifications.router)
|
||||
app.include_router(messages.router)
|
||||
app.include_router(channels.router)
|
||||
app.include_router(playlists.router)
|
||||
app.include_router(saved_views.router)
|
||||
|
||||
+1
-62
@@ -59,7 +59,7 @@ POSITION_MAX_S = 2**31 - 1
|
||||
|
||||
class TimestampMixin:
|
||||
"""Adds a server-defaulted `created_at`. Models that index created_at or order it
|
||||
differently (QuotaEvent, Notification, Message) keep their own column on purpose."""
|
||||
differently (QuotaEvent, Notification) keep their own column on purpose."""
|
||||
|
||||
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
|
||||
|
||||
@@ -814,44 +814,6 @@ class Notification(Base):
|
||||
)
|
||||
|
||||
|
||||
class Message(Base):
|
||||
"""A message in the notification module (phase 2).
|
||||
|
||||
Two kinds share this table:
|
||||
- `kind="user"`: a private, END-TO-END ENCRYPTED direct message between two users. The
|
||||
server only ever stores `ciphertext` + `iv` (base64); it never holds the plaintext or
|
||||
the keys, so not even an admin can read it. A conversation is just the set of messages
|
||||
between two users — no separate thread entity; queries group by the {sender, recipient}
|
||||
pair.
|
||||
- `kind="system"`: a server-authored plaintext message (e.g. the welcome, or future admin
|
||||
announcements). `sender_id` is NULL and `body` holds the text. These are non-private
|
||||
boilerplate, readable without any E2EE key setup, and surface as a "Siftlode" conversation.
|
||||
|
||||
`read_at` is stamped when the recipient opens the thread (NULL = unread, feeds the nav
|
||||
badge). The recipient FK cascades (and sender, for user messages) so a GDPR account erasure
|
||||
removes that user's messages outright."""
|
||||
|
||||
__tablename__ = "messages"
|
||||
|
||||
id: Mapped[int] = mapped_column(primary_key=True)
|
||||
kind: Mapped[str] = mapped_column(String(16), default="user", server_default="user")
|
||||
# NULL for system messages (no human sender).
|
||||
sender_id: Mapped[int | None] = mapped_column(
|
||||
ForeignKey("users.id", ondelete="CASCADE"), index=True
|
||||
)
|
||||
recipient_id: Mapped[int] = mapped_column(
|
||||
ForeignKey("users.id", ondelete="CASCADE"), index=True
|
||||
)
|
||||
# System messages carry plaintext `body`; user messages carry `ciphertext`+`iv` (base64).
|
||||
body: Mapped[str | None] = mapped_column(Text)
|
||||
ciphertext: Mapped[str | None] = mapped_column(Text)
|
||||
iv: Mapped[str | None] = mapped_column(String(32))
|
||||
read_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
|
||||
created_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), server_default=func.now(), index=True
|
||||
)
|
||||
|
||||
|
||||
class MediaAsset(Base, TimestampMixin):
|
||||
"""A physical downloaded media file — the SHARED cache layer of the download center.
|
||||
|
||||
@@ -1094,29 +1056,6 @@ class DownloadCredential(Base, TimestampMixin, UpdatedAtMixin):
|
||||
cookies_enc: Mapped[str] = mapped_column(Text) # Fernet ciphertext — never returned/logged
|
||||
|
||||
|
||||
class MessageKey(Base, TimestampMixin):
|
||||
"""A user's end-to-end-encryption key material for direct messaging (phase 2).
|
||||
|
||||
One row per user. `public_key` (base64 SPKI, ECDH P-256) is shared with others so they can
|
||||
derive the pairwise conversation key. `wrapped_private_key` is the user's private key
|
||||
encrypted IN THE BROWSER with a key derived (PBKDF2) from a message passphrase the server
|
||||
never sees — so the server stores only an opaque blob it cannot unwrap. `salt`/`wrap_iv`
|
||||
parameterise that derivation; `key_check` is a small ciphertext the client decrypts to
|
||||
verify a passphrase on unlock. The server is purely a key directory + blob store; it can
|
||||
never read any message."""
|
||||
|
||||
__tablename__ = "message_keys"
|
||||
|
||||
user_id: Mapped[int] = mapped_column(
|
||||
ForeignKey("users.id", ondelete="CASCADE"), primary_key=True
|
||||
)
|
||||
public_key: Mapped[str] = mapped_column(Text)
|
||||
wrapped_private_key: Mapped[str] = mapped_column(Text)
|
||||
salt: Mapped[str] = mapped_column(String(64))
|
||||
wrap_iv: Mapped[str] = mapped_column(String(32))
|
||||
key_check: Mapped[str] = mapped_column(Text)
|
||||
|
||||
|
||||
# --- Plex integration (optional) -------------------------------------------------------------
|
||||
# A dedicated, self-contained catalog mirrored from a locally-reachable Plex server. Playback is
|
||||
# from the LOCAL physical file (Plex is metadata only); per-user watch state lives in plex_states
|
||||
|
||||
@@ -1,68 +0,0 @@
|
||||
"""In-process WebSocket connection registry for live server→client push.
|
||||
|
||||
Generic groundwork (notification module phase 2 is the first user): connections are kept in
|
||||
this process's memory, keyed by user_id, so a request handler can push an event to all of a
|
||||
user's open tabs/devices. Single-uvicorn-process assumption — if we ever run multiple workers,
|
||||
replace the in-memory map with a Redis pub/sub fan-out behind the same `push()` interface (same
|
||||
note as app.ratelimit).
|
||||
|
||||
`push()` is intentionally SYNC-callable: ordinary (sync) route handlers run in a threadpool, so
|
||||
they schedule the actual send onto the WebSocket event loop via run_coroutine_threadsafe. The
|
||||
connect/disconnect/_push coroutines all run on that one loop, so the connection map is only ever
|
||||
mutated from a single thread.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
from collections import defaultdict
|
||||
|
||||
from starlette.websockets import WebSocket
|
||||
from app import logs
|
||||
|
||||
log = logs.get_logger(__name__)
|
||||
|
||||
|
||||
class ConnectionManager:
|
||||
def __init__(self) -> None:
|
||||
self._conns: dict[int, set[WebSocket]] = defaultdict(set)
|
||||
self._loop: asyncio.AbstractEventLoop | None = None
|
||||
|
||||
async def connect(self, user_id: int, ws: WebSocket) -> None:
|
||||
self._loop = asyncio.get_running_loop()
|
||||
await ws.accept()
|
||||
self._conns[user_id].add(ws)
|
||||
|
||||
async def disconnect(self, user_id: int, ws: WebSocket) -> None:
|
||||
conns = self._conns.get(user_id)
|
||||
if conns:
|
||||
conns.discard(ws)
|
||||
if not conns:
|
||||
self._conns.pop(user_id, None)
|
||||
|
||||
async def _push(self, user_id: int, payload: dict) -> None:
|
||||
for ws in list(self._conns.get(user_id, ())):
|
||||
try:
|
||||
await ws.send_json(payload)
|
||||
except Exception as exc: # noqa: BLE001 — one dead socket must not stop the others
|
||||
# Stale socket; it'll be cleaned up by its own disconnect handler. But this is
|
||||
# also where a live update goes missing: the UI simply never changes, and there
|
||||
# is no request behind it whose failure anyone could look up.
|
||||
log.debug(
|
||||
"a live update could not be pushed to a socket",
|
||||
extra=logs.event(
|
||||
"realtime.push_failed", user_id=user_id, reason=type(exc).__name__
|
||||
),
|
||||
)
|
||||
|
||||
def push(self, user_id: int, payload: dict) -> None:
|
||||
"""Schedule a push onto the WebSocket event loop. Safe to call from a sync route
|
||||
handler (threadpool). No-op if no WebSocket has ever connected (no loop captured)."""
|
||||
loop = self._loop
|
||||
if loop is None:
|
||||
return
|
||||
try:
|
||||
asyncio.run_coroutine_threadsafe(self._push(user_id, payload), loop)
|
||||
except Exception: # noqa: BLE001 — a closed loop or socket must not fail the caller
|
||||
log.debug("ws push failed", exc_info=True)
|
||||
|
||||
|
||||
manager = ConnectionManager()
|
||||
@@ -1,479 +0,0 @@
|
||||
"""Direct messaging (notification module, phase 2): end-to-end encrypted user-to-user chat
|
||||
plus server-authored system messages, with live WebSocket delivery.
|
||||
|
||||
E2EE model: every private message is encrypted IN THE BROWSER under a key both parties derive
|
||||
from ECDH(their private key, the other's public key). The server only ever stores ciphertext +
|
||||
iv and acts as a key directory (public keys) and an opaque blob store (each user's private key,
|
||||
wrapped client-side with a passphrase the server never sees). So not even an admin can read a
|
||||
user-to-user conversation. `kind="system"` messages (the welcome; future announcements) are
|
||||
plain server boilerplate — non-private, readable without any key setup, shown as a "Siftlode"
|
||||
conversation.
|
||||
|
||||
Every endpoint is gated by `require_human` (the shared demo account can't message). Metadata
|
||||
(who messaged whom, when, read state) is NOT encrypted — standard for E2EE and needed for
|
||||
routing and unread counts.
|
||||
"""
|
||||
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from fastapi import APIRouter, Depends, WebSocket, WebSocketDisconnect
|
||||
from pydantic import BaseModel
|
||||
from sqlalchemy import and_, case, delete, func, or_, select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.mailtext import lang_for_user
|
||||
from app.collation import collated
|
||||
from app.auth import require_human, resolved_user_id
|
||||
from app.db import SessionLocal, get_db
|
||||
from app.models import Message, MessageKey, User
|
||||
from app.ratelimit import RateLimiter
|
||||
from app.realtime import manager
|
||||
from app.security import verify_password
|
||||
from app.userscope import is_messageable_user, messageable_clauses
|
||||
from app.utils import iso
|
||||
from app.refusal import refused
|
||||
from app.routes._common import not_found
|
||||
|
||||
router = APIRouter(prefix="/api/messages", tags=["messages"])
|
||||
|
||||
MAX_CIPHERTEXT = 8000 # base64 of a generous plaintext + AES-GCM overhead
|
||||
SYSTEM_PARTNER_ID = 0 # synthetic conversation partner for system messages
|
||||
SYSTEM_NAME = "Siftlode"
|
||||
_send_limiter = RateLimiter(max_events=30, window_seconds=60)
|
||||
_reset_key_limiter = RateLimiter(max_events=5, window_seconds=300)
|
||||
|
||||
# The welcome is rendered server-side at creation time, so it carries its own translations
|
||||
# (one per supported UI language). Kept short and non-private (every new user gets the same text).
|
||||
WELCOME = {
|
||||
"en": (
|
||||
"Welcome to Siftlode! 👋\n\n"
|
||||
"This is your private message inbox. Conversations with other members are end-to-end "
|
||||
"encrypted — only you and the person you're chatting with can read them, not even an "
|
||||
"admin. To start messaging someone, set up secure messaging with a passphrase first.\n\n"
|
||||
"Enjoy your Feed!"
|
||||
),
|
||||
"hu": (
|
||||
"Üdv a Siftlode-on! 👋\n\n"
|
||||
"Ez a privát üzenet-postafiókod. A többi taggal folytatott beszélgetések végpontig "
|
||||
"titkosítottak — csak te és a beszélgetőpartnered olvashatja őket, még az admin sem. "
|
||||
"Ha üzenni szeretnél valakinek, előbb állítsd be a biztonságos üzenetküldést egy "
|
||||
"jelszóval.\n\n"
|
||||
"Jó böngészést!"
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
class KeySetupIn(BaseModel):
|
||||
public_key: str
|
||||
wrapped_private_key: str
|
||||
salt: str
|
||||
wrap_iv: str
|
||||
key_check: str
|
||||
|
||||
|
||||
class KeyResetIn(BaseModel):
|
||||
# Present only for password accounts; a Google-only account has no password to re-verify.
|
||||
current_password: str | None = None
|
||||
|
||||
|
||||
class SendIn(BaseModel):
|
||||
recipient_id: int
|
||||
ciphertext: str
|
||||
iv: str
|
||||
|
||||
|
||||
def _user_label(u: User) -> str:
|
||||
return u.display_name or u.email.split("@")[0]
|
||||
|
||||
|
||||
def _serialize_user(u: User) -> dict:
|
||||
return {"id": u.id, "name": _user_label(u), "avatar_url": u.avatar_url}
|
||||
|
||||
|
||||
def _system_partner() -> dict:
|
||||
return {"id": SYSTEM_PARTNER_ID, "name": SYSTEM_NAME, "avatar_url": None}
|
||||
|
||||
|
||||
# The welcome's opening line in every language — how a stored copy is recognised. It was rendered
|
||||
# once, in the language of the day it was created, so an account that later switched language kept
|
||||
# reading it in the old one (E31 S1: the owner's, in English). Recognised here, it is served in the
|
||||
# reader's language instead. Derived from WELCOME itself, so a change to an OPENING line, or a
|
||||
# language taken out of WELCOME, orphans the copies already stored: they are served as stored. The
|
||||
# one retired language (German, 2026-07) left none on prod (checked 2026-10-02).
|
||||
_WELCOME_OPENINGS = {text.partition("\n")[0] for text in WELCOME.values()}
|
||||
|
||||
|
||||
def _system_body(body: str | None, lang: str | None) -> str | None:
|
||||
"""A system message's text for a reader of `lang`: the welcome in their language, any other as
|
||||
stored."""
|
||||
if body is None or lang is None or body.partition("\n")[0] not in _WELCOME_OPENINGS:
|
||||
return body
|
||||
return WELCOME.get(lang, WELCOME["en"])
|
||||
|
||||
|
||||
def _serialize_msg(m: Message, lang: str | None = None) -> dict:
|
||||
"""`lang`: the reader's language, for a system message's text (a user message is ciphertext)."""
|
||||
return {
|
||||
"id": m.id,
|
||||
"kind": m.kind,
|
||||
"sender_id": m.sender_id,
|
||||
"recipient_id": m.recipient_id,
|
||||
"body": _system_body(m.body, lang) if m.kind == "system" else m.body,
|
||||
"ciphertext": m.ciphertext,
|
||||
"iv": m.iv,
|
||||
"read_at": iso(m.read_at),
|
||||
"created_at": iso(m.created_at),
|
||||
}
|
||||
|
||||
|
||||
def ensure_welcome(db: Session, user: User) -> None:
|
||||
"""Create the one-time system welcome message for a (human) user the first time it's needed.
|
||||
Idempotent: guarded on whether any system message already exists for them."""
|
||||
if user.is_demo:
|
||||
return
|
||||
exists = db.scalar(
|
||||
select(Message.id).where(Message.recipient_id == user.id, Message.kind == "system").limit(1)
|
||||
)
|
||||
if exists:
|
||||
return
|
||||
lang = (user.preferences or {}).get("language")
|
||||
body = WELCOME.get(lang, WELCOME["en"])
|
||||
m = Message(kind="system", sender_id=None, recipient_id=user.id, body=body)
|
||||
db.add(m)
|
||||
db.commit()
|
||||
db.refresh(m)
|
||||
manager.push(user.id, {"type": "message", "message": _serialize_msg(m, lang_for_user(user))})
|
||||
|
||||
|
||||
# --- E2EE key directory / blob store -----------------------------------------
|
||||
|
||||
|
||||
@router.get("/keys/me")
|
||||
def my_key(user: User = Depends(require_human), db: Session = Depends(get_db)) -> dict:
|
||||
"""My own key bundle. `configured` says whether secure messaging is set up; if so the
|
||||
wrapped blob + params are returned so this device can unlock with the passphrase."""
|
||||
k = db.get(MessageKey, user.id)
|
||||
if k is None:
|
||||
return {"configured": False}
|
||||
return {
|
||||
"configured": True,
|
||||
"public_key": k.public_key,
|
||||
"wrapped_private_key": k.wrapped_private_key,
|
||||
"salt": k.salt,
|
||||
"wrap_iv": k.wrap_iv,
|
||||
"key_check": k.key_check,
|
||||
}
|
||||
|
||||
|
||||
@router.post("/keys")
|
||||
def setup_keys(
|
||||
payload: KeySetupIn,
|
||||
user: User = Depends(require_human),
|
||||
db: Session = Depends(get_db),
|
||||
) -> dict:
|
||||
"""Store this user's public key + passphrase-wrapped private key. Set-once: changing the
|
||||
key would orphan every existing conversation key, so re-setup is rejected here."""
|
||||
if db.get(MessageKey, user.id) is not None:
|
||||
raise refused(409, "messaging_set_up")
|
||||
db.add(
|
||||
MessageKey(
|
||||
user_id=user.id,
|
||||
public_key=payload.public_key,
|
||||
wrapped_private_key=payload.wrapped_private_key,
|
||||
salt=payload.salt,
|
||||
wrap_iv=payload.wrap_iv,
|
||||
key_check=payload.key_check,
|
||||
)
|
||||
)
|
||||
db.commit()
|
||||
return {"ok": True, "configured": True}
|
||||
|
||||
|
||||
@router.post("/keys/reset")
|
||||
def reset_keys(
|
||||
payload: KeyResetIn,
|
||||
user: User = Depends(require_human),
|
||||
db: Session = Depends(get_db),
|
||||
) -> dict:
|
||||
"""Forgotten-passphrase escape hatch: drop this user's secure-messaging key so they can set a
|
||||
NEW passphrase (the set-once guard in setup_keys otherwise makes a lost passphrase a permanent
|
||||
wall). Destroys the user's entire E2EE history BY DESIGN — every conversation key is derived
|
||||
from this keypair, so replacing it orphans every stored ciphertext in both directions. Those
|
||||
now-undecryptable `user` messages are deleted here (a clean slate for both parties) rather than
|
||||
left as permanent "can't decrypt" clutter; `system` messages (plaintext) are untouched.
|
||||
|
||||
An account WITH a password must re-authenticate: a hijacked session must not be able to silently
|
||||
reset messaging and then read the victim's future messages under a new passphrase. A password-
|
||||
less (Google-only) account has nothing to re-verify, so it relies on its live session plus the
|
||||
client-side danger confirmation."""
|
||||
if not _reset_key_limiter.allow(str(user.id)):
|
||||
raise refused(429, "too_many_attempts")
|
||||
if user.password_hash and not verify_password(
|
||||
payload.current_password or "", user.password_hash
|
||||
):
|
||||
raise refused(403, "wrong_password")
|
||||
# Delete the now-dead ciphertext first (both directions), then the wrapped keypair itself so a
|
||||
# fresh setup is allowed. Order doesn't matter (no FK between them), but do it in one commit.
|
||||
db.execute(
|
||||
delete(Message).where(
|
||||
Message.kind == "user",
|
||||
or_(Message.sender_id == user.id, Message.recipient_id == user.id),
|
||||
)
|
||||
)
|
||||
key = db.get(MessageKey, user.id)
|
||||
if key is not None:
|
||||
db.delete(key)
|
||||
db.commit()
|
||||
return {"ok": True, "configured": False}
|
||||
|
||||
|
||||
@router.get("/keys/{user_id}")
|
||||
def public_key(
|
||||
user_id: int, user: User = Depends(require_human), db: Session = Depends(get_db)
|
||||
) -> dict:
|
||||
"""A member's public key, for deriving the pairwise conversation key. MB2 (like get_thread):
|
||||
resolve the target only if they're messageable OR we already share a thread, else the same
|
||||
"User not found" as a non-existent id — so this can't be probed to enumerate user_ids or to
|
||||
fetch keys for suspended/deactivated/demo accounts."""
|
||||
cond = and_(
|
||||
Message.kind == "user",
|
||||
or_(
|
||||
and_(Message.sender_id == user.id, Message.recipient_id == user_id),
|
||||
and_(Message.sender_id == user_id, Message.recipient_id == user.id),
|
||||
),
|
||||
)
|
||||
p = db.get(User, user_id)
|
||||
has_history = db.scalar(select(Message.id).where(cond).limit(1))
|
||||
if p is None or (not is_messageable_user(p) and not has_history):
|
||||
raise not_found("user")
|
||||
k = db.get(MessageKey, user_id)
|
||||
if k is None:
|
||||
raise refused(404, "no_messaging")
|
||||
return {"user_id": user_id, "public_key": k.public_key}
|
||||
|
||||
|
||||
# --- directory + conversations -----------------------------------------------
|
||||
|
||||
|
||||
@router.get("/users")
|
||||
def list_recipients(
|
||||
user: User = Depends(require_human), db: Session = Depends(get_db)
|
||||
) -> list[dict]:
|
||||
"""Member directory for the recipient picker: messageable members (except me) who have set
|
||||
up secure messaging (so a conversation key can be derived)."""
|
||||
rows = (
|
||||
db.execute(
|
||||
select(User)
|
||||
.join(MessageKey, MessageKey.user_id == User.id)
|
||||
.where(User.id != user.id, *messageable_clauses())
|
||||
.order_by(collated(func.coalesce(User.display_name, User.email), lang_for_user(user)))
|
||||
)
|
||||
.scalars()
|
||||
.all()
|
||||
)
|
||||
return [_serialize_user(u) for u in rows]
|
||||
|
||||
|
||||
@router.get("/conversations")
|
||||
def list_conversations(user: User = Depends(require_human), db: Session = Depends(get_db)) -> dict:
|
||||
"""One entry per conversation: each user partner (E2EE) plus the system "Siftlode" thread,
|
||||
with the latest message + my unread count, newest first. Lazily creates the welcome on the
|
||||
first open. User-message previews are ciphertext — the client decrypts them."""
|
||||
ensure_welcome(db, user)
|
||||
# The conversation partner for a message, from this user's perspective (system → 0).
|
||||
partner_expr = case(
|
||||
(Message.kind == "system", SYSTEM_PARTNER_ID),
|
||||
(Message.sender_id == user.id, Message.recipient_id),
|
||||
else_=Message.sender_id,
|
||||
)
|
||||
mine = or_(Message.sender_id == user.id, Message.recipient_id == user.id)
|
||||
# Latest message per conversation in ONE query (DISTINCT ON, Postgres) instead of loading the
|
||||
# user's whole message history into Python — bounded by the number of conversations.
|
||||
latest = (
|
||||
db.execute(
|
||||
select(Message)
|
||||
.where(mine)
|
||||
.distinct(partner_expr)
|
||||
.order_by(partner_expr, Message.created_at.desc())
|
||||
)
|
||||
.scalars()
|
||||
.all()
|
||||
)
|
||||
# Unread incoming, grouped per conversation.
|
||||
unread = dict(
|
||||
db.execute(
|
||||
select(partner_expr, func.count())
|
||||
.where(mine, Message.recipient_id == user.id, Message.read_at.is_(None))
|
||||
.group_by(partner_expr)
|
||||
).all()
|
||||
)
|
||||
|
||||
def _pid(m: Message) -> int:
|
||||
if m.kind == "system":
|
||||
return SYSTEM_PARTNER_ID
|
||||
return m.recipient_id if m.sender_id == user.id else m.sender_id
|
||||
|
||||
user_ids = [_pid(m) for m in latest if m.kind != "system"]
|
||||
partners = {}
|
||||
if user_ids:
|
||||
partners = {
|
||||
u.id: u for u in db.execute(select(User).where(User.id.in_(user_ids))).scalars().all()
|
||||
}
|
||||
items = []
|
||||
for m in latest:
|
||||
pid = _pid(m)
|
||||
if pid == SYSTEM_PARTNER_ID:
|
||||
partner = _system_partner()
|
||||
elif pid in partners:
|
||||
partner = _serialize_user(partners[pid])
|
||||
else:
|
||||
continue
|
||||
items.append(
|
||||
{
|
||||
"partner": partner,
|
||||
"last_message": _serialize_msg(m, lang_for_user(user)),
|
||||
"unread": int(unread.get(pid, 0)),
|
||||
}
|
||||
)
|
||||
items.sort(key=lambda x: x["last_message"]["created_at"] or "", reverse=True)
|
||||
return {"items": items}
|
||||
|
||||
|
||||
@router.get("/conversations/{partner_id}")
|
||||
def get_thread(
|
||||
partner_id: int,
|
||||
mark_read: bool = True,
|
||||
user: User = Depends(require_human),
|
||||
db: Session = Depends(get_db),
|
||||
) -> dict:
|
||||
"""Full thread with a partner (or the system thread when partner_id == 0), oldest first.
|
||||
Stamps unread incoming messages read unless mark_read=false."""
|
||||
if partner_id == SYSTEM_PARTNER_ID:
|
||||
ensure_welcome(db, user)
|
||||
partner = _system_partner()
|
||||
cond = and_(Message.recipient_id == user.id, Message.kind == "system")
|
||||
else:
|
||||
p = db.get(User, partner_id)
|
||||
cond = and_(
|
||||
Message.kind == "user",
|
||||
or_(
|
||||
and_(Message.sender_id == user.id, Message.recipient_id == partner_id),
|
||||
and_(Message.sender_id == partner_id, Message.recipient_id == user.id),
|
||||
),
|
||||
)
|
||||
# MB2: don't leak arbitrary user identities. Resolve a partner only if they're messageable
|
||||
# OR we already share a thread (so a conversation with a since-suspended/deactivated user
|
||||
# still opens). Otherwise the same 404 as a non-existent id → no enumeration oracle.
|
||||
has_history = db.scalar(select(Message.id).where(cond).limit(1))
|
||||
if p is None or (not is_messageable_user(p) and not has_history):
|
||||
raise not_found("user")
|
||||
partner = _serialize_user(p)
|
||||
msgs = (
|
||||
db.execute(select(Message).where(cond).order_by(Message.created_at.asc())).scalars().all()
|
||||
)
|
||||
if mark_read:
|
||||
now = datetime.now(timezone.utc)
|
||||
changed = False
|
||||
for m in msgs:
|
||||
if m.recipient_id == user.id and m.read_at is None:
|
||||
m.read_at = now
|
||||
changed = True
|
||||
if changed:
|
||||
db.commit()
|
||||
# MB3: ping this user's OTHER tabs/devices that their unread dropped so the badge
|
||||
# refreshes live instead of waiting for the next 20s poll. The client re-fetches the
|
||||
# count on this signal, so no number is carried here.
|
||||
manager.push(user.id, {"type": "unread"})
|
||||
return {"partner": partner, "items": [_serialize_msg(m, lang_for_user(user)) for m in msgs]}
|
||||
|
||||
|
||||
@router.post("")
|
||||
def send_message(
|
||||
payload: SendIn,
|
||||
user: User = Depends(require_human),
|
||||
db: Session = Depends(get_db),
|
||||
) -> dict:
|
||||
"""Send an end-to-end-encrypted message. The server stores only the ciphertext the client
|
||||
produced, then pushes it live to the recipient (and the sender's other devices)."""
|
||||
if not _send_limiter.allow(f"msg:{user.id}"):
|
||||
raise refused(429, "message_rate")
|
||||
ct = (payload.ciphertext or "").strip()
|
||||
iv = (payload.iv or "").strip()
|
||||
if not ct or not iv:
|
||||
raise refused(400, "message_empty")
|
||||
if len(ct) > MAX_CIPHERTEXT:
|
||||
raise refused(400, "message_too_long")
|
||||
if payload.recipient_id == user.id:
|
||||
raise refused(400, "message_self")
|
||||
recipient = db.get(User, payload.recipient_id)
|
||||
if not is_messageable_user(recipient):
|
||||
raise not_found("recipient")
|
||||
if db.get(MessageKey, recipient.id) is None:
|
||||
raise refused(404, "no_messaging")
|
||||
m = Message(kind="user", sender_id=user.id, recipient_id=recipient.id, ciphertext=ct, iv=iv)
|
||||
db.add(m)
|
||||
db.commit()
|
||||
db.refresh(m)
|
||||
# Carry both parties so each recipient can open/flash the right dock window (the partner is
|
||||
# whichever of from/to isn't them).
|
||||
event = {
|
||||
"type": "message",
|
||||
"message": _serialize_msg(m),
|
||||
"from": _serialize_user(user),
|
||||
"to": _serialize_user(recipient),
|
||||
}
|
||||
manager.push(recipient.id, event)
|
||||
manager.push(user.id, event) # the sender's other open devices
|
||||
return _serialize_msg(m)
|
||||
|
||||
|
||||
@router.get("/unread_count")
|
||||
def unread_count(user: User = Depends(require_human), db: Session = Depends(get_db)) -> dict:
|
||||
n = db.scalar(
|
||||
select(func.count())
|
||||
.select_from(Message)
|
||||
.where(Message.recipient_id == user.id, Message.read_at.is_(None))
|
||||
)
|
||||
return {"count": n or 0}
|
||||
|
||||
|
||||
# --- live delivery -----------------------------------------------------------
|
||||
|
||||
|
||||
@router.websocket("/ws")
|
||||
async def messages_ws(ws: WebSocket) -> None:
|
||||
"""Live push channel: authenticated by the session cookie, registers the connection so
|
||||
sent messages reach this user's open tabs instantly. We don't consume client→server frames
|
||||
(sending goes through POST /api/messages); the receive loop only detects disconnect."""
|
||||
# Which signed-in account this socket acts as — the SHARED per-tab resolution (a wallet-gated
|
||||
# ?account=, the session default when none is named, nobody for an account no longer signed in
|
||||
# here). A browser WebSocket can't send the X-Siftlode-Account header, so resolved_user_id falls
|
||||
# through to the ?account= query param.
|
||||
uid, _ = resolved_user_id(ws)
|
||||
if not uid:
|
||||
await ws.close(code=1008)
|
||||
return
|
||||
db = SessionLocal()
|
||||
try:
|
||||
user = db.get(User, uid)
|
||||
# SA4: honour server-side session revocation on the live channel too — reject a cookie whose
|
||||
# recorded epoch is behind the account's current one (mirrors current_user). Without this a
|
||||
# copied cookie could keep receiving pushes after a password reset / "log out everywhere".
|
||||
epochs = ws.session.get("epochs") or {}
|
||||
epoch_ok = user is not None and int(epochs.get(str(uid), 0)) == (user.session_epoch or 0)
|
||||
ok = epoch_ok and is_messageable_user(user)
|
||||
finally:
|
||||
db.close()
|
||||
if not ok:
|
||||
await ws.close(code=1008)
|
||||
return
|
||||
await manager.connect(uid, ws)
|
||||
try:
|
||||
while True:
|
||||
await ws.receive_text()
|
||||
# silent-ok: the ordinary way a chat socket ends — a closed tab. The `finally` below is the
|
||||
# whole handling, and nothing failed.
|
||||
except WebSocketDisconnect:
|
||||
pass
|
||||
finally:
|
||||
await manager.disconnect(uid, ws)
|
||||
@@ -11,7 +11,7 @@ Design:
|
||||
instantly (no re-download); a fresh asset is claimed pending→downloading by exactly one
|
||||
worker (a DB compare-and-set), so two jobs for the same video+format never double-download.
|
||||
* Live progress is written to the job row (short-lived sessions, throttled ~1/s) — the
|
||||
frontend polls it; the worker can't reach the API's in-process WebSocket registry.
|
||||
frontend polls it (the worker is a separate process with no channel to the browser).
|
||||
* Pause/cancel are cooperative: the progress hook re-reads the job status and aborts the
|
||||
yt-dlp download if it's no longer `running`.
|
||||
* Crash-safe: on startup, orphaned `running` jobs are requeued and `downloading` assets reset.
|
||||
|
||||
+1
-1
@@ -31,7 +31,7 @@
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "E2EE material and cookies",
|
||||
"name": "ciphertext, passphrases and cookies",
|
||||
"key": "",
|
||||
"value": {
|
||||
"ciphertext": "AAAA",
|
||||
|
||||
@@ -221,30 +221,13 @@ def _others(db, n_offset: int = 0) -> list[User]:
|
||||
return users
|
||||
|
||||
|
||||
def test_the_share_and_message_recipients_are_in_the_readers_alphabet(
|
||||
db, monkeypatch, override_dep
|
||||
):
|
||||
from app.models import MessageKey
|
||||
|
||||
def test_the_share_recipients_are_in_the_readers_alphabet(db, monkeypatch, override_dep):
|
||||
user = _user(db, "hu")
|
||||
others = _others(db)
|
||||
db.add_all(
|
||||
MessageKey(
|
||||
user_id=u.id,
|
||||
public_key="k",
|
||||
wrapped_private_key="w",
|
||||
salt="s",
|
||||
wrap_iv="i",
|
||||
key_check="c",
|
||||
)
|
||||
for u in others
|
||||
)
|
||||
_others(db)
|
||||
db.commit()
|
||||
client = _client(db, user, monkeypatch, override_dep)
|
||||
emails = [r["email"] for r in client.get("/api/downloads/recipients").json()]
|
||||
assert emails == [f"{w.lower()}@example.com" for w in HU]
|
||||
names = [r["name"] for r in client.get("/api/messages/users").json()]
|
||||
assert names == HU
|
||||
|
||||
|
||||
def test_the_blocked_channels_are_in_the_readers_alphabet(db, monkeypatch, override_dep):
|
||||
|
||||
@@ -219,11 +219,11 @@ def _client() -> TestClient:
|
||||
raise HTTPException(status_code=400, detail={"code": "bad_cursor"})
|
||||
return {"ok": True}
|
||||
|
||||
@app.get("/api/messages/unread_count")
|
||||
@app.get("/api/me/notifications/unread_count")
|
||||
def _unread():
|
||||
return {"count": 0}
|
||||
|
||||
@app.post("/api/messages/unread_count")
|
||||
@app.post("/api/me/notifications/unread_count")
|
||||
def _unread_write():
|
||||
return {"ok": True}
|
||||
|
||||
@@ -337,11 +337,11 @@ def test_a_polled_read_is_debug_a_write_to_the_same_route_is_not(sink):
|
||||
http = logging.getLogger("siftlode.http")
|
||||
http.setLevel(logs.TRACE)
|
||||
client = _client()
|
||||
r = client.get("/api/messages/unread_count")
|
||||
r = client.get("/api/me/notifications/unread_count")
|
||||
assert sink.objects()[-1]["level"] == "DEBUG"
|
||||
# ...and says so, for the browser's copy of the line (frontend lib/http.ts).
|
||||
assert r.headers["X-Log-Quiet"] == "1"
|
||||
r = client.post("/api/messages/unread_count")
|
||||
r = client.post("/api/me/notifications/unread_count")
|
||||
assert sink.objects()[-1]["level"] == "INFO"
|
||||
assert "X-Log-Quiet" not in r.headers
|
||||
assert client.get("/api/videos/abc").status_code == 200
|
||||
|
||||
@@ -1,53 +0,0 @@
|
||||
"""E31 S1: the welcome message is rendered once, when the account is created, in the language of that day.
|
||||
An account that switched language afterwards kept reading it in the old one (the owner's: English under a
|
||||
Hungarian UI). It is served in the reader's language now; any other system message stays as stored.
|
||||
Needs the DB."""
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
import app.state as state
|
||||
from app.auth import current_user
|
||||
from app.db import get_db
|
||||
from app.main import app
|
||||
from app.models import Message, User
|
||||
from app.routes.messages import WELCOME
|
||||
|
||||
|
||||
def _client(db, user, monkeypatch, override_dep) -> TestClient:
|
||||
monkeypatch.setattr(state, "_configured_cache", True)
|
||||
override_dep(get_db, db.info["request"])
|
||||
override_dep(current_user, lambda: user)
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
def _user_with_english_welcome(db) -> User:
|
||||
"""An account whose welcome was written in English, then switched to Hungarian."""
|
||||
user = User(email="welcome-hu@example.com", preferences={"language": "hu"})
|
||||
db.add(user)
|
||||
db.flush()
|
||||
db.add(Message(kind="system", sender_id=None, recipient_id=user.id, body=WELCOME["en"]))
|
||||
db.commit()
|
||||
return user
|
||||
|
||||
|
||||
def test_the_welcome_is_read_in_the_readers_language(db, monkeypatch, override_dep):
|
||||
user = _user_with_english_welcome(db)
|
||||
client = _client(db, user, monkeypatch, override_dep)
|
||||
|
||||
thread = client.get("/api/messages/conversations/0").json()["items"]
|
||||
assert [m["body"] for m in thread] == [WELCOME["hu"]]
|
||||
|
||||
convs = client.get("/api/messages/conversations").json()["items"]
|
||||
assert [c["last_message"]["body"] for c in convs if c["partner"]["id"] == 0] == [WELCOME["hu"]]
|
||||
|
||||
|
||||
def test_another_system_message_is_served_as_stored(db, monkeypatch, override_dep):
|
||||
user = _user_with_english_welcome(db)
|
||||
db.add(
|
||||
Message(kind="system", sender_id=None, recipient_id=user.id, body="Maintenance tonight.")
|
||||
)
|
||||
db.commit()
|
||||
client = _client(db, user, monkeypatch, override_dep)
|
||||
|
||||
bodies = [m["body"] for m in client.get("/api/messages/conversations/0").json()["items"]]
|
||||
assert bodies == [WELCOME["hu"], "Maintenance tonight."]
|
||||
@@ -23,11 +23,9 @@ COMMITTING_HELPERS = {
|
||||
"search.py:_ingest_candidates": "stages around YouTube calls: the stubs must survive an enrich "
|
||||
"that fails",
|
||||
"search.py:_classify_shorts": "the last stage of that ingest, after its probes",
|
||||
"auth.py:upsert_pending_invite": "commits only when it writes — the contract its three "
|
||||
"auth.py:upsert_pending_invite": "commits only when it writes — the contract its two "
|
||||
"callers rely on (`_register_account` persists its own flushed user because a no-op here "
|
||||
"commits nothing)",
|
||||
"messages.py:ensure_welcome": "the websocket push after it must announce a row the client "
|
||||
"can already fetch",
|
||||
"debug_sessions.py:_end_send": "gives back the claim a send took, in a `finally`: it runs when "
|
||||
"the handler is already unwinding and has no commit left to reach",
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user