# aukpad.py from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request, HTTPException from fastapi.responses import HTMLResponse, RedirectResponse, PlainTextResponse, FileResponse, JSONResponse from fastapi.staticfiles import StaticFiles import json, re, secrets, string, time, os, threading, asyncio, hashlib from collections import defaultdict, deque from typing import Optional app = FastAPI() application = app # alias if you prefer "application" # Vendored highlight.js ES modules, fetched lazily by the editor and only for pads # that actually set a language. Mounted before the catch-all /{doc_id}/ route so # pad-id routing cannot swallow it; the cost is that "static" is not a usable pad id. app.mount("/static", StaticFiles(directory=os.path.join(os.path.dirname(os.path.abspath(__file__)), "vendor")), name="vendor") # Minimum gap between accepted pings, per connection. A ping is one inbound message # fanned out to every peer, so this bounds that amplification. Low enough that rapid # human clicking gets through. PING_MIN_INTERVAL = 0.15 # Languages offered in the editor dropdown. This is a security boundary as much as a # feature list: the value is interpolated into a module URL client-side, so it must # never be free-form. Keep in sync with vendor/fetch.sh. LANGS = ("accesslog", "apache", "bash", "cpp", "css", "diff", "dockerfile", "go", "graphql", "http", "ini", "java", "javascript", "json", "markdown", "nginx", "php", "powershell", "python", "rust", "sql", "typescript", "xml", "yaml") # Environment variables USE_VALKEY = os.getenv("USE_VALKEY", "false").lower() == "true" VALKEY_URL = os.getenv("VALKEY_URL", "redis://localhost:6379/0") MAX_TEXT_SIZE = int(os.getenv("MAX_TEXT_SIZE", "5")) * 1024 * 1024 # 5MB default MAX_CONNECTIONS_PER_IP = int(os.getenv("MAX_CONNECTIONS_PER_IP", "10")) RETENTION_HOURS = int(os.getenv("RETENTION_HOURS", "48")) # Default 48 hours MAX_ROOMS = int(os.getenv("MAX_ROOMS", "10000")) TRUST_PROXY = os.getenv("TRUST_PROXY", "false").lower() == "true" DESCRIPTION = os.getenv("DESCRIPTION", "powered by aukpad.com") ENABLE_METRICS = os.getenv("ENABLE_METRICS", "false").lower() == "true" METRICS_KEY = os.getenv("METRICS_KEY", "") # empty = no auth on /system/metrics START_TIME = time.time() DOC_ID_RE = re.compile(r"^[a-zA-Z0-9_-]{1,64}$") def is_valid_doc_id(doc_id: str) -> bool: return bool(DOC_ID_RE.match(doc_id)) # --- Operational transform core ------------------------------------------------ # An op is a single splice: at `pos`, remove `dele` code units, insert `ins`. Every # textarea interaction - typing, deleting, pasting, Tab, autocorrect - reduces to one # splice, so this is the whole edit vocabulary. Ops travel over the wire instead of # document snapshots because two snapshots taken concurrently cannot be combined: one # has to win, which is what used to drop a peer's text and yank carets between lines. # # Offsets are UTF-16 code units, because that is what a textarea reports and what JS # string indexing uses. Python indexes by code point; the two diverge only on pads # holding astral characters (emoji), which is what splice()'s second path is for. # # diffOp/applyOp/transform in the client mirror this. The two copies MUST agree: # test_ot.py fuzzes convergence, and every update carries the server's length so a # disagreement resyncs instead of silently drifting. MAX_OP_LOG = 256 # ops kept per room, for rebasing edits from stale clients MAX_OP_LOG_BYTES = 262144 # ...and a size ceiling, so one big paste cannot pin memory def count_astral(s: str) -> int: """Chars needing two UTF-16 code units. isascii() is a flag on the string object, so the common case is O(1) and never walks the document.""" if s.isascii(): return 0 return sum(1 for ch in s if ord(ch) > 0xFFFF) def u16len(s: str) -> int: return len(s) + count_astral(s) def splice(text: str, op, u16: bool): """Apply an op; returns (new_text, removed_text). `u16` says the text holds astral characters, so code-unit offsets no longer line up with Python's code-point indexing and the splice has to happen in UTF-16 space. That path costs an encode/decode of the document, which is why it is gated rather than unconditional.""" pos, dele, ins = op if not u16: return text[:pos] + ins + text[pos + dele:], text[pos:pos + dele] b = text.encode("utf-16-le") lo, hi = pos * 2, (pos + dele) * 2 return ((b[:lo] + ins.encode("utf-16-le") + b[hi:]).decode("utf-16-le"), b[lo:hi].decode("utf-16-le")) def transform(a, b, a_first: bool): """Rewrite op `a` so it applies to a text that already has op `b` applied. `a_first` says whether `a` precedes `b` in the server's total order - which is not the same question as which one is already applied. The server rebases an incoming op onto ops it has already ordered (a_first=False); a client rebases an op the server has already ordered onto its own uncommitted edits (a_first=True). Both sides must agree on the resulting order or their documents drift apart, so the flag is required rather than defaulted.""" apos, adel, ains = a bpos, bdel, bins = b aend, bend = apos + adel, bpos + bdel # Two pure inserts at one position do not overlap - only the order of the two # insertions is ambiguous - so settle it on the server's ordering. Without this each # side puts the other's text first and two people typing at the same spot diverge # immediately. if adel == 0 and bdel == 0 and apos == bpos: return a if a_first else (apos + u16len(bins), 0, ains) if bend <= apos: # b entirely before a return (apos + u16len(bins) - bdel, adel, ains) if aend <= bpos: # b entirely after a return a # The two edits touch the same characters. A single splice cannot express the honest # result - a's surviving fragments are no longer contiguous - so clamp, identically on # both sides so they still converge. Only reachable when two people edit the same # characters inside one round trip, where any answer is arbitrary. if apos < bpos: return (apos, bpos - apos, ains) return (bpos + u16len(bins), max(0, aend - bend), ains) def parse_op(data): """Validate a wire op into (pos, dele, ins), or None. Structural checks only: the range is checked after rebasing, since these offsets describe the client's older version of the text, and the size limit is checked against the resulting document so that an oversize paste gets told it was too large rather than malformed.""" if not isinstance(data, dict): return None pos, dele, ins = data.get("pos"), data.get("del"), data.get("ins") # bool subclasses int, so True would otherwise pass as position 1 if isinstance(pos, bool) or isinstance(dele, bool): return None if not isinstance(pos, int) or not isinstance(dele, int) or not isinstance(ins, str): return None if pos < 0 or dele < 0: return None if "\x00" in ins: # matches the POST / check return None try: ins.encode("utf-8") # rejects lone surrogates, which cannot be stored except UnicodeEncodeError: return None return (pos, dele, ins) # --- end operational transform core -------------------------------------------- # test_ot.py slices the block between these two markers and execs it, so that the # fuzzer can run on a bare python with no fastapi installed. Keep them in place. def get_client_ip(conn) -> str: # Only honor proxy headers when explicitly opted in — otherwise attackers # can spoof them to bypass per-IP limits. if TRUST_PROXY: xff = conn.headers.get("x-forwarded-for") if xff: return xff.split(",")[0].strip() xri = conn.headers.get("x-real-ip") if xri: return xri.strip() return conn.client.host if conn.client else "unknown" # Valkey/Redis client (initialized later if enabled) redis_client = None # In-memory rooms: {doc_id: {"text": str, "ver": int, "peers": set[WebSocket], "authed_peers": set[WebSocket], "last_access": float, "pw_hash": bytes|None, "pw_salt": bytes|None, "pw_iter": int|None, "lang": str|None}} rooms: dict[str, dict] = {} # Rate limiting: {ip: [timestamp, timestamp, ...]} rate_limits: dict[str, list] = defaultdict(list) # Failed password attempts per IP (for auth brute-force / CPU-DoS limiting) failed_auth_attempts: dict[str, list] = defaultdict(list) # Connection tracking: {ip: connection_count} connections_per_ip: dict[str, int] = defaultdict(int) # Password hashing parameters PBKDF2_ITERATIONS = 600_000 # OWASP 2023 recommendation for PBKDF2-SHA256 LEGACY_PBKDF2_ITERATIONS = 200_000 # backward-compat for pads hashed before the bump MAX_AUTH_FAILURES = 10 # failed password attempts allowed per window AUTH_FAILURE_WINDOW = 60 # seconds def random_id(n: int = 8) -> str: alphabet = string.ascii_lowercase + string.digits return "".join(secrets.choice(alphabet) for _ in range(n)) def init_valkey(): global redis_client if USE_VALKEY: try: import redis redis_client = redis.from_url(VALKEY_URL, decode_responses=True) redis_client.ping() # Test connection print(f"Valkey/Redis connected: {VALKEY_URL}") except ImportError as e: print(f"Warning: redis package import failed ({e}), falling back to memory-only storage") redis_client = None except Exception as e: print(f"Warning: Failed to connect to Valkey/Redis: {e}") redis_client = None def get_room_data_from_cache(doc_id: str) -> Optional[dict]: if redis_client: try: data = redis_client.get(f"room:{doc_id}") if data: cached = json.loads(data) # Convert hex strings back to bytes if cached.get("pw_hash"): cached["pw_hash"] = bytes.fromhex(cached["pw_hash"]) if cached.get("pw_salt"): cached["pw_salt"] = bytes.fromhex(cached["pw_salt"]) return cached except Exception as e: print(f"Cache read error for {doc_id}: {e}") return None def save_room_data_to_cache(doc_id: str, room: dict): if redis_client: try: data = { "text": room["text"], "ver": room["ver"], "last_access": room.get("last_access", time.time()), "pw_hash": room["pw_hash"].hex() if room.get("pw_hash") else None, "pw_salt": room["pw_salt"].hex() if room.get("pw_salt") else None, "pw_iter": room.get("pw_iter"), "lang": room.get("lang"), } redis_client.setex(f"room:{doc_id}", RETENTION_HOURS * 3600, json.dumps(data)) except Exception as e: print(f"Cache write error for {doc_id}: {e}") def update_room_access_time(doc_id: str): now = time.time() if doc_id in rooms: rooms[doc_id]["last_access"] = now if redis_client: try: data = redis_client.get(f"room:{doc_id}") if data: room_data = json.loads(data) room_data["last_access"] = now redis_client.setex(f"room:{doc_id}", RETENTION_HOURS * 3600, json.dumps(room_data)) # Reset TTL except Exception as e: print(f"Cache access update error for {doc_id}: {e}") def cleanup_old_rooms(): while True: try: now = time.time() cutoff = now - (RETENTION_HOURS * 3600) # Convert hours to seconds # Clean in-memory rooms to_remove = [] for doc_id, room in rooms.items(): if room.get("last_access", 0) < cutoff and len(room.get("peers", set())) == 0: to_remove.append(doc_id) for doc_id in to_remove: del rooms[doc_id] print(f"Cleaned up inactive room: {doc_id}") # Valkey/Redis has TTL, so it cleans up automatically except Exception as e: print(f"Cleanup error: {e}") time.sleep(3600) # Run every hour def check_rate_limit(client_ip: str) -> bool: now = time.time() hour_ago = now - 3600 # Clean old entries rate_limits[client_ip] = [t for t in rate_limits[client_ip] if t > hour_ago] # Check limit (50 per hour) if len(rate_limits[client_ip]) >= 50: return False # Add current request rate_limits[client_ip].append(now) return True def check_auth_rate_limit(client_ip: str) -> bool: now = time.time() cutoff = now - AUTH_FAILURE_WINDOW failed_auth_attempts[client_ip] = [t for t in failed_auth_attempts[client_ip] if t > cutoff] return len(failed_auth_attempts[client_ip]) < MAX_AUTH_FAILURES def record_auth_failure(client_ip: str): failed_auth_attempts[client_ip].append(time.time()) HTML = """ aukpad
disconnected 
1

Content is deleted after __RETENTION_HOURS__ hours of inactivity and can be accessed by the server and anyone with the link and optional password.

This pad is password protected

""" # RETENTION_HOURS is fixed at startup, so bake it in once rather than per request. HTML = HTML.replace("__RETENTION_HOURS__", str(RETENTION_HOURS)) @app.get("/favicon.ico", include_in_schema=False) def favicon(): return FileResponse("favicon.ico") @app.get("/health", include_in_schema=False) def health(): checks = {"app": "ok"} healthy = True # Writable = can we still take new pads, or are we at room capacity? checks["write"] = "ok" if len(rooms) < MAX_ROOMS else "at capacity" healthy = checks["write"] == "ok" # Cache: real round-trip (write, read back, delete) so a half-dead # connection shows up instead of just "configured". if not USE_VALKEY: checks["cache"] = "disabled" elif redis_client is None: checks["cache"] = "unavailable" healthy = False else: key = f"health:{random_id()}" try: redis_client.setex(key, 10, "ok") checks["cache"] = "ok" if redis_client.get(key) == "ok" else "readback failed" redis_client.delete(key) except Exception as e: # Type only — the exception text can contain the connection URL. checks["cache"] = f"error: {type(e).__name__}" if checks["cache"] != "ok": healthy = False return JSONResponse({ "status": "ok" if healthy else "degraded", "checks": checks, "rooms": len(rooms), "uptime_seconds": round(time.time() - START_TIME), }, status_code=200 if healthy else 503) @app.get("/system/info", response_class=HTMLResponse) def get_system_info(): max_text_size_mb = int(os.getenv("MAX_TEXT_SIZE", "5")) html_content = f""" aukpad - System Info

System Information

Instance

{DESCRIPTION}

Simple temporary live collaboration notepad with websockets and FastAPI — pads expire automatically after a configurable retention period.

Features

Endpoints

Configuration

Valkey/Redis: {'Enabled' if USE_VALKEY else 'Disabled'}
Max text size: {max_text_size_mb} MB
Max connections per IP: {MAX_CONNECTIONS_PER_IP}
Retention time: {RETENTION_HOURS} hours

Open Source

Source Code: https://git.uphillsecurity.com/cf7/aukpad
License: Apache License Version 2.0, January 2004
License URL: http://www.apache.org/licenses/
""" return HTMLResponse(html_content) def _metric(out: list, name: str, value, help_text: str, labels: str = ""): out.append(f"# HELP {name} {help_text}") out.append(f"# TYPE {name} gauge") out.append(f"{name}{labels} {value}") def _esc(v: str) -> str: # Prometheus label values escape backslash, double quote and newline. return v.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n") @app.get("/system/metrics", include_in_schema=False) def metrics(request: Request): # Off by default: these numbers are usage data, not for public instances. if not ENABLE_METRICS: raise HTTPException(status_code=404, detail="Not Found") if METRICS_KEY: scheme, _, token = request.headers.get("authorization", "").partition(" ") # encode() both sides: compare_digest rejects non-ASCII str, and the # header is attacker-controlled. if scheme.lower() != "bearer" or not secrets.compare_digest(token.encode(), METRICS_KEY.encode()): return PlainTextResponse("Unauthorized", status_code=401, headers={"WWW-Authenticate": "Bearer"}) now = time.time() # Snapshot: the cleanup thread mutates rooms without a lock. snapshot = list(rooms.values()) peers = authed = active = protected = versions = 0 total_bytes = max_bytes = 0 idle_max = 0.0 for room in snapshot: n = len(room.get("peers", ())) peers += n authed += len(room.get("authed_peers", ())) active += 1 if n else 0 protected += 1 if room.get("pw_hash") else 0 versions += room.get("ver", 0) size = len(room.get("text", "").encode("utf-8")) total_bytes += size max_bytes = max(max_bytes, size) idle_max = max(idle_max, now - room.get("last_access", now)) ip_counts = [c for c in list(connections_per_ip.values()) if c > 0] hour_ago = now - 3600 auth_cutoff = now - AUTH_FAILURE_WINDOW # Both dicts prune lazily (only when that IP acts), so filter by age here. creations = sum(len([t for t in ts if t > hour_ago]) for ts in list(rate_limits.values())) auth_fails = sum(len([t for t in ts if t > auth_cutoff]) for ts in list(failed_auth_attempts.values())) out: list[str] = [] _metric(out, "aukpad_rooms", len(snapshot), "Pads currently held in memory") _metric(out, "aukpad_rooms_active", active, "Pads with at least one connected peer") _metric(out, "aukpad_rooms_protected", protected, "Pads with a password set") _metric(out, "aukpad_peers_connected", peers, "Live WebSocket peers across all pads") _metric(out, "aukpad_peers_authenticated", authed, "Peers that passed pad authentication") _metric(out, "aukpad_stored_bytes", total_bytes, "Total UTF-8 bytes of all pad text") _metric(out, "aukpad_pad_bytes_mean", total_bytes // len(snapshot) if snapshot else 0, "Mean pad size in UTF-8 bytes") _metric(out, "aukpad_pad_bytes_max", max_bytes, "Largest single pad in UTF-8 bytes") _metric(out, "aukpad_pad_versions", versions, "Sum of all pad version counters; drops when pads are evicted") _metric(out, "aukpad_room_idle_seconds_max", round(idle_max, 1), "Seconds since last access of the least recently used pad") _metric(out, "aukpad_client_ips", len(ip_counts), "Distinct IPs holding WebSocket connections") _metric(out, "aukpad_ip_connections_max", max(ip_counts) if ip_counts else 0, "Connection count of the busiest single IP") _metric(out, "aukpad_pads_created_last_hour", creations, "Pad creations via POST in the last hour, across all IPs") _metric(out, "aukpad_auth_failures_recent", auth_fails, f"Failed pad password attempts in the last {AUTH_FAILURE_WINDOW}s") _metric(out, "aukpad_cache_enabled", 1 if USE_VALKEY else 0, "Valkey/Redis cache configured") cache_up, cache_keys = 0, None if redis_client: try: redis_client.ping() cache_up = 1 cache_keys = redis_client.dbsize() except Exception: cache_up = 0 _metric(out, "aukpad_cache_up", cache_up, "Valkey/Redis answered PING") if cache_keys is not None: _metric(out, "aukpad_cache_keys", cache_keys, "Keys in the Valkey/Redis database (all keys, if the DB is shared)") _metric(out, "aukpad_max_rooms", MAX_ROOMS, "Configured pad limit") _metric(out, "aukpad_max_text_bytes", MAX_TEXT_SIZE, "Configured per-pad size limit in bytes") _metric(out, "aukpad_max_connections_per_ip", MAX_CONNECTIONS_PER_IP, "Configured WebSocket connection limit per IP") _metric(out, "aukpad_retention_seconds", RETENTION_HOURS * 3600, "Configured pad retention") _metric(out, "aukpad_start_time_seconds", round(START_TIME, 3), "Unix start time; uptime is time() minus this") _metric(out, "aukpad_build_info", 1, "Instance description as a label", labels=f'{{description="{_esc(DESCRIPTION)}"}}') return PlainTextResponse("\n".join(out) + "\n", media_type="text/plain; version=0.0.4; charset=utf-8") @app.get("/", include_in_schema=False) def root(): return RedirectResponse(url=f"/{random_id()}/", status_code=307) @app.post("/", include_in_schema=False) async def create_pad_with_content(request: Request): # Get client IP client_ip = get_client_ip(request) # Check rate limit if not check_rate_limit(client_ip): raise HTTPException(status_code=429, detail="Rate limit exceeded. Max 50 requests per hour.") # Get and validate content content = await request.body() if not content: raise HTTPException(status_code=400, detail="Empty content not allowed") try: text_content = content.decode('utf-8') except UnicodeDecodeError: raise HTTPException(status_code=400, detail="Content must be valid UTF-8") # Check for null bytes if '\x00' in text_content: raise HTTPException(status_code=400, detail="Null bytes not allowed") # Check text size limit if len(text_content.encode('utf-8')) > MAX_TEXT_SIZE: raise HTTPException(status_code=413, detail=f"Content too large. Max size: {MAX_TEXT_SIZE} bytes") doc_id = random_id() rooms[doc_id] = {"text": text_content, "ver": 1, "peers": set(), "authed_peers": set(), "last_access": time.time(), "pw_hash": None, "pw_salt": None, "pw_iter": None, "lang": None} # Save to cache if enabled save_room_data_to_cache(doc_id, rooms[doc_id]) # Return URL instead of redirect for CLI usage base_url = str(request.base_url).rstrip('/') return PlainTextResponse(f"{base_url}/{doc_id}/\n") @app.get("/{doc_id}/", response_class=HTMLResponse) def pad(doc_id: str): if not is_valid_doc_id(doc_id): raise HTTPException(status_code=400, detail="Invalid pad ID") # Update access time when pad is accessed update_room_access_time(doc_id) return HTMLResponse(HTML) @app.get("/{doc_id}/raw", response_class=PlainTextResponse) def get_raw_pad_content(doc_id: str, request: Request, pw: str = ""): if not is_valid_doc_id(doc_id): raise HTTPException(status_code=400, detail="Invalid pad ID") # Load room into memory if needed if doc_id not in rooms: cached_data = get_room_data_from_cache(doc_id) if cached_data: rooms[doc_id] = { "text": cached_data.get("text", ""), "ver": cached_data.get("ver", 0), "peers": set(), "authed_peers": set(), "last_access": time.time(), "pw_hash": cached_data.get("pw_hash"), "pw_salt": cached_data.get("pw_salt"), "pw_iter": cached_data.get("pw_iter"), "lang": cached_data.get("lang"), } if doc_id not in rooms: return PlainTextResponse("") room = rooms[doc_id] # Enforce password protection if room.get("pw_hash"): if not pw: raise HTTPException(status_code=403, detail="This pad is password protected. Use ?pw=") client_ip = get_client_ip(request) if not check_auth_rate_limit(client_ip): raise HTTPException(status_code=429, detail="Too many failed attempts. Try again later.") iters = room.get("pw_iter") or LEGACY_PBKDF2_ITERATIONS candidate = hashlib.pbkdf2_hmac("sha256", pw.encode(), room["pw_salt"], iters) if candidate != room["pw_hash"]: record_auth_failure(client_ip) raise HTTPException(status_code=403, detail="Wrong password") update_room_access_time(doc_id) return PlainTextResponse(room["text"]) def init_payload(room: dict, authed: bool) -> dict: """The full-snapshot message. Used for the first connect, after a successful unlock, and for every resync, so a client resets its sync state through one code path.""" return { "type": "init", "text": room["text"] if authed else "", "ver": room["ver"], "protected": room["pw_hash"] is not None, "peers": len(room["peers"]), # Withheld until authed: the language is content metadata and would tell # someone sitting at the unlock overlay what kind of file this is. "lang": room.get("lang") if authed else None, } async def _broadcast(doc_id: str, message: dict, exclude: WebSocket | None = None, authed_only: bool = False): room = rooms.get(doc_id) if not room: return dead = [] payload = json.dumps(message) targets = room["authed_peers"] if authed_only else room["peers"] for peer in list(targets): if peer is exclude: continue try: await peer.send_text(payload) except Exception: dead.append(peer) for d in dead: room["peers"].discard(d) room["authed_peers"].discard(d) @app.websocket("/ws/{doc_id}") async def ws(doc_id: str, ws: WebSocket): # Validate doc_id format if not is_valid_doc_id(doc_id): await ws.close(code=1008, reason="Invalid pad ID") return # Get client IP for connection limiting client_ip = get_client_ip(ws) # Check connection limit per IP if connections_per_ip[client_ip] >= MAX_CONNECTIONS_PER_IP: await ws.close(code=1008, reason="Too many connections from this IP") return await ws.accept() connections_per_ip[client_ip] += 1 # Try to load room from cache first if doc_id not in rooms: cached_data = get_room_data_from_cache(doc_id) if cached_data: rooms[doc_id] = { "text": cached_data.get("text", ""), "ver": cached_data.get("ver", 0), "peers": set(), "authed_peers": set(), "last_access": time.time(), "pw_hash": cached_data.get("pw_hash"), "pw_salt": cached_data.get("pw_salt"), "pw_iter": cached_data.get("pw_iter"), "lang": cached_data.get("lang"), } # Refuse to create new rooms when at capacity if doc_id not in rooms and len(rooms) >= MAX_ROOMS: await ws.close(code=1008, reason="Server at capacity") connections_per_ip[client_ip] = max(0, connections_per_ip[client_ip] - 1) return room = rooms.setdefault(doc_id, {"text": "", "ver": 0, "peers": set(), "authed_peers": set(), "last_access": time.time(), "pw_hash": None, "pw_salt": None, "pw_iter": None, "lang": None}) room["peers"].add(ws) # Update access time update_room_access_time(doc_id) # Notify all peers of updated count await _broadcast(doc_id, {"type": "peers_changed", "count": len(room["peers"])}) # Per-connection auth state: already authed if pad has no password last_ping = 0.0 # per-connection ping throttle; see the "ping" branch below authed = room["pw_hash"] is None if authed: room["authed_peers"].add(ws) # Send init; withhold text if protected and not yet authed await ws.send_text(json.dumps(init_payload(room, authed))) try: while True: msg = await ws.receive_text() data = json.loads(msg) if data.get("type") == "auth": if room["pw_hash"] is None: authed = True room["authed_peers"].add(ws) await ws.send_text(json.dumps({"type": "auth_ok"})) await ws.send_text(json.dumps(init_payload(room, True))) else: if not check_auth_rate_limit(client_ip): await ws.send_text(json.dumps({"type": "error", "message": "Too many failed attempts. Try again later."})) continue iters = room.get("pw_iter") or LEGACY_PBKDF2_ITERATIONS candidate = hashlib.pbkdf2_hmac( "sha256", str(data.get("password", "")).encode(), room["pw_salt"], iters ) if candidate == room["pw_hash"]: authed = True room["authed_peers"].add(ws) await ws.send_text(json.dumps({"type": "auth_ok"})) await ws.send_text(json.dumps(init_payload(room, True))) else: record_auth_failure(client_ip) await ws.send_text(json.dumps({"type": "error", "message": "Wrong password"})) continue if not authed: await ws.send_text(json.dumps({"type": "error", "message": "Authentication required"})) continue if data.get("type") == "edit": op = parse_op(data.get("op")) base = data.get("base") if op is None or isinstance(base, bool) or not isinstance(base, int) or base < 0: await ws.send_text(json.dumps({"type": "error", "message": "Malformed edit"})) continue # Derived state, initialized on the first edit rather than in the three # places a room gets built. Both are then O(1) to read and to maintain, # which is what keeps a keystroke off the length of the document. log = room.setdefault("ops", deque(maxlen=MAX_OP_LOG)) astral = room.setdefault("astral", count_astral(room["text"])) nbytes = room.setdefault("bytes", len(room["text"].encode("utf-8"))) # Rebase onto everything that landed since the client's base version. If # the log no longer reaches that far back the client cannot be rebased and # has to start from a snapshot - the same path a room restored from cache # takes, since its op history did not survive. if base < room["ver"] - len(log): await ws.send_text(json.dumps(init_payload(room, True))) continue for v, prev in log: if v > base: # The logged op was ordered first, so the incoming one follows it. op = transform(op, prev, False) if op[0] + op[1] > len(room["text"]) + astral: await ws.send_text(json.dumps(init_payload(room, True))) continue try: new_text, removed = splice(room["text"], op, astral > 0) delta = len(op[2].encode("utf-8")) - len(removed.encode("utf-8")) except UnicodeEncodeError: # A clamped op split a surrogate pair, so the result is not storable. # Rare, and cheaper to resync than to repair. await ws.send_text(json.dumps(init_payload(room, True))) continue if nbytes + delta > MAX_TEXT_SIZE: # The client rolls its own text back to ours, so the two cannot # diverge silently the way they used to when this was ignored. await ws.send_text(json.dumps({"type": "error", "message": f"Text too large. Max size: {MAX_TEXT_SIZE} bytes"})) continue room["text"] = new_text room["bytes"] = nbytes + delta room["astral"] = astral - count_astral(removed) + count_astral(op[2]) room["ver"] += 1 room["last_access"] = time.time() log.append((room["ver"], op)) while len(log) > 1 and sum(len(o[2]) for _, o in log) > MAX_OP_LOG_BYTES: log.popleft() # Save to cache save_room_data_to_cache(doc_id, room) # Goes to everyone including the author: the author recognises its own # clientId and treats the echo as the ack, which carries the op as it was # actually applied - rebased, if it arrived stale. `len` is the divergence # tripwire: a client whose own length disagrees asks for a snapshot. await _broadcast(doc_id, { "type": "update", "op": {"pos": op[0], "del": op[1], "ins": op[2]}, "ver": room["ver"], "len": len(room["text"]) + room["astral"], "clientId": data.get("clientId"), }, authed_only=True) elif data.get("type") == "resync": # A client whose shadow no longer matches us asks for a fresh snapshot. # Unthrottled on purpose: unlike a ping this fans out to nobody, so the # cost lands on the asker's own connection, same as hitting /raw. await ws.send_text(json.dumps(init_payload(room, True))) elif data.get("type") == "set_password": pw = str(data.get("password", "")) if pw: salt = os.urandom(16) room["pw_hash"] = hashlib.pbkdf2_hmac("sha256", pw.encode(), salt, PBKDF2_ITERATIONS) room["pw_salt"] = salt room["pw_iter"] = PBKDF2_ITERATIONS else: room["pw_hash"] = None room["pw_salt"] = None room["pw_iter"] = None save_room_data_to_cache(doc_id, room) await _broadcast(doc_id, {"type": "protected_changed", "protected": room["pw_hash"] is not None}) elif data.get("type") == "set_lang": lang = str(data.get("lang", "")) if lang and lang not in LANGS: continue # unknown language: ignore rather than echo it back room["lang"] = lang or None save_room_data_to_cache(doc_id, room) await _broadcast(doc_id, {"type": "lang_changed", "lang": room["lang"]}, authed_only=True) elif data.get("type") == "ping": # Flash a line for everyone. Stateless: nothing is stored, so there is # nothing to drift, expire or clean up on disconnect. now = time.monotonic() if now - last_ping < PING_MIN_INTERVAL: continue # one inbound message fans out to every peer - throttle it last_ping = now line = data.get("line") # bool subclasses int, so True would otherwise pass as line 1 if isinstance(line, bool) or not isinstance(line, int): continue if not 1 <= line <= room["text"].count("\n") + 1: continue await _broadcast(doc_id, {"type": "ping", "line": line}, authed_only=True) except WebSocketDisconnect: pass finally: room["peers"].discard(ws) room["authed_peers"].discard(ws) await _broadcast(doc_id, {"type": "peers_changed", "count": len(room["peers"])}) # Decrement connection count for this IP connections_per_ip[client_ip] = max(0, connections_per_ip[client_ip] - 1) # Initialize Valkey/Redis and cleanup thread on startup @app.on_event("startup") async def startup_event(): init_valkey() # Start cleanup thread cleanup_thread = threading.Thread(target=cleanup_old_rooms, daemon=True) cleanup_thread.start() print("Aukpad started with cleanup routine") # Run locally: uvicorn aukpad:app --reload