api: restructure user authentication modelling to support other user backends

api/src/dooris_api/deps.py

@@ -1,4 +1,4 @@
-from typing import Annotated, Optional
+from typing import Annotated, Optional, Tuple
 import logging
 from datetime import datetime, UTC, timedelta
 from fastapi import Request, Depends, Response
@@ -20,9 +20,9 @@ async def get_oidc_client(req: Request) -> OpenidClient:
 OpenidClient = Annotated[OpenidClient, Depends(get_oidc_client)]
 
 
-async def get_current_user(
+async def get_logged_in_oidc_user(
     req: Request, resp: Response, oidc_client: OpenidClient
-) -> Optional[models.CurrentUser]:
+) -> 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")):
@@ -30,11 +30,10 @@ async def get_current_user(
             "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.CurrentUser(
-            id_token=id_token, raw_id_token=req.cookies["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",)):
@@ -45,24 +44,26 @@ async def get_current_user(
         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_auth_state(oidc_client, resp, token_resp, auth_start_time, None)
+            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.CurrentUser(
-                id_token=id_token, raw_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)
+            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 user")
+        logger.debug("no currently authenticated oidc user")
 
-    raise exceptions.HttpProblemException.unauthorized(req.url)
+    return None
 
 
-def persist_auth_state(
+def persist_oidc_auth_state(
     oidc_client: OpenidClient,
     resp: Response,
     tokens: TokenSuccessResponse,
@@ -118,7 +119,7 @@ def persist_auth_state(
     )
 
 
-def clear_auth_state(resp: Response):
+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)
@@ -128,7 +129,39 @@ def clear_auth_state(resp: Response):
     resp.set_cookie("auth_start_time", "", max_age=0)
 
 
-CurrentUser = Annotated[Optional[models.CurrentUser], Depends(get_current_user)]
+async def get_api_user(
+    req: Request, resp: Response, oidc_client: OpenidClient
+) -> models.ApiUser:
+    oidc_user = await get_logged_in_oidc_user(req, resp, oidc_client)
+    # TODO: Implement API user based on static tokens
+    if oidc_user is not None:
+        return oidc_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: