from typing import Annotated, Optional import logging from datetime import datetime, UTC, timedelta from fastapi import Request, Depends, Response, Header from fastapi.security import APIKeyHeader from simple_openid_connect.data import TokenSuccessResponse from simple_openid_connect.client import OpenidClient from dooris_api import app_config from dooris_api import models, exceptions from dooris_api.ccujack import CCUJackClient logger = logging.getLogger(__name__) api_key_security_scheme = APIKeyHeader(name="Authorization", scheme_name="Static-Token", auto_error=False) async def get_oidc_client(req: Request) -> OpenidClient: return req.app.extra["oidc_client"] OpenidClient = Annotated[OpenidClient, Depends(get_oidc_client)] async def get_logged_in_oidc_user( req: Request, resp: Response, oidc_client: OpenidClient ) -> Optional[models.ApiUser]: # easiest case: we still have an access token (which is the most fleeting component) # everything else should still be valid so we can just use it if all(i in req.cookies for i in ("access_token", "id_token")): logger.debug( "user is fully authenticated, returning current user from existing id_token" ) id_token = oidc_client.decode_id_token( req.cookies["id_token"], nonce=req.cookies.get("auth_nonce", None), ) return models.ApiUser.from_id_token(id_token, req.cookies["id_token"]) # if we have a refresh token, try to get new tokens elif all(i in req.cookies for i in ("refresh_token",)): logger.debug( "user has been previously authenticated, trying to recover with refresh_token" ) auth_start_time = datetime.now(UTC) token_resp = oidc_client.exchange_refresh_token(req.cookies["refresh_token"]) if isinstance(token_resp, TokenSuccessResponse): logger.debug("successfully got new tokens from refresh token") persist_oidc_auth_state( oidc_client, resp, token_resp, auth_start_time, None ) # return the newly gotten info id_token = oidc_client.decode_id_token(token_resp.id_token) return models.ApiUser.from_id_token(id_token, token_resp.id_token) else: logger.debug( "failed to exchange refresh token for new access token: %s", token_resp ) # otherwise we can't meaningfully recover any user information or the user is simply not authenticated else: logger.debug("no currently authenticated oidc user") return None def persist_oidc_auth_state( oidc_client: OpenidClient, resp: Response, tokens: TokenSuccessResponse, auth_start_time: datetime, token_nonce: Optional[str] = None, ): now = datetime.now(UTC) # extract the ID token now to validate its authenticity and properly set the cookie lifetime id_token = oidc_client.decode_id_token(tokens.id_token, nonce=token_nonce) # calculate how long each token is valid at_max_age = auth_start_time - now + timedelta(seconds=tokens.expires_in) id_max_age = datetime.fromtimestamp(id_token.exp, UTC) - now nonce_max_age = max(at_max_age, id_max_age) if tokens.refresh_token is not None and tokens.refresh_expires_in is not None: rt_max_age = ( auth_start_time - now + timedelta(seconds=tokens.refresh_expires_in) ) nonce_max_age = max(at_max_age, rt_max_age, id_max_age) if token_nonce is None: nonce_max_age = timedelta(0) # update cookies resp.set_cookie( "access_token", tokens.access_token, max_age=int(at_max_age.total_seconds()), httponly=True, secure=True, ) if tokens.refresh_token is not None and tokens.refresh_expires_in is not None: resp.set_cookie( "refresh_token", tokens.refresh_token, max_age=int(rt_max_age.total_seconds()), httponly=True, secure=True, ) resp.set_cookie( "id_token", tokens.id_token, max_age=int(id_max_age.total_seconds()), httponly=True, secure=True, ) resp.set_cookie( "auth_nonce", token_nonce, max_age=int(nonce_max_age.total_seconds()), httponly=True, secure=True, ) def clear_oidc_auth_state(resp: Response): resp.set_cookie("access_token", "", max_age=0) resp.set_cookie("refresh_token", "", max_age=0) resp.set_cookie("id_token", "", max_age=0) resp.set_cookie("auth_nonce", "", max_age=0) resp.set_cookie("auth_next", "", max_age=0) resp.set_cookie("auth_state", "", max_age=0) resp.set_cookie("auth_start_time", "", max_age=0) def get_logged_in_token_user(req: Request, token: Optional[str]): if not token or not token.startswith("Static-Token "): logger.debug("No static API-Token was part of the request") return None token = token.removeprefix("Static-Token ") valid_tokens = app_config.get().static_api_tokens if any((i == token for i in valid_tokens)): logger.debug("Successfully authenticated a static API-Token") return models.ApiUser( is_anonymous=False, is_ccchh_user=False, is_token_user=True, may_operate_locks=True, username="static-token", guaranteed_session_until=None, raw_id_token=None, ) return None async def get_api_user( req: Request, resp: Response, oidc_client: OpenidClient, token: Annotated[Optional[str], Depends(api_key_security_scheme)] = None ) -> models.ApiUser: oidc_user = await get_logged_in_oidc_user(req, resp, oidc_client) token_user = get_logged_in_token_user(req, token) if oidc_user is not None: return oidc_user elif token_user is not None: return token_user else: return models.ApiUser( is_anonymous=True, is_ccchh_user=False, is_token_user=False, may_operate_locks=False, username="anonymous", guaranteed_session_until=None, raw_id_token=None, ) ApiUser = Annotated[models.ApiUser, Depends(get_api_user)] async def get_authenticated_user( req: Request, resp: Response, oidc_client: OpenidClient ) -> models.ApiUser: user = await get_api_user(req, resp, oidc_client) if user.is_anonymous: raise exceptions.HttpProblemException.unauthorized(req.url) else: return user AuthenticatedUser = Annotated[models.ApiUser, Depends(get_authenticated_user)] def get_ccujack(req: Request) -> CCUJackClient: return req.app.extra["ccujack"] CCUJackClient = Annotated[CCUJackClient, Depends(get_ccujack)]