File size: 5,685 Bytes
b387e01 4c5fda9 430efad 4c5fda9 b387e01 430efad 4c5fda9 430efad b387e01 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 | """Supabase Auth JWT verification + PostgREST helpers."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
import httpx
import jwt
from fastapi import HTTPException, status
from jwt import PyJWKClient
from app.config import (
AUTH_ENABLED,
SUPABASE_ANON_KEY,
SUPABASE_JWT_SECRET,
SUPABASE_SERVICE_ROLE_KEY,
SUPABASE_URL,
)
@dataclass(frozen=True)
class AuthUser:
id: str
email: str | None
_jwks_client: PyJWKClient | None = None
def _get_jwks_client() -> PyJWKClient:
global _jwks_client
if _jwks_client is None:
# Newer Supabase projects sign with asymmetric keys
_jwks_client = PyJWKClient(f"{SUPABASE_URL}/auth/v1/.well-known/jwks.json")
return _jwks_client
def verify_access_token(token: str) -> AuthUser:
"""Validate a Supabase access token and return the user id/email."""
if not token:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing access token.")
payload: dict[str, Any] | None = None
errors: list[str] = []
if SUPABASE_JWT_SECRET:
try:
payload = jwt.decode(
token,
SUPABASE_JWT_SECRET,
algorithms=["HS256"],
audience="authenticated",
)
except jwt.PyJWTError as exc:
errors.append(f"hs256:{exc}")
if payload is None:
try:
signing_key = _get_jwks_client().get_signing_key_from_jwt(token)
payload = jwt.decode(
token,
signing_key.key,
algorithms=["ES256", "RS256"],
audience="authenticated",
)
except Exception as exc: # noqa: BLE001 — fall through to Auth API
errors.append(f"jwks:{exc}")
if payload is None:
# Last resort: ask Supabase Auth
try:
with httpx.Client(timeout=10.0) as client:
res = client.get(
f"{SUPABASE_URL}/auth/v1/user",
headers={
"Authorization": f"Bearer {token}",
"apikey": SUPABASE_ANON_KEY,
},
)
if res.status_code == 200:
data = res.json()
uid = data.get("id")
if uid:
return AuthUser(id=str(uid), email=data.get("email"))
errors.append(f"auth_api:{res.status_code}")
except Exception as exc: # noqa: BLE001
errors.append(f"auth_api:{exc}")
if not payload or not payload.get("sub"):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid or expired session. Please sign in again.",
)
return AuthUser(id=str(payload["sub"]), email=payload.get("email"))
def rest_headers(*, service: bool = True) -> dict[str, str]:
key = SUPABASE_SERVICE_ROLE_KEY if service else SUPABASE_ANON_KEY
return {
"apikey": key,
"Authorization": f"Bearer {key}",
"Content-Type": "application/json",
"Prefer": "return=representation",
}
def rest_get(path: str, params: dict[str, str] | None = None) -> Any:
url = f"{SUPABASE_URL}/rest/v1/{path.lstrip('/')}"
with httpx.Client(timeout=15.0) as client:
res = client.get(url, headers=rest_headers(service=True), params=params or {})
if res.status_code >= 400:
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail=f"Database error ({res.status_code}).",
)
return res.json()
def rest_patch(path: str, body: dict[str, Any], params: dict[str, str] | None = None) -> Any:
url = f"{SUPABASE_URL}/rest/v1/{path.lstrip('/')}"
with httpx.Client(timeout=15.0) as client:
res = client.patch(
url,
headers=rest_headers(service=True),
params=params or {},
json=body,
)
if res.status_code >= 400:
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail=f"Database error ({res.status_code}).",
)
return res.json()
def rest_post(path: str, body: dict[str, Any] | list[dict[str, Any]], *, upsert: bool = False) -> Any:
url = f"{SUPABASE_URL}/rest/v1/{path.lstrip('/')}"
headers = rest_headers(service=True)
if upsert:
headers["Prefer"] = "resolution=merge-duplicates,return=representation"
with httpx.Client(timeout=15.0) as client:
res = client.post(url, headers=headers, json=body)
if res.status_code >= 400:
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail=f"Database error ({res.status_code}).",
)
return res.json()
def auth_public_config() -> dict[str, Any]:
from app.billing.quota import plans_catalog
from app.config import (
GUEST_DAILY_REWRITES,
GUEST_DAILY_WORD_CAP,
GUEST_MAX_WORDS,
SESSION_IDLE_MINUTES,
)
return {
"enabled": AUTH_ENABLED,
"supabase_url": SUPABASE_URL if AUTH_ENABLED else "",
"supabase_anon_key": SUPABASE_ANON_KEY if AUTH_ENABLED else "",
"session_idle_minutes": SESSION_IDLE_MINUTES,
"guest": {
"daily_rewrites": GUEST_DAILY_REWRITES,
"max_words_per_request": GUEST_MAX_WORDS,
"daily_word_cap": GUEST_DAILY_WORD_CAP,
},
"plans": plans_catalog(),
}
|