Spaces:
Sleeping
Sleeping
| """Admin-key authentication and small in-memory rate limiters (ADR 0004, T-04..T-06).""" | |
| from __future__ import annotations | |
| import hashlib | |
| import hmac | |
| import time | |
| from collections import defaultdict, deque | |
| from fastapi import Header, HTTPException, Request | |
| from . import constants as C | |
| from .config import get_settings | |
| from . import firebase_auth | |
| _fails: dict[str, deque] = defaultdict(deque) | |
| _hits: dict[str, deque] = defaultdict(deque) | |
| def client_ip(request: Request) -> str: | |
| fwd = request.headers.get("x-forwarded-for") | |
| if fwd: | |
| return fwd.split(",")[0].strip() | |
| return request.client.host if request.client else "unknown" | |
| def _prune(q: deque, window: float) -> None: | |
| cutoff = time.monotonic() - window | |
| while q and q[0] < cutoff: | |
| q.popleft() | |
| def reset_limits() -> None: | |
| _fails.clear() | |
| _hits.clear() | |
| def admin_enabled() -> bool: | |
| if get_settings().production: # named, signed-in staff only in production | |
| return False | |
| return len(get_settings().admin_key) >= C.ADMIN_KEY_MIN_LENGTH | |
| def keys_match(supplied: str, expected: str) -> bool: | |
| a = hashlib.sha256(supplied.encode()).digest() | |
| b = hashlib.sha256(expected.encode()).digest() | |
| return hmac.compare_digest(a, b) | |
| class Admin: | |
| def __init__(self, name: str): | |
| self.name = name | |
| def require_admin(request: Request, | |
| x_admin_key: str | None = Header(default=None), | |
| x_admin_name: str | None = Header(default=None), | |
| authorization: str | None = Header(default=None)) -> Admin: | |
| # Preferred: a named staff member signed in with Firebase (email must be verified and on the list). | |
| claims = firebase_auth.claims_from_header(authorization) | |
| if claims: | |
| if not claims["email_verified"]: | |
| raise HTTPException(403, "This account is not staff: its email address is not verified.") | |
| if claims["email"] not in get_settings().staff_emails: | |
| raise HTTPException(403, "This account is not staff: its email address is not on the staff list.") | |
| return Admin(claims["email"]) | |
| if not admin_enabled(): | |
| raise HTTPException(503, "Admin console is disabled: sign in as staff or set ADMIN_KEY (16+ characters).") | |
| ip = client_ip(request) | |
| q = _fails[ip] | |
| _prune(q, C.ADMIN_FAIL_WINDOW_SECONDS) | |
| if len(q) >= C.ADMIN_FAIL_LIMIT: | |
| raise HTTPException(429, "Too many failed attempts. Try again later.") | |
| if not x_admin_key or not keys_match(x_admin_key, get_settings().admin_key): | |
| q.append(time.monotonic()) | |
| raise HTTPException(401, "Invalid admin key.") | |
| name = (x_admin_name or "").strip()[:60] or "admin" | |
| return Admin(name) | |
| def allow_correction(request: Request) -> bool: | |
| ip = client_ip(request) | |
| q = _hits[ip] | |
| _prune(q, 3600) | |
| if len(q) >= C.CORRECTION_RATE_PER_HOUR: | |
| return False | |
| q.append(time.monotonic()) | |
| return True | |