import os import secrets from urllib.parse import urlencode import httpx from fastapi import APIRouter, Depends, HTTPException, Query, Request from fastapi.responses import RedirectResponse from app.auth import ( COOKIE_NAME, OIDC_CLIENT_ID, OIDC_CLIENT_SECRET, SESSION_TTL_SECONDS, get_current_user, is_allowed_user, mint_session_jwt, oidc_config, verify_oidc_id_token, ) router = APIRouter() ENV = os.getenv("ENV", "production") COOKIE_SECURE = os.getenv("COOKIE_SECURE", "true").lower() == "true" PUBLIC_BASE_URL = os.getenv("PUBLIC_BASE_URL", "http://localhost").rstrip("/") STATE_COOKIE = "hk_oauth_state" CALLBACK_PATH = "/api/v1/auth/callback" def _set_session_cookie(resp, token: str): resp.set_cookie( key=COOKIE_NAME, value=token, max_age=SESSION_TTL_SECONDS, path="/", httponly=True, secure=COOKIE_SECURE, samesite="lax", ) @router.get("/login") def login(): config = oidc_config() state = secrets.token_urlsafe(24) params = { "client_id": OIDC_CLIENT_ID, "redirect_uri": f"{PUBLIC_BASE_URL}{CALLBACK_PATH}", "response_type": "code", "scope": "openid profile", "state": state, } resp = RedirectResponse(f"{config['authorization_endpoint']}?{urlencode(params)}") resp.set_cookie(STATE_COOKIE, state, max_age=600, path="/", httponly=True, secure=COOKIE_SECURE, samesite="lax") return resp @router.get("/callback") def callback(request: Request, code: str | None = Query(None), state: str | None = Query(None), error: str | None = Query(None)): if error: raise HTTPException(status_code=400, detail=f"OIDC error: {error}") expected_state = request.cookies.get(STATE_COOKIE) if not code or not state or not expected_state or state != expected_state: raise HTTPException(status_code=400, detail="Invalid OAuth state") config = oidc_config() token_resp = httpx.post( config["token_endpoint"], data={ "grant_type": "authorization_code", "code": code, "redirect_uri": f"{PUBLIC_BASE_URL}{CALLBACK_PATH}", "client_id": OIDC_CLIENT_ID, "client_secret": OIDC_CLIENT_SECRET, }, timeout=10, ) token_resp.raise_for_status() id_token = token_resp.json()["id_token"] claims = verify_oidc_id_token(id_token) username = claims.get("preferred_username") or claims.get("name") if not username or not is_allowed_user(username): raise HTTPException(status_code=403, detail=f"User '{username}' is not allowed to access Homekeeper") resp = RedirectResponse("/") _set_session_cookie(resp, mint_session_jwt(username)) resp.delete_cookie(STATE_COOKIE, path="/") return resp @router.get("/logout") def logout(): resp = RedirectResponse("/") resp.delete_cookie(COOKIE_NAME, path="/") return resp @router.get("/me") def me(username: str = Depends(get_current_user)): return {"username": username} @router.get("/dev-login") def dev_login(username: str = Query(...)): """Local-dev-only bypass: mints a session for `username` without touching the OIDC provider. Only reachable when ENV=development.""" if ENV != "development": raise HTTPException(status_code=404) if not is_allowed_user(username): raise HTTPException(status_code=403, detail=f"User '{username}' is not in ALLOWED_USERS") resp = RedirectResponse("/") _set_session_cookie(resp, mint_session_jwt(username)) return resp