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