api: implement logout endpoint

api/src/dooris_api/app.py

@@ -8,11 +8,11 @@ from fastapi import FastAPI, Request, Response, status
 from fastapi.responses import RedirectResponse
 from contextlib import asynccontextmanager
 from simple_openid_connect.client import OpenidClient
-from simple_openid_connect.data import TokenSuccessResponse
+from simple_openid_connect.data import TokenSuccessResponse, RpInitiatedLogoutRequest
 from cachetools import TTLCache
 from aiohttp import BasicAuth
 
-from dooris_api import deps, models, exceptions
+from dooris_api import deps, models, exceptions, app_config
 from dooris_api.ccujack import CCUJackClient
 
 
@@ -21,54 +21,79 @@ logger = logging.getLogger(__name__)
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    from dooris_api import app_config
+    app_cfg = app_config.get()
-    app_config = app_config.get()
+
-    
     root_logger = logging.getLogger("")
     root_logger.setLevel(logging.INFO)
     root_logger.addHandler(logging.StreamHandler(sys.stderr))
     app_logger = logging.getLogger("dooris_api")
     app_logger.setLevel(logging.DEBUG)
-    
+
     app.extra["oidc_client"] = OpenidClient.from_issuer_url(
-        url=app_config.openid_issuer,
+        url=app_cfg.openid_issuer,
-        authentication_redirect_uri=f"{app_config.base_url}/auth/login-callback",
+        authentication_redirect_uri=f"{app_cfg.base_url}/auth/login-callback",
-        client_id=app_config.openid_client_id,
+        client_id=app_cfg.openid_client_id,
-        client_secret=app_config.openid_client_secret,
+        client_secret=app_cfg.openid_client_secret,
-        scope=app_config.openid_scope,
+        scope=app_cfg.openid_scope,
     )
 
     app.extra["cache"] = TTLCache(maxsize=64, ttl=30 * 60)
 
-    app.extra["ccujack"] = CCUJackClient("https://hmdooris-ccu.ccchh.net:2122", auth=BasicAuth("dooris", os.environ["HMDOORIS_PW"]))
+    app.extra["ccujack"] = CCUJackClient(
-    
+        "https://hmdooris-ccu.ccchh.net:2122",
+        auth=BasicAuth("dooris", os.environ["HMDOORIS_PW"]),
+    )
+
     yield
 
 
 app = FastAPI(
-    title="Dooris", summary="Server to interact with CCCHH doors via locks over CCUJACK over HomeMatic over WiFi", docs_url="/api/docs",
+    title="Dooris",
+    summary="Server to interact with CCCHH doors via locks over CCUJACK over HomeMatic over WiFi",
+    docs_url="/api/docs",
     lifespan=lifespan,
 )
-app.add_exception_handler(exceptions.HttpProblemException, exceptions.problem_exception_handler)
+app.add_exception_handler(
+    exceptions.HttpProblemException, exceptions.problem_exception_handler
+)
 
 
-@app.get("/api/user-info/", name="get-user-info", tags=["auth"], responses={status.HTTP_401_UNAUTHORIZED: {"model": models.HttpProblemDetail}})
+@app.get(
-async def get_user_info(req: Request, current_user: deps.CurrentUser) -> models.UserStatus:
+    "/api/user-info/",
+    name="get-user-info",
+    tags=["auth"],
+    responses={status.HTTP_401_UNAUTHORIZED: {"model": models.HttpProblemDetail}},
+)
+async def get_user_info(
+    req: Request, current_user: deps.CurrentUser
+) -> models.UserStatus:
     if current_user is None:
-        return models.UserStatus(is_logged_in=False, is_authorized=False, username=None, guaranteed_session_until=None)
+        return models.UserStatus(
+            is_logged_in=False,
+            is_authorized=False,
+            username=None,
+            guaranteed_session_until=None,
+        )
     else:
         return models.UserStatus(
             is_logged_in=True,
             is_authorized=True,
-            guaranteed_session_until=datetime.fromtimestamp(current_user.id_token.exp, UTC),
+            guaranteed_session_until=datetime.fromtimestamp(
+                current_user.id_token.exp, UTC
+            ),
             username=current_user.id_token.preferred_username,
         )
 
 
 @app.get("/auth/login", tags=["auth"], response_class=RedirectResponse, status_code=302)
-async def login_init(req: Request, resp: Response, oidc_client: deps.OpenidClient, next: Optional[str] = "") -> str:
+async def login_init(
+    req: Request,
+    resp: Response,
+    oidc_client: deps.OpenidClient,
+    next: Optional[str] = "",
+) -> str:
     logger.debug("starting user authentication with upstream identity provider")
-    
+
     # save the ?next url for later redirection if the user requested that
     if next:
         resp.set_cookie("auth_next", next, max_age=60 * 10, httponly=True, secure=True)
@@ -77,49 +102,94 @@ async def login_init(req: Request, resp: Response, oidc_client: deps.OpenidClien
     # ref: https://simple-openid-connect.readthedocs.io/en/stable/nonce_and_state.html
     state = secrets.token_urlsafe(32)
     resp.set_cookie("auth_state", state, max_age=60 * 10, httponly=True, secure=True)
-    resp.set_cookie("auth_start_time", datetime.now(UTC).timestamp(), max_age=60 * 10, httponly=True, secure=True)
+    resp.set_cookie(
+        "auth_start_time",
+        datetime.now(UTC).timestamp(),
+        max_age=60 * 10,
+        httponly=True,
+        secure=True,
+    )
 
     # prevent replay attacks by generating and specifying a nonce
     # ref: https://simple-openid-connect.readthedocs.io/en/stable/nonce_and_state.html
     nonce = secrets.token_urlsafe(32)
-    
+
     # assemble the response
     resp.set_cookie("auth_nonce", nonce, max_age=60 * 10, httponly=True, secure=True)
 
-    return oidc_client.authorization_code_flow.start_authentication(state=state, nonce=nonce)
+    return oidc_client.authorization_code_flow.start_authentication(
+        state=state, nonce=nonce
+    )
 
 
-@app.get("/auth/login-callback", tags=["auth"], response_class=RedirectResponse, status_code=302)
+@app.get(
+    "/auth/login-callback",
+    tags=["auth"],
+    response_class=RedirectResponse,
+    status_code=302,
+)
 async def login_callback(req: Request, resp: Response, oidc_client: deps.OpenidClient):
     # check that the user is currently in an authenticating state
     # these cookies are set by the login_init() view
-    if "auth_state" not in req.cookies or "auth_nonce" not in req.cookies or "auth_start_time" not in req.cookies:
+    if (
-        logger.debug("user tried to log in but cookies indicate they are in a wrong state; redirecting to error view")
+        "auth_state" not in req.cookies
+        or "auth_nonce" not in req.cookies
+        or "auth_start_time" not in req.cookies
+    ):
+        logger.debug(
+            "user tried to log in but cookies indicate they are in a wrong state; redirecting to error view"
+        )
         return "/auth/login-error?error=todo"
 
     # ensure cookies are always cleared in the response
     resp.set_cookie("auth_state", "", max_age=0)
     resp.set_cookie("auth_start_time", "", max_age=0)
-    
+
     # perform the actual authentication via OIDC token exchange
-    auth_result = oidc_client.authorization_code_flow.handle_authentication_result(current_url=req.url, state=req.cookies["auth_state"])
+    auth_result = oidc_client.authorization_code_flow.handle_authentication_result(
-    
+        current_url=req.url, state=req.cookies["auth_state"]
+    )
+
     # save the authentication result for later reuse
     if isinstance(auth_result, TokenSuccessResponse):
-        auth_start_time = datetime.fromtimestamp(float(req.cookies["auth_start_time"]), UTC)
+        auth_start_time = datetime.fromtimestamp(
-        deps.persist_auth_state(oidc_client, resp, auth_result, auth_start_time, req.cookies["auth_nonce"])
+            float(req.cookies["auth_start_time"]), UTC
+        )
+        deps.persist_auth_state(
+            oidc_client, resp, auth_result, auth_start_time, req.cookies["auth_nonce"]
+        )
 
         # redirect the user to the page they wanted to visit
         # TODO: respect "auth_next" cookie to redirect the user to a specific url
         logger.debug("successfully authenticated user")
         return str(req.url_for("get-user-info"))
     else:
-        logger.debug("could not authenticate user because of OIDC error; redirecting to error page with error messages intact")
+        logger.debug(
+            "could not authenticate user because of OIDC error; redirecting to error page with error messages intact"
+        )
         return f"/auth/login-error?{req.query_params}"
 
 
-@app.get("/api/locks/", tags=["locks"], responses={status.HTTP_401_UNAUTHORIZED: {"model": models.HttpProblemDetail}})
+@app.get(
-async def list_locks(ccujack: deps.CCUJackClient, cache: deps.Cache) -> List[models.Lock]:
+    "/auth/logout", tags=["auth"], response_class=RedirectResponse, status_code=302
+)
+async def logout(
+    resp: Response, oidc_client: deps.OpenidClient, current_user: deps.CurrentUser
+):
+    deps.clear_auth_state(resp)
+    return oidc_client.initiate_logout(
+        RpInitiatedLogoutRequest(id_token_hint=current_user.raw_id_token, post_logout_redirect_uri=f"{app_config.get().base_url}/")
+    )
+
+
+@app.get(
+    "/api/locks/",
+    tags=["locks"],
+    responses={status.HTTP_401_UNAUTHORIZED: {"model": models.HttpProblemDetail}},
+)
+async def list_locks(
+    ccujack: deps.CCUJackClient, cache: deps.Cache
+) -> List[models.Lock]:
     # discover locks from ccujack
     CACHE_KEY = "ccu-find-locks"
     if CACHE_KEY in cache:
@@ -134,7 +204,9 @@ async def list_locks(ccujack: deps.CCUJackClient, cache: deps.Cache) -> List[mod
         status_data = dict()
         for i_channel, channel_params in lock_channels:
             for i_param in channel_params:
-                value = await ccujack.query_param_value(f"{i_lock.address}/{i_channel.index}/{i_param.id}")
+                value = await ccujack.query_param_value(
+                    f"{i_lock.address}/{i_channel.index}/{i_param.id}"
+                )
                 match i_param.id:
                     case "LOCK_STATE":
                         match value.v:
@@ -167,9 +239,8 @@ async def list_locks(ccujack: deps.CCUJackClient, cache: deps.Cache) -> List[mod
                     case "UNREACH":
                         status_data["is_unreachable"] = value.v
 
-        result.append(models.Lock(name=i_lock.title, status=models.LockStatus(**status_data)))
+        result.append(
-                            
+            models.Lock(name=i_lock.title, status=models.LockStatus(**status_data))
+        )
 
     return result
-
-

api/src/dooris_api/deps.py

@@ -20,17 +20,25 @@ async def get_oidc_client(req: Request) -> OpenidClient:
 OpenidClient = Annotated[OpenidClient, Depends(get_oidc_client)]
 
 
-async def get_current_user(req: Request, resp: Response, oidc_client: OpenidClient) -> Optional[models.CurrentUser]:
+async def get_current_user(
+    req: Request, resp: Response, oidc_client: OpenidClient
+) -> Optional[models.CurrentUser]:
     # 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", "auth_nonce")):
-        logger.debug("user is fully authenticated, returning current user from existing id_token")
+        logger.debug(
-        id_token = oidc_client.decode_id_token(req.cookies["id_token"], nonce=req.cookies["auth_nonce"])
+            "user is fully authenticated, returning current user from existing id_token"
-        return models.CurrentUser(id_token=id_token)
+        )
+        id_token = oidc_client.decode_id_token(
+            req.cookies["id_token"], nonce=req.cookies["auth_nonce"]
+        )
+        return models.CurrentUser(id_token=id_token, raw_id_token=req.cookies["id_token"])
 
     # if we have a refresh token, try to get new tokens
     if all(i in req.cookies for i in ("refresh_token", "auth_nonce")):
-        logger.debug("user has been previously authenticated, trying to recover with 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):
@@ -38,16 +46,24 @@ async def get_current_user(req: Request, resp: Response, oidc_client: OpenidClie
 
             # return the newly gotten info
             id_token = oidc_client.decode_id_token(token_resp.id_token)
-            return models.CurrentUser(id_token=id_token)
+            return models.CurrentUser(id_token=id_token, raw_id_token=token_resp.id_token)
-    
+
     # otherwise we can't meaningfully recover any user information or the user is simply not authenticated
     logger.debug("no currently authenticated user")
-    raise exceptions.HttpProblemException(models.HttpProblemDetail.new_unauthorized(req.url))
+    raise exceptions.HttpProblemException(
+        models.HttpProblemDetail.new_unauthorized(req.url)
+    )
 
 
-def persist_auth_state(oidc_client: OpenidClient, resp: Response, tokens: TokenSuccessResponse, auth_start_time: datetime, token_nonce: Optional[str] = None):
+def persist_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)
 
@@ -56,17 +72,53 @@ def persist_auth_state(oidc_client: OpenidClient, resp: Response, tokens: TokenS
     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)
+        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)
+    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(
-    resp.set_cookie("id_token", tokens.id_token, max_age=int(id_max_age.total_seconds()), httponly=True, secure=True)
+            "refresh_token",
-    resp.set_cookie("auth_nonce", token_nonce, max_age=int(nonce_max_age.total_seconds()), httponly=True, secure=True)
+            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_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)
 
 
 CurrentUser = Annotated[Optional[models.CurrentUser], Depends(get_current_user)]
@@ -84,4 +136,3 @@ def get_ccujack(req: Request) -> CCUJackClient:
 
 
 CCUJackClient = Annotated[CCUJackClient, Depends(get_ccujack)]
-

api/src/dooris_api/models.py

@@ -1,4 +1,4 @@
-from typing import Optional, Self, Mapping, Any, Literal
+from typing import Optional, Self, Literal
 from datetime import datetime
 from pydantic import BaseModel, HttpUrl
 from enum import Enum
@@ -9,8 +9,9 @@ from fastapi import status
 
 class HttpProblemType(Enum):
     """
-    Statically known HTTP problem types using the [type URI scheme](https://datatracker.ietf.org/doc/rfc4151/)  
+    Statically known HTTP problem types using the [type URI scheme](https://datatracker.ietf.org/doc/rfc4151/)
     """
+
     UNAUTHORIZED = "type:noc@hamburg.ccc.de,2026:UNAUTHORIZED"
     DOOR_NOT_FOUND = "type:noc@hamburg.ccc.de,2026:DOOR_NOT_FOUND"
 
@@ -19,6 +20,7 @@ class HttpProblemDetail(BaseModel):
     """
     API Error modeled after [RFC9475](https://www.rfc-editor.org/rfc/rfc9457.html).
     """
+
     status: int
     type: HttpProblemType
     title: str
@@ -27,15 +29,28 @@ class HttpProblemDetail(BaseModel):
 
     @classmethod
     def new_unauthorized(cls, request_uri: str | URL) -> Self:
-        return cls(type=HttpProblemType.UNAUTHORIZED, status=status.HTTP_401_UNAUTHORIZED, title="Unauthorized", detail="You tried to access a ressource which requires authentication but you are not authenticated", instance=HttpUrl(str(request_uri)))
+        return cls(
+            type=HttpProblemType.UNAUTHORIZED,
+            status=status.HTTP_401_UNAUTHORIZED,
+            title="Unauthorized",
+            detail="You tried to access a ressource which requires authentication but you are not authenticated",
+            instance=HttpUrl(str(request_uri)),
+        )
 
     @classmethod
     def new_door_not_found(cls, requested_door: str, request_uri: str | URL) -> Self:
-        return cls(type=HttpProblemType.DOOR_NOT_FOUND, status=status.HTTP_404_NOT_FOUND, title="Door not found", detail=f"You tried to interact with door {requested_door!r} that is not known to dooris", instance=str(request_uri))
+        return cls(
+            type=HttpProblemType.DOOR_NOT_FOUND,
+            status=status.HTTP_404_NOT_FOUND,
+            title="Door not found",
+            detail=f"You tried to interact with door {requested_door!r} that is not known to dooris",
+            instance=str(request_uri),
+        )
 
 
 class CurrentUser(BaseModel):
     id_token: IdToken
+    raw_id_token: str
 
 
 class UserStatus(BaseModel):
@@ -52,9 +67,8 @@ class LockStatus(BaseModel):
     lock_target_level: Literal["locked", "unlocked", "open"]
     lock_state: Literal["unknown", "locked", "unlocked"]
     activity_state: Literal["unknown", "locking", "unlocking", "stable"]
-    
+
 
 class Lock(BaseModel):
     name: str
     status: LockStatus
-