lilly
ecd6c2c48f
All checks were successful
Build Container / Build Container (push) Successful in 4m38s
225 lines
7.2 KiB
Python
225 lines
7.2 KiB
Python
from typing import Annotated, Optional
|
|
import logging
|
|
from datetime import datetime, UTC, timedelta
|
|
from fastapi import Request, Depends, Response
|
|
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
|
|
from dooris_api.spaceapi import SpaceApiClient
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
api_key_security_scheme = APIKeyHeader(
|
|
name="authorization",
|
|
scheme_name="Static-Token",
|
|
description="Set the Authorization header to 'Static-Token foobar123' with a token that is statically configured at application parameter",
|
|
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]):
|
|
print(req.headers)
|
|
print(token)
|
|
|
|
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,
|
|
token: Annotated[Optional[str], Depends(api_key_security_scheme)] = None,
|
|
) -> models.ApiUser:
|
|
user = await get_api_user(req, resp, oidc_client, token)
|
|
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)]
|
|
|
|
|
|
|
|
def get_spaceapi_client(req: Request) -> SpaceApiClient:
|
|
return req.app.extra["spaceapi"]
|
|
|
|
|
|
SpaceApiClient = Annotated[SpaceApiClient, Depends(get_spaceapi_client)]
|
|
|