From 0e758e99a615ea606cc29147cc9ffeb02ff3721c Mon Sep 17 00:00:00 2001 From: FrenchToblerone54 Date: Mon, 6 Jul 2026 16:34:51 +0000 Subject: [PATCH 1/2] Port backend from Flask to FastAPI/Starlette Replaces Flask/waitress with Starlette/uvicorn behind a thin Flask-compat shim (app/router.py, app/compat.py, app/responses.py), so existing route bodies stay unchanged. Business logic stays synchronous and runs via Starlette's threadpool; the /stream SSE endpoint is upgraded to a real async asyncio.Queue + StreamingResponse so many long-lived connections don't rely on a thread-per-connection poll loop. Fixes a status-code bug in the compat shim where FileResponse status codes were never applied, which silently turned custom error pages (404.html, /pfp and /attachment 404s) into HTTP 200 responses. --- Dockerfile | 1 + api/__init__.py | 35 ---- app/__init__.py | 0 api/utils.py => app/api_utils.py | 9 +- app/compat.py | 161 ++++++++++++++++ utils.py => app/config.py | 4 +- db.py => app/db.py | 2 +- app/main.py | 311 +++++++++++++++++++++++++++++++ app/responses.py | 82 ++++++++ app/router.py | 11 ++ app/routers/__init__.py | 0 {api => app/routers}/auth.py | 18 +- {api => app/routers}/bans.py | 10 +- {api => app/routers}/calls.py | 14 +- {api => app/routers}/channels.py | 16 +- {api => app/routers}/keys.py | 12 +- {api => app/routers}/members.py | 12 +- {api => app/routers}/messages.py | 16 +- {api => app/routers}/pins.py | 9 +- {api => app/routers}/stream.py | 81 +++++--- {api => app/routers}/users.py | 12 +- {api => app/routers}/webhooks.py | 14 +- cli.py | 2 +- main.py | 170 +---------------- migrations.py | 4 +- requirements.txt | 7 +- 26 files changed, 710 insertions(+), 303 deletions(-) delete mode 100644 api/__init__.py create mode 100644 app/__init__.py rename api/utils.py => app/api_utils.py (98%) create mode 100644 app/compat.py rename utils.py => app/config.py (94%) rename db.py => app/db.py (99%) create mode 100644 app/main.py create mode 100644 app/responses.py create mode 100644 app/router.py create mode 100644 app/routers/__init__.py rename {api => app/routers}/auth.py (96%) rename {api => app/routers}/bans.py (96%) rename {api => app/routers}/calls.py (94%) rename {api => app/routers}/channels.py (98%) rename {api => app/routers}/keys.py (96%) rename {api => app/routers}/members.py (97%) rename {api => app/routers}/messages.py (98%) rename {api => app/routers}/pins.py (97%) rename {api => app/routers}/stream.py (91%) rename {api => app/routers}/users.py (97%) rename {api => app/routers}/webhooks.py (96%) diff --git a/Dockerfile b/Dockerfile index 9867376..07b6da6 100644 --- a/Dockerfile +++ b/Dockerfile @@ -2,6 +2,7 @@ FROM python:3.13-slim ARG DEBIAN_FRONTEND=noninteractive ENV TZ=Etc/UTC +ENV PYTHONUNBUFFERED=1 RUN python -m pip install --upgrade pip && \ apt-get update && \ diff --git a/api/__init__.py b/api/__init__.py deleted file mode 100644 index 33bd802..0000000 --- a/api/__init__.py +++ /dev/null @@ -1,35 +0,0 @@ -from flask import Blueprint -from .auth import auth_bp -from .channels import channels_bp -from .keys import keys_bp -from .members import members_bp -from .bans import bans_bp -from .messages import messages_bp -from .users import users_bp -from .pins import pins_bp -from .stream import stream_bp -from .calls import calls_bp -from .webhooks import webhooks_bp -from .utils import process_cors_headers, cleaner -from threading import Thread - -api_bp=Blueprint("API", __name__) - -@api_bp.after_request -def add_cors_headers(resp): - process_cors_headers(resp) - return resp - -api_bp.register_blueprint(auth_bp) -api_bp.register_blueprint(channels_bp) -api_bp.register_blueprint(keys_bp) -api_bp.register_blueprint(members_bp) -api_bp.register_blueprint(bans_bp) -api_bp.register_blueprint(messages_bp) -api_bp.register_blueprint(users_bp) -api_bp.register_blueprint(pins_bp) -api_bp.register_blueprint(stream_bp) -api_bp.register_blueprint(calls_bp) -api_bp.register_blueprint(webhooks_bp) - -Thread(target=cleaner, daemon=True).start() diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/api/utils.py b/app/api_utils.py similarity index 98% rename from api/utils.py rename to app/api_utils.py index c9792ee..c574d6e 100644 --- a/api/utils.py +++ b/app/api_utils.py @@ -1,5 +1,6 @@ -from flask import request, make_response, jsonify -from db import SQLite +from app.compat import request +from app.responses import make_response, jsonify +from app.db import SQLite import base64 from PIL import Image from cryptography.hazmat.primitives import serialization, hashes @@ -12,7 +13,7 @@ from functools import wraps import inspect from threading import Lock -from utils import config, generate +from app.config import config, generate import math os.makedirs(config["data_dir"]["pfps"], exist_ok=True) @@ -206,7 +207,7 @@ def regex_first_group_encrypted(match, public_key): return rsa_encrypt(public_ke all_sliding_window_ratelimits=[] def cleaner(): - from utils import stopping + from app.config import stopping while not stopping.is_set(): now=timestamp() with challenges_lock: diff --git a/app/compat.py b/app/compat.py new file mode 100644 index 0000000..55785b0 --- /dev/null +++ b/app/compat.py @@ -0,0 +1,161 @@ +import io +import json as _json +import os +from contextvars import ContextVar +from urllib.parse import parse_qsl +from starlette.requests import Request +from starlette.datastructures import UploadFile + +_ctx: ContextVar["RequestProxy"]=ContextVar("request_proxy") +event_loop_var: ContextVar=ContextVar("event_loop", default=None) + +class MultiDict: + def __init__(self, items=None): + self._items=list(items) if items else [] + def __contains__(self, key): return any(k==key for k, _ in self._items) + def __getitem__(self, key): + for k, v in self._items: + if k==key: return v + raise BadRequestKeyError(key) + def get(self, key, default=None): + for k, v in self._items: + if k==key: return v + return default + def getlist(self, key): return [v for k, v in self._items if k==key] + def keys(self): return [k for k, _ in self._items] + def items(self): return list(self._items) + def __iter__(self): return iter(k for k, _ in self._items) + def __bool__(self): return bool(self._items) + def __len__(self): return len(self._items) + +class BadRequestKeyError(KeyError): + pass + +class FileStorage: + """Werkzeug FileStorage-compatible wrapper backed by an in-memory stream.""" + def __init__(self, stream: io.BytesIO, filename: str, content_type: str, size: int): + self.stream=stream + self.filename=filename or "" + self.content_type=content_type + self.mimetype=(content_type or "").split(";")[0].strip() or None + self.content_length=size + def save(self, dst): + self.stream.seek(0) + with open(dst, "wb") as f: + while True: + chunk=self.stream.read(1024*1024) + if not chunk: break + f.write(chunk) + self.stream.seek(0) + def read(self, *args): return self.stream.read(*args) + +class RequestProxy: + def __init__(self, scope, headers, method, path, remote_addr, host_url, script_root, args, form, files, raw_body, body_size, over_limit): + self.headers=headers + self.method=method + self.path=path + self.remote_addr=remote_addr + self.host_url=host_url + self.script_root=script_root + self.args=args + self._form=form + self._files=files + self._raw_body=raw_body + self._body_size=body_size + self._over_limit=over_limit + self._json_cached=False + self._json_value=None + + def _check_limit(self): + if self._over_limit: + from app.responses import HTTPAbort + raise HTTPAbort(413) + + @property + def form(self): + self._check_limit() + return self._form + + @property + def files(self): + self._check_limit() + return self._files + + @property + def json(self): + return self.get_json(silent=True) + + def get_json(self, force=False, silent=False): + if self._json_cached: return self._json_value + self._check_limit() + ctype=self.headers.get("content-type", "") + if not force and "application/json" not in ctype.lower(): + if silent: + self._json_cached=True + self._json_value=None + return None + try: + value=_json.loads(self._raw_body) if self._raw_body else None + except Exception: + value=None + if not silent: raise + self._json_cached=True + self._json_value=value + return value + +class _RequestLocal: + def __getattr__(self, name): return getattr(_ctx.get(), name) + +request=_RequestLocal() + +def _build_host_url(scope, headers): + scheme=headers.get("x-forwarded-proto") or scope.get("scheme", "http") + host=headers.get("host", "") + root=scope.get("root_path", "") or "" + return f"{scheme}://{host}{root}/" + +async def build_proxy(req: Request) -> RequestProxy: + headers={k.decode().lower(): v.decode() for k, v in req.scope.get("headers", [])} + method=req.method + root_path=req.scope.get("root_path", "") or "" + raw_path=req.scope.get("path", "") + path=root_path+raw_path + client=req.client + remote_addr=client.host if client else None + forwarded=headers.get("x-forwarded-for") + if forwarded: remote_addr=forwarded.split(",")[0].strip() + qs=req.scope.get("query_string", b"").decode() + args=MultiDict(parse_qsl(qs, keep_blank_values=True)) + ctype=headers.get("content-type", "") + from app.config import config as _config + max_len=_config["server"]["max_content_length"] + form_items=[] + file_items=[] + raw_body=b"" + is_form="multipart/form-data" in ctype or "application/x-www-form-urlencoded" in ctype + if method in ("GET", "HEAD", "DELETE") and not headers.get("content-length") and not is_form: + body_size=0 + else: + raw_body=await req.body() + body_size=len(raw_body) + over_limit=max_len is not None and body_size>max_len + if is_form and not over_limit: + async with req.form(max_files=10000, max_fields=10000) as form_data: + for key, value in form_data.multi_items(): + if isinstance(value, UploadFile): + data=await value.read() + file_items.append((key, FileStorage(io.BytesIO(data), value.filename, value.content_type or "application/octet-stream", len(data)))) + else: + form_items.append((key, value)) + raw_body=b"" + host_url=_build_host_url(req.scope, headers) + return RequestProxy(req.scope, _Headers(headers), method, path, remote_addr, host_url, root_path, args, MultiDict(form_items), MultiDict(file_items), raw_body, body_size, over_limit) + +class _Headers: + def __init__(self, d): self._d=d + def __contains__(self, key): return key.lower() in self._d + def __getitem__(self, key): return self._d[key.lower()] + def get(self, key, default=None): return self._d.get(key.lower(), default) + +def set_request(proxy): return _ctx.set(proxy) +def reset_request(token): _ctx.reset(token) diff --git a/utils.py b/app/config.py similarity index 94% rename from utils.py rename to app/config.py index ac70ba8..0bc11da 100644 --- a/utils.py +++ b/app/config.py @@ -8,7 +8,7 @@ def generate(size=20): return nanoid.generate(size=size) stopping=Event() -_data_dir=sys._MEIPASS if getattr(sys, "frozen", False) else os.path.dirname(os.path.abspath(__file__)) +_data_dir=sys._MEIPASS if getattr(sys, "frozen", False) else os.path.dirname(os.path.dirname(os.path.abspath(__file__))) with open(os.path.join(_data_dir, "version.toml"), "rb") as f: version_data=tomllib.load(f) @@ -35,4 +35,4 @@ def generate(size=20): return nanoid.generate(size=size) RED="\033[31m" RESET="\033[0m" -def colored_log(color, tag, text): print(f"{color}[{tag}]{RESET} {text}") \ No newline at end of file +def colored_log(color, tag, text): print(f"{color}[{tag}]{RESET} {text}") diff --git a/db.py b/app/db.py similarity index 99% rename from db.py rename to app/db.py index 6d3a6b8..b071600 100644 --- a/db.py +++ b/app/db.py @@ -6,7 +6,7 @@ import time import math from typing import List, Dict, Any, Union, Tuple, Optional -from utils import config +from app.config import config logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger=logging.getLogger(__name__) diff --git a/app/main.py b/app/main.py new file mode 100644 index 0000000..73e92d1 --- /dev/null +++ b/app/main.py @@ -0,0 +1,311 @@ +import os +import sys +import traceback +os.chdir(os.path.dirname(os.path.abspath(sys.argv[0])) if getattr(sys, "frozen", False) else os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +from app.config import config, dev_mode, db_version, stopping, BLUE, YELLOW, RED, colored_log +os.makedirs(os.path.dirname(config["data_dir"]["database"]), exist_ok=True) +from threading import Thread +from starlette.applications import Starlette +from starlette.routing import Route +from starlette.responses import Response, PlainTextResponse +from starlette.exceptions import HTTPException as StarletteHTTPException +from starlette.concurrency import run_in_threadpool +from app.db import SQLite +from migrations import run_migrations +from app.compat import build_proxy, set_request, reset_request, event_loop_var +from app.responses import to_starlette, send_from_directory, HTTPAbort, redirect, make_response, jsonify +from app.api_utils import process_cors_headers, cleaner +from app.routers.auth import auth_bp +from app.routers.channels import channels_bp +from app.routers.keys import keys_bp +from app.routers.members import members_bp +from app.routers.bans import bans_bp +from app.routers.messages import messages_bp +from app.routers.users import users_bp +from app.routers.pins import pins_bp +from app.routers.stream import stream_bp +from app.routers.calls import calls_bp +from app.routers.webhooks import webhooks_bp +import asyncio + +try: run_migrations() +except Exception as e: + colored_log(RED, "ERROR", f"Migration failed: {e}") + sys.exit(1) + +with SQLite() as db: + db.create_table("users", {"seq": "INTEGER PRIMARY KEY AUTOINCREMENT", "id": "TEXT UNIQUE NOT NULL", "username": "TEXT UNIQUE NOT NULL", "display_name": "TEXT", "pfp": "TEXT", "passkey": "TEXT NOT NULL", "public_key": "TEXT NOT NULL", "created_at": "INTEGER NOT NULL", "FOREIGN KEY (pfp)": "REFERENCES files (id) ON DELETE SET NULL"}) + db.create_table("session", {"seq": "INTEGER PRIMARY KEY AUTOINCREMENT", "user": "TEXT NOT NULL", "token_hash": "TEXT UNIQUE NOT NULL", "id": "TEXT UNIQUE NOT NULL", "device": "TEXT", "browser": "TEXT", "logged_in_at": "INTEGER NOT NULL", "next_challenge": "INTEGER", "FOREIGN KEY (user)": "REFERENCES users (id) ON DELETE CASCADE"}) + db.create_table("channels", {"id": "TEXT PRIMARY KEY", "name": "TEXT", "pfp": "TEXT", "type": "INTEGER NOT NULL CHECK (type IN (1, 2, 3))", "permissions": "INTEGER NOT NULL DEFAULT 0", "dm": "TEXT", "invite_code": "TEXT UNIQUE", "created_at": "INTEGER NOT NULL", "FOREIGN KEY (pfp)": "REFERENCES files (id) ON DELETE SET NULL"}) + db.create_table("members", {"seq": "INTEGER PRIMARY KEY AUTOINCREMENT", "user_id": "TEXT", "channel_id": "TEXT", "joined_at": "INTEGER NOT NULL", "permissions": "INTEGER", "message_seq": "INTEGER DEFAULT 0", "hidden": "INTEGER CHECK (hidden IS NULL OR hidden = 1)", "UNIQUE": "(user_id, channel_id)", "FOREIGN KEY (user_id)": "REFERENCES users (id) ON DELETE CASCADE", "FOREIGN KEY (channel_id)": "REFERENCES channels (id) ON DELETE CASCADE"}) + db.create_table("messages", {"seq": "INTEGER PRIMARY KEY AUTOINCREMENT", "id": "TEXT UNIQUE NOT NULL", "channel_id": "TEXT NOT NULL", "user_id": "TEXT NOT NULL", "content": "TEXT NOT NULL", "key": "TEXT", "iv": "TEXT", "timestamp": "INTEGER NOT NULL", "edited_at": "INTEGER", "replied_to": "TEXT", "signature": "TEXT", "signed_timestamp": "INTEGER", "nonce": "TEXT", "webhook_id": "TEXT", "webhook_name": "TEXT", "webhook_pfp": "TEXT", "FOREIGN KEY (channel_id)": "REFERENCES channels (id) ON DELETE CASCADE", "FOREIGN KEY (user_id)": "REFERENCES users (id) ON DELETE CASCADE"}) + db.create_table("message_pins", {"seq": "INTEGER PRIMARY KEY AUTOINCREMENT", "id": "TEXT UNIQUE NOT NULL", "FOREIGN KEY (id)": "REFERENCES messages (id) ON DELETE CASCADE"}) + db.create_table("files", {"id": "TEXT PRIMARY KEY", "filename": "TEXT", "hash": "TEXT NOT NULL", "size": "INTEGER NOT NULL", "mimetype": "TEXT", "file_type": "TEXT NOT NULL CHECK (file_type IN ('attachment', 'pfp'))", "UNIQUE": "(hash, file_type)"}) + db.create_table("attachment_message", {"file_id": "TEXT NOT NULL", "message_id": "TEXT NOT NULL", "encrypted": "INTEGER NOT NULL DEFAULT 0", "iv": "TEXT", "PRIMARY KEY": "(file_id, message_id)", "FOREIGN KEY (file_id)": "REFERENCES files (id) ON DELETE CASCADE", "FOREIGN KEY (message_id)": "REFERENCES messages (id) ON DELETE CASCADE"}) + db.create_table("channels_keys", {"id": "TEXT NOT NULL", "channel_id": "TEXT", "user_id": "TEXT", "key": "TEXT", "FOREIGN KEY (channel_id)": "REFERENCES channels (id) ON DELETE CASCADE", "FOREIGN KEY (user_id)": "REFERENCES users (id) ON DELETE CASCADE"}) + db.create_table("channels_keys_info", {"seq": "INTEGER PRIMARY KEY AUTOINCREMENT", "key_id": "TEXT UNIQUE NOT NULL", "channel_id": "TEXT", "by": "TEXT", "timestamp": "INTEGER NOT NULL", "expires_at": "INTEGER NOT NULL", "FOREIGN KEY (channel_id)": "REFERENCES channels (id) ON DELETE CASCADE", "FOREIGN KEY (by)": "REFERENCES users (id) ON DELETE SET NULL"}) + db.create_table("message_reads", {"seq": "INTEGER PRIMARY KEY AUTOINCREMENT", "user_id": "TEXT NOT NULL", "channel_id": "TEXT NOT NULL", "last_message_id": "TEXT NOT NULL", "read_at": "INTEGER NOT NULL", "UNIQUE": "(user_id, channel_id)", "FOREIGN KEY (user_id)": "REFERENCES users (id) ON DELETE CASCADE", "FOREIGN KEY (channel_id)": "REFERENCES channels (id) ON DELETE CASCADE", "FOREIGN KEY (last_message_id)": "REFERENCES messages (id) ON DELETE CASCADE"}) + db.create_table("bans", {"seq": "INTEGER PRIMARY KEY AUTOINCREMENT", "user_id": "TEXT NOT NULL", "channel_id": "TEXT NOT NULL", "banned_by": "TEXT NOT NULL", "banned_at": "INTEGER NOT NULL", "reason": "TEXT", "UNIQUE": "(user_id, channel_id)", "FOREIGN KEY (user_id)": "REFERENCES users (id) ON DELETE CASCADE", "FOREIGN KEY (channel_id)": "REFERENCES channels (id) ON DELETE CASCADE", "FOREIGN KEY (banned_by)": "REFERENCES users (id) ON DELETE CASCADE"}) + db.create_table("blocks", {"seq": "INTEGER PRIMARY KEY AUTOINCREMENT", "blocker_id": "TEXT NOT NULL", "blocked_id": "TEXT NOT NULL", "blocked_at": "INTEGER NOT NULL", "UNIQUE": "(blocker_id, blocked_id)", "FOREIGN KEY (blocker_id)": "REFERENCES users (id) ON DELETE CASCADE", "FOREIGN KEY (blocked_id)": "REFERENCES users (id) ON DELETE CASCADE"}) + db.create_table("calls", {"channel_id": "TEXT PRIMARY KEY", "started_by": "TEXT NOT NULL", "started_at": "INTEGER NOT NULL", "FOREIGN KEY (channel_id)": "REFERENCES channels (id) ON DELETE CASCADE", "FOREIGN KEY (started_by)": "REFERENCES users (id) ON DELETE CASCADE"}) + db.create_table("call_participants", {"channel_id": "TEXT NOT NULL", "user_id": "TEXT NOT NULL", "joined_at": "INTEGER NOT NULL", "left_at": "INTEGER", "PRIMARY KEY": "(channel_id, user_id)", "FOREIGN KEY (channel_id)": "REFERENCES calls (channel_id) ON DELETE CASCADE", "FOREIGN KEY (user_id)": "REFERENCES users (id) ON DELETE CASCADE"}) + db.create_table("webhooks", {"id": "TEXT PRIMARY KEY", "channel_id": "TEXT NOT NULL", "name": "TEXT NOT NULL", "pfp": "TEXT", "token": "TEXT NOT NULL", "created_by": "TEXT", "created_at": "INTEGER NOT NULL", "last_used_at": "INTEGER", "FOREIGN KEY (channel_id)": "REFERENCES channels (id) ON DELETE CASCADE", "FOREIGN KEY (created_by)": "REFERENCES users (id) ON DELETE SET NULL"}) + db.create_index("session", "user") + db.create_index("members", "channel_id") + db.create_index("members", "message_seq") + db.create_index("messages", "channel_id") + db.create_index("messages", "user_id") + db.create_index("messages", "timestamp") + db.create_index("files", "file_type") + db.create_index("attachment_message", "message_id") + db.create_index("channels_keys", "id") + db.create_index("channels_keys", "channel_id") + db.create_index("channels_keys", "user_id") + db.create_index("channels_keys_info", "channel_id") + db.create_index("message_reads", "user_id") + db.create_index("message_reads", "channel_id") + db.create_index("call_participants", "channel_id") + db.create_index("call_participants", "user_id") + db.create_index("webhooks", "channel_id") + db.create_index("webhooks", "token", unique=True) + if not db.exists("users", {"id": "0"}): db.insert_data("users", {"id": "0", "username": "__parley_webhooks_system_account_do_not_use__", "display_name": "System", "pfp": None, "passkey": "system", "public_key": "system", "created_at": 0}) + if db.execute_raw_sql("PRAGMA user_version;")[0]["user_version"]!=db_version: db.execute_raw_sql(f"PRAGMA user_version={db_version};") + +uri_prefix="/"+config["uri_prefix"] if config["uri_prefix"] else "" +def route_rule(rule: str): return uri_prefix+rule + +frontend_hosted=config["frontend"]["hosted"] +frontend_present=os.path.isdir(config["frontend"]["frontend_directory"]) +frontend=frontend_hosted and frontend_present + +error_text={ + "404": "not found", + "405": "method not allowed", + "400": "bad request", + "413": "content too big", + "415": "unsupported media type", + "500": "internal server error" +} + +api_url=route_rule("/api/v1/") + +def _flask_to_starlette_rule(rule): + out="" + i=0 + while i", i) + inner=rule[i+1:j] + if ":" in inner: + converter, var=inner.split(":", 1) + else: + converter, var="default", inner + out+="{"+var+(":path" if converter=="path" else "")+"}" + i=j+1 + else: + out+=rule[i] + i+=1 + return out + +def make_endpoint(method_map, is_api, allow_methods, explicit_options): + allow_header=", ".join(allow_methods) + async def endpoint(req): + method=req.method + if method=="OPTIONS" and not explicit_options: + resp=Response(status_code=200, headers={"Allow": allow_header}) + if is_api: process_cors_headers(resp) + return resp + handler=method_map.get(method) or (method_map.get("GET") if method=="HEAD" else None) + loop=asyncio.get_running_loop() + proxy=await build_proxy(req) + token=set_request(proxy) + loop_token=event_loop_var.set(loop) + try: + path_params=req.path_params + rv=await run_in_threadpool(handler, **path_params) + resp=to_starlette(rv) + except HTTPAbort as e: + resp=_error_response(e.code, req) + finally: + event_loop_var.reset(loop_token) + reset_request(token) + if is_api: process_cors_headers(resp) + return resp + return endpoint + +def _error_response(code, req): + path=req.scope.get("root_path", "")+req.scope.get("path", "") + if path.startswith(uri_prefix): + if path==api_url or (path+"/").startswith(api_url): + resp=to_starlette((jsonify({"error": error_text[str(code)], "success": False}), code)) + process_cors_headers(resp) + return resp + if frontend: + try: return to_starlette((send_from_directory(config["frontend"]["frontend_directory"], f"{code}.html"), code)) + except HTTPAbort: return to_starlette((error_text[str(code)], code)) + return to_starlette((error_text[str(code)], code)) + return to_starlette(({"error": error_text[str(code)]}, code)) + +routes=[] +_path_groups={} +_path_order=[] + +def register_router(router, prefix): + for rule, methods, func in router.routes: + full=route_rule(prefix+rule) + path=_flask_to_starlette_rule(full) + if path not in _path_groups: + _path_groups[path]={} + _path_order.append(path) + for m in methods: + _path_groups[path][m]=func + +API_PREFIX="/api/v1" +for r in [auth_bp, channels_bp, keys_bp, members_bp, bans_bp, messages_bp, users_bp, pins_bp, stream_bp, calls_bp, webhooks_bp]: + register_router(r, API_PREFIX) + +for path in _path_order: + method_map=_path_groups[path] + explicit_options="OPTIONS" in method_map + all_methods=set(method_map.keys()) + if "GET" in all_methods: all_methods.add("HEAD") + all_methods.add("OPTIONS") + allow=[m for m in ["HEAD", "OPTIONS", "GET", "POST", "PUT", "PATCH", "DELETE"] if m in all_methods] + routes.append(Route(path, make_endpoint(method_map, True, allow, explicit_options), methods=sorted(all_methods))) + +async def api_index(req): + resp=to_starlette(redirect(api_url, 301)) + process_cors_headers(resp) + return resp +routes.append(Route(route_rule("/api/v1"), api_index, methods=["GET"])) + +def serve_pfp_handler(pfp): + db=SQLite() + try: + pfp_data=db.select_data("files", ["id", "mimetype"], {"id": pfp, "file_type": "pfp"}) + if not pfp_data: raise HTTPAbort(404) + try: + resp=send_from_directory(config["data_dir"]["pfps"], f"{pfp_data[0]['id']}.webp", mimetype=pfp_data[0]["mimetype"], as_attachment=False) + process_cors_headers(resp) + return resp + except HTTPAbort: + db.cleanup_unused_files() + raise HTTPAbort(404) + finally: db.close() + +def serve_attachment_handler(file_id): + db=SQLite() + try: + file_data=db.select_data("files", ["id", "filename", "mimetype"], {"id": file_id, "file_type": "attachment"}) + if not file_data: raise HTTPAbort(404) + filename=file_data[0]["filename"] or "attachment" + try: + resp=send_from_directory(config["data_dir"]["attachments"], file_data[0]["id"], mimetype=file_data[0]["mimetype"], as_attachment=True, download_name=filename) + process_cors_headers(resp) + return resp + except HTTPAbort: + db.cleanup_unused_files() + raise HTTPAbort(404) + finally: db.close() + +async def serve_pfp(req): + try: return to_starlette(await run_in_threadpool(serve_pfp_handler, req.path_params["pfp"])) + except HTTPAbort as e: return _error_response(e.code, req) + +async def serve_attachment(req): + try: return to_starlette(await run_in_threadpool(serve_attachment_handler, req.path_params["file_id"])) + except HTTPAbort as e: return _error_response(e.code, req) + +routes.append(Route(route_rule("/pfp/{pfp}"), serve_pfp, methods=["GET"])) +routes.append(Route(route_rule("/attachment/{file_id}"), serve_attachment, methods=["GET"])) + +async def health(req): return to_starlette({"status": "ok"}) +routes.append(Route(route_rule("/health"), health, methods=["GET"])) + +if frontend: + colored_log(BLUE, "INFO", "Frontend directory present, serving it") + excluded=[i.lower() for i in config["frontend"]["excluded_frontend_root_paths"]] + + def safe_join(directory, *pathnames): + parts=[directory] + for filename in pathnames: + if filename!="": + filename=os.path.normpath(filename) + if (os.path.isabs(filename) or filename==".." or filename.startswith("../") or os.sep+".."+os.sep in os.sep+filename+os.sep): + return None + parts.append(filename) + return os.path.join(*parts) + + def index_handler(): + fdir=config["frontend"]["frontend_directory"] + path=os.path.join(fdir, "index.html") + if not os.path.isfile(path): raise HTTPAbort(404) + with open(path, encoding="utf-8") as f: html=f.read() + return html + + def serve_static_handler(path): + if "." not in path: path+=".html" + safe_path=safe_join(config["frontend"]["frontend_directory"], path) + if not safe_path: raise HTTPAbort(404) + safe_path=os.path.relpath(safe_path, config["frontend"]["frontend_directory"]).lower() + for exclude in excluded: + if safe_path.startswith(exclude+os.sep) or safe_path==exclude: raise HTTPAbort(404) + return send_from_directory(config["frontend"]["frontend_directory"], path) + + async def index(req): + try: resp=to_starlette(await run_in_threadpool(index_handler)); resp.headers["Cache-Control"]="no-cache"; return resp + except HTTPAbort as e: return _error_response(e.code, req) + async def serve_static(req): + try: return to_starlette(await run_in_threadpool(serve_static_handler, req.path_params["path"])) + except HTTPAbort as e: return _error_response(e.code, req) + routes.append(Route(route_rule("/"), index, methods=["GET"])) + routes.append(Route(route_rule("/{path:path}"), serve_static, methods=["GET"])) +elif not frontend_hosted: + colored_log(BLUE, "INFO", "Frontend directory isn't hosted") +elif not frontend_present: + colored_log(RED, "ERROR", "Frontend directory isn't present") + +async def starlette_http_exc(req, exc): + code=exc.status_code + if str(code) in error_text: return _error_response(code, req) + return PlainTextResponse(getattr(exc, "detail", "") or "", status_code=code) + +async def server_error(req, exc): + colored_log(RED, "ERROR", f"Unhandled error: {traceback.format_exc()}") + return _error_response(500, req) + +exception_handlers={StarletteHTTPException: starlette_http_exc, Exception: server_error} + +HTTP_STATUS_CODES={100: "Continue", 101: "Switching Protocols", 102: "Processing", 103: "Early Hints", 200: "OK", 201: "Created", 202: "Accepted", 203: "Non Authoritative Information", 204: "No Content", 205: "Reset Content", 206: "Partial Content", 207: "Multi Status", 208: "Already Reported", 226: "IM Used", 300: "Multiple Choices", 301: "Moved Permanently", 302: "Found", 303: "See Other", 304: "Not Modified", 305: "Use Proxy", 306: "Switch Proxy", 307: "Temporary Redirect", 308: "Permanent Redirect", 400: "Bad Request", 401: "Unauthorized", 402: "Payment Required", 403: "Forbidden", 404: "Not Found", 405: "Method Not Allowed", 406: "Not Acceptable", 407: "Proxy Authentication Required", 408: "Request Timeout", 409: "Conflict", 410: "Gone", 411: "Length Required", 412: "Precondition Failed", 413: "Request Entity Too Large", 414: "Request URI Too Long", 415: "Unsupported Media Type", 416: "Requested Range Not Satisfiable", 417: "Expectation Failed", 418: "I'm a teapot", 421: "Misdirected Request", 422: "Unprocessable Entity", 423: "Locked", 424: "Failed Dependency", 425: "Too Early", 426: "Upgrade Required", 428: "Precondition Required", 429: "Too Many Requests", 431: "Request Header Fields Too Large", 449: "Retry With", 451: "Unavailable For Legal Reasons", 500: "Internal Server Error", 501: "Not Implemented", 502: "Bad Gateway", 503: "Service Unavailable", 504: "Gateway Timeout", 505: "HTTP Version Not Supported", 506: "Variant Also Negotiates", 507: "Insufficient Storage", 508: "Loop Detected", 510: "Not Extended", 511: "Network Authentication Failed"} + +def _patch_uvicorn_reason_phrases(): + """Match waitress/werkzeug reason phrases (uppercased) for byte-identical status lines.""" + def phrase(code): return (HTTP_STATUS_CODES.get(code, "UNKNOWN")).upper().encode() + try: + import uvicorn.protocols.http.httptools_impl as ht + ht.STATUS_LINE={code: b"".join([b"HTTP/1.1 ", str(code).encode(), b" ", phrase(code), b"\r\n"]) for code in range(100, 600)} + except Exception: pass + try: + import uvicorn.protocols.http.h11_impl as h11i + h11i.STATUS_PHRASES={code: phrase(code) for code in range(100, 600)} + except Exception: pass + +_patch_uvicorn_reason_phrases() + +app=Starlette(routes=routes, exception_handlers=exception_handlers) + +Thread(target=cleaner, daemon=True).start() + +colored_log(BLUE, "INFO", f"Access instance at http://{config['server']['host']}:{config['server']['port']}{uri_prefix}/") +if dev_mode: colored_log(YELLOW, "WARNING", "Dev mode is enabled, please disable this mode if you're running this in production") +if dev_mode: colored_log(BLUE, "DEV MODE INFO", f"Access instance at http://localhost:{config['server']['port']}{uri_prefix}/ for local access") + +def run(): + import uvicorn + proxy_headers=config["server"]["proxy"] + try: + uvicorn.run(app, host=config["server"]["host"], port=config["server"]["port"], log_level="debug" if dev_mode else "info", proxy_headers=proxy_headers, forwarded_allow_ips="*" if proxy_headers else None) + except KeyboardInterrupt: pass + finally: + colored_log(BLUE, "LOG", "Exiting...") + stopping.set() diff --git a/app/responses.py b/app/responses.py new file mode 100644 index 0000000..baa3089 --- /dev/null +++ b/app/responses.py @@ -0,0 +1,82 @@ +import json as _json +import os +import html as _html +from starlette.responses import Response, FileResponse, PlainTextResponse + +class FlaskJSONResponse(Response): + media_type="application/json" + def render(self, content) -> bytes: + return (_json.dumps(content, ensure_ascii=True, separators=(",", ":"), sort_keys=True)+"\n").encode("utf-8") + +class HTTPAbort(Exception): + def __init__(self, code): self.code=code + +def abort(code): raise HTTPAbort(code) + +class FlaskResponse: + """Mutable response holder mirroring the parts of a Flask Response the handlers use.""" + def __init__(self, body=None, status=200, mimetype=None, json_body=None, is_json=False, file_path=None, file_kwargs=None, redirect_to=None, redirect_code=302): + self.body=body + self.status_code=status + self.mimetype=mimetype + self.json_body=json_body + self.is_json=is_json + self.file_path=file_path + self.file_kwargs=file_kwargs or {} + self.redirect_to=redirect_to + self.redirect_code=redirect_code + self.headers={} + +def jsonify(*args, **kwargs): + if args and kwargs: raise TypeError("jsonify behavior undefined for both args and kwargs") + if len(args)==1: data=args[0] + elif args: data=list(args) + else: data=kwargs + return FlaskResponse(json_body=data, is_json=True, mimetype="application/json") + +def make_response(rv): + if isinstance(rv, (FlaskResponse, Response)): return rv + if isinstance(rv, tuple): + resp=make_response(rv[0]) + if len(rv)>=2 and rv[1] is not None: resp.status_code=rv[1] + if len(rv)>=3 and rv[2]: resp.headers.update(rv[2]) + return resp + if isinstance(rv, (dict, list)): return FlaskResponse(json_body=rv, is_json=True, mimetype="application/json") + if isinstance(rv, str): return FlaskResponse(body=rv, mimetype="text/html") + return FlaskResponse(body=rv) + +def redirect(location, code=302): return FlaskResponse(redirect_to=location, redirect_code=code) + +def send_from_directory(directory, filename, mimetype=None, as_attachment=False, download_name=None): + path=os.path.join(directory, filename) + if not os.path.isfile(path): raise HTTPAbort(404) + return FlaskResponse(file_path=path, file_kwargs={"mimetype": mimetype, "as_attachment": as_attachment, "download_name": download_name}) + +def to_starlette(rv) -> Response: + resp=make_response(rv) + if isinstance(resp, Response): return resp + if resp.redirect_to is not None: + loc=resp.redirect_to + display=_html.escape(loc) + body=(f"\n\nRedirecting...\n" + f"

Redirecting...

\n

You should be redirected automatically to the target URL: " + f"{display}. If not, click the link.\n") + out=Response(content=body, status_code=resp.redirect_code, media_type="text/html; charset=utf-8") + out.headers["Location"]=loc + elif resp.file_path is not None: + fk=resp.file_kwargs + disposition="attachment" if fk.get("as_attachment") else "inline" + out=FileResponse(resp.file_path, media_type=fk.get("mimetype"), filename=fk.get("download_name") if fk.get("as_attachment") else None, content_disposition_type=disposition) + elif resp.is_json: + out=FlaskJSONResponse(content=resp.json_body, status_code=resp.status_code) + elif resp.mimetype and resp.mimetype.startswith("text/") and not isinstance(resp.body, (bytes, bytearray)): + out=PlainTextResponse(resp.body if resp.body is not None else "", status_code=resp.status_code, media_type=resp.mimetype) + else: + body=resp.body if resp.body is not None else b"" + if isinstance(body, str): body=body.encode() + out=Response(content=body, status_code=resp.status_code, media_type=resp.mimetype) + if resp.redirect_to is None: + out.status_code=resp.status_code + for k, v in resp.headers.items(): + out.headers[k]=v + return out diff --git a/app/router.py b/app/router.py new file mode 100644 index 0000000..afb08c2 --- /dev/null +++ b/app/router.py @@ -0,0 +1,11 @@ +class Router: + """Minimal Flask-Blueprint-like collector of routes for later registration.""" + def __init__(self, name): + self.name=name + self.routes=[] + def route(self, rule, methods=None, **options): + methods=methods or ["GET"] + def decorator(f): + self.routes.append((rule, list(methods), f)) + return f + return decorator diff --git a/app/routers/__init__.py b/app/routers/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/api/auth.py b/app/routers/auth.py similarity index 96% rename from api/auth.py rename to app/routers/auth.py index 0bff801..6dce2c4 100644 --- a/api/auth.py +++ b/app/routers/auth.py @@ -1,19 +1,21 @@ -from flask import Blueprint, request, jsonify -from utils import version, dev_mode -from .utils import ( +from app.router import Router +from app.compat import request +from app.responses import jsonify +from app.config import version, dev_mode +from app.api_utils import ( make_json_error, logged_in, pass_db, validate_request_data, sliding_window_rate_limiter, public_key_open, get_challenge, timestamp, challenges, challenges_lock, regex_first_group_encrypted, browser_regex, device_regex, rsa_encrypt, get_channel_last_message_seq, hash_token ) -from utils import generate -from db import SQLite +from app.config import generate +from app.db import SQLite import bcrypt import re -from utils import config, RED, colored_log -from .stream import channel_added, member_join +from app.config import config, RED, colored_log +from app.routers.stream import channel_added, member_join -auth_bp=Blueprint("auth", __name__) +auth_bp=Router("auth") @auth_bp.route("/") def index(): return {"running": "Parley", "version": version, diff --git a/api/bans.py b/app/routers/bans.py similarity index 96% rename from api/bans.py rename to app/routers/bans.py index 41794c8..e984308 100644 --- a/api/bans.py +++ b/app/routers/bans.py @@ -1,11 +1,13 @@ -from flask import Blueprint, request, jsonify -from .utils import ( +from app.router import Router +from app.compat import request +from app.responses import jsonify +from app.api_utils import ( make_json_error, logged_in, sliding_window_rate_limiter, timestamp, get_args_int, perm, has_permission, get_pagination_params ) -from db import SQLite +from app.db import SQLite -bans_bp=Blueprint("bans", __name__) +bans_bp=Router("bans") @bans_bp.route("/channel//bans") @logged_in() diff --git a/api/calls.py b/app/routers/calls.py similarity index 94% rename from api/calls.py rename to app/routers/calls.py index 3ee730c..efe483d 100644 --- a/api/calls.py +++ b/app/routers/calls.py @@ -1,10 +1,12 @@ -from flask import Blueprint, request, jsonify -from .utils import make_json_error, logged_in, sliding_window_rate_limiter, timestamp, validate_request_data -from .stream import call_start, call_join, call_left, call_signal -from utils import config -from db import SQLite +from app.router import Router +from app.compat import request +from app.responses import jsonify +from app.api_utils import make_json_error, logged_in, sliding_window_rate_limiter, timestamp, validate_request_data +from app.routers.stream import call_start, call_join, call_left, call_signal +from app.config import config +from app.db import SQLite -calls_bp=Blueprint("calls", __name__) +calls_bp=Router("calls") @calls_bp.route("/channel//call", methods=["POST"]) @logged_in() diff --git a/api/channels.py b/app/routers/channels.py similarity index 98% rename from api/channels.py rename to app/routers/channels.py index 7e76577..44b4840 100644 --- a/api/channels.py +++ b/app/routers/channels.py @@ -1,17 +1,19 @@ import re import json -from flask import Blueprint, request, jsonify -from .utils import ( +from app.router import Router +from app.compat import request +from app.responses import jsonify +from app.api_utils import ( make_json_error, logged_in, sliding_window_rate_limiter, timestamp, handle_pfp, create_dm_id, perm, has_permission, validate_request_data, check_user_channel_limit, get_channel_last_message_seq ) -from utils import generate -from .stream import channel_added, channel_edited, channel_deleted, member_join, member_leave, emit, dm_unhide -from utils import config -from db import SQLite +from app.config import generate +from app.routers.stream import channel_added, channel_edited, channel_deleted, member_join, member_leave, emit, dm_unhide +from app.config import config +from app.db import SQLite -channels_bp=Blueprint("channels", __name__) +channels_bp=Router("channels") @channels_bp.route("/channels") @logged_in() diff --git a/api/keys.py b/app/routers/keys.py similarity index 96% rename from api/keys.py rename to app/routers/keys.py index b386c43..7300db6 100644 --- a/api/keys.py +++ b/app/routers/keys.py @@ -1,12 +1,14 @@ -from flask import Blueprint, request, jsonify -from .utils import ( +from app.router import Router +from app.compat import request +from app.responses import jsonify +from app.api_utils import ( make_json_error, logged_in, sliding_window_rate_limiter, timestamp, has_permission, perm ) -from utils import generate -from db import SQLite +from app.config import generate +from app.db import SQLite -keys_bp=Blueprint("keys", __name__) +keys_bp=Router("keys") @keys_bp.route("/channel//key") @logged_in() diff --git a/api/members.py b/app/routers/members.py similarity index 97% rename from api/members.py rename to app/routers/members.py index 8f92880..3f335c1 100644 --- a/api/members.py +++ b/app/routers/members.py @@ -1,12 +1,14 @@ -from flask import Blueprint, request, jsonify -from .utils import ( +from app.router import Router +from app.compat import request +from app.responses import jsonify +from app.api_utils import ( make_json_error, logged_in, sliding_window_rate_limiter, perm, has_permission, get_pagination_params ) -from .stream import member_leave, member_perms_changed -from db import SQLite +from app.routers.stream import member_leave, member_perms_changed +from app.db import SQLite -members_bp=Blueprint("members", __name__) +members_bp=Router("members") PERM_BITS=perm.mask.bit_length() @members_bp.route("/channel//members") diff --git a/api/messages.py b/app/routers/messages.py similarity index 98% rename from api/messages.py rename to app/routers/messages.py index dc79a16..72505d7 100644 --- a/api/messages.py +++ b/app/routers/messages.py @@ -1,20 +1,22 @@ -from flask import Blueprint, request, jsonify +from app.router import Router +from app.compat import request +from app.responses import jsonify import json -from .utils import ( +from app.api_utils import ( make_json_error, logged_in, sliding_window_rate_limiter, timestamp, perm, has_permission, validate_request_data, get_file_size_chunked ) -from utils import generate -from .stream import message_sent, message_edited, message_deleted, dm_unhide -from utils import config -from db import SQLite +from app.config import generate +from app.routers.stream import message_sent, message_edited, message_deleted, dm_unhide +from app.config import config +from app.db import SQLite import os import math max_encrypted_msg_len=math.ceil((config["messages"]["max_message_length"]+16)/3)*4 -messages_bp=Blueprint("messages", __name__) +messages_bp=Router("messages") os.makedirs(config["data_dir"]["attachments"], exist_ok=True) diff --git a/api/pins.py b/app/routers/pins.py similarity index 97% rename from api/pins.py rename to app/routers/pins.py index 33c45e0..ee682a2 100644 --- a/api/pins.py +++ b/app/routers/pins.py @@ -1,12 +1,13 @@ -from flask import Blueprint, jsonify +from app.router import Router +from app.responses import jsonify import json -from .utils import ( +from app.api_utils import ( make_json_error, logged_in, sliding_window_rate_limiter, get_pagination_params, has_permission, perm ) -from db import SQLite +from app.db import SQLite -pins_bp=Blueprint("pins", __name__) +pins_bp=Router("pins") @pins_bp.route("/channel//pins") @logged_in() diff --git a/api/stream.py b/app/routers/stream.py similarity index 91% rename from api/stream.py rename to app/routers/stream.py index 9a7ae22..d1a3609 100644 --- a/api/stream.py +++ b/app/routers/stream.py @@ -1,18 +1,25 @@ -from flask import Blueprint, Response, stream_with_context -from .utils import ( +from app.router import Router +from app.api_utils import ( logged_in, sliding_window_rate_limiter, timestamp, perm, has_permission ) -from utils import generate -import time +from app.config import generate +from app.compat import event_loop_var +from starlette.responses import StreamingResponse +import asyncio import json from threading import Lock -from db import SQLite +from app.db import SQLite -stream_bp=Blueprint("stream", __name__) +stream_bp=Router("stream") streams={} streams_lock=Lock() +def _push(stream_data, event_data): + loop=stream_data["loop"] + queue=stream_data["queue"] + loop.call_soon_threadsafe(queue.put_nowait, event_data) + def emit(event_type, data, conditions=None): """Emit event to all matching streams with thread safety""" with streams_lock: @@ -36,7 +43,7 @@ def emit(event_type, data, conditions=None): "data": data, "timestamp": timestamp(True) } - stream_data["pending"].append(event_data) + _push(stream_data, event_data) except: streams_to_remove.append(i) for i in streams_to_remove: @@ -487,10 +494,16 @@ def call_signal(channel_id, from_user_id, signal_type, signal_data, db): "exclude_user": from_user_id }) +def _session_valid(session_id): + db=SQLite() + try: return db.exists("session", {"id": session_id}) + finally: db.close() + @stream_bp.route("/stream") @logged_in(True) @sliding_window_rate_limiter(limit=10, window=60, user_limit=5) def stream(db:SQLite, session_id, id): + loop=event_loop_var.get() channel_ids=db.execute_raw_sql(""" SELECT c.id FROM channels c JOIN members m ON c.id=m.channel_id @@ -515,18 +528,23 @@ def stream(db:SQLite, session_id, id): } }) + queue=asyncio.Queue() + for event in active_call_events: + queue.put_nowait(event) + stream_data={ "channel_ids": channel_ids, "user_id": id, - "pending": active_call_events, + "queue": queue, + "loop": loop, "lock": Lock() } client=generate() with streams_lock: streams[client]=stream_data - def generator(): + async def generator(): try: - yield f": heartbeat\n\n" + yield ": heartbeat\n\n" next_heartbeat=timestamp()+10 session_check_time=timestamp()+60 while True: @@ -534,32 +552,33 @@ def generator(): # Check session validity every 60s if current_time>=session_check_time: - if not db.exists("session", {"id": session_id}): - yield f"event: error\ndata: {{\"error\": \"Invalid_session\"}}\n\n" + valid=await asyncio.to_thread(_session_valid, session_id) + if not valid: + yield "event: error\ndata: {\"error\": \"Invalid_session\"}\n\n" break session_check_time=current_time+60 - # Send heartbeat every 10s - if current_time>=next_heartbeat: - yield f": heartbeat\n\n" - next_heartbeat=current_time+10 - - # Process pending events - with stream_data["lock"]: - pending_events=stream_data["pending"].copy() - stream_data["pending"].clear() - - for event in pending_events: - event_str=json.dumps(event["data"]) - yield f"event: {event['event']}\ndata: {event_str}\n\n" - - time.sleep(0.1) - except Exception as e: - yield f"event: error\ndata: {{\"error\": \"connection_error\"}}\n\n" + timeout=min(next_heartbeat, session_check_time)-current_time + if timeout<0: timeout=0 + try: + event=await asyncio.wait_for(queue.get(), timeout=timeout) + except asyncio.TimeoutError: + if timestamp()>=next_heartbeat: + yield ": heartbeat\n\n" + next_heartbeat=timestamp()+10 + continue + + event_str=json.dumps(event["data"]) + yield f"event: {event['event']}\ndata: {event_str}\n\n" + if timestamp()>=next_heartbeat: + yield ": heartbeat\n\n" + next_heartbeat=timestamp()+10 + except Exception: + yield "event: error\ndata: {\"error\": \"connection_error\"}\n\n" finally: with streams_lock: del streams[client] - resp=Response(stream_with_context(generator()), mimetype="text/event-stream") - resp.headers["Cache-Control"] = "no-cache" + resp=StreamingResponse(generator(), media_type="text/event-stream") + resp.headers["Cache-Control"]="no-cache" return resp diff --git a/api/users.py b/app/routers/users.py similarity index 97% rename from api/users.py rename to app/routers/users.py index 5fa68e1..a64c01b 100644 --- a/api/users.py +++ b/app/routers/users.py @@ -1,13 +1,15 @@ -from flask import Blueprint, request, jsonify -from .utils import ( +from app.router import Router +from app.compat import request +from app.responses import jsonify +from app.api_utils import ( logged_in, sliding_window_rate_limiter, make_json_error, handle_pfp, perm, has_permission, timestamp, hash_token, public_key_open, get_challenge, challenges_lock, challenges ) -from .stream import member_info_changed, member_leave, channel_deleted -from db import SQLite +from app.routers.stream import member_info_changed, member_leave, channel_deleted +from app.db import SQLite -users_bp=Blueprint("users", __name__) +users_bp=Router("users") @users_bp.route("/me") @logged_in() diff --git a/api/webhooks.py b/app/routers/webhooks.py similarity index 96% rename from api/webhooks.py rename to app/routers/webhooks.py index c8ca17a..b1bdc60 100644 --- a/api/webhooks.py +++ b/app/routers/webhooks.py @@ -1,12 +1,14 @@ -from flask import Blueprint, request, jsonify +from app.router import Router +from app.compat import request +from app.responses import jsonify import json import re -from .utils import make_json_error, logged_in, sliding_window_rate_limiter, timestamp, perm, has_permission -from .stream import message_sent -from utils import generate, config -from db import SQLite +from app.api_utils import make_json_error, logged_in, sliding_window_rate_limiter, timestamp, perm, has_permission +from app.routers.stream import message_sent +from app.config import generate, config +from app.db import SQLite -webhooks_bp=Blueprint("webhooks", __name__) +webhooks_bp=Router("webhooks") data_uri_regex=re.compile(r"^data:image\/(png|jpeg|jpg|webp|gif|x-icon);base64,[A-Za-z0-9+/=]+$") def _get_manageable_channel(db, id, channel_id): diff --git a/cli.py b/cli.py index 7187d96..454620e 100644 --- a/cli.py +++ b/cli.py @@ -8,7 +8,7 @@ from rich.panel import Panel from rich import box from rich.text import Text -from db import SQLite +from app.db import SQLite console=Console() diff --git a/main.py b/main.py index dc0b1d9..956d1e6 100644 --- a/main.py +++ b/main.py @@ -1,168 +1,4 @@ -import os, sys -os.chdir(os.path.dirname(os.path.abspath(sys.argv[0])) if getattr(sys, "frozen", False) else os.path.dirname(os.path.abspath(__file__))) -from utils import stopping, db_version, dev_mode, config, BLUE, YELLOW, RED, colored_log -os.makedirs(os.path.dirname(config["data_dir"]["database"]), exist_ok=True) -from flask import Flask, send_from_directory, abort, request, jsonify, redirect, make_response -from api import api_bp -from api.utils import make_json_error, pass_db, process_cors_headers -from werkzeug.utils import safe_join -from db import SQLite -from migrations import run_migrations -import sys +from app.main import run -try: run_migrations() -except Exception as e: - colored_log(RED, "ERROR", f"Migration failed: {e}") - sys.exit(1) - -with SQLite() as db: - db.create_table("users", {"seq": "INTEGER PRIMARY KEY AUTOINCREMENT", "id": "TEXT UNIQUE NOT NULL", "username": "TEXT UNIQUE NOT NULL", "display_name": "TEXT", "pfp": "TEXT", "passkey": "TEXT NOT NULL", "public_key": "TEXT NOT NULL", "created_at": "INTEGER NOT NULL", "FOREIGN KEY (pfp)": "REFERENCES files (id) ON DELETE SET NULL"}) - db.create_table("session", {"seq": "INTEGER PRIMARY KEY AUTOINCREMENT", "user": "TEXT NOT NULL", "token_hash": "TEXT UNIQUE NOT NULL", "id": "TEXT UNIQUE NOT NULL", "device": "TEXT", "browser": "TEXT", "logged_in_at": "INTEGER NOT NULL", "next_challenge": "INTEGER", "FOREIGN KEY (user)": "REFERENCES users (id) ON DELETE CASCADE"}) - db.create_table("channels", {"id": "TEXT PRIMARY KEY", "name": "TEXT", "pfp": "TEXT", "type": "INTEGER NOT NULL CHECK (type IN (1, 2, 3))", "permissions": "INTEGER NOT NULL DEFAULT 0", "dm": "TEXT", "invite_code": "TEXT UNIQUE", "created_at": "INTEGER NOT NULL", "FOREIGN KEY (pfp)": "REFERENCES files (id) ON DELETE SET NULL"}) - db.create_table("members", {"seq": "INTEGER PRIMARY KEY AUTOINCREMENT", "user_id": "TEXT", "channel_id": "TEXT", "joined_at": "INTEGER NOT NULL", "permissions": "INTEGER", "message_seq": "INTEGER DEFAULT 0", "hidden": "INTEGER CHECK (hidden IS NULL OR hidden = 1)", "UNIQUE": "(user_id, channel_id)", "FOREIGN KEY (user_id)": "REFERENCES users (id) ON DELETE CASCADE", "FOREIGN KEY (channel_id)": "REFERENCES channels (id) ON DELETE CASCADE"}) - db.create_table("messages", {"seq": "INTEGER PRIMARY KEY AUTOINCREMENT", "id": "TEXT UNIQUE NOT NULL", "channel_id": "TEXT NOT NULL", "user_id": "TEXT NOT NULL", "content": "TEXT NOT NULL", "key": "TEXT", "iv": "TEXT", "timestamp": "INTEGER NOT NULL", "edited_at": "INTEGER", "replied_to": "TEXT", "signature": "TEXT", "signed_timestamp": "INTEGER", "nonce": "TEXT", "webhook_id": "TEXT", "webhook_name": "TEXT", "webhook_pfp": "TEXT", "FOREIGN KEY (channel_id)": "REFERENCES channels (id) ON DELETE CASCADE", "FOREIGN KEY (user_id)": "REFERENCES users (id) ON DELETE CASCADE"}) - db.create_table("message_pins", {"seq": "INTEGER PRIMARY KEY AUTOINCREMENT", "id": "TEXT UNIQUE NOT NULL", "FOREIGN KEY (id)": "REFERENCES messages (id) ON DELETE CASCADE"}) - db.create_table("files", {"id": "TEXT PRIMARY KEY", "filename": "TEXT", "hash": "TEXT NOT NULL", "size": "INTEGER NOT NULL", "mimetype": "TEXT", "file_type": "TEXT NOT NULL CHECK (file_type IN ('attachment', 'pfp'))", "UNIQUE": "(hash, file_type)"}) - db.create_table("attachment_message", {"file_id": "TEXT NOT NULL", "message_id": "TEXT NOT NULL", "encrypted": "INTEGER NOT NULL DEFAULT 0", "iv": "TEXT", "PRIMARY KEY": "(file_id, message_id)", "FOREIGN KEY (file_id)": "REFERENCES files (id) ON DELETE CASCADE", "FOREIGN KEY (message_id)": "REFERENCES messages (id) ON DELETE CASCADE"}) - db.create_table("channels_keys", {"id": "TEXT NOT NULL", "channel_id": "TEXT", "user_id": "TEXT", "key": "TEXT", "FOREIGN KEY (channel_id)": "REFERENCES channels (id) ON DELETE CASCADE", "FOREIGN KEY (user_id)": "REFERENCES users (id) ON DELETE CASCADE"}) - db.create_table("channels_keys_info", {"seq": "INTEGER PRIMARY KEY AUTOINCREMENT", "key_id": "TEXT UNIQUE NOT NULL", "channel_id": "TEXT", "by": "TEXT", "timestamp": "INTEGER NOT NULL", "expires_at": "INTEGER NOT NULL", "FOREIGN KEY (channel_id)": "REFERENCES channels (id) ON DELETE CASCADE", "FOREIGN KEY (by)": "REFERENCES users (id) ON DELETE SET NULL"}) - db.create_table("message_reads", {"seq": "INTEGER PRIMARY KEY AUTOINCREMENT", "user_id": "TEXT NOT NULL", "channel_id": "TEXT NOT NULL", "last_message_id": "TEXT NOT NULL", "read_at": "INTEGER NOT NULL", "UNIQUE": "(user_id, channel_id)", "FOREIGN KEY (user_id)": "REFERENCES users (id) ON DELETE CASCADE", "FOREIGN KEY (channel_id)": "REFERENCES channels (id) ON DELETE CASCADE", "FOREIGN KEY (last_message_id)": "REFERENCES messages (id) ON DELETE CASCADE"}) - db.create_table("bans", {"seq": "INTEGER PRIMARY KEY AUTOINCREMENT", "user_id": "TEXT NOT NULL", "channel_id": "TEXT NOT NULL", "banned_by": "TEXT NOT NULL", "banned_at": "INTEGER NOT NULL", "reason": "TEXT", "UNIQUE": "(user_id, channel_id)", "FOREIGN KEY (user_id)": "REFERENCES users (id) ON DELETE CASCADE", "FOREIGN KEY (channel_id)": "REFERENCES channels (id) ON DELETE CASCADE", "FOREIGN KEY (banned_by)": "REFERENCES users (id) ON DELETE CASCADE"}) - db.create_table("blocks", {"seq": "INTEGER PRIMARY KEY AUTOINCREMENT", "blocker_id": "TEXT NOT NULL", "blocked_id": "TEXT NOT NULL", "blocked_at": "INTEGER NOT NULL", "UNIQUE": "(blocker_id, blocked_id)", "FOREIGN KEY (blocker_id)": "REFERENCES users (id) ON DELETE CASCADE", "FOREIGN KEY (blocked_id)": "REFERENCES users (id) ON DELETE CASCADE"}) - db.create_table("calls", {"channel_id": "TEXT PRIMARY KEY", "started_by": "TEXT NOT NULL", "started_at": "INTEGER NOT NULL", "FOREIGN KEY (channel_id)": "REFERENCES channels (id) ON DELETE CASCADE", "FOREIGN KEY (started_by)": "REFERENCES users (id) ON DELETE CASCADE"}) - db.create_table("call_participants", {"channel_id": "TEXT NOT NULL", "user_id": "TEXT NOT NULL", "joined_at": "INTEGER NOT NULL", "left_at": "INTEGER", "PRIMARY KEY": "(channel_id, user_id)", "FOREIGN KEY (channel_id)": "REFERENCES calls (channel_id) ON DELETE CASCADE", "FOREIGN KEY (user_id)": "REFERENCES users (id) ON DELETE CASCADE"}) - db.create_table("webhooks", {"id": "TEXT PRIMARY KEY", "channel_id": "TEXT NOT NULL", "name": "TEXT NOT NULL", "pfp": "TEXT", "token": "TEXT NOT NULL", "created_by": "TEXT", "created_at": "INTEGER NOT NULL", "last_used_at": "INTEGER", "FOREIGN KEY (channel_id)": "REFERENCES channels (id) ON DELETE CASCADE", "FOREIGN KEY (created_by)": "REFERENCES users (id) ON DELETE SET NULL"}) - db.create_index("session", "user") - db.create_index("members", "channel_id") - db.create_index("members", "message_seq") - db.create_index("messages", "channel_id") - db.create_index("messages", "user_id") - db.create_index("messages", "timestamp") - db.create_index("files", "file_type") - db.create_index("attachment_message", "message_id") - db.create_index("channels_keys", "id") - db.create_index("channels_keys", "channel_id") - db.create_index("channels_keys", "user_id") - db.create_index("channels_keys_info", "channel_id") - db.create_index("message_reads", "user_id") - db.create_index("message_reads", "channel_id") - db.create_index("call_participants", "channel_id") - db.create_index("call_participants", "user_id") - db.create_index("webhooks", "channel_id") - db.create_index("webhooks", "token", unique=True) - if not db.exists("users", {"id": "0"}): db.insert_data("users", {"id": "0", "username": "__parley_webhooks_system_account_do_not_use__", "display_name": "System", "pfp": None, "passkey": "system", "public_key": "system", "created_at": 0}) - if db.execute_raw_sql("PRAGMA user_version;")[0]["user_version"]!=db_version: db.execute_raw_sql(f"PRAGMA user_version={db_version};") - -uri_prefix="/"+config["uri_prefix"] if config["uri_prefix"] else "" -def route_rule(rule: str): return uri_prefix+rule - -app=Flask(__name__, static_folder=None) - -if config["server"]["proxy"]: - from werkzeug.middleware.proxy_fix import ProxyFix - app.wsgi_app=ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1, x_port=1, x_prefix=1) - -app.config["MAX_CONTENT_LENGTH"]=config["server"]["max_content_length"] - -def app_route(rule: str, **options): return app.route(uri_prefix+rule, **options) - -frontend_hosted=config["frontend"]["hosted"] -frontend_present=os.path.isdir(config["frontend"]["frontend_directory"]) -frontend=frontend_hosted and frontend_present - -error_text={ - "404": "not found", - "405": "method not allowed", - "400": "bad request", - "413": "content too big", - "415": "unsupported media type", - "500": "internal server error" -} - -api_url=route_rule("/api/v1/") -@app.errorhandler(404) -@app.errorhandler(405) -@app.errorhandler(400) -@app.errorhandler(413) -@app.errorhandler(415) -@app.errorhandler(500) -def error_handler(error): - if request.path.startswith(uri_prefix): - if request.path==api_url or (request.path+"/").startswith(api_url): return make_json_error(error.code, error_text[str(error.code)]) - try: return send_from_directory(config["frontend"]["frontend_directory"], f"{error.code}.html") if frontend else error_text[str(error.code)], error.code - except: return error_text[str(error.code)], error.code - else: return jsonify({"error": error_text[str(error.code)]}), error.code - -if frontend: - colored_log(BLUE, "INFO", "Frontend directory present, serving it") - - excluded=[i.lower() for i in config["frontend"]["excluded_frontend_root_paths"]] - - @app_route("/") - def index(): return send_from_directory(config["frontend"]["frontend_directory"], "index.html") - - @app_route("/") - def serve_static(path): - if "." not in path: path+=".html" - safe_path=safe_join(config["frontend"]["frontend_directory"], path) - if not safe_path: abort(404) - safe_path=os.path.relpath(safe_path,config["frontend"]["frontend_directory"]).lower() - for exclude in excluded: - if safe_path.startswith(exclude+os.sep) or safe_path==exclude: abort(404) - return send_from_directory(config["frontend"]["frontend_directory"], path) -elif not frontend_hosted: - colored_log(BLUE, "INFO", "Frontend directory isn't hosted") -elif not frontend_present: - colored_log(RED, "ERROR", "Frontend directory isn't present") - -app.register_blueprint(api_bp, url_prefix=route_rule("/api/v1")) - -api_url=route_rule("/api/v1/") -@app_route("/api/v1") -def api_index(): - resp=make_response(redirect(api_url, 301)) - process_cors_headers(resp) - return resp - -@app_route("/pfp/", methods=["GET"]) -@pass_db -def serve_pfp(db:SQLite, pfp:str): - pfp_data=db.select_data("files", ["id", "mimetype"], {"id": pfp, "file_type": "pfp"}) - if not pfp_data: abort(404) - try: - resp=send_from_directory(config["data_dir"]["pfps"], f"{pfp_data[0]["id"]}.webp", mimetype=pfp_data[0]["mimetype"], as_attachment=False) - process_cors_headers(resp) - return resp - except: - db.cleanup_unused_files() - abort(404) - -@app_route("/attachment/", methods=["GET"]) -@pass_db -def serve_attachment(db:SQLite, file_id:str): - file_data=db.select_data("files", ["id", "filename", "mimetype"], {"id": file_id, "file_type": "attachment"}) - if not file_data: abort(404) - filename=file_data[0]["filename"] or "attachment" - try: - resp=send_from_directory(config["data_dir"]["attachments"], file_data[0]["id"], mimetype=file_data[0]["mimetype"], as_attachment=True, download_name=filename) - process_cors_headers(resp) - return resp - except: - db.cleanup_unused_files() - abort(404) - -@app_route("/health") -def health(): return jsonify({"status": "ok"}) - -colored_log(BLUE, "INFO", f"Access instance at http://{config["server"]["host"]}:{config["server"]["port"]}{uri_prefix}/") -if dev_mode: colored_log(YELLOW, "WARNING", "Dev mode is enabled, please disable this mode if you're running this in production") -if dev_mode: colored_log(BLUE, "DEV MODE INFO", f"Access instance at http://localhost:{config["server"]["port"]}{uri_prefix}/ for local access") -try: - if dev_mode: app.run(host=config["server"]["host"], port=config["server"]["port"], debug=dev_mode, threaded=True) - else: - from waitress import serve - serve(app, host=config["server"]["host"], port=config["server"]["port"], threads=config["server"]["threads"]) -except KeyboardInterrupt: pass -finally: - colored_log(BLUE, "LOG", "Exiting...") - stopping.set() +if __name__=="__main__": + run() diff --git a/migrations.py b/migrations.py index 9ac8788..9f5a661 100644 --- a/migrations.py +++ b/migrations.py @@ -1,7 +1,7 @@ import os import glob -from db import SQLite -from utils import colored_log, BLUE, RED +from app.db import SQLite +from app.config import colored_log, BLUE, RED def run_migrations(): with SQLite() as db: diff --git a/requirements.txt b/requirements.txt index 986eaf1..c0b58aa 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,8 +1,9 @@ -flask -waitress +fastapi>=0.115 +uvicorn[standard]>=0.30 +python-multipart>=0.0.9 nanoid Pillow bcrypt cryptography cffi -rich \ No newline at end of file +rich From a34520a94072122d75ab676b1a0e710ce4abe697 Mon Sep 17 00:00:00 2001 From: FrenchToblerone54 Date: Tue, 7 Jul 2026 19:39:00 +0000 Subject: [PATCH 2/2] Fix E2EE key validation rejecting valid non-latest keys Accept any non-expired channel key on message send instead of only the latest-by-seq key, so clients holding a still-valid older key aren't rejected with 'Invalid or outdated encryption key'. Guard key rotation to return the existing valid key instead of minting a duplicate, closing the race that produced coexisting valid keys. Clarify member-change key expiry comments. --- app/routers/keys.py | 2 ++ app/routers/messages.py | 12 +++++------- app/routers/stream.py | 6 +++--- 3 files changed, 10 insertions(+), 10 deletions(-) diff --git a/app/routers/keys.py b/app/routers/keys.py index 7300db6..4b9187e 100644 --- a/app/routers/keys.py +++ b/app/routers/keys.py @@ -56,6 +56,8 @@ def store_channel_keys(db:SQLite, id, channel_id): if username not in user_keys: return make_json_error(400, "User missing from keys") key_id=generate() with db: + current_key_info=db.execute_raw_sql("SELECT cki.key_id, expires_at FROM channels_keys_info cki WHERE cki.channel_id=? ORDER BY cki.seq DESC LIMIT 1", (channel_id,)) + if current_key_info and current_key_info[0]["expires_at"]>=timestamp(True): return jsonify({"key_id": current_key_info[0]["key_id"], "success": True}), 200 db.insert_data("channels_keys_info", {"key_id": key_id, "channel_id": channel_id, "by": id, "timestamp": timestamp(), "expires_at": timestamp(True)+86400000}) for username, encrypted_key in user_keys.items(): db.insert_data("channels_keys", {"id": key_id, "channel_id": channel_id, "user_id": user_ids[username], "key": encrypted_key}) diff --git a/app/routers/messages.py b/app/routers/messages.py index 72505d7..ddfa95f 100644 --- a/app/routers/messages.py +++ b/app/routers/messages.py @@ -160,14 +160,12 @@ def sending_messages(db:SQLite, id, channel_id): if data["type"]!=3: if "key" not in request.form or "iv" not in request.form: return make_json_error(400, "key and iv is required in non-broadcast channels") key=request.form["key"] - latest_user_key=db.execute_raw_sql( - "SELECT cki.key_id, expires_at FROM channels_keys_info cki " - "WHERE cki.channel_id=? " - "ORDER BY cki.seq DESC LIMIT 1", - (channel_id,) + key_info=db.execute_raw_sql( + "SELECT expires_at FROM channels_keys_info WHERE channel_id=? AND key_id=?", + (channel_id, key) ) - if not latest_user_key or latest_user_key[0]["expires_at"]=?", (channel_id, timestamp()) @@ -269,7 +269,7 @@ def member_join(channel_id, user_data, db): """Emit member join event and update user's channel_ids""" user_id=user_data["id"] - # Update channels_keys_info to expire the latest entry + # Expire all live keys so the next send rotates to one the changed member set can't decrypt update_channel_keys_on_member_change(channel_id, db) # Update the user's channel_ids in their active streams @@ -328,7 +328,7 @@ def member_leave(channel_id, user_data, db): """Emit member leave event and update user's channel_ids""" user_id=user_data["id"] - # Update channels_keys_info to expire the latest entry + # Expire all live keys so the next send rotates to one the changed member set can't decrypt update_channel_keys_on_member_change(channel_id, db) # Create user data without id for the event