330 lines
10 KiB
Python
330 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import re
|
|
import secrets
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any
|
|
|
|
import httpx
|
|
from fastapi import Cookie, Depends, Header, HTTPException, status
|
|
from jose import JWTError, jwt
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.config import Settings, get_settings
|
|
from app.database import get_db
|
|
from app.models import AVATAR_ANIMALS, AVATAR_COLORS, ApiKey, User, utcnow
|
|
|
|
SESSION_COOKIE_NAME = "tryme_session"
|
|
SESSION_ALGORITHM = "HS256"
|
|
|
|
_jwks_cache: dict[str, tuple[float, dict[str, Any]]] = {}
|
|
|
|
|
|
def hash_api_key(raw_key: str) -> str:
|
|
return hashlib.sha256(raw_key.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def generate_api_key_pair() -> tuple[str, str, str]:
|
|
token = secrets.token_urlsafe(28)
|
|
raw_key = f"tm_live_{token}"
|
|
prefix = raw_key[:16]
|
|
return raw_key, prefix, hash_api_key(raw_key)
|
|
|
|
|
|
def create_session_token(user_id: int, auth_sub: str, settings: Settings) -> str:
|
|
now = utcnow()
|
|
exp = now + timedelta(hours=settings.session_ttl_hours)
|
|
payload = {
|
|
"sub": auth_sub,
|
|
"uid": user_id,
|
|
"iat": int(now.timestamp()),
|
|
"exp": int(exp.timestamp()),
|
|
"iss": "trymeapp",
|
|
}
|
|
return jwt.encode(payload, settings.session_secret, algorithm=SESSION_ALGORITHM)
|
|
|
|
|
|
def _slugify_nickname(seed: str) -> str:
|
|
cleaned = re.sub(r"[^A-Za-z0-9_\-]", "", seed.strip())
|
|
if len(cleaned) < 2:
|
|
cleaned = "player"
|
|
return cleaned[:36]
|
|
|
|
|
|
def ensure_unique_nickname(db: Session, desired: str) -> str:
|
|
base = _slugify_nickname(desired)
|
|
candidate = base
|
|
suffix = 1
|
|
while db.scalar(select(User.id).where(func.lower(User.nickname) == candidate.lower())) is not None:
|
|
suffix += 1
|
|
candidate = f"{base[:30]}_{suffix}"
|
|
return candidate
|
|
|
|
|
|
def matches_configured_admin(
|
|
settings: Settings,
|
|
*,
|
|
auth_sub: str,
|
|
email: str | None,
|
|
nickname: str,
|
|
) -> bool:
|
|
target = settings.admin_user.strip().lower()
|
|
if not target:
|
|
return False
|
|
candidates = {auth_sub.lower(), nickname.lower()}
|
|
if email:
|
|
candidates.add(email.lower())
|
|
return target in candidates
|
|
|
|
|
|
def get_or_create_user_from_identity(
|
|
db: Session,
|
|
settings: Settings,
|
|
*,
|
|
auth_sub: str,
|
|
email: str | None = None,
|
|
preferred_nickname: str | None = None,
|
|
first_name: str = "",
|
|
last_name: str = "",
|
|
) -> User:
|
|
user = db.scalar(select(User).where(User.auth_sub == auth_sub))
|
|
if user is None:
|
|
seed = preferred_nickname or (email.split("@")[0] if email else auth_sub.split("|")[-1])
|
|
nickname = ensure_unique_nickname(db, seed)
|
|
idx = abs(hash(auth_sub))
|
|
user = User(
|
|
auth_sub=auth_sub,
|
|
email=email,
|
|
nickname=nickname,
|
|
first_name=first_name,
|
|
last_name=last_name,
|
|
avatar_animal=AVATAR_ANIMALS[idx % len(AVATAR_ANIMALS)],
|
|
avatar_color=AVATAR_COLORS[idx % len(AVATAR_COLORS)],
|
|
is_admin=matches_configured_admin(
|
|
settings, auth_sub=auth_sub, email=email, nickname=nickname
|
|
),
|
|
)
|
|
db.add(user)
|
|
db.commit()
|
|
db.refresh(user)
|
|
else:
|
|
promoted = matches_configured_admin(
|
|
settings, auth_sub=user.auth_sub, email=user.email or email, nickname=user.nickname
|
|
)
|
|
changed = False
|
|
if email and not user.email:
|
|
user.email = email
|
|
changed = True
|
|
if promoted and not user.is_admin:
|
|
user.is_admin = True
|
|
changed = True
|
|
if changed:
|
|
db.commit()
|
|
db.refresh(user)
|
|
return user
|
|
|
|
|
|
async def verify_auth0_jwt(token: str, settings: Settings) -> dict[str, Any]:
|
|
domain = settings.auth0_domain.strip().rstrip("/")
|
|
if not domain.startswith("http"):
|
|
issuer_base = f"https://{domain}"
|
|
else:
|
|
issuer_base = domain
|
|
jwks_url = f"{issuer_base}/.well-known/jwks.json"
|
|
|
|
now_ts = datetime.now(timezone.utc).timestamp()
|
|
cached = _jwks_cache.get(jwks_url)
|
|
if cached and (now_ts - cached[0]) < 3600:
|
|
jwks = cached[1]
|
|
else:
|
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
|
resp = await client.get(jwks_url)
|
|
resp.raise_for_status()
|
|
jwks = resp.json()
|
|
_jwks_cache[jwks_url] = (now_ts, jwks)
|
|
|
|
unverified_header = jwt.get_unverified_header(token)
|
|
rsa_key: dict[str, Any] = {}
|
|
for key in jwks.get("keys", []):
|
|
if key.get("kid") == unverified_header.get("kid"):
|
|
rsa_key = {
|
|
"kty": key["kty"],
|
|
"kid": key["kid"],
|
|
"use": key.get("use", "sig"),
|
|
"n": key["n"],
|
|
"e": key["e"],
|
|
}
|
|
break
|
|
if not rsa_key:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Unable to match Auth0 signing key.",
|
|
)
|
|
|
|
audience = settings.auth0_audience or settings.auth0_client_id
|
|
options = {"verify_aud": bool(audience)}
|
|
try:
|
|
payload = jwt.decode(
|
|
token,
|
|
rsa_key,
|
|
algorithms=["RS256"],
|
|
audience=audience or None,
|
|
issuer=f"{issuer_base}/",
|
|
options=options,
|
|
)
|
|
except JWTError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=f"Invalid Auth0 token: {exc}",
|
|
) from exc
|
|
return payload
|
|
|
|
|
|
def _authenticate_api_key(raw_key: str, db: Session) -> User:
|
|
digest = hash_api_key(raw_key.strip())
|
|
api_key = db.scalar(select(ApiKey).where(ApiKey.key_hash == digest))
|
|
if api_key is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid API key.",
|
|
)
|
|
now = utcnow()
|
|
if api_key.revoked_at is not None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="This API key has been expired.",
|
|
)
|
|
if api_key.expires_at is not None:
|
|
exp = api_key.expires_at
|
|
if exp.tzinfo is None:
|
|
exp = exp.replace(tzinfo=timezone.utc)
|
|
if exp <= now:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="This API key has reached its expiration date.",
|
|
)
|
|
api_key.last_used_at = now
|
|
db.commit()
|
|
user = db.get(User, api_key.user_id)
|
|
if user is None:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found.")
|
|
return user
|
|
|
|
|
|
async def get_optional_user(
|
|
db: Session = Depends(get_db),
|
|
settings: Settings = Depends(get_settings),
|
|
authorization: str | None = Header(default=None),
|
|
x_api_key: str | None = Header(default=None, alias="X-API-Key"),
|
|
session_cookie: str | None = Cookie(default=None, alias=SESSION_COOKIE_NAME),
|
|
) -> User | None:
|
|
if x_api_key:
|
|
return _authenticate_api_key(x_api_key, db)
|
|
|
|
token: str | None = None
|
|
if authorization and authorization.lower().startswith("bearer "):
|
|
candidate = authorization[7:].strip()
|
|
if candidate.startswith("tm_live_"):
|
|
return _authenticate_api_key(candidate, db)
|
|
token = candidate
|
|
elif session_cookie:
|
|
token = session_cookie
|
|
|
|
if not token:
|
|
return None
|
|
|
|
# First check app-issued HS256 session JWT
|
|
try:
|
|
payload = jwt.decode(
|
|
token,
|
|
settings.session_secret,
|
|
algorithms=[SESSION_ALGORITHM],
|
|
options={"verify_aud": False},
|
|
)
|
|
if payload.get("iss") == "trymeapp" and payload.get("uid"):
|
|
user = db.get(User, int(payload["uid"]))
|
|
if user is not None:
|
|
if matches_configured_admin(
|
|
settings,
|
|
auth_sub=user.auth_sub,
|
|
email=user.email,
|
|
nickname=user.nickname,
|
|
) and not user.is_admin:
|
|
user.is_admin = True
|
|
db.commit()
|
|
db.refresh(user)
|
|
return user
|
|
except JWTError:
|
|
pass
|
|
|
|
# Fallback: federated Auth0 RS256 JWT
|
|
if settings.auth0_configured:
|
|
claims = await verify_auth0_jwt(token, settings)
|
|
sub = str(claims.get("sub", ""))
|
|
if not sub:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Auth0 token missing subject claim.",
|
|
)
|
|
return get_or_create_user_from_identity(
|
|
db,
|
|
settings,
|
|
auth_sub=sub,
|
|
email=claims.get("email"),
|
|
preferred_nickname=claims.get("nickname"),
|
|
first_name=claims.get("given_name", "") or "",
|
|
last_name=claims.get("family_name", "") or "",
|
|
)
|
|
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid or expired session token.",
|
|
)
|
|
|
|
|
|
async def get_current_user(
|
|
user: User | None = Depends(get_optional_user),
|
|
) -> User:
|
|
if user is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Sign in or provide a valid X-API-Key header to continue.",
|
|
)
|
|
return user
|
|
|
|
|
|
async def get_active_user(
|
|
user: User = Depends(get_current_user),
|
|
) -> User:
|
|
if user.is_banned:
|
|
reason_suffix = f" Reason: {user.ban_reason}" if user.ban_reason else ""
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail=f"Your account has been suspended by a moderator.{reason_suffix}",
|
|
)
|
|
return user
|
|
|
|
|
|
async def get_consented_user(
|
|
user: User = Depends(get_active_user),
|
|
) -> User:
|
|
if not user.public_website_consent:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Please complete profile setup and confirm permission to use public websites.",
|
|
)
|
|
return user
|
|
|
|
|
|
async def get_admin_user(
|
|
user: User = Depends(get_active_user),
|
|
) -> User:
|
|
if not user.is_admin:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Moderator privileges are required for this action.",
|
|
)
|
|
return user
|