| """ |
| webapp/auth.py |
| Two ways to authenticate to the dashboard: |
| |
| 1. Telegram Mini App `initData` (verify_init_data) β used when the |
| dashboard is opened from inside Telegram via the /dashboard button. |
| Verified per Telegram's documented algorithm: |
| - Parse initData as a query string. |
| - Remove the "hash" field; sort the remaining key=value pairs |
| alphabetically; join with "\n" -> data_check_string. |
| - secret_key = HMAC_SHA256(key="WebAppData", msg=bot_token) |
| - computed_hash = HMAC_SHA256(key=secret_key, msg=data_check_string).hexdigest() |
| - computed_hash must equal the received "hash" field. |
| We additionally reject stale data via `auth_date` (replay protection) |
| and reject anyone whose Telegram user id doesn't match |
| Config.TELEGRAM_OWNER_ID. |
| |
| 2. A plain access token (verify_browser_token) β used when the |
| dashboard is opened directly in a regular browser (no Telegram |
| context at all), which is what happens when a Hugging Face Space |
| serves the dashboard as its landing page. This token is generated |
| once per process start (or pinned via DASHBOARD_ACCESS_TOKEN in |
| .env) and is pushed to the owner over Telegram at startup β see |
| main.py β so it never needs to live in a public README or log |
| that a stranger could stumble onto. |
| |
| Both paths grant the exact same level of access, so both are held to |
| the same bar: a secret only the real owner has, verified server-side |
| on every single request, never a persistent session cookie. |
| """ |
| import hashlib |
| import hmac |
| import json |
| import time |
| from urllib.parse import parse_qsl |
|
|
| from fastapi import HTTPException |
|
|
| from config import Config |
|
|
| MAX_AUTH_AGE_SECONDS = 3600 |
|
|
|
|
| def verify_init_data(init_data: str) -> dict: |
| """Returns the parsed initData dict (with 'user' JSON-decoded) if valid. |
| Raises HTTPException(403) otherwise. Never trust anything from the |
| client without calling this first.""" |
| if not Config.TELEGRAM_BOT_TOKEN: |
| raise HTTPException(status_code=500, detail="Server misconfigured: no bot token") |
|
|
| pairs = parse_qsl(init_data, keep_blank_values=True) |
| data = dict(pairs) |
| received_hash = data.pop("hash", None) |
| if not received_hash: |
| raise HTTPException(status_code=403, detail="Init data missing signature") |
|
|
| data_check_string = "\n".join(f"{k}={v}" for k, v in sorted(data.items())) |
|
|
| secret_key = hmac.new(b"WebAppData", Config.TELEGRAM_BOT_TOKEN.encode(), hashlib.sha256).digest() |
| computed_hash = hmac.new(secret_key, data_check_string.encode(), hashlib.sha256).hexdigest() |
|
|
| if not hmac.compare_digest(computed_hash, received_hash): |
| raise HTTPException(status_code=403, detail="Init data signature invalid") |
|
|
| auth_date = data.get("auth_date") |
| if auth_date: |
| age = time.time() - int(auth_date) |
| if age > MAX_AUTH_AGE_SECONDS: |
| raise HTTPException(status_code=403, detail="Init data expired β reopen the dashboard") |
|
|
| user_raw = data.get("user") |
| user = json.loads(user_raw) if user_raw else {} |
| user_id = user.get("id") |
|
|
| if not Config.TELEGRAM_OWNER_ID or user_id != Config.TELEGRAM_OWNER_ID: |
| raise HTTPException(status_code=403, detail="Not authorized β this dashboard is owner-only") |
|
|
| data["user"] = user |
| data["auth_via"] = "telegram" |
| return data |
|
|
|
|
| def verify_browser_token(token: str) -> dict: |
| """Returns a minimal identity dict if the plain access token matches. |
| Raises HTTPException(403) otherwise.""" |
| if not token or not Config.DASHBOARD_ACCESS_TOKEN: |
| raise HTTPException(status_code=403, detail="Missing or invalid access token") |
| if not hmac.compare_digest(token, Config.DASHBOARD_ACCESS_TOKEN): |
| raise HTTPException(status_code=403, detail="Invalid access token") |
| return {"user": {"id": Config.TELEGRAM_OWNER_ID}, "auth_via": "browser_token"} |
|
|
|
|
| def authorize(init_data: str = None, access_token: str = None) -> dict: |
| """Single entry point every route should call. Tries Telegram initData |
| first (if provided), otherwise falls back to the plain browser token. |
| Raises HTTPException(403) if neither is present/valid.""" |
| if init_data: |
| return verify_init_data(init_data) |
| if access_token: |
| return verify_browser_token(access_token) |
| raise HTTPException(status_code=403, detail="No credentials provided") |
|
|