"""Optional Supabase sign-in and math-class storage.""" from __future__ import annotations import json import logging import os import re import threading from concurrent.futures import ThreadPoolExecutor from datetime import datetime import time import urllib.error import urllib.parse import urllib.request log = logging.getLogger("mathly.auth") TIMEOUT = 10 CACHE_TTL = 300 CLASSES_TTL = 60 class AuthUnavailable(RuntimeError): pass class ClassConflict(ValueError): pass class SupabaseAuth: def __init__(self, url: str, key: str, teachers: frozenset[str] = frozenset(), secret: str = "", developers: frozenset[str] = frozenset()): self.url, self.key, self.secret = url.rstrip("/"), key, secret self.teachers, self.developers = teachers, developers self._users: dict[str, tuple[float, dict]] = {} self._classes: tuple[float, list[dict]] | None = None self._access: dict[str, tuple[float, dict | None]] = {} self._lock = threading.Lock() def user(self, token: str) -> dict | None: now = time.monotonic() with self._lock: cached = self._users.get(token) if cached and cached[0] > now: return cached[1] request = urllib.request.Request(self.url + "/auth/v1/user", headers={"Authorization": f"Bearer {token}", "apikey": self.key, "User-Agent": "mathly/1.0"}) try: with urllib.request.urlopen(request, timeout=TIMEOUT) as response: data = json.load(response) except urllib.error.HTTPError as exc: if exc.code in (400, 401, 403): return None raise AuthUnavailable(f"supabase answered {exc.code}") from exc except (urllib.error.URLError, TimeoutError, ValueError) as exc: raise AuthUnavailable(str(exc)) from exc if not data.get("id"): return None user = {"id": data["id"], "email": data.get("email", "")} with self._lock: self._users[token] = (now + CACHE_TTL, user) return user def _admin(self, path: str, method: str = "GET", body=None, prefer: str = "return=representation"): if not self.secret: raise AuthUnavailable("SUPABASE_SECRET_KEY is not set") request = urllib.request.Request(self.url + path, method=method, data=json.dumps(body).encode() if body is not None else None, headers={"Authorization": f"Bearer {self.secret}", "apikey": self.secret, "Content-Type": "application/json", "Prefer": prefer, "User-Agent": "mathly/1.0"}) try: with urllib.request.urlopen(request, timeout=TIMEOUT) as response: return json.loads(response.read() or b"null") except urllib.error.HTTPError as exc: detail = exc.read().decode(errors="replace")[:300] if exc.code == 409: raise ClassConflict(detail) from exc raise AuthUnavailable(f"supabase answered {exc.code}: {detail}") from exc except (urllib.error.URLError, TimeoutError, ValueError) as exc: raise AuthUnavailable(str(exc)) from exc def classes(self, fresh: bool = False) -> list[dict]: now = time.monotonic() with self._lock: cached = self._classes if cached and cached[0] > now and not fresh: return cached[1] rows: list[dict] = [] if self.secret: try: rows = self._admin("/rest/v1/classes?select=*&order=created_at.desc") or [] except AuthUnavailable as exc: if "answered 404" not in str(exc): raise log.warning("the classes table is missing; every account is allowed") for row in rows: row["emails"] = [str(value).lower() for value in row.get("emails") or []] row["teachers"] = [str(value).lower() for value in row.get("teachers") or []] row["lessons"] = row.get("lessons") or [] row["nicknames"] = row.get("nicknames") or {} row["exams"] = row.get("exams") or [] row["exam_note"] = str(row.get("exam_note") or "") row["exam_title"] = str(row.get("exam_title") or "") with self._lock: self._classes = (now + CLASSES_TTL, rows) return rows def save_class(self, class_id: int | None, name: str, emails: list[str], nicknames: dict[str, str], lessons: list[str], *, teachers: list[str] | None = None, **_ignored) -> None: body = {"name": name, "emails": emails, "nicknames": nicknames, "lessons": lessons, "teachers": list(teachers or [])} if class_id is None: self._admin("/rest/v1/classes", "POST", body) elif not self._admin(f"/rest/v1/classes?id=eq.{int(class_id)}", "PATCH", body): raise KeyError(class_id) self._classes = None def save_class_exams(self, class_id: int, exams: list[dict]) -> None: """The class's exam board (each exam: subject, date, and the units and note it may name).""" if not self._admin(f"/rest/v1/classes?id=eq.{int(class_id)}", "PATCH", {"exams": exams}): raise KeyError(class_id) self._classes = None def save_class_settings(self, class_id: int, fields: dict) -> None: """A class's AI settings (app/ai_batches.py): automatic practice checks, and its daily AI budget.""" if not self._admin(f"/rest/v1/classes?id=eq.{int(class_id)}", "PATCH", fields): raise KeyError(class_id) self._classes = None def delete_class(self, class_id: int) -> None: self._admin(f"/rest/v1/classes?id=eq.{int(class_id)}", "DELETE") self._classes = None def ping(self, timeout: float = 8) -> dict: """How Supabase answers right now, for Monitoring: its sign-in service and its database, each timed in ms. A part that fails holds {"ok": False, "error"} and does not hide the other.""" def timed(path: str, key: str) -> dict: request = urllib.request.Request(self.url + path, headers={ "apikey": key, "Authorization": f"Bearer {key}", "User-Agent": "mathly/1.0"}) start = time.monotonic() try: with urllib.request.urlopen(request, timeout=timeout) as response: response.read() return {"ok": True, "ms": round((time.monotonic() - start) * 1000)} except urllib.error.HTTPError as exc: return {"ok": False, "ms": round((time.monotonic() - start) * 1000), "error": f"HTTP {exc.code}"} except (urllib.error.URLError, TimeoutError, OSError, ValueError) as exc: return {"ok": False, "error": str(exc) or type(exc).__name__} with ThreadPoolExecutor(2) as pool: auth = pool.submit(timed, "/auth/v1/health", self.key) db = pool.submit(timed, "/rest/v1/classes?select=id&limit=1", self.secret) if self.secret else None return {"auth": auth.result(), "db": db.result() if db else {"ok": False, "error": "SUPABASE_SECRET_KEY is not set"}} def answer_activity(self, since: datetime, limit: int = 10_000) -> dict | None: """Open-answer attempts and photo checks since a time: user, time and verdict only, never the answer. None when the answer tables are not installed.""" stamp = urllib.parse.quote(since.isoformat()) try: attempts = self._admin(f"/rest/v1/math_question_attempts?select=user_id,question_id,created_at" f"&created_at=gte.{stamp}&order=created_at.asc&limit={limit}") or [] except AuthUnavailable as exc: if "answered 404" in str(exc): return None raise try: grades = self._admin(f"/rest/v1/math_answer_grades?select=user_id,state,created_at,verdict:result->>verdict" f"&created_at=gte.{stamp}&order=created_at.asc&limit={limit}") or [] except AuthUnavailable as exc: if "answered 404" not in str(exc): raise grades = [] return {"attempts": attempts, "grades": grades} def user_accounts(self, emails: list[str]) -> dict[str, dict]: """The id and Google name of each email that has signed in at least once (Supabase's admin user list), by email. The name is "" when the account has none.""" wanted, found = {email.lower() for email in emails}, {} page = 1 while wanted - found.keys(): data = self._admin(f"/auth/v1/admin/users?page={page}&per_page=1000") or {} users = data.get("users") if isinstance(data, dict) else data for user in users or []: email = str(user.get("email") or "").lower() if email in wanted and user.get("id"): meta = user.get("user_metadata") if isinstance(user.get("user_metadata"), dict) else {} name = " ".join(str(meta.get("full_name") or meta.get("name") or "").split()) found[email] = {"id": user["id"], "name": name} if len(users or []) < 1000 or page >= 50: break page += 1 return found def user_email(self, user_id: str) -> str: """The email of one account (Supabase's admin API), "" when it has none.""" data = self._admin(f"/auth/v1/admin/users/{user_id}") or {} return str(data.get("email") or "").lower() if isinstance(data, dict) else "" def user_ids(self, emails: list[str]) -> dict[str, str]: """The account id of each email that has signed in at least once, by email.""" return {email: account["id"] for email, account in self.user_accounts(emails).items()} def access_for(self, user_id: str, fresh: bool = False) -> dict | None: if not self.secret: return None now = time.monotonic() with self._lock: cached = self._access.get(user_id) if cached and cached[0] > now and not fresh: return cached[1] try: rows = self._admin("/rest/v1/user_access?" + urllib.parse.urlencode( {"user_id": f"eq.{user_id}", "select": "user_id,email,roles,updated_at"})) value = rows[0] if rows else None except AuthUnavailable as exc: if "answered 404" not in str(exc): raise value = None with self._lock: self._access[user_id] = (now + CLASSES_TTL, value) return value def load_auth() -> SupabaseAuth | None: url = os.environ.get("SUPABASE_URL", "").strip() key = os.environ.get("SUPABASE_PUBLISHABLE_KEY", "").strip() if not url or not key: log.info("sign-in is disabled; the math path is public") return None emails = lambda name: frozenset(re.split(r"[\s,;]+", os.environ.get(name, "").strip().lower())) - {""} return SupabaseAuth(url, key, emails("TEACHER_EMAILS"), os.environ.get("SUPABASE_SECRET_KEY", "").strip(), emails("DEVELOPER_EMAILS"))