| """HTTP API and dashboard. Routes and responses match the Node version.""" |
|
|
| from __future__ import annotations |
|
|
| import hmac |
| import logging |
| import re |
| import secrets |
| import time |
| from urllib.parse import urlsplit |
|
|
| import asyncpg |
| from fastapi import FastAPI, Request |
| from fastapi.responses import FileResponse, JSONResponse |
| from fastapi.staticfiles import StaticFiles |
|
|
| log = logging.getLogger("web") |
|
|
| ROLES = ("free", "premium", "admin") |
| CODE_RE = re.compile(r"^[A-Za-z0-9_-]{3,20}$") |
| ALPHABET = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789" |
| LOCK_AFTER = 5 |
| LOCK_SECONDS = 10 * 60 |
|
|
|
|
| def _err(status: int, message: str) -> JSONResponse: |
| return JSONResponse({"error": message}, status_code=status) |
|
|
|
|
| def _digits(value) -> str: |
| return re.sub(r"\D", "", str(value or "")) |
|
|
|
|
| def _valid_url(url: str) -> bool: |
| try: |
| parts = urlsplit(url) |
| except ValueError: |
| return False |
| return parts.scheme in ("http", "https") and bool(parts.netloc) |
|
|
|
|
| def create_app(static_dir: str) -> FastAPI: |
| app = FastAPI(title="File Desk", docs_url=None, redoc_url=None, openapi_url=None) |
| failed: dict[str, dict] = {} |
|
|
| def svc(request: Request): |
| return request.app.state.services |
|
|
| def client_ip(request: Request) -> str: |
| forwarded = request.headers.get("x-forwarded-for", "") |
| if forwarded: |
| return forwarded.split(",")[0].strip() |
| return request.client.host if request.client else "unknown" |
|
|
| @app.middleware("http") |
| async def admin_auth(request: Request, call_next): |
| path = request.url.path |
| if not path.startswith("/api/") or path == "/api/catalog": |
| return await call_next(request) |
| s = getattr(request.app.state, "services", None) |
| if s is None: |
| return _err(503, "Starting. Try again in a moment.") |
| ip = client_ip(request) |
| record = failed.get(ip, {"n": 0, "until": 0.0}) |
| if time.time() < record["until"]: |
| return _err(429, "Too many wrong attempts. Try again in a few minutes.") |
| given = request.headers.get("x-admin-key", "").encode("utf-8") |
| if hmac.compare_digest(given, s.cfg.admin_password.encode("utf-8")): |
| failed.pop(ip, None) |
| return await call_next(request) |
| record["n"] += 1 |
| if record["n"] >= LOCK_AFTER: |
| record = {"n": 0, "until": time.time() + LOCK_SECONDS} |
| failed[ip] = record |
| return _err(401, "Wrong password.") |
|
|
| @app.exception_handler(Exception) |
| async def unhandled(request: Request, exc: Exception): |
| log.error("API error on %s: %s", request.url.path, exc) |
| return _err(500, "Server error. Check the console log.") |
|
|
| async def body(request: Request) -> dict: |
| try: |
| data = await request.json() |
| except Exception: |
| return {} |
| return data if isinstance(data, dict) else {} |
|
|
| |
| @app.get("/health") |
| async def health(request: Request): |
| return {"status": "ok", "whatsapp": svc(request).wa.connected} |
|
|
| @app.get("/api/status") |
| async def status(request: Request): |
| return svc(request).wa.status() |
|
|
| @app.post("/api/pair") |
| async def pair(request: Request): |
| s = svc(request) |
| phone = _digits((await body(request)).get("phone")) |
| if len(phone) < 10 or len(phone) > 15: |
| return _err(400, "Use country code + number, e.g. 94771234567.") |
| if s.wa.client is None: |
| return _err(503, "Bot is starting. Try again in 10 seconds.") |
| if s.wa.connected or s.wa.me_jid is not None: |
| return _err(409, 'Already linked. Use "Unlink" first to link a different number.') |
| try: |
| code = await s.wa.pair_phone(phone) |
| except Exception as e: |
| return _err(500, f"Could not create pairing code: {e}") |
| return {"code": code} |
|
|
| @app.post("/api/unlink") |
| async def unlink(request: Request): |
| await svc(request).wa.unlink() |
| return {"ok": True} |
|
|
| |
| @app.get("/api/files") |
| async def list_files(request: Request): |
| return await svc(request).db.list_files() |
|
|
| @app.post("/api/files") |
| async def add_file(request: Request): |
| s = svc(request) |
| data = await body(request) |
| title = str(data.get("title") or "").strip() |
| url = str(data.get("url") or "").strip() |
| type_ = "video" if data.get("type") == "video" else "document" |
| code = str(data.get("code") or "").strip() |
| delete_after = data.get("deleteAfterSend") is True |
|
|
| if not title or len(title) > 100: |
| return _err(400, "Title is required (max 100 characters).") |
| if not _valid_url(url): |
| return _err(400, "Enter a valid http(s) file URL.") |
| if code: |
| if not CODE_RE.match(code): |
| return _err(400, "Code must be 3-20 letters, numbers, - or _.") |
| if await s.db.find_file(code): |
| return _err(409, "That code is already used.") |
| else: |
| code = await _new_code(s.db) |
| try: |
| item = await s.db.add_file(code, title, url, type_, delete_after) |
| except asyncpg.UniqueViolationError: |
| return _err(409, "That code is already used.") |
| return JSONResponse(item, status_code=201) |
|
|
| @app.patch("/api/files/{code}") |
| async def patch_file(code: str, request: Request): |
| listed = (await body(request)).get("listed") |
| if not isinstance(listed, bool): |
| return _err(400, "listed must be true or false.") |
| item = await svc(request).db.set_listed(code, listed) |
| return item if item else _err(404, "Code not found.") |
|
|
| @app.delete("/api/files/{code}") |
| async def delete_file(code: str, request: Request): |
| if await svc(request).db.delete_file(code) == 0: |
| return _err(404, "Code not found.") |
| return {"ok": True} |
|
|
| |
| @app.get("/api/stats") |
| async def stats(request: Request): |
| return await svc(request).db.stats() |
|
|
| @app.get("/api/users") |
| async def users(request: Request): |
| from .bot import limit_for |
|
|
| s = svc(request) |
| q = _digits(request.query_params.get("q", "")) |
| rows = await s.db.list_users(q) |
| settings = await s.db.get_settings() |
| out = [] |
| for u in rows: |
| limit = limit_for(u, settings["limits"]) |
| out.append({**u, "limit": None if limit == float("inf") else limit}) |
| return out |
|
|
| @app.patch("/api/users/{user_id}") |
| async def patch_user(user_id: str, request: Request): |
| s = svc(request) |
| data = await body(request) |
| phone = _digits(user_id) |
| patch: dict = {} |
| if data.get("role") is not None: |
| if data["role"] not in ROLES: |
| return _err(400, "Role must be free, premium or admin.") |
| patch["role"] = data["role"] |
| if data.get("blocked") is not None: |
| patch["blocked"] = data["blocked"] is True |
| if "dailyLimit" in data: |
| value = data["dailyLimit"] |
| if value is None or value == "": |
| patch["dailyLimit"] = None |
| else: |
| try: |
| n = int(value) |
| except (TypeError, ValueError): |
| n = -1 |
| if n < 0 or n > 10000: |
| return _err(400, "Daily limit must be a whole number from 0 to 10000.") |
| patch["dailyLimit"] = n |
| user = await s.db.update_user(phone, patch) if patch else await s.db.get_user(phone) |
| return user if user else _err(404, "User not found.") |
|
|
| @app.get("/api/settings") |
| async def get_settings(request: Request): |
| return await svc(request).db.get_settings() |
|
|
| @app.put("/api/settings") |
| async def put_settings(request: Request): |
| s = svc(request) |
| data = await body(request) |
| patch: dict = {} |
| limits = data.get("limits") |
| if isinstance(limits, dict): |
| patch["limits"] = {} |
| for role in ROLES: |
| if role not in limits: |
| continue |
| value = limits[role] |
| if value is None or value == "": |
| if role != "admin": |
| return _err(400, f"{role} needs a number.") |
| patch["limits"][role] = None |
| continue |
| try: |
| n = int(value) |
| except (TypeError, ValueError): |
| n = -1 |
| if n < 0 or n > 10000: |
| return _err(400, f"{role} limit must be a whole number from 0 to 10000.") |
| patch["limits"][role] = n |
| if "autoDeleteHours" in data: |
| try: |
| h = int(data["autoDeleteHours"]) |
| except (TypeError, ValueError): |
| h = -1 |
| if h < 0 or h > 720: |
| return _err(400, "Auto-delete hours must be a whole number from 0 (never) to 720.") |
| patch["autoDeleteHours"] = h |
| return await s.db.set_settings(patch) |
|
|
| |
| @app.get("/api/catalog") |
| async def catalog(request: Request): |
| s = svc(request) |
| number = s.wa.bot_number |
| items = [] |
| for row in await s.db.list_catalog(): |
| link = None |
| if number: |
| link = f"https://wa.me/{number}?text=" + _quote(".dl " + row["code"]) |
| items.append({"code": row["code"], "title": row["title"], "type": row["type"], "link": link}) |
| return JSONResponse({"botNumber": number, "items": items}, headers={"Cache-Control": "no-store"}) |
|
|
| @app.get("/catalog") |
| async def catalog_page(): |
| return FileResponse(f"{static_dir}/catalog.html") |
|
|
| app.mount("/", StaticFiles(directory=static_dir, html=True), name="static") |
| return app |
|
|
|
|
| def _quote(text: str) -> str: |
| from urllib.parse import quote |
|
|
| return quote(text, safe="") |
|
|
|
|
| async def _new_code(db) -> str: |
| while True: |
| code = "".join(secrets.choice(ALPHABET) for _ in range(5)) |
| if not await db.find_file(code): |
| return code |
|
|
