ryan
c6d9a3b8b6
- Auth via Authelia OIDC (public client + PKCE) with signed session cookies - SQLite DB for users, rides, and insights (SQLAlchemy) - Garage S3 for GPX file storage (boto3) - All API endpoints scoped to authenticated user - Dockerfile + docker-compose.yml with Garage service - Frontend auth check: redirects to /auth/login on 401, shows user badge Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
106 lines
3.2 KiB
Python
106 lines
3.2 KiB
Python
import os
|
|
import logging
|
|
|
|
from authlib.integrations.starlette_client import OAuth
|
|
from fastapi import APIRouter, HTTPException, Request
|
|
from fastapi.responses import RedirectResponse
|
|
from itsdangerous import BadSignature, SignatureExpired, URLSafeTimedSerializer
|
|
|
|
from db import User, get_user_by_id, upsert_user
|
|
|
|
log = logging.getLogger("bikeslop")
|
|
|
|
OIDC_ISSUER = os.getenv("OIDC_ISSUER", "")
|
|
OIDC_CLIENT_ID = os.getenv("OIDC_CLIENT_ID", "")
|
|
APP_BASE_URL = os.getenv("APP_BASE_URL", "http://localhost:8000")
|
|
SECRET_KEY = os.getenv("SECRET_KEY", "dev-secret-change-me")
|
|
|
|
SESSION_COOKIE = "bs_session"
|
|
SESSION_MAX_AGE = 30 * 24 * 3600 # 30 days
|
|
|
|
_signer = URLSafeTimedSerializer(SECRET_KEY)
|
|
|
|
oauth = OAuth()
|
|
oauth.register(
|
|
name="authelia",
|
|
client_id=OIDC_CLIENT_ID,
|
|
client_secret=None, # public client — no secret
|
|
server_metadata_url=f"{OIDC_ISSUER}/.well-known/openid-configuration" if OIDC_ISSUER else None,
|
|
client_kwargs={
|
|
"scope": "openid email profile",
|
|
"code_challenge_method": "S256", # PKCE required by Authelia config
|
|
"token_endpoint_auth_method": "none", # public client
|
|
},
|
|
)
|
|
|
|
|
|
def _sign_session(user_id: str) -> str:
|
|
return _signer.dumps({"uid": user_id})
|
|
|
|
|
|
def _unsign_session(token: str) -> str | None:
|
|
try:
|
|
data = _signer.loads(token, max_age=SESSION_MAX_AGE)
|
|
return data.get("uid")
|
|
except (BadSignature, SignatureExpired):
|
|
return None
|
|
|
|
|
|
def get_session_user(request: Request) -> User | None:
|
|
token = request.cookies.get(SESSION_COOKIE)
|
|
if not token:
|
|
return None
|
|
user_id = _unsign_session(token)
|
|
if not user_id:
|
|
return None
|
|
return get_user_by_id(user_id)
|
|
|
|
|
|
def require_user(request: Request) -> User:
|
|
user = get_session_user(request)
|
|
if user is None:
|
|
raise HTTPException(status_code=401, detail="Not authenticated")
|
|
return user
|
|
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/auth/login")
|
|
async def login(request: Request):
|
|
if not OIDC_ISSUER:
|
|
raise HTTPException(status_code=503, detail="OIDC not configured (set OIDC_ISSUER)")
|
|
redirect_uri = f"{APP_BASE_URL}/auth/callback"
|
|
return await oauth.authelia.authorize_redirect(request, redirect_uri)
|
|
|
|
|
|
@router.get("/auth/callback")
|
|
async def callback(request: Request):
|
|
token = await oauth.authelia.authorize_access_token(request)
|
|
userinfo = token.get("userinfo") or await oauth.authelia.userinfo(token=token)
|
|
sub = userinfo.get("sub")
|
|
if not sub:
|
|
raise HTTPException(status_code=400, detail="OIDC userinfo missing 'sub'")
|
|
email = userinfo.get("email")
|
|
display_name = userinfo.get("name") or userinfo.get("preferred_username") or email
|
|
user = upsert_user(oidc_sub=sub, email=email, display_name=display_name)
|
|
log.info("User logged in: %s (%s)", display_name, user.id)
|
|
session_token = _sign_session(user.id)
|
|
response = RedirectResponse(url="/", status_code=302)
|
|
response.set_cookie(
|
|
SESSION_COOKIE,
|
|
session_token,
|
|
max_age=SESSION_MAX_AGE,
|
|
httponly=True,
|
|
samesite="lax",
|
|
secure=APP_BASE_URL.startswith("https"),
|
|
)
|
|
return response
|
|
|
|
|
|
@router.get("/auth/logout")
|
|
def logout():
|
|
response = RedirectResponse(url="/", status_code=302)
|
|
response.delete_cookie(SESSION_COOKIE)
|
|
return response
|