Cleanup

TODO

@@ -1,3 +1,5 @@
 https://docs.authlib.org/en/latest/oauth/2/intro.html#intro-oauth2
 
 https://www.keycloak.org/docs/latest/authorization_services/index.html
+
+https://thinhdanggroup.github.io/oauth2-python/

src/oidc_test/auth_utils.py

@@ -1,6 +1,5 @@
 from typing import Union
 from functools import wraps
-from datetime import datetime
 import logging
 
 from fastapi import HTTPException, Request, status
@@ -54,7 +53,7 @@ async def get_current_user(request: Request) -> User:
         logger.info(f"Token expired for user {user.name}")
         try:
             userinfo = await oidc_provider.fetch_access_token(
-                refresh_token=token.refresh_token
+                refresh_token=token.get("refresh_token")
             )
         except OAuthError as err:
             logger.exception(err)

src/oidc_test/main.py

@@ -66,7 +66,7 @@ for provider in settings.oidc.providers:
         name=provider.id,
         server_metadata_url=provider.openid_configuration,
         client_kwargs={
-            "scope": "openid email",  # offline_access profile",
+            "scope": "openid email offline_access profile",
         },
         client_id=provider.client_id,
         client_secret=provider.client_secret,
@@ -170,7 +170,7 @@ async def auth(request: Request, oidc_provider_id: str) -> RedirectResponse:
     except OAuthError as error:
         raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail=error.error)
     # Remember the oidc_provider in the session
-    # logger.debug(f"Scope: {token['scope']}")
+    # logger.info(f"Scope: {token['scope']}")
     request.session["oidc_provider_id"] = oidc_provider_id
     #
     # One could process the full decoded token which contains extra information
@@ -351,9 +351,14 @@ async def get_introspect(
     token: Annotated[OAuth2Token, Depends(get_token)],
 ) -> JSONResponse:
     assert request is not None  # Just to keep QA checks happy
+    if (url := oidc_provider.server_metadata.get("introspection_endpoint")) is None:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="No intrispection endpoint found for the OIDC provider",
+        )
     if (
         response := await oidc_provider.post(
-            oidc_provider.server_metadata["introspection_endpoint"],
+            url,
             token=token,
             data={"token": token["access_token"]},
         )

src/oidc_test/models.py

@@ -6,11 +6,8 @@ from pydantic import (
     AnyHttpUrl,
     EmailStr,
     ConfigDict,
-    GetCoreSchemaHandler,
 )
-from pydantic_core import CoreSchema, core_schema
 from authlib.integrations.starlette_client.apps import StarletteOAuth2App
-from authlib.oauth2.rfc6749 import OAuth2Token as OAuth2Token_authlib
 from sqlmodel import SQLModel, Field
 
 
@@ -19,10 +16,9 @@ class Role(SQLModel, extra="ignore"):
 
 
 class UserBase(SQLModel, extra="ignore"):
-
     id: str | None = None
     sid: str | None = None
-    name: str
+    name: str | None = None
     email: EmailStr | None = None
     picture: AnyHttpUrl | None = None
     roles: list[Role] = []

src/oidc_test/settings.py

@@ -45,6 +45,12 @@ class OIDCProvider(BaseModel):
         return "auth/" + self.id
 
 
+class ResourceProvider(BaseModel):
+    id: str
+    name: str
+    resources: list[Resource] = []
+
+
 class OIDCSettings(BaseModel):
     show_session_details: bool = False
     providers: list[OIDCProvider] = []