From 002a79949bd55386b5acd5975df6db3f7270f2f3 Mon Sep 17 00:00:00 2001 From: npeter83 Date: Thu, 25 Jun 2026 22:05:35 +0200 Subject: [PATCH] feat(messages): end-to-end encrypted real-time direct messaging backend MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Private user-to-user messages are end-to-end encrypted: the server only ever stores ciphertext + iv and acts as a key directory (public keys) plus an opaque store for each user's private key, wrapped client-side with a passphrase the server never sees — so not even an admin can read a conversation. A separate kind=system message (plaintext, no sender) powers a server-authored Siftlode welcome shown on first open, reusable later for announcements. - models: rework Message (kind, nullable sender/body, ciphertext+iv) + MessageKey; migrations 0026 (table) + 0027 (E2EE rework). - routes/messages.py: key directory/blob endpoints, ciphertext send, conversations + threads (system + user), lazy welcome, all gated by require_human. - realtime.py: in-process WebSocket connection registry; /ws delivers sent messages to a user's open tabs instantly (sync-callable push, single-process). --- backend/alembic/versions/0026_messages.py | 56 +++ backend/alembic/versions/0027_message_e2ee.py | 68 ++++ backend/app/main.py | 2 + backend/app/models.py | 66 ++++ backend/app/realtime.py | 60 +++ backend/app/routes/messages.py | 373 ++++++++++++++++++ 6 files changed, 625 insertions(+) create mode 100644 backend/alembic/versions/0026_messages.py create mode 100644 backend/alembic/versions/0027_message_e2ee.py create mode 100644 backend/app/realtime.py create mode 100644 backend/app/routes/messages.py diff --git a/backend/alembic/versions/0026_messages.py b/backend/alembic/versions/0026_messages.py new file mode 100644 index 0000000..0ebb22b --- /dev/null +++ b/backend/alembic/versions/0026_messages.py @@ -0,0 +1,56 @@ +"""user-to-user messages + +Revision ID: 0026_messages +Revises: 0025_install_wizard +Create Date: 2026-06-25 + +Adds the `messages` table for direct user-to-user messaging (notification module phase 2). +A conversation is the set of rows between two users; both FKs cascade so a GDPR account +erasure removes that user's messages. +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0026_messages" +down_revision: Union[str, None] = "0025_install_wizard" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> 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=False, + ), + sa.Column( + "recipient_id", + sa.Integer(), + sa.ForeignKey("users.id", ondelete="CASCADE"), + nullable=False, + ), + sa.Column("body", sa.Text(), nullable=False), + 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, + ), + ) + 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"]) + + +def downgrade() -> None: + 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") diff --git a/backend/alembic/versions/0027_message_e2ee.py b/backend/alembic/versions/0027_message_e2ee.py new file mode 100644 index 0000000..2d72fa0 --- /dev/null +++ b/backend/alembic/versions/0027_message_e2ee.py @@ -0,0 +1,68 @@ +"""end-to-end encrypted messages + system messages + +Revision ID: 0027_message_e2ee +Revises: 0026_messages +Create Date: 2026-06-25 + +Reworks direct messaging for end-to-end encryption. The `messages` table now stores ciphertext +(+ iv) for private user-to-user messages instead of plaintext `body`; `body` is kept (nullable) +for server-authored `kind="system"` messages (welcome / announcements). Adds `message_keys` to +hold each user's public key + passphrase-wrapped private key (the server can never decrypt). + +Any pre-existing rows were plaintext with no keys, so they're dropped (the feature is unshipped; +prod has none). +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0027_message_e2ee" +down_revision: Union[str, None] = "0026_messages" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # Old rows were plaintext with no E2EE keys — can't be migrated; drop them. + op.execute("DELETE FROM messages") + op.add_column( + "messages", + sa.Column("kind", sa.String(length=16), nullable=False, server_default="user"), + ) + op.add_column("messages", sa.Column("ciphertext", sa.Text(), nullable=True)) + op.add_column("messages", sa.Column("iv", sa.String(length=32), nullable=True)) + # System messages have no human sender and carry plaintext `body`. + op.alter_column("messages", "sender_id", existing_type=sa.Integer(), nullable=True) + op.alter_column("messages", "body", existing_type=sa.Text(), nullable=True) + + 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, + ), + ) + + +def downgrade() -> None: + op.drop_table("message_keys") + op.execute("DELETE FROM messages") + op.alter_column("messages", "body", existing_type=sa.Text(), nullable=False) + op.alter_column("messages", "sender_id", existing_type=sa.Integer(), nullable=False) + op.drop_column("messages", "iv") + op.drop_column("messages", "ciphertext") + op.drop_column("messages", "kind") diff --git a/backend/app/main.py b/backend/app/main.py index 4fc3803..f530847 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -34,6 +34,7 @@ from app.routes import ( feed, health, me, + messages, notifications, playlists, quota, @@ -114,6 +115,7 @@ app.include_router(tags.router) app.include_router(feed.router) app.include_router(me.router) app.include_router(notifications.router) +app.include_router(messages.router) app.include_router(channels.router) app.include_router(playlists.router) app.include_router(admin.router) diff --git a/backend/app/models.py b/backend/app/models.py index 623da7b..86226b3 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -518,3 +518,69 @@ class Notification(Base): created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), index=True ) + + +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 MessageKey(Base): + """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) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now() + ) diff --git a/backend/app/realtime.py b/backend/app/realtime.py new file mode 100644 index 0000000..4899cfc --- /dev/null +++ b/backend/app/realtime.py @@ -0,0 +1,60 @@ +"""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 +import logging +from collections import defaultdict + +from starlette.websockets import WebSocket + +log = logging.getLogger("siftlode.realtime") + + +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: + # Stale socket; it'll be cleaned up by its own disconnect handler. + pass + + 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: + log.debug("ws push failed", exc_info=True) + + +manager = ConnectionManager() diff --git a/backend/app/routes/messages.py b/backend/app/routes/messages.py new file mode 100644 index 0000000..f7240ae --- /dev/null +++ b/backend/app/routes/messages.py @@ -0,0 +1,373 @@ +"""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, HTTPException, WebSocket, WebSocketDisconnect +from pydantic import BaseModel +from sqlalchemy import and_, func, or_, select +from sqlalchemy.orm import Session + +from app.auth import require_human +from app.db import SessionLocal, get_db +from app.models import Message, MessageKey, User +from app.ratelimit import RateLimiter +from app.realtime import manager + +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) + +# 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!" + ), + "de": ( + "Willkommen bei Siftlode! 👋\n\n" + "Das ist dein privater Nachrichten-Posteingang. Unterhaltungen mit anderen Mitgliedern " + "sind Ende-zu-Ende-verschlüsselt — nur du und dein Gegenüber könnt sie lesen, nicht " + "einmal ein Admin. Um jemandem zu schreiben, richte zuerst die sichere " + "Nachrichtenübermittlung mit einem Passwort ein.\n\n" + "Viel Spaß mit deinem Feed!" + ), +} + + +class KeySetupIn(BaseModel): + public_key: str + wrapped_private_key: str + salt: str + wrap_iv: str + key_check: str + + +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} + + +def _serialize_msg(m: Message) -> dict: + return { + "id": m.id, + "kind": m.kind, + "sender_id": m.sender_id, + "recipient_id": m.recipient_id, + "body": m.body, # only set for system messages + "ciphertext": m.ciphertext, + "iv": m.iv, + "read_at": m.read_at.isoformat() if m.read_at else None, + "created_at": m.created_at.isoformat() if m.created_at else None, + } + + +def _messageable() -> list: + return [ + User.is_demo.is_(False), + User.is_active.is_(True), + User.is_suspended.is_(False), + ] + + +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)}) + + +# --- 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 HTTPException(status_code=409, detail="Secure messaging is already 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 {"configured": True} + + +@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.""" + k = db.get(MessageKey, user_id) + if k is None: + raise HTTPException(status_code=404, detail="That member hasn't set up messaging yet.") + 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()) + .order_by(func.lower(func.coalesce(User.display_name, User.email))) + ) + .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) + msgs = ( + db.execute( + select(Message) + .where(or_(Message.sender_id == user.id, Message.recipient_id == user.id)) + .order_by(Message.created_at.desc()) + ) + .scalars() + .all() + ) + convs: dict[int, dict] = {} + for m in msgs: + if m.kind == "system": + pid = SYSTEM_PARTNER_ID + else: + pid = m.recipient_id if m.sender_id == user.id else m.sender_id + c = convs.setdefault(pid, {"last": m, "unread": 0}) + if m.recipient_id == user.id and m.read_at is None: + c["unread"] += 1 + user_ids = [pid for pid in convs if pid != SYSTEM_PARTNER_ID] + 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 pid, c in convs.items(): + 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(c["last"]), "unread": c["unread"]} + ) + 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) + if p is None: + raise HTTPException(status_code=404, detail="User not found") + partner = _serialize_user(p) + 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), + ), + ) + 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() + return {"partner": partner, "items": [_serialize_msg(m) 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 HTTPException(status_code=429, detail="Too many messages — slow down a moment.") + ct = (payload.ciphertext or "").strip() + iv = (payload.iv or "").strip() + if not ct or not iv: + raise HTTPException(status_code=400, detail="Message is empty.") + if len(ct) > MAX_CIPHERTEXT: + raise HTTPException(status_code=400, detail="Message is too long.") + if payload.recipient_id == user.id: + raise HTTPException(status_code=400, detail="You can't message yourself.") + recipient = db.get(User, payload.recipient_id) + if recipient is None or recipient.is_demo or not recipient.is_active or recipient.is_suspended: + raise HTTPException(status_code=404, detail="Recipient not available.") + if db.get(MessageKey, recipient.id) is None: + raise HTTPException(status_code=404, detail="That member hasn't set up messaging yet.") + m = Message( + kind="user", sender_id=user.id, recipient_id=recipient.id, ciphertext=ct, iv=iv + ) + db.add(m) + db.commit() + db.refresh(m) + event = {"type": "message", "message": _serialize_msg(m)} + 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.""" + uid = ws.session.get("user_id") if "session" in ws.scope else None + if not uid: + await ws.close(code=1008) + return + db = SessionLocal() + try: + u = db.get(User, uid) + ok = bool(u and not u.is_demo and not u.is_suspended and u.is_active) + finally: + db.close() + if not ok: + await ws.close(code=1008) + return + await manager.connect(uid, ws) + try: + while True: + await ws.receive_text() + except WebSocketDisconnect: + pass + finally: + await manager.disconnect(uid, ws)