.forgejo/workflows/build.yaml

@@ -19,7 +19,7 @@ jobs:
       - name: Install the latest version of uv
         uses: astral-sh/setup-uv@v4
         with:
-          version: "0.6.9"
+          version: "0.5.16"
 
       - name: Install
         run: uv sync
@@ -27,26 +27,32 @@ jobs:
       - name: Run tests (API call)
         run: .venv/bin/pytest -s tests/basic.py
 
-      - name: Get version
+      - name: Get version with git describe
-        run: echo "VERSION=$(.venv/bin/dunamai from any --style semver)" >> $GITHUB_ENV
+        id: version
+        run: |
+          echo "version=$(git describe)" >> $GITHUB_OUTPUT
+          echo "$VERSION"
 
-      - name: Version
+      - name: Check if the container should be built
-        run: echo $VERSION
+        id: builder
+        env:
+          RUN: ${{ toJSON(inputs.build || !contains(steps.version.outputs.version, '-')) }}
+        run: |
+          echo "run=$RUN" >> $GITHUB_OUTPUT
+          echo "Run build: $RUN"
 
-      - name: Get distance from tag
+      - name: Set the version in pyproject.toml (workaround for uv not supporting dynamic version)
-        run: echo "DISTANCE=$(.venv/bin/dunamai from any --format '{distance}')" >> $GITHUB_ENV
+        if: fromJSON(steps.builder.outputs.run)
-
+        env:
-      - name: Distance
+          VERSION: ${{ steps.version.outputs.version }}
-        run: echo $DISTANCE
+        run: sed "s/0.0.0/$VERSION/" -i pyproject.toml
 
       - name: Workaround for bug of podman-login
-        if: env.DISTANCE == '0'
         run: |
           mkdir -p $HOME/.docker
           echo "{ \"auths\": {} }" > $HOME/.docker/config.json
 
       - name: Log in to the container registry (with another workaround)
-        if: env.DISTANCE == '0'
         uses: actions/podman-login@v1
         with:
           registry: ${{ vars.REGISTRY }}
@@ -55,31 +61,26 @@ jobs:
           auth_file_path: /tmp/auth.json
 
       - name: Build the container image
-        if: env.DISTANCE == '0'
         uses: actions/buildah-build@v1
         with:
           image: oidc-fastapi-test
           oci: true
           labels: oidc-fastapi-test
-          tags: "latest ${{ env.VERSION }}"
+          tags: latest ${{ steps.version.outputs.version }}
           containerfiles: |
             ./Containerfile
 
       - name: Push the image to the registry
-        if: env.DISTANCE == '0'
         uses: actions/push-to-registry@v2
         with:
           registry: "docker://${{ vars.REGISTRY }}/${{ vars.ORGANISATION }}"
           image: oidc-fastapi-test
-          tags: "latest ${{ env.VERSION }}"
+          tags: latest ${{ steps.version.outputs.version }}
 
       - name: Build wheel
-        if: env.DISTANCE == '0'
         run: uv build --wheel
 
       - name: Publish Python package (home)
-        if: env.DISTANCE == '0'
         env:
           LOCAL_PYPI_TOKEN: ${{ secrets.LOCAL_PYPI_TOKEN }}
         run: uv publish --publish-url https://code.philo.ydns.eu/api/packages/philorg/pypi --token $LOCAL_PYPI_TOKEN
-        continue-on-error: true

.forgejo/workflows/test.yaml

@@ -19,7 +19,7 @@ jobs:
       - name: Install the latest version of uv
         uses: astral-sh/setup-uv@v4
         with:
-          version: "0.6.3"
+          version: "0.5.16"
 
       - name: Install
         run: uv sync

Containerfile

@@ -1,4 +1,4 @@
-FROM docker.io/library/python:latest
+FROM docker.io/library/python:alpine
 
 COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /usr/local/bin/
 
@@ -9,9 +9,6 @@ WORKDIR /app
 
 RUN uv pip install --system .
 
-# Add demo plugin
-RUN PIP_EXTRA_INDEX_URL=https://pypi.org/simple/ uv pip install --system --index-url https://code.philo.ydns.eu/api/packages/philorg/pypi/simple/ oidc-fastapi-test-resource-provider-demo
-
 # Possible to run with:
 #CMD ["oidc-test", "--port", "80"]
 #CMD ["fastapi", "run", "src/oidc_test/main.py", "--port", "8873", "--root-path", "/oidc-test"]
@@ -22,5 +19,5 @@ CMD [ \
   "oidc_test.main:app", \
   "--host", "0.0.0.0", \
   "--port", "80", \
-  "--forwarded-allow-ips", "*" \
+  "--forwarded-allow-ips", "*", \
   ]

README.md

@@ -16,12 +16,6 @@ as a template for integration in other FastAPI/SQLModel applications.
 
 Feedback welcome.
 
-## Resource server
-
-It also functions as a resource server in a OAuth architecture.
-See a sibling test project, a web based OIDC/OAuth:
-[oidc-vue-test](https://code.philo.ydns.eu/philorg/oidc-vue-test).
-
 ## RBAC
 
 The application is also a playground for RBAC (Role Based Access control)
@@ -51,78 +45,36 @@ given by the OIDC providers.
 
 For example:
 
-```yaml
+```text
-secret_key: AVeryWellKeptSecret
+oidc:
-debug_token: no
+  secret_key: "ASecretNoOneKnows"
-show_token: yes
+  show_session_details: yes
-log: yes
-
-auth:
   providers:
     - id: auth0
       name: Okta / Auth0
-      url: https://<your_auth0_app_URL>
+      url: "https://<your_auth0_app_URL>"
-      public_key_url: https://<your_auth0_app_URL>/pem
+      client_id: "<your_auth0_client_id>"
-      client_id: <your_auth0_client_id>
+      client_secret: "client_secret_generated_by_auth0"
-      client_secret: client_secret_generated_by_auth0
+      hint: "A hint for test credentials"
-      hint: A hint for test credentials
 
     - id: keycloak
       name: Keycloak at somewhere
-      url: https://<the_keycloak_realm_url>
+      url: "https://<the_keycloak_realm_url>"
-      info_url: https://philo.ydns.eu/auth/realms/test
+      client_id: "<your_keycloak_client_id>"
-      account_url_template: /account
+      client_secret: "client_secret_generated_by_keycloak"
-      client_id: <your_keycloak_client_id>
+      hint: "User: foo, password: foofoo"
-      client_secret: <client_secret_generated_by_keycloak>
-      hint: A hint for test credentials
-      code_challenge_method: S256
-      resource_provider_scopes:
-        - get:time
-        - get:bs
-      resource_providers:
-        - id: <third_party_resource_provider_id>
-          name: A third party resource provider
-          base_url: https://some.example.com/
-          verify_ssl: yes
-          resources:
-            - name: Public RS2
-              resource_name: public
-              url: resource/public
-            - name: BS RS2
-              resource_name: bs
-              url: resource/bs
-            - name: Time RS2
-              resource_name: time
-              url: resource/time
 
     - id: codeberg
-      disabled: no
       name: Codeberg
-      url: https://codeberg.org
+      url: "https://codeberg.org"
-      account_url_template: /user/settings
+      client_id: "<your_codeberg_client_id>"
-      client_id: <your_codeberg_client_id>
+      client_secret: "client_secret_generated_by_codeberg"
-      client_secret: client_secret_generated_by_codeberg
-      info_url: https://codeberg.org/login/oauth/keys
-      session_key: sub
-      skip_verify_signature: no
-      resources:
-        - name: List of repos
-          id: repos
-          url: /api/v1/user/repos
-        - name: List of OAuth2 applications
-          id: oauth2_applications
-          url: /api/v1/user/applications/oauth2
-
-cors_origins:
-  - https://some.client
-  - https://localhost:8000
 ```
 
 The application reads the `OIDC_TEST_SETTINGS_FILE` environment variable
 to determine the location of this file at startup.
 
-For example, to run on port 8000 in a container,
+For example, to run on port 8000 in a container, with the setting file in the current working directory:
-with the setting file in the current working directory:
 
 ```sh
 podman run -p 8000:80 --env OIDC_TEST_CONFIG_FILE=/app/settings.yaml --mount type=bind,source=settings.yaml,destination=/app/settings.yaml code.philo.ydns.eu/philorg/oidc-fastapi-test:latest

TODO

@@ -1,5 +0,0 @@
-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/

pyproject.toml

@@ -1,7 +1,7 @@
 [project]
 name = "oidc-fastapi-test"
-#version = "0.0.0"
+version = "0.0.0"
-dynamic = ["version"]
+# dynamic = ["version"]
 description = "Add your description here"
 readme = "README.md"
 requires-python = ">=3.13"
@@ -9,12 +9,9 @@ dependencies = [
   "authlib>=1.4.0",
   "cachetools>=5.5.0",
   "fastapi[standard]>=0.115.6",
-  "httpx>=0.28.1",
   "itsdangerous>=2.2.0",
   "passlib[bcrypt]>=1.7.4",
-  "pkce>=1.0.3",
   "pydantic-settings>=2.7.1",
-  "pyjwt>=2.10.1",
   "python-jose[cryptography]>=3.3.0",
   "requests>=2.32.3",
   "sqlmodel>=0.0.22",
@@ -24,24 +21,14 @@ dependencies = [
 oidc-test = "oidc_test.main:main"
 
 [dependency-groups]
-dev = ["dunamai>=1.23.0", "ipdb>=0.13.13", "pytest>=8.3.4"]
+dev = ["ipdb>=0.13.13", "pytest>=8.3.4"]
 
 [build-system]
-requires = ["hatchling", "uv-dynamic-versioning"]
+requires = ["hatchling"]
 build-backend = "hatchling.build"
 
-[tool.hatch.version]
-source = "uv-dynamic-versioning"
-
 [tool.hatch.build.targets.wheel]
 packages = ["src/oidc_test"]
-package = true
-
-[tool.uv-dynamic-versioning]
-style = "semver"
 
 [tool.uv]
 package = true
-
-[tool.black]
-line-length = 98

src/oidc_test/__init__.py

@@ -1,11 +0,0 @@
-import importlib.metadata
-
-try:
-    from dunamai import Version, Style
-
-    __version__ = Version.from_git().serialize(style=Style.SemVer, dirty=True)
-except ImportError:
-    # __name__ could be used if the package name is the same
-    # as the directory - not the case here
-    # __version__ = importlib.metadata.version(__name__)
-    __version__ = importlib.metadata.version("oidc-fastapi-test")

src/oidc_test/auth/provider.py

@@ -1,113 +0,0 @@
-from json import JSONDecodeError
-from typing import Any
-from jwt import decode
-import logging
-
-from pydantic import ConfigDict
-from authlib.integrations.starlette_client.apps import StarletteOAuth2App
-from httpx import AsyncClient
-
-from oidc_test.settings import AuthProviderSettings, ResourceProvider, Resource, settings
-from oidc_test.models import User
-
-logger = logging.getLogger("oidc-test")
-
-
-class NoPublicKey(Exception):
-    pass
-
-
-class Provider(AuthProviderSettings):
-    # To allow authlib_client as StarletteOAuth2App
-    model_config = ConfigDict(arbitrary_types_allowed=True)  # type:ignore
-
-    authlib_client: StarletteOAuth2App = StarletteOAuth2App(None)
-    info: dict[str, Any] = {}
-    unknown_auth_user: User
-    logout_with_id_token_hint: bool = True
-
-    def decode(self, token: str, verify_signature: bool | None = None) -> dict[str, Any]:
-        """Decode the token with signature check"""
-        if self.public_key is None:
-            raise NoPublicKey
-        if verify_signature is None:
-            verify_signature = self.skip_verify_signature
-        if settings.debug_token:
-            decoded = decode(
-                token,
-                self.public_key,
-                algorithms=[self.signature_alg],
-                audience=["account", "oidc-test", "oidc-test-web"],
-                options={
-                    "verify_signature": False,
-                    "verify_aud": False,
-                },  # not settings.insecure.skip_verify_signature},
-            )
-            logger.debug(str(decoded))
-        return decode(
-            token,
-            self.public_key,
-            algorithms=[self.signature_alg],
-            audience=["account", "oidc-test", "oidc-test-web"],
-            options={
-                "verify_signature": verify_signature,
-            },  # not settings.insecure.skip_verify_signature},
-        )
-
-    async def get_info(self):
-        # Get the public key:
-        async with AsyncClient() as client:
-            public_key: str | None = None
-            if self.info_url is not None:
-                try:
-                    provider_info = await client.get(self.info_url)
-                except Exception as err:
-                    logger.debug("Provider_info: cannot connect")
-                    logger.exception(err)
-                    raise NoPublicKey
-                try:
-                    self.info = provider_info.json()
-                except JSONDecodeError:
-                    logger.debug("Provider_info: cannot decode json response")
-                    raise NoPublicKey
-                if "public_key" in self.info:
-                    # For Keycloak
-                    try:
-                        public_key = str(self.info["public_key"])
-                    except KeyError:
-                        logger.debug("Provider_info: cannot get public_key")
-                        raise NoPublicKey
-                elif "keys" in self.info:
-                    # For Forgejo/Gitea
-                    try:
-                        public_key = str(self.info["keys"][0]["n"])
-                    except KeyError:
-                        logger.debug("Provider_info: cannot get key 0.n")
-                        raise NoPublicKey
-            if self.public_key_url is not None:
-                resp = await client.get(self.public_key_url)
-                public_key = resp.text
-            if public_key is None:
-                logger.debug("Provider_info: cannot determine public key")
-                raise NoPublicKey
-            self.public_key = "\n".join(
-                ["-----BEGIN PUBLIC KEY-----", public_key, "-----END PUBLIC KEY-----"]
-            )
-
-    def get_session_key(self, userinfo):
-        return userinfo[self.session_key]
-
-    def get_resource(self, resource_name: str) -> Resource:
-        return [
-            resource for resource in self.resources if resource.resource_name == resource_name
-        ][0]
-
-    def get_resource_url(self, resource_name: str) -> str:
-        return self.url + self.get_resource(resource_name).url
-
-    def get_resource_provider(self, resource_provider_id: str) -> ResourceProvider:
-        return [
-            provider
-            for provider in self.resource_providers
-            if provider.id == resource_provider_id
-        ][0]

src/oidc_test/auth/utils.py

@@ -1,305 +0,0 @@
-from typing import Union, Annotated
-from functools import wraps
-import logging
-
-from fastapi import HTTPException, Request, Depends, status
-from fastapi.security import OAuth2PasswordBearer
-from authlib.integrations.starlette_client import OAuth, OAuthError, StarletteOAuth2App
-from authlib.oauth2.rfc6749 import OAuth2Token
-from jwt import ExpiredSignatureError, InvalidKeyError, DecodeError, PyJWTError
-
-from oidc_test.auth.provider import Provider
-from oidc_test.models import User
-from oidc_test.database import db, TokenNotInDb, UserNotInDB
-from oidc_test.settings import settings
-from oidc_test.auth_providers import providers
-
-logger = logging.getLogger("oidc-test")
-
-
-async def fetch_token(name, request):
-    assert name is not None
-    assert request is not None
-    logger.warning("TODO: fetch_token")
-    ...
-    # if name in oidc_providers:
-    #    model = OAuth2Token
-    # else:
-    #    model = OAuthToken
-
-    # token = model.find(name=name, user=request.user)
-    # return token.to_token()
-
-
-async def update_token(
-    provider_id,
-    token,
-    refresh_token: str | None = None,
-    access_token: str | None = None,
-):
-    """Update the token in the database"""
-    provider = providers[provider_id]
-    sid: str = provider.get_session_key(provider.decode(token["id_token"]))
-    item = await db.get_token(provider, sid)
-    # update old token
-    item["access_token"] = token["access_token"]
-    item["refresh_token"] = token["refresh_token"]
-    item["id_token"] = token["id_token"]
-    item["expires_at"] = token["expires_at"]
-    logger.info(f"Token {sid} refreshed")
-    # It's a fake db and only in memory, so there's nothing to save
-    # await item.save()
-
-
-def init_providers():
-    """Add oidc providers to authlib from the settings
-    and build the providers dict"""
-    for provider_settings in settings.auth.providers:
-        provider_settings_dict = provider_settings.model_dump()
-        # Add an anonymous user, that cannot be identified but has provided a valid access token
-        provider_settings_dict["unknown_auth_user"] = User(
-            sub="", auth_provider_id=provider_settings.id
-        )
-        provider = Provider(**provider_settings_dict)
-        if provider.disabled:
-            logger.info(f"{provider_settings.name} is disabled, skipping")
-        else:
-            authlib_oauth.register(
-                name=provider.id,
-                server_metadata_url=provider.openid_configuration,
-                client_kwargs={
-                    "scope": " ".join(
-                        ["openid", "email", "offline_access", "profile"]
-                        + provider.resource_provider_scopes
-                    ),
-                },
-                client_id=provider.client_id,
-                client_secret=provider.client_secret,
-                api_base_url=provider.url,
-                # For PKCE (not implemented yet):
-                # code_challenge_method="S256",
-                fetch_token=fetch_token,
-                update_token=update_token,
-                # client_id="some-client-id", # if enabled, authlib will also check that the access token belongs to this client id (audience)
-            )
-            provider.authlib_client = getattr(authlib_oauth, provider.id)
-        providers[provider.id] = provider
-
-
-authlib_oauth = OAuth(cache=None, fetch_token=fetch_token, update_token=update_token)
-oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
-oauth2_scheme_optional = OAuth2PasswordBearer(tokenUrl="token", auto_error=False)
-
-
-def get_auth_provider_client_or_none(request: Request) -> StarletteOAuth2App | None:
-    """Return the oidc_provider from a request object, from the session.
-    It can be used in Depends()"""
-    if (auth_provider_id := request.session.get("auth_provider_id")) is None:
-        return
-    return getattr(authlib_oauth, str(auth_provider_id), None)
-
-
-def get_auth_provider_client(request: Request) -> StarletteOAuth2App:
-    if (oidc_provider := get_auth_provider_client_or_none(request)) is None:
-        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "No such provider")
-    else:
-        return oidc_provider
-
-
-def get_auth_provider_or_none(request: Request) -> Provider | None:
-    """Return the oidc_provider settings from a request object, from the session.
-    It can be used in Depends()"""
-    if (auth_provider_id := request.session.get("auth_provider_id")) is None:
-        return
-    return providers.get(auth_provider_id)
-
-
-def get_auth_provider(request: Request) -> Provider:
-    if (provider := get_auth_provider_or_none(request)) is None:
-        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "No such provider")
-    return provider
-
-
-async def get_current_user(request: Request) -> User:
-    """Get the current user from a request object.
-    Also validates the token expiration time.
-    ... TODO: complete about refresh token
-    """
-    if (user_sub := request.session.get("user_sub")) is None:
-        raise HTTPException(status.HTTP_401_UNAUTHORIZED)
-    token = await get_token_from_session(request)
-    user = await db.get_user(user_sub)
-    ## Check if the token is expired
-    if token.is_expired():
-        provider = get_auth_provider(request=request)
-        ## Ask a new refresh token from the provider
-        logger.info(f"Token expired for user {user.name}")
-        try:
-            userinfo = await provider.authlib_client.fetch_access_token(
-                refresh_token=token.get("refresh_token")
-            )
-            assert userinfo is not None
-        except OAuthError as err:
-            logger.exception(err)
-            # raise HTTPException(
-            #    status.HTTP_401_UNAUTHORIZED, "Token expired, cannot refresh"
-            # )
-
-    return user
-
-
-async def get_token_from_session_or_none(request: Request) -> OAuth2Token | None:
-    """Return the auth token from the session or None.
-    Can be used in Depends()"""
-    try:
-        return await get_token_from_session(request)
-    except HTTPException:
-        return None
-
-
-async def get_token_from_session(request: Request) -> OAuth2Token:
-    """Return the token from the session.
-    Can be used in Depends()"""
-    try:
-        provider = providers[request.session.get("auth_provider_id", "")]
-    except KeyError:
-        request.session.pop("auth_provider_id", None)
-        request.session.pop("user_sub", None)
-        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Invalid provider")
-    try:
-        return await db.get_token(
-            provider,
-            request.session.get("sid"),
-        )
-    except (TokenNotInDb, InvalidKeyError, DecodeError) as err:
-        raise HTTPException(status.HTTP_401_UNAUTHORIZED, err.__class__.__name__)
-
-
-async def get_current_user_or_none(request: Request) -> User | None:
-    """Return the user from a request object, from the session.
-    It can be used in Depends()"""
-    try:
-        return await get_current_user(request)
-    except HTTPException:
-        return None
-
-
-def hasrole(required_roles: Union[str, list[str]] = []):
-    """Decorator for RBAC permissions"""
-    required_roles_set: set[str]
-    if isinstance(required_roles, str):
-        required_roles_set = set([required_roles])
-    else:
-        required_roles_set = set(required_roles)
-
-    def decorator(func):
-        @wraps(func)
-        async def wrapper(request=None, *args, **kwargs):
-            if request is None:
-                raise HTTPException(
-                    500,
-                    "Functions decorated with hasrole must have a request:Request argument",
-                )
-            user: User = await get_current_user(request)
-            if not any(required_roles_set.intersection(user.roles_as_set)):
-                raise HTTPException(status.HTTP_401_UNAUTHORIZED)
-            return await func(request, *args, **kwargs)
-
-        return wrapper
-
-    return decorator
-
-
-def get_token_info(token: dict) -> dict:
-    token_info = dict()
-    for key in token:
-        if key != "userinfo":
-            token_info[key] = token[key]
-    return token_info
-
-
-async def get_user_from_token(
-    token: Annotated[str, Depends(oauth2_scheme)],
-    request: Request,
-) -> User:
-    try:
-        auth_provider_id = request.headers["auth_provider"]
-    except KeyError:
-        raise HTTPException(
-            status.HTTP_401_UNAUTHORIZED,
-            "Request headers must have a 'auth_provider' field",
-        )
-    try:
-        provider = providers[auth_provider_id]
-    except KeyError:
-        if auth_provider_id == "":
-            raise HTTPException(status.HTTP_401_UNAUTHORIZED, "No auth provider")
-        else:
-            raise HTTPException(
-                status.HTTP_401_UNAUTHORIZED, f"Unknown auth provider '{auth_provider_id}'"
-            )
-    if token == "None":
-        request.session.pop("auth_provider_id", None)
-        request.session.pop("user_sub", None)
-        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "No token")
-    try:
-        payload = provider.decode(token)
-    except ExpiredSignatureError:
-        raise HTTPException(
-            status.HTTP_401_UNAUTHORIZED,
-            "Expired signature (token refresh not implemented yet)",
-        )
-    except InvalidKeyError:
-        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Invalid auth provider key")
-    except PyJWTError as err:
-        raise HTTPException(
-            status.HTTP_401_UNAUTHORIZED, f"Cannot decode token: {err.__class__.__name__}"
-        )
-    try:
-        user_id = payload["sub"]
-    except KeyError:
-        return provider.unknown_auth_user
-    try:
-        user = await db.get_user(user_id)
-        if user.access_token != token:
-            user.access_token = token
-    except UserNotInDB:
-        logger.info(
-            f"User {user_id} not found in DB, creating it (real apps can behave differently)"
-        )
-        user = await db.add_user(
-            sub=payload["sub"],
-            user_info=payload,
-            auth_provider=providers[auth_provider_id],
-            access_token=token,
-        )
-    return user
-
-
-async def get_user_from_token_or_none(
-    token: Annotated[str | None, Depends(oauth2_scheme_optional)],
-    request: Request,
-) -> User | None:
-    if token is None:
-        return None
-    try:
-        return await get_user_from_token(token, request)
-    except HTTPException:
-        return None
-
-
-class UserWithRole:
-    roles: set[str]
-
-    def __init__(self, roles: str | list[str] | tuple[str] | set[str]):
-        if isinstance(roles, str):
-            self.roles = set([roles])
-        elif isinstance(roles, (list, tuple, set)):
-            self.roles = set(roles)
-
-    def __call__(self, user: User = Depends(get_user_from_token)) -> User:
-        if not any(self.roles.intersection(user.roles_as_set)):
-            raise HTTPException(
-                status.HTTP_401_UNAUTHORIZED, f"Not of any required role {', '.join(self.roles)}"
-            )
-        return user

src/oidc_test/auth_misc.py

@@ -0,0 +1,29 @@
+from datetime import datetime, timedelta
+from collections import OrderedDict
+
+from .models import User
+
+time_keys = set(("iat", "exp", "auth_time", "updated_at"))
+
+
+def pretty_details(user: User, now: datetime) -> OrderedDict:
+    details = OrderedDict()
+    # breakpoint()
+    for key in sorted(time_keys):
+        try:
+            dt = datetime.fromtimestamp(user.userinfo[key])
+        except (KeyError, TypeError):
+            pass
+        else:
+            td = now - dt
+            td = timedelta(days=td.days, seconds=td.seconds)
+            if td.days < 0:
+                ptd = f"in {-td} h:m:s"
+            else:
+                ptd = f"{td} h:m:s ago"
+            details[key] = f"{user.userinfo[key]} - {dt} ({ptd})"
+    for key in sorted(user.userinfo):
+        if key in time_keys:
+            continue
+        details[key] = user.userinfo[key]
+    return details

src/oidc_test/auth_providers.py

@@ -1,5 +0,0 @@
-from collections import OrderedDict
-
-from oidc_test.auth.provider import Provider
-
-providers: OrderedDict[str, Provider] = OrderedDict()

src/oidc_test/auth_utils.py

@@ -0,0 +1,120 @@
+from typing import Union
+from functools import wraps
+from datetime import datetime
+import logging
+
+from fastapi import HTTPException, Request, status
+from authlib.integrations.starlette_client import OAuth, OAuthError, StarletteOAuth2App
+
+# from authlib.oauth1.auth import OAuthToken
+# from authlib.oauth2.auth import OAuth2Token
+
+from .models import OAuth2Token, User
+from .database import db
+from .settings import settings
+
+logger = logging.getLogger(__name__)
+
+OIDC_PROVIDERS = set([provider.id for provider in settings.oidc.providers])
+
+
+def get_provider(request: Request) -> StarletteOAuth2App:
+    """Return the oidc_provider from a request object, from the session.
+    It can be used in Depends()"""
+    if (oidc_provider_id := request.session.get("oidc_provider_id")) is None:
+        raise HTTPException(
+            status.HTTP_500_INTERNAL_SERVER_ERROR,
+            "Not logged in (no provider in session)",
+        )
+    try:
+        return getattr(authlib_oauth, str(oidc_provider_id))
+    except AttributeError:
+        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "No such provider")
+
+
+async def get_current_user(request: Request) -> User:
+    """Get the current user from a request object.
+    Also validates the token expiration time.
+    ... TODO: complete about refresh token
+    """
+    if (user_sub := request.session.get("user_sub")) is None:
+        raise HTTPException(status.HTTP_401_UNAUTHORIZED)
+    if (token := await db.get_token(request.session["token"])) is None:
+        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Token unknown")
+    user = await db.get_user(user_sub)
+    ## Check if the token is expired
+    if token.expires_at < datetime.timestamp(datetime.now()):
+        oidc_provider = get_provider(request=request)
+        ## Ask a new refresh token from the provider
+        logger.info(f"Token expired for user {user.name}")
+        try:
+            userinfo = await oidc_provider.fetch_access_token(
+                refresh_token=token.refresh_token
+            )
+        except OAuthError as err:
+            logger.exception(err)
+            # raise HTTPException(
+            #    status.HTTP_401_UNAUTHORIZED, "Token expired, cannot refresh"
+            # )
+
+    return user
+
+
+async def get_current_user_or_none(request: Request) -> User | None:
+    try:
+        return await get_current_user(request)
+    except HTTPException:
+        return None
+
+
+def hasrole(required_roles: Union[str, list[str]] = []):
+    required_roles_set: set[str]
+    if isinstance(required_roles, str):
+        required_roles_set = set([required_roles])
+    else:
+        required_roles_set = set(required_roles)
+
+    def decorator(func):
+        @wraps(func)
+        async def wrapper(request=None, *args, **kwargs):
+            if request is None:
+                raise HTTPException(
+                    500,
+                    "Functions decorated with hasrole must have a request:Request argument",
+                )
+            user: User = await get_current_user(request)
+            if not any(required_roles_set.intersection(user.roles_as_set)):
+                raise HTTPException(status.HTTP_401_UNAUTHORIZED)
+            return await func(request, *args, **kwargs)
+
+        return wrapper
+
+    return decorator
+
+
+def get_token_info(token: dict) -> dict:
+    token_info = dict()
+    for key in token:
+        if key != "userinfo":
+            token_info[key] = token[key]
+    return token_info
+
+
+def fetch_token(name, request):
+    breakpoint()
+    ...
+    # if name in OIDC_PROVIDERS:
+    #    model = OAuth2Token
+    # else:
+    #    model = OAuthToken
+
+    # token = model.find(name=name, user=request.user)
+    # return token.to_token()
+
+
+def update_token(*args, **kwargs):
+    breakpoint()
+    ...
+
+
+authlib_oauth = OAuth(cache=None, fetch_token=fetch_token, update_token=update_token)

src/oidc_test/database.py

@@ -1,29 +1,15 @@
-"""Fake in-memory database interface for demo purpose"""
+# Implement a fake in-memory database interface for demo purpose
-
 import logging
 
-from authlib.oauth2.rfc6749 import OAuth2Token
+from authlib.integrations.starlette_client.apps import StarletteOAuth2App
-from jwt import PyJWTError
 
-from oidc_test.auth.provider import Provider
+from .models import User, OAuth2Token, Role
 
-from oidc_test.models import User, Role
+logger = logging.getLogger(__name__)
-from oidc_test.auth_providers import providers
-
-logger = logging.getLogger("oidc-test")
-
-
-class UserNotInDB(Exception):
-    pass
-
-
-class TokenNotInDb(Exception):
-    pass
 
 
 class Database:
     users: dict[str, User] = {}
-    # TODO: key of the token table should be provider: sid
     tokens: dict[str, OAuth2Token] = {}
 
     # Last sessions for the user (key: users's subject id (sub))
@@ -32,65 +18,30 @@ class Database:
         self,
         sub: str,
         user_info: dict,
-        auth_provider: Provider,
+        oidc_provider: StarletteOAuth2App,
-        access_token: str,
+        user_info_from_endpoint: dict,
-        access_token_decoded: dict | None = None,
     ) -> User:
-        if access_token_decoded is None:
+        user = User.from_auth(userinfo=user_info, oidc_provider=oidc_provider)
-            assert auth_provider.name is not None
-            provider = providers[auth_provider.id]
-            try:
-                access_token_decoded = provider.decode(access_token)
-            except PyJWTError:
-                access_token_decoded = {}
-        user_info["auth_provider_id"] = auth_provider.id
-        user = User(**user_info)
-        user.userinfo = user_info
-        # user.access_token = access_token
-        # user.access_token_decoded = access_token_decoded
-        # Add roles provided in the access token
-        roles = set()
         try:
-            r = access_token_decoded["resource_access"][auth_provider.client_id]["roles"]
+            raw_roles = user_info_from_endpoint["resource_access"][
-            roles.update(r)
+                oidc_provider.client_id
-        except KeyError:
+            ]["roles"]
-            pass
+        except Exception as err:
-        try:
+            logger.debug(f"Cannot read additional roles: {err}")
-            r = access_token_decoded["realm_access"]["roles"]
+            raw_roles = []
-            if isinstance(r, str):
+        for raw_role in raw_roles:
-                roles.add(r)
+            user.roles.append(Role(name=raw_role))
-            else:
-                roles.update(r)
-        except KeyError:
-            pass
-        user.roles = [Role(name=role_name) for role_name in roles]
         self.users[sub] = user
         return user
 
     async def get_user(self, sub: str) -> User:
-        if sub not in self.users:
-            raise UserNotInDB
         return self.users[sub]
 
-    async def add_token(self, provider: Provider, token: OAuth2Token) -> None:
+    async def add_token(self, token_dict: dict, user: User) -> None:
-        """Store a token using as key the sid (auth provider's session id)
+        self.tokens[token_dict['id_token']] = OAuth2Token.from_dict(token_dict=token_dict, user=user)
-        in the id_token"""
-        sid = provider.get_session_key(token["userinfo"])
-        self.tokens[sid] = token
 
-    async def get_token(
+    async def get_token(self, name) -> OAuth2Token | None:
-        self,
+        return self.tokens.get(name)
-        provider: Provider,
-        sid: str | None,
-    ) -> OAuth2Token:
-        # TODO: key of the token table should be provider: sid
-        assert isinstance(provider, Provider)
-        if sid is None:
-            raise TokenNotInDb
-        try:
-            return self.tokens[sid]
-        except KeyError:
-            raise TokenNotInDb
 
 
 db = Database()

src/oidc_test/log_conf.yaml

@@ -1,34 +0,0 @@
-version: 1
-disable_existing_loggers: False
-formatters:
-  default:
-    "()": uvicorn.logging.DefaultFormatter
-    format: "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
-  access:
-    "()": uvicorn.logging.AccessFormatter
-    format: "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
-handlers:
-  default:
-    formatter: default
-    class: logging.StreamHandler
-    stream: ext://sys.stderr
-  access:
-    formatter: access
-    class: logging.StreamHandler
-    stream: ext://sys.stdout
-loggers:
-  uvicorn.error:
-    level: INFO
-    handlers:
-      - default
-    propagate: no
-  uvicorn.access:
-    level: INFO
-    handlers:
-      - access
-    propagate: no
-  "oidc-test":
-    level: DEBUG
-    handlers:
-      - default
-    propagate: yes

src/oidc_test/main.py

@@ -6,324 +6,270 @@ from typing import Annotated
 from pathlib import Path
 from datetime import datetime
 import logging
-import logging.config
-import importlib.resources
-from yaml import safe_load
 from urllib.parse import urlencode
-from contextlib import asynccontextmanager
 
 from httpx import HTTPError
-from fastapi import Depends, FastAPI, HTTPException, Request, status
+from fastapi import Depends, FastAPI, HTTPException, Request, Response, status
-from fastapi.staticfiles import StaticFiles
 from fastapi.responses import HTMLResponse, RedirectResponse
 from fastapi.templating import Jinja2Templates
-from fastapi.middleware.cors import CORSMiddleware
+from fastapi.security import OpenIdConnect
-from jwt import PyJWTError
 from starlette.middleware.sessions import SessionMiddleware
 from authlib.integrations.starlette_client.apps import StarletteOAuth2App
-from authlib.integrations.base_client import OAuthError
+from authlib.integrations.starlette_client import OAuth, OAuthError
-from authlib.oauth2.rfc6749 import OAuth2Token
 
-# TODO: PKCE
+# authlib startlette integration does not support revocation: using requests
-# from authlib.integrations.httpx_client import AsyncOAuth2Client
+# from authlib.integrations.requests_client import OAuth2Session
-# from fastapi.security import OpenIdConnect
-# from pkce import generate_code_verifier, generate_pkce_pair
 
-from oidc_test import __version__
+from .settings import settings
-from oidc_test.registry import registry
+from .models import User
-from oidc_test.auth.provider import NoPublicKey, Provider
+from .auth_utils import (
-from oidc_test.auth.utils import (
+    get_provider,
-    get_auth_provider,
+    hasrole,
-    get_auth_provider_or_none,
     get_current_user_or_none,
+    get_current_user,
     authlib_oauth,
-    get_token_from_session_or_none,
-    get_token_from_session,
-    update_token,
 )
-from oidc_test.auth.utils import init_providers
+from .auth_misc import pretty_details
-from oidc_test.settings import settings
+from .database import db
-from oidc_test.auth_providers import providers
-from oidc_test.models import User
-from oidc_test.database import TokenNotInDb, db
-from oidc_test.resource_server import resource_server
 
-logger = logging.getLogger("oidc-test")
+logger = logging.getLogger("uvicorn.error")
-
-if settings.log:
-    assert __package__ is not None
-    with (
-        importlib.resources.path(__package__) as package_path,
-        open(package_path / settings.log_config_file) as f,
-    ):
-        logging_config = safe_load(f)
-        logging.config.dictConfig(logging_config)
 
 templates = Jinja2Templates(Path(__file__).parent / "templates")
 
 
-@asynccontextmanager
+app = FastAPI(
-async def lifespan(app: FastAPI):
+    title="OIDC auth test",
-    assert app is not None
-    init_providers()
-    registry.make_registry()
-    for provider in list(providers.values()):
-        if provider.disabled:
-            continue
-        try:
-            await provider.get_info()
-        except NoPublicKey:
-            logger.warning(f"Disable {provider.id}: public key not found")
-            del providers[provider.id]
-    yield
-
-
-app = FastAPI(title="OIDC auth test", lifespan=lifespan)
-
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=settings.cors_origins,
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
 )
 
+
 # SessionMiddleware is required by authlib
 app.add_middleware(
     SessionMiddleware,
     secret_key=settings.secret_key,
 )
 
-app.mount("/static", StaticFiles(directory=Path(__file__).parent / "static"), name="static")
+# Add oidc providers to authlib from the settings
-app.mount("/resource", resource_server, name="resource_server")
 
+fastapi_providers = {}
+_providers = {}
 
-@app.get("/")
+for provider in settings.oidc.providers:
-async def home(
+    authlib_oauth.register(
-    request: Request,
+        name=provider.id,
-    user: Annotated[User | None, Depends(get_current_user_or_none)],
+        server_metadata_url=provider.openid_configuration,
-    provider: Annotated[Provider | None, Depends(get_auth_provider_or_none)],
+        client_kwargs={
-    token: Annotated[OAuth2Token | None, Depends(get_token_from_session_or_none)],
+            "scope": "openid email offline_access profile",
-) -> HTMLResponse:
+        },
-    context = {
+        client_id=provider.client_id,
-        "show_token": settings.show_token,
+        client_secret=provider.client_secret,
-        "user": user,
+        api_base_url=provider.url,
-        "now": datetime.now(),
+        # fetch_token=fetch_token,
-        "__version__": __version__,
+        # update_token=update_token,
-    }
+        # client_id="some-client-id", # if enabled, authlib will also check that the access token belongs to this client id (audience)
-    if provider is None or token is None:
+    )
-        context["providers"] = providers
+    fastapi_providers[provider.id] = OpenIdConnect(
-        context["access_token"] = None
+        openIdConnectUrl=provider.openid_configuration
-        context["id_token_parsed"] = None
+    )
-        context["access_token_parsed"] = None
+    _providers[provider.id] = provider
-        context["refresh_token_parsed"] = None
-        context["resources"] = None
-        context["auth_provider"] = None
-    else:
-        context["auth_provider"] = provider
-        context["access_token"] = token["access_token"]
-        try:
-            access_token_parsed = provider.decode(token["access_token"], verify_signature=False)
-            context["access_token_parsed"] = access_token_parsed
-            context["access_token_scope"] = access_token_parsed.get("scope")
-        except PyJWTError as err:
-            context["access_token_parsed"] = {"Cannot parse": err.__class__.__name__}
-            context["access_token_scope"] = None
-        try:
-            id_token_parsed = provider.decode(token["id_token"], verify_signature=False)
-            context["id_token_parsed"] = id_token_parsed
-        except PyJWTError as err:
-            context["id_token_parsed"] = {"Cannot parse": err.__class__.__name__}
-        try:
-            refresh_token_parsed = provider.decode(token["refresh_token"], verify_signature=False)
-            context["refresh_token_parsed"] = refresh_token_parsed
-        except PyJWTError as err:
-            context["refresh_token_parsed"] = {"Cannot parse": err.__class__.__name__}
-        context["resources"] = registry.resources
-        context["resource_providers"] = provider.resource_providers
-    return templates.TemplateResponse(name="home.html", request=request, context=context)
 
 
 # Endpoints for the login / authorization process
 
 
-@app.get("/login/{auth_provider_id}")
+@app.get("/login/{oidc_provider_id}")
-async def login(request: Request, auth_provider_id: str) -> RedirectResponse:
+async def login(request: Request, oidc_provider_id: str) -> RedirectResponse:
-    """Login with the provider id, giving the browser a redirect to its authorize page.
+    """Login with the provider id,
-    The provider is expected to send the browser back to our own /auth/{auth_provider_id} url
+    by giving the browser a redirect to its authorize page.
-    with the token.
+    After successful authentification, the provider replies with an encrypted
+    auth token that only we can decode and contains userinfo,
+    and a redirect to our own /auth/{oidc_provider_id} url
     """
-    redirect_uri = request.url_for("auth", auth_provider_id=auth_provider_id)
+    redirect_uri = request.url_for("auth", oidc_provider_id=oidc_provider_id)
     try:
-        provider: StarletteOAuth2App = getattr(authlib_oauth, auth_provider_id)
+        provider_: StarletteOAuth2App = getattr(authlib_oauth, oidc_provider_id)
     except AttributeError:
         raise HTTPException(status.HTTP_401_UNAUTHORIZED, "No such provider")
-    # if (
-    #    code_challenge_method := providers[
-    #        auth_provider_id
-    #    ].code_challenge_method
-    # ) is not None:
-    #    #client = AsyncOAuth2Client(..., code_challenge_method=code_challenge_method)
-    #    code_verifier = generate_code_verifier()
-    #    logger.debug("TODO: PKCE")
-    # else:
-    #    code_verifier = None
     try:
-        response = await provider.authorize_redirect(
+        response = await provider_.authorize_redirect(
-            request,
+            request, redirect_uri, access_type="offline"
-            redirect_uri,
-            access_type="offline",
-            code_verifier=None,
         )
         return response
     except HTTPError:
         raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Cannot reach provider")
 
 
-@app.get("/auth/{auth_provider_id}")
+@app.get("/auth/{oidc_provider_id}")
-async def auth(
+async def auth(request: Request, oidc_provider_id: str) -> RedirectResponse:
-    request: Request,
-    auth_provider_id: str,
-) -> RedirectResponse:
     """Decrypt the auth token, store it to the session (cookie based)
     and response to the browser with a redirect to a "welcome user" page.
     """
     try:
-        provider = providers[auth_provider_id]
+        oidc_provider: StarletteOAuth2App = getattr(authlib_oauth, oidc_provider_id)
-    except KeyError:
+    except AttributeError:
         raise HTTPException(status.HTTP_401_UNAUTHORIZED, "No such provider")
     try:
-        token: OAuth2Token = await provider.authlib_client.authorize_access_token(request)
+        token = await oidc_provider.authorize_access_token(request)
     except OAuthError as error:
         raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail=error.error)
-    # Remember the authlib_client in the session
+    # Remember the oidc_provider in the session
-    # logger.info(f"Scope: {token['scope']}")
+    # logger.debug(f"Scope: {token['scope']}")
-    request.session["auth_provider_id"] = auth_provider_id
+    request.session["oidc_provider_id"] = oidc_provider_id
     #
     # One could process the full decoded token which contains extra information
     # eg for updates. Here we are only interested in roles
     #
     if userinfo := token.get("userinfo"):
-        # Remember the authlib_client in the session
+        # Remember the oidc_provider in the session
-        request.session["auth_provider_id"] = auth_provider_id
+        request.session["oidc_provider_id"] = oidc_provider_id
-        # User id (sub) given by auth provider
+        # User id (sub) given by oidc provider
         sub = userinfo["sub"]
-        ## Get additional data from userinfo endpoint
+        # Get additional data from userinfo endpoint
-        # try:
+        try:
-        #    user_info_from_endpoint = await authlib_client.userinfo(
+            user_info_url = oidc_provider.server_metadata["userinfo_endpoint"]
-        #        token=token, follow_redirects=True
+            user_info_from_endpoint = (
-        #    )
+                await oidc_provider.get(
-        # except Exception as err:
+                    user_info_url, token=token, follow_redirects=True
-        #    logger.warn(f"Cannot get userinfo from endpoint: {err}")
+                )
-        #    user_info_from_endpoint = {}
+            ).json()
+        except Exception as err:
+            logger.info(f"Cannot get userinfo from endpoint: {err}")
+            user_info_from_endpoint = {}
         # Build and remember the user in the session
         request.session["user_sub"] = sub
-        # Store the user in the database, which also verifies the token validity and signature
+        # Store the user in the database
-        try:
+        user = await db.add_user(
-            user = await db.add_user(
+            sub,
-                sub,
+            user_info=userinfo,
-                user_info=userinfo,
+            oidc_provider=oidc_provider,
-                auth_provider=providers[auth_provider_id],
+            user_info_from_endpoint=user_info_from_endpoint,
-                access_token=token["access_token"],
+        )
-            )
+        # Add the id_token to the session
-        except PyJWTError as err:
+        request.session["token"] = token["id_token"]
-            raise HTTPException(
-                status.HTTP_401_UNAUTHORIZED,
-                detail=f"Token invalid: {err.__class__.__name__}",
-            )
-        assert isinstance(user, User)
-        # Add the provider session id to the session
-        request.session["sid"] = provider.get_session_key(userinfo)
         # Add the token to the db because it is used for logout
-        await db.add_token(provider, token)
+        await db.add_token(token, user)
         # Send the user to the home: (s)he is authenticated
         return RedirectResponse(url=request.url_for("home"))
     else:
         # Not sure if it's correct to redirect to plain login
         # if no userinfo is provided
-        return RedirectResponse(url=request.url_for("login", auth_provider_id=auth_provider_id))
+        return RedirectResponse(
-
+            url=request.url_for("login", oidc_provider_id=oidc_provider_id)
-
+        )
-@app.get("/account")
-async def account(
-    provider: Annotated[Provider, Depends(get_auth_provider)],
-) -> RedirectResponse:
-    """Redirect to the auth provider account management,
-    if account_url_template is in the provider's settings"""
-    return RedirectResponse(f"{provider.account_url_template}")
 
 
 @app.get("/logout")
 async def logout(
     request: Request,
-    provider: Annotated[Provider, Depends(get_auth_provider)],
+    provider: Annotated[StarletteOAuth2App, Depends(get_provider)],
 ) -> RedirectResponse:
-    # Get provider's endpoint
-    if (
-        provider_logout_uri := provider.authlib_client.server_metadata.get("end_session_endpoint")
-    ) is None:
-        logger.warning(f"Cannot find end_session_endpoint for provider {provider.id}")
-        return RedirectResponse(request.url_for("non_compliant_logout"))
-    post_logout_uri = request.url_for("home")
     # Clear session
     request.session.pop("user_sub", None)
-    request.session.pop("auth_provider_id", None)
+    # Get provider's endpoint
-    try:
+    if (
-        token = await db.get_token(provider, request.session.pop("sid", None))
+        provider_logout_uri := provider.server_metadata.get("end_session_endpoint")
-    except TokenNotInDb:
+    ) is None:
-        logger.warning("No session in db for the token or no token")
+        logger.warn(f"Cannot find end_session_endpoint for provider {provider.name}")
+        return RedirectResponse(request.url_for("non_compliant_logout"))
+    post_logout_uri = request.url_for("home")
+    if (id_token := await db.get_token(request.session.pop("token", None))) is None:
+        logger.warn("No session in db for the token")
         return RedirectResponse(request.url_for("home"))
-    url_query = {
+    logout_url = (
-        "post_logout_redirect_uri": post_logout_uri,
+        provider_logout_uri
-        "client_id": provider.client_id,
+        + "?"
-    }
+        + urlencode(
-    if provider.logout_with_id_token_hint:
+            {
-        url_query["id_token_hint"] = token["id_token"]
+                "post_logout_redirect_uri": post_logout_uri,
-    logout_url = f"{provider_logout_uri}?{urlencode(url_query)}"
+                "id_token_hint": id_token.raw_id_token,
+                "cliend_id": "oidc_local_test",
+            }
+        )
+    )
     return RedirectResponse(logout_url)
 
 
 @app.get("/non-compliant-logout")
 async def non_compliant_logout(
     request: Request,
-    provider: Annotated[Provider, Depends(get_auth_provider)],
+    provider: Annotated[StarletteOAuth2App, Depends(get_provider)],
 ):
     """A page for non-compliant OAuth2 servers that we cannot log out."""
-    # Clear session
-    request.session.pop("user_sub", None)
-    request.session.pop("auth_provider_id", None)
     return templates.TemplateResponse(
         name="non_compliant_logout.html",
         request=request,
-        context={"auth_provider": provider, "home_url": request.url_for("home")},
+        context={"provider": provider, "home_url": request.url_for("home")},
     )
 
 
-@app.get("/refresh")
+# Home URL
-async def refresh(
+
-    request: Request,
+
-    provider: Annotated[Provider, Depends(get_auth_provider)],
+@app.get("/")
-    token: Annotated[OAuth2Token, Depends(get_token_from_session)],
+async def home(
-) -> RedirectResponse:
+    request: Request, user: Annotated[User, Depends(get_current_user_or_none)]
-    """Manually refresh token"""
+) -> HTMLResponse:
-    new_token = await provider.authlib_client.fetch_access_token(
+    now = datetime.now()
-        refresh_token=token["refresh_token"],
+    return templates.TemplateResponse(
-        grant_type="refresh_token",
+        name="home.html",
+        request=request,
+        context={
+            "settings": settings.model_dump(),
+            "user": user,
+            "now": now,
+            "user_info_details": (
+                pretty_details(user, now)
+                if user and settings.oidc.show_session_details
+                else None
+            ),
+        },
     )
-    try:
-        await update_token(provider.id, new_token)
-    except PyJWTError as err:
-        logger.info(f"Cannot refresh token: {err.__class__.__name__}")
-        raise HTTPException(
-            status.HTTP_510_NOT_EXTENDED, f"Token refresh error: {err.__class__.__name__}"
-        )
-    return RedirectResponse(url=request.url_for("home"))
 
 
-# Snippet for running standalone
+@app.get("/public")
-# Mostly useful for the --version option,
+async def public() -> HTMLResponse:
-# as running with uvicorn is easy and provides better flexibility, eg.
+    return HTMLResponse("<h1>Not protected</h1>")
-# uvicorn --host foo oidc_test.main:app --reload
+
+
+# Some URIs for testing the permissions
+
+
+@app.get("/protected")
+async def get_protected(
+    user: Annotated[User, Depends(get_current_user)]
+) -> HTMLResponse:
+    return HTMLResponse("<h1>Only authenticated users can see this</h1>")
+
+
+@app.get("/protected-by-foorole")
+@hasrole("foorole")
+async def get_protected_by_foorole(request: Request) -> HTMLResponse:
+    return HTMLResponse("<h1>Only users with foorole can see this</h1>")
+
+
+@app.get("/protected-by-barrole")
+@hasrole("barrole")
+async def get_protected_by_barrole(request: Request) -> HTMLResponse:
+    return HTMLResponse("<h1>Protected by barrole</h1>")
+
+
+@app.get("/protected-by-foorole-and-barrole")
+@hasrole("barrole")
+@hasrole("foorole")
+async def get_protected_by_foorole_and_barrole(request: Request) -> HTMLResponse:
+    return HTMLResponse("<h1>Only users with foorole and barrole can see this</h1>")
+
+
+@app.get("/protected-by-foorole-or-barrole")
+@hasrole(["foorole", "barrole"])
+async def get_protected_by_foorole_or_barrole(request: Request) -> HTMLResponse:
+    return HTMLResponse("<h1>Only users with foorole or barrole can see this</h1>")
+
+
+# @app.get("/fast_api_depends")
+# def fast_api_depends(
+#    token: Annotated[str, Depends(fastapi_providers["Keycloak"])]
+# ) -> HTMLResponse:
+#    return HTMLResponse("You're Authenticated")
 
 
 def main():
@@ -341,7 +287,9 @@ def main():
     parser.add_argument(
         "-p", "--port", type=int, default=80, help="Port to listen to (default: 80)"
     )
-    parser.add_argument("-v", "--version", action="store_true", help="Print version and exit")
+    parser.add_argument(
+        "-v", "--version", action="store_true", help="Print version and exit"
+    )
     args = parser.parse_args()
 
     if args.version:

src/oidc_test/models.py

@@ -1,66 +1,77 @@
-import logging
 from functools import cached_property
-from typing import Any
+from typing import Self
 
-from pydantic import (
+from pydantic import computed_field, AnyHttpUrl, EmailStr, ConfigDict
-    computed_field,
+from authlib.integrations.starlette_client.apps import StarletteOAuth2App
-    AnyHttpUrl,
-    EmailStr,
-    ConfigDict,
-)
 from sqlmodel import SQLModel, Field
 
-logger = logging.getLogger("oidc-test")
-
 
 class Role(SQLModel, extra="ignore"):
     name: str
 
 
 class UserBase(SQLModel, extra="ignore"):
+
     id: str | None = None
     sid: str | None = None
-    name: str | None = None
+    name: str
     email: EmailStr | None = None
     picture: AnyHttpUrl | None = None
     roles: list[Role] = []
 
 
 class User(UserBase):
-    model_config = ConfigDict(arbitrary_types_allowed=True)  # type:ignore
+    model_config = ConfigDict(arbitrary_types_allowed=True)
 
     sub: str = Field(
         description="""subject id of the user given by the oidc provider,
                             also the key for the database 'table'""",
     )
     userinfo: dict = {}
-    access_token: str | None = None
+    oidc_provider: StarletteOAuth2App | None = None
-    access_token_decoded: dict[str, Any] | None = None
+
-    auth_provider_id: str
+    @classmethod
+    def from_auth(cls, userinfo: dict, oidc_provider: StarletteOAuth2App) -> Self:
+        user = cls(**userinfo)
+        user.userinfo = userinfo
+        user.oidc_provider = oidc_provider
+        # Add roles if they are provided in the token
+        if raw_ra := userinfo.get("realm_access"):
+            if raw_roles := raw_ra.get("roles"):
+                user.roles = [Role(name=raw_role) for raw_role in raw_roles]
+        return user
 
     @computed_field
     @cached_property
     def roles_as_set(self) -> set[str]:
         return set([role.name for role in self.roles])
 
-    def has_scope(self, scope: str) -> bool:
-        """Check if the scope is present in user info or access token"""
-        info_scopes = self.userinfo.get("scope", "").split(" ")
-        try:
-            access_token_scopes = self.decode_access_token().get("scope", "").split(" ")
-        except Exception as err:
-            logger.debug(f"Cannot find scope because the access token cannot be decoded: {err}")
-            access_token_scopes = []
-        return scope in set(info_scopes + access_token_scopes)
 
-    def decode_access_token(self, verify_signature: bool = True):
+class OAuth2Token(SQLModel):
-        assert self.access_token is not None, "no access_token"
+    name: str = Field(primary_key=True)
-        assert self.auth_provider_id is not None, "no auth_provider_id"
+    token_type: str  # = Field(max_length=40)
-        from .auth_providers import providers
+    access_token: str  # = Field(max_length=2000)
+    raw_id_token: str
+    refresh_token: str  # = Field(max_length=200)
+    expires_at: int  # = PositiveIntegerField()
+    user: User  # = ForeignKey(User)
 
-        return providers[self.auth_provider_id].decode(
+    def to_token(self):
-            self.access_token, verify_signature=verify_signature
+        return dict(
+            access_token=self.access_token,
+            token_type=self.token_type,
+            refresh_token=self.refresh_token,
+            expires_at=self.expires_at,
         )
 
-    def get_scope(self, verify_signature: bool = True):
+    @classmethod
-        return self.decode_access_token(verify_signature=verify_signature)["scope"]
+    def from_dict(cls, token_dict: dict, user: User) -> Self:
+        return cls(
+            name=token_dict["access_token"],
+            access_token=token_dict["access_token"],
+            raw_id_token=token_dict["id_token"],
+            token_type=token_dict["token_type"],
+            refresh_token=token_dict["refresh_token"],
+            expires_at=token_dict["expires_at"],
+            user=user,
+        )

src/oidc_test/registry.py

@@ -1,47 +0,0 @@
-from importlib.metadata import entry_points
-import logging
-
-from pydantic import BaseModel, ConfigDict
-
-from oidc_test.models import User
-
-logger = logging.getLogger("registry")
-
-
-class ProcessResult(BaseModel):
-    model_config = ConfigDict(
-        extra="allow",
-    )
-
-
-class ProcessError(Exception):
-    pass
-
-
-class Resource(BaseModel):
-    name: str
-    scope_required: str | None = None
-    role_required: str | None = None
-    is_public: bool = False
-    default_resource_id: str | None = None
-
-    def __init__(self, name: str):
-        super().__init__()
-        self.__id__ = name
-
-    async def process(self, user: User | None, resource_id: str | None = None) -> ProcessResult:
-        logger.warning(f"{self.__id__} should define a process method")
-        return ProcessResult()
-
-
-class ResourceRegistry(BaseModel):
-    resources: dict[str, Resource] = {}
-
-    def make_registry(self):
-        for ep in entry_points().select(group="oidc_test.resource_provider"):
-            ResourceClass = ep.load()
-            if issubclass(ResourceClass, Resource):
-                self.resources[ep.name] = ResourceClass(ep.name)
-
-
-registry = ResourceRegistry()

src/oidc_test/resource_server.py

@@ -1,343 +0,0 @@
-from typing import Annotated, Any
-import logging
-from json import JSONDecodeError
-
-from authlib.oauth2.rfc6749 import OAuth2Token
-from httpx import AsyncClient, HTTPError
-from jwt.exceptions import DecodeError, ExpiredSignatureError, InvalidTokenError
-from fastapi import FastAPI, HTTPException, Depends, Request, status
-from fastapi.middleware.cors import CORSMiddleware
-
-# from starlette.middleware.sessions import SessionMiddleware
-# from authlib.integrations.starlette_client.apps import StarletteOAuth2App
-# from authlib.oauth2.rfc6749 import OAuth2Token
-
-from oidc_test.auth.provider import Provider
-from oidc_test.auth.utils import (
-    get_user_from_token_or_none,
-    oauth2_scheme_optional,
-)
-from oidc_test.auth_providers import providers
-from oidc_test.settings import ResourceProvider, settings
-from oidc_test.models import User
-from oidc_test.registry import ProcessError, ProcessResult, registry
-
-logger = logging.getLogger("oidc-test")
-
-resource_server = FastAPI()
-
-
-resource_server.add_middleware(
-    CORSMiddleware,
-    allow_origins=settings.cors_origins,
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
-
-# SessionMiddleware is required by authlib
-# resource_server.add_middleware(
-#    SessionMiddleware,
-#    secret_key=settings.secret_key,
-# )
-
-# Route for OAuth resource server
-
-
-# Routes for RBAC based tests
-
-
-@resource_server.get("/")
-async def resources() -> dict[str, dict[str, Any]]:
-    return {"internal": {}, "plugins": registry.resources}
-
-
-@resource_server.get("/{resource_name}")
-@resource_server.get("/{resource_name}/{resource_id}")
-async def get_resource(
-    resource_name: str,
-    user: Annotated[User | None, Depends(get_user_from_token_or_none)],
-    token: Annotated[OAuth2Token | None, Depends(oauth2_scheme_optional)],
-    resource_id: str | None = None,
-):
-    """Generic path for testing a resource provided by a provider.
-    There's no field validation (response type of ProcessResult) on purpose,
-    leaving the responsibility of the response validation to resource providers"""
-    # Get the resource if it's defined in user auth provider's resources (external)
-    if user is not None:
-        provider = providers[user.auth_provider_id]
-        if ":" in resource_name:
-            # Third-party resource provider: send the request with the request token
-            resource_provider_id, resource_name = resource_name.split(":", 1)
-            provider = providers[user.auth_provider_id]
-            resource_provider: ResourceProvider = provider.get_resource_provider(
-                resource_provider_id
-            )
-            resource_url = resource_provider.get_resource_url(resource_name)
-            async with AsyncClient(verify=resource_provider.verify_ssl) as client:
-                try:
-                    logger.debug(f"GET request to {resource_url}")
-                    resp = await client.get(
-                        resource_url,
-                        headers={
-                            "Content-type": "application/json",
-                            "Authorization": f"Bearer {token}",
-                            "auth_provider": user.auth_provider_id,
-                        },
-                    )
-                except HTTPError as err:
-                    raise HTTPException(
-                        status.HTTP_503_SERVICE_UNAVAILABLE, err.__class__.__name__
-                    )
-                except Exception as err:
-                    raise HTTPException(
-                        status.HTTP_500_INTERNAL_SERVER_ERROR, err.__class__.__name__
-                    )
-                else:
-                    if resp.is_success:
-                        return resp.json()
-                    else:
-                        reason_str: str
-                        try:
-                            reason_str = resp.json().get("detail", str(resp))
-                        except Exception:
-                            reason_str = str(resp.text)
-                        raise HTTPException(resp.status_code, reason_str)
-        # Third party resource (provided through the auth provider)
-        # The token is just passed on
-        # XXX: is this branch valid anymore?
-        if resource_name in [r.resource_name for r in provider.resources]:
-            return await get_auth_provider_resource(
-                provider=provider,
-                resource_name=resource_name,
-                token=token,
-                user=user,
-            )
-    # Internal resource (provided here)
-    if resource_name in registry.resources:
-        resource = registry.resources[resource_name]
-        reason: dict[str, str] = {}
-        if not resource.is_public:
-            if user is None:
-                raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Resource is not public")
-            else:
-                if resource.scope_required is not None and not user.has_scope(
-                    resource.scope_required
-                ):
-                    reason["scope"] = (
-                        f"No scope {resource.scope_required} in the access token "
-                        "but it is required for accessing this resource"
-                    )
-                if (
-                    resource.role_required is not None
-                    and resource.role_required not in user.roles_as_set
-                ):
-                    reason["role"] = (
-                        f"You don't have the role {resource.role_required} "
-                        "but it is required for accessing this resource"
-                    )
-        if len(reason) == 0:
-            try:
-                resp = await resource.process(user=user, resource_id=resource_id)
-                return resp
-            except ProcessError as err:
-                raise HTTPException(
-                    status.HTTP_401_UNAUTHORIZED, f"Cannot process resource: {err}"
-                )
-        else:
-            raise HTTPException(status.HTTP_401_UNAUTHORIZED, ", ".join(reason.values()))
-    else:
-        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Unknown resource")
-    # return await get_resource_(resource_name, user, **request.query_params)
-
-
-async def get_auth_provider_resource(
-    provider: Provider, resource_name: str, token: OAuth2Token | None, user: User
-) -> ProcessResult:
-    if token is None:
-        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "No auth token")
-    access_token = token
-    async with AsyncClient() as client:
-        resp = await client.get(
-            url=provider.get_resource_url(resource_name),
-            headers={
-                "Content-type": "application/json",
-                "Authorization": f"Bearer {access_token}",
-            },
-        )
-    if resp.is_error:
-        raise HTTPException(resp.status_code, f"Cannot fetch resource: {resp.reason_phrase}")
-    # Only a demo, real application would really process the response
-    resp_length = len(resp.text)
-    if resp_length > 1024:
-        return ProcessResult(
-            msg=f"The resource is too long ({resp_length} bytes) to show in this demo, here is just the begining in raw format",
-            start=resp.text[:100] + "...",
-        )
-    else:
-        try:
-            resp_json = resp.json()
-        except JSONDecodeError:
-            return ProcessResult(msg="The resource is not formatted in JSON", text=resp.text)
-        if isinstance(resp_json, dict):
-            return ProcessResult(**resp.json())
-        elif isinstance(resp_json, list):
-            return ProcessResult(**{str(i): line for i, line in enumerate(resp_json)})
-
-
-# @resource_server.get("/public")
-# async def public() -> dict:
-#    return {"msg": "Not protected"}
-#
-#
-# @resource_server.get("/protected")
-# async def get_protected(user: Annotated[User, Depends(get_user_from_token)]):
-#    assert user is not None  # Just to keep QA checks happy
-#    return {"msg": "Only authenticated users can see this"}
-#
-#
-# @resource_server.get("/protected-by-foorole")
-# async def get_protected_by_foorole(
-#    user: Annotated[User, Depends(UserWithRole("foorole"))],
-# ):
-#    assert user is not None
-#    return {"msg": "Only users with foorole can see this"}
-#
-#
-# @resource_server.get("/protected-by-barrole")
-# async def get_protected_by_barrole(
-#    user: Annotated[User, Depends(UserWithRole("barrole"))],
-# ):
-#    assert user is not None
-#    return {"msg": "Protected by barrole"}
-#
-#
-# @resource_server.get("/protected-by-foorole-and-barrole")
-# async def get_protected_by_foorole_and_barrole(
-#    user: Annotated[User, Depends(UserWithRole("foorole")), Depends(UserWithRole("barrole"))],
-# ):
-#    assert user is not None  # Just to keep QA checks happy
-#    return {"msg": "Only users with foorole and barrole can see this"}
-#
-#
-# @resource_server.get("/protected-by-foorole-or-barrole")
-# async def get_protected_by_foorole_or_barrole(
-#    user: Annotated[User, Depends(UserWithRole(["foorole", "barrole"]))],
-# ):
-#    assert user is not None  # Just to keep QA checks happy
-#    return {"msg": "Only users with foorole or barrole can see this"}
-
-# async def get_resource_(resource_id: str, user: User, **kwargs) -> dict:
-#    """
-#    Resource processing: build an informative rely as a simple showcase
-#    """
-#    if resource_id == "petition":
-#        return await sign(user, kwargs["petition_id"])
-#    provider = providers[user.auth_provider_id]
-#    try:
-#        pname = provider.name
-#    except KeyError:
-#        pname = "?"
-#    resp = {
-#        "hello": f"Hi {user.name} from an OAuth resource provider",
-#        "comment": f"I received a request for '{resource_id}' "
-#        + f"with an access token signed by {pname}",
-#    }
-#    # For the demo, resource resource_id matches a scope get:resource_id,
-#    # but this has to be refined for production
-#    required_scope = f"get:{resource_id}"
-#    # Check if the required scope is in the scopes allowed in userinfo
-#    try:
-#        if user.has_scope(required_scope):
-#            await process(user, resource_id, resp)
-#        else:
-#            ## For the showcase, giving a explanation.
-#            ## Alternatively, raise HTTP_401_UNAUTHORIZED
-#            raise HTTPException(
-#                status.HTTP_401_UNAUTHORIZED,
-#                f"No scope {required_scope} in the access token "
-#                + "but it is required for accessing this resource",
-#            )
-#    except ExpiredSignatureError:
-#        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "The token's signature has expired")
-#    except InvalidTokenError:
-#        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "The token is invalid")
-#    return resp
-
-
-# async def process(user, resource_id, resp):
-#    """
-#    Too simple to be serious.
-#    It's a good fit for a plugin architecture for production
-#    """
-#    if resource_id == "time":
-#        resp["time"] = datetime.now().strftime("%c")
-#    elif resource_id == "bs":
-#        async with AsyncClient() as client:
-#            bs = await client.get("https://corporatebs-generator.sameerkumar.website/")
-#            resp["bs"] = bs.json().get("phrase", "Sorry, i am out of BS today.")
-#    else:
-#        raise HTTPException(
-#            status.HTTP_401_UNAUTHORIZED, f"I don't known how to give '{resource_id}'."
-#        )
-
-
-# @resource_server.get("/introspect")
-# async def get_introspect(
-#    request: Request,
-#    oidc_provider: Annotated[StarletteOAuth2App, Depends(get_oidc_provider)],
-#    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 introspection endpoint found for the OIDC provider",
-#        )
-#    if (
-#        response := await oidc_provider.post(
-#            url,
-#            token=token,
-#            data={"token": token["access_token"]},
-#        )
-#    ).is_success:
-#        return response.json()
-#    else:
-#        raise HTTPException(status_code=response.status_code, detail=response.text)
-
-# assert user.oidc_provider is not None
-### Get some info (TODO: refactor)
-# if (auth_provider_id := user.oidc_provider.name) is None:
-#    raise HTTPException(
-#        status.HTTP_401_UNAUTHORIZED,
-#        "Request headers must have a 'auth_provider' field",
-#    )
-# if (
-#    auth_provider_settings := oidc_providers_settings.get(auth_provider_id)
-# ) is None:
-#    raise HTTPException(
-#        status.HTTP_401_UNAUTHORIZED, f"Unknown auth provider '{auth_provider_id}'"
-#    )
-# if (key := auth_provider_settings.get_public_key()) is None:
-#    raise HTTPException(
-#        status.HTTP_401_UNAUTHORIZED,
-#        f"Key for provider '{auth_provider_id}' unknown",
-#    )
-# logger.warn(f"refresh with scope {scope}")
-# breakpoint()
-# refreshed_auth_info = await user.oidc_provider.fetch_access_token(scope=scope)
-### Decode the new token
-# try:
-#    payload = decode(
-#        refreshed_auth_info["access_token"],
-#        key=key,
-#        algorithms=["RS256"],
-#        audience="account",
-#        options={"verify_signature": not settings.insecure.skip_verify_signature},
-#    )
-# except ExpiredSignatureError as err:
-#    logger.info(f"Expired signature: {err}")
-#    raise HTTPException(
-#        status.HTTP_401_UNAUTHORIZED,
-#        "Expired signature (refresh not implemented yet)",
-#    )

src/oidc_test/settings.py

@@ -4,64 +4,22 @@ import random
 from typing import Type, Tuple
 from pathlib import Path
 
-from pydantic import AnyHttpUrl, BaseModel, computed_field, AnyUrl
+from pydantic import BaseModel, computed_field
 from pydantic_settings import (
     BaseSettings,
     SettingsConfigDict,
     PydanticBaseSettingsSource,
     YamlConfigSettingsSource,
 )
-from starlette.requests import Request
 
 
-class Resource(BaseModel):
+class OIDCProvider(BaseModel):
-    """A resource with an URL that can be accessed with an OAuth2 access token"""
-
-    resource_name: str
-    name: str
-    url: str
-
-
-class ResourceProvider(BaseModel):
-    id: str
-    name: str
-    base_url: AnyUrl
-    resources: list[Resource] = []
-    verify_ssl: bool = True
-
-    def get_resource(self, resource_name: str) -> Resource:
-        return [
-            resource for resource in self.resources if resource.resource_name == resource_name
-        ][0]
-
-    def get_resource_url(self, resource_name: str) -> str:
-        return f"{self.base_url}{self.get_resource(resource_name).url}"
-
-
-class AuthProviderSettings(BaseModel):
-    """Auth provider, can also be a resource server"""
-
     id: str
     name: str
     url: str
     client_id: str
     client_secret: str = ""
-    # For PKCE (not implemented yet)
-    code_challenge_method: str | None = None
     hint: str = "No hint"
-    resources: list[Resource] = []
-    account_url_template: str | None = None
-    info_url: str | None = (
-        None  # Used eg. for Keycloak's public key (see https://stackoverflow.com/questions/54318633/getting-keycloaks-public-key)
-    )
-    public_key: str | None = None
-    public_key_url: str | None = None
-    signature_alg: str = "RS256"
-    resource_provider_scopes: list[str] = []
-    session_key: str = "sid"
-    skip_verify_signature: bool = True
-    disabled: bool = False
-    resource_providers: list[ResourceProvider] = []
 
     @computed_field
     @property
@@ -73,43 +31,21 @@ class AuthProviderSettings(BaseModel):
     def token_url(self) -> str:
         return "auth/" + self.id
 
-    def get_account_url(self, request: Request, user: dict) -> str | None:
-        if self.account_url_template:
-            if not (self.url.endswith("/") or self.account_url_template.startswith("/")):
-                sep = "/"
-            else:
-                sep = ""
-            return self.url + sep + self.account_url_template.format(request=request, user=user)
-        else:
-            return None
 
-
+class OIDCSettings(BaseModel):
-class AuthSettings(BaseModel):
     show_session_details: bool = False
-    providers: list[AuthProviderSettings] = []
+    providers: list[OIDCProvider] = []
     swagger_provider: str = ""
 
 
-class Insecure(BaseModel):
-    """Warning: changing these defaults are only suitable for debugging"""
-
-    skip_verify_signature: bool = False
-
-
 class Settings(BaseSettings):
     """Settings wil be read from an .env file"""
 
-    model_config = SettingsConfigDict(env_nested_delimiter="__")
+    oidc: OIDCSettings = OIDCSettings()
-
-    auth: AuthSettings = AuthSettings()
     secret_key: str = "".join(random.choice(string.ascii_letters) for _ in range(16))
     log: bool = False
-    log_config_file: str = "log_conf.yaml"
+
-    insecure: Insecure = Insecure()
+    model_config = SettingsConfigDict(env_nested_delimiter="__")
-    cors_origins: list[str] = []
-    debug_token: bool = False
-    show_token: bool = False
-    show_external_resource_providers_links: bool = False
 
     @classmethod
     def settings_customise_sources(
@@ -128,7 +64,9 @@ class Settings(BaseSettings):
                 settings_cls,
                 Path(
                     Path(
-                        environ.get("OIDC_TEST_SETTINGS_FILE", Path.cwd() / "settings.yaml"),
+                        environ.get(
+                            "OIDC_TEST_SETTINGS_FILE", Path.cwd() / "settings.yaml"
+                        ),
                     )
                 ),
             ),

src/oidc_test/static/styles.css

@@ -1,239 +0,0 @@
-body {
-  font-family: Arial, Helvetica, sans-serif;
-  background-color: floralwhite;
-  margin: 0;
-  font-family: system-ui;
-  text-align: center;
-}
-h1 {
-  background-color: #f7c7867d;
-  margin: 0 0 0.2em 0;
-  box-shadow: 0px 0.2em 0.2em #f7c7867d;
-  text-shadow: 0 0 2px #00000080;
-  font-weight: 200;
-}
-p {
-  margin: 0.2em;
-}
-hr {
-  margin: 0.2em;
-}
-.hidden {
-  display: none;
-}
-.version {
-  position: absolute;
-  font-size: 75%;
-  top: 0.3em;
-  right: 0.3em;
-}
-.center {
-  text-align: center;
-}
-.error {
-  color: darkred;
-}
-.content {
-  width: 100%;
-  display: flex;
-  flex-direction: column;
-  align-items: center;
-  justify-content: center;
-}
-.user-info {
-  padding: 0.5em;
-  display: flex;
-  gap: 0.5em;
-  flex-direction: column;
-  width: fit-content;
-  align-items: center;
-  margin: 5px auto;
-  box-shadow: 0px 0px 10px lightgreen;
-  background-color: lightgreen;
-  border-radius: 8px;
-}
-.user-info * {
-  flex: 2 1 auto;
-  margin: 0;
-}
-.user-info .picture {
-  max-width: 3em;
-  max-height: 3em
-}
-.user-info a.logout {
-  border: 2px solid darkkhaki;
-  padding: 3px 6px;
-  text-decoration: none;
-  color: black;
-}
-.user-info a.logout:hover {
-  background-color: orange;
-}
-.debug-auth {
-  font-size: 90%;
-  background-color: #d8bebc75;
-  padding: 6px;
-}
-.debug-auth * {
-  margin: 0;
-}
-.debug-auth p {
-  border-bottom: 1px solid black;
-}
-.debug-auth ul {
-  padding: 0;
-  list-style: none;
-}
-.debug-auth p, .debug-auth .key {
-  font-weight: bold;
-}
-.content {
-  text-align: left;
-}
-.hasResponseStatus {
-  background-color: #88888840;
-}
-.hasResponseStatus.status-200 {
-  background-color: #00ff0040;
-}
-.hasResponseStatus.status-401 {
-  background-color: #ff000040;
-}
-.hasResponseStatus.status-403 {
-  background-color: #ff990040;
-}
-.hasResponseStatus.status-404 {
-  background-color: #ffCC0040;
-}
-.hasResponseStatus.status-503 {
-  background-color: #ffA88050;
-}
-
-.role, .scope {
-  padding: 3px 6px;
-  margin: 3px;
-  border-radius: 6px;
-}
-
-.role {
-  background-color: #44228840;
-}
-
-.scope {
-  background-color: #8888FF80;
-}
-
-
-/* For home */
-
-.login-box {
-  background-color: antiquewhite;
-  margin: 0.5em auto;
-  width: fit-content;
-  box-shadow: 0 0 10px #49759b88;
-  border-radius: 8px;
-}
-.login-box .description {
-  font-style: italic;
-  font-weight: bold;
-  background-color: #f7c7867d;
-  padding: 6px;
-  margin: 0;
-  border-radius: 8px 8px 0 0;
-}
-.providers {
-  justify-content: center;
-  padding: 0.8em;
-}
-.providers .provider {
-  min-height: 2em;
-}
-.providers .provider .link {
-  text-decoration: none;
-  max-height: 2em;
-}
-.providers .provider .link {
-  background-color: #f7c7867d;
-  border-radius: 8px;
-  padding: 6px;
-  text-align: center;
-  color: black;
-  font-weight: 400;
-  cursor: pointer;
-  border: 0;
-  width: 100%;
-}
-
-.providers .provider .link.disabled {
-  color: gray; 
-  cursor: not-allowed;
-}
-
-.providers .provider .hint {
-  font-size: 80%;
-  max-width: 13em;
-}
-.providers .error {
-  padding: 3px 6px;
-  font-weight: bold;
-  flex: 1 1 auto;
-}
-.content .links-to-check {
-  display: flex;
-  justify-content: center;
-  gap: 0.5em;
-  flex-flow: wrap;
-}
-.content .links-to-check button {
-  color: black;
-  padding: 5px 10px;
-  text-decoration: none;
-  border-radius: 8px;
-  border: none;
-  cursor: pointer;
-}
-
-.token {
-  overflow-wrap: anywhere;
-  font-family: monospace;
-}
-
-.resourceResult {
-  padding: 0.5em;
-  display: flex;
-  gap: 0.5em;
-  width: fit-content;
-  align-items: center;
-  margin: 5px auto;
-  box-shadow: 0px 0px 10px #90c3eeA0;
-  background-color: #90c3eeA0;
-  border-radius: 8px;
-}
-
-.resources {
-  display: flex;
-}
-
-.resource {
-  text-align: center;
-}
-
-.token-info {
-  margin: 0 1em;
-}
-
-.key {
-  font-weight: bold;
-}
-
-.token .key, .token .value {
-  display: inline;
-}
-.token .value {
-  padding-left: 1em;
-}
-
-.msg {
-  text-align: center;
-  font-weight: bold;
-}

src/oidc_test/static/utils.js

@@ -1,90 +0,0 @@
-async function checkHref(elem, token, authProvider) {
-  const msg = document.getElementById("msg")
-  const resourceName = elem.getAttribute("resource-name")
-  const resourceId = elem.getAttribute("resource-id")
-  const resourceProviderId = elem.getAttribute("resource-provider-id") ? elem.getAttribute("resource-provider-id") : ""
-  const fqResourceName = resourceProviderId ? `${resourceProviderId}:${resourceName}` : resourceName
-  const url = resourceId ? `resource/${fqResourceName}/${resourceId}` : `resource/${fqResourceName}`
-  const resp = await fetch(url, {
-    method: "GET",
-    headers: new Headers({
-      "Content-type": "application/json",
-      "Authorization": `Bearer ${token}`,
-      "auth_provider": authProvider,
-    }),
-  }).catch(err => {
-    msg.innerHTML = "Cannot fetch resource: " + err.message
-    resourceElem.innerHTML = ""
-  })
-  if (resp === undefined) {
-    return
-  } else {
-    elem.classList.add("hasResponseStatus")
-    elem.classList.add("status-" + resp.status)
-    elem.title = "Response code: " + resp.status + " - " + resp.statusText
-  }
-}
-
-function checkPerms(className, token, authProvider) {
-  var rootElems = document.getElementsByClassName(className)
-  Array.from(rootElems).forEach(elem =>
-    Array.from(elem.children).forEach(elem => checkHref(elem, token, authProvider))
-  )
-}
-
-async function get_resource(resourceName, token, authProvider, resourceId, resourceProviderId) {
-  // BaseUrl for an external resource provider
-  //if (!keycloak.keycloak) { return }
-  const msg = document.getElementById("msg")
-  const resourceElem = document.getElementById('resource')
-  const fqResourceName = resourceProviderId ? `${resourceProviderId}:${resourceName}` : resourceName
-  const url = resourceId ? `resource/${fqResourceName}/${resourceId}` : `resource/${fqResourceName}`
-  const resp = await fetch(url, {
-    method: "GET",
-    headers: new Headers({
-      "Content-type": "application/json",
-      "Authorization": `Bearer ${token}`,
-      "auth_provider": authProvider,
-    }),
-  }).catch(err => {
-    msg.innerHTML = "Cannot fetch resource: " + err.message
-    resourceElem.innerHTML = ""
-  })
-  if (resp === undefined) {
-    return
-  }
-  const resource = await resp.json()
-  if (!resp.ok) {
-    msg.innerHTML = resource["detail"]
-    resourceElem.innerHTML = ""
-    return
-  }
-  msg.innerHTML = ""
-  resourceElem.innerHTML = ""
-  Object.entries(resource).forEach(
-    ([key, value]) => {
-      let r = document.createElement('div')
-      let kElem = document.createElement('div')
-      kElem.innerText = key
-      kElem.className = "key"
-      let vElem = document.createElement('div')
-      if (typeof value == "object") {
-        Object.entries(value).forEach(v => {
-          const ne = document.createElement('div')
-          ne.innerHTML = `<span class="key">${v[0]}</span>: <span class="value">${v[1]}</span>`
-          vElem.appendChild(ne)
-        })
-      }
-      else {
-        vElem.innerText = value
-      }
-      vElem.className = "value"
-      if (key == "sorry") {
-        vElem.classList.add("error")
-      }
-      r.appendChild(kElem)
-      r.appendChild(vElem)
-      resourceElem.appendChild(r)
-    }
-  )
-}

src/oidc_test/templates/base.html

@@ -1,12 +1,139 @@
 <html>
   <head>
-    <title>OIDC (FastAPI) test</title>
+    <title>FastAPI OIDC test</title>
-    <link href="{{ url_for('static', path='/styles.css') }}" rel="stylesheet">
+    <style>
-    <script src="{{ url_for('static', path='/utils.js') }}"></script>
+      body {
+        font-family: Arial, Helvetica, sans-serif;
+        background-color: floralwhite;
+        margin: 0;
+      }
+      h1 {
+        text-align: center;
+        background-color: #f7c7867d;
+        margin: 0 0 0.2em 0;
+      }
+      p {
+        margin: 0.2em;
+      }
+      hr {
+        margin: 0.2em;
+      }
+      .hidden {
+        display: none;
+      }
+      .center {
+        text-align: center;
+      }
+      .content {
+        width: 100%;
+        display: flex;
+        flex-direction: column;
+        align-items: center;
+        justify-content: center;
+      }
+      .user-info {
+        padding: 0.5em;
+        display: flex;
+        gap: 0.5em;
+        flex-direction: column;
+        width: fit-content;
+        align-items: center;
+        margin: 5px auto;
+        box-shadow: 0px 0px 10px lightgreen;
+        background-color: lightgreen;
+        border-radius: 8px;
+      }
+      .user-info * {
+        flex: 2 1 auto;
+        margin: 0;
+      }
+      .user-info .picture {
+        max-width: 3em;
+        max-height: 3em
+      }
+      .user-info a.logout {
+        border: 2px solid darkkhaki;
+        padding: 3px 6px;
+        text-decoration: none;
+        text-align: center;
+        color: black;
+      }
+      .user-info a.logout:hover {
+        background-color: orange;
+      }
+     .debug-auth {
+        font-size: 90%;
+        background-color: #d8bebc75;
+        padding: 6px;
+      }
+      .debug-auth * {
+        margin: 0;
+      }
+      .debug-auth p {
+        text-align: center;
+        border-bottom: 1px solid black;
+      }
+      .debug-auth ul {
+        padding: 0;
+        list-style: none;
+      }
+      .debug-auth p, .debug-auth .key {
+        font-weight: bold;
+      }
+      .content {
+        text-align: left;
+      }
+      .content #links-to-check {
+        display: flex;
+        text-align: center;
+        justify-content: center;
+        gap: 0.5em;
+        flex-flow: wrap;
+      }
+      .content #links-to-check a {
+        color: black;
+        padding: 5px 10px;
+        text-decoration: none;
+        border-radius: 8px;
+      }
+      .hasResponseStatus {
+        background-color: #88888840;
+      }
+      .hasResponseStatus.status-200 {
+        background-color: #00ff0040;
+      }
+      .hasResponseStatus.status-401 {
+        background-color: #ff000040;
+      }
+      .hasResponseStatus.status-403 {
+        background-color: #ff990040;
+      }
+      .role {
+        padding: 3px 6px;
+        background-color: #44228840;
+      }
+    </style>
+    <script>
+    function checkHref(elem) {
+      var xmlHttp = new XMLHttpRequest()
+      xmlHttp.onreadystatechange = function() { 
+        if (xmlHttp.readyState == 4) {
+          elem.classList.add("hasResponseStatus")
+          elem.classList.add("status-" + xmlHttp.status)
+          elem.title = "Response code: " + xmlHttp.status
+        }
+      }
+      xmlHttp.open("GET", elem.href, true) // true for asynchronous 
+      xmlHttp.send(null)
+    }
+    function checkPerms(rootId) {
+      var rootElem = document.getElementById(rootId)
+      Array.from(rootElem.children).forEach(elem => checkHref(elem))
+    }
+    </script>
   </head>
-  <body onload="checkPerms('links-to-check', '{{ access_token }}', '{{ auth_provider.id }}')">
+  <body onload="checkPerms('links-to-check')">
-    <div class="version">v. {{ __version__}}</div>
+    <h1>OIDC-test</h1>
-    <h1>OIDC-test - FastAPI client</h1>
     {% block content %}
     {% endblock %}
   </body>

src/oidc_test/templates/home.html

@@ -1,42 +1,85 @@
 {% extends "base.html" %}
 {% block content %}
+<style>
+ .login-box {
+   text-align: center;
+   background-color: antiquewhite;
+   margin: 0.5em auto;
+   width: fit-content;
+   box-shadow: 0 0 10px #49759b88;
+   border-radius: 8px;
+ }
+ .login-box .description {
+   font-style: italic;
+   font-weight: bold;
+   background-color: #f7c7867d;
+   padding: 6px;
+   margin: 0;
+   border-radius: 8px 8px 0 0;
+ }
+ .providers {
+   justify-content: center;
+   padding: 0.8em;
+ }
+ .providers .provider {
+   min-height: 2em;
+ }
+ .providers .provider a.link {
+   text-decoration: none;
+   max-height: 2em;
+ }
+ .providers .provider .link div {
+   text-align: center;
+   background-color: #f7c7867d;
+   border-radius: 8px;
+   padding: 6px;
+   text-align: center;
+   color: black;
+   font-weight: bold;
+   cursor: pointer;
+ }
+ .providers .provider .hint {
+   font-size: 80%;
+   max-width: 13em;
+ }
+ .providers .error {
+   color: darkred;
+   padding: 3px 6px;
+   text-align: center;
+   font-weight: bold;
+   flex: 1 1 auto;
+ }
+</style>
   <p class="center">
     Test the authentication and authorization,
     with OpenID Connect and OAuth2 with different providers.
   </p>
   {% if not user %} 
-    <div class="login-box">
+  <div class="login-box">
-      <p class="description">Log in with:</p>
+    <p class="description">Log in with:</p>
-      <table class="providers">
+    <table class="providers">
-        {% for provider in providers.values() %}
+      {% for provider in settings.oidc.providers %}
-        <tr class="provider">
+      <tr class="provider">
-          <td>
+        <td>
-            <button class="link{% if provider.disabled %} disabled{% endif %}"
+          <a class="link" href="login/{{ provider.id }}"><div>{{ provider.name }}</div></a>
-                    {% if provider.disabled %}disabled{% endif %}
+        </td>
-                    onclick="location.href='login/{{ provider.id }}'">
+        <td class="hint">{{ provider.hint }}</div>
-              {{ provider.name }}
+        </td>
-            </button>
+      </tr>
-          </td>
+      {% else %}
-          <td class="hint">{{ provider.hint }}</div>
+        <div class="error">There is no authentication provider defined.
-          </td>
+        Hint: check the settings.yaml file.</div>
-        </tr>
+      {% endfor %}
-        {% else %}
+    </table>
-          <div class="error">There is no authentication provider defined.
+  </div>
-          Hint: check the settings.yaml file.</div>
+  {% endif %}
-        {% endfor %}
+  {% if user %}
-      </table>
-    </div>
-  {% else %}
     <div class="user-info">
       <p>Hey, {{ user.name }}</p>
       {% if user.picture %}
         <img src="{{ user.picture }}" class="picture"></img>
       {% endif %}
       <div>{{ user.email }}</div>
-      <div>
-        <span>Provider:</span>
-        {{ auth_provider.name }} 
-      </div>
       {% if user.roles %}
         <div>
         <span>Roles:</span>
@@ -45,125 +88,40 @@
         {% endfor %}
         </div>
       {% endif %}
-      {% if access_token_scope %}
+      <div>
-        <div>
+        <span>Provider:</span>
-          <span>Scopes</span>:
+        {{ user.oidc_provider.name }} 
-          {% for scope in access_token_scope.split(' ') %}
+      </div>
-            <span class="scope">{{ scope }}</span>
+      <a href="logout" class="logout">Logout</a>
-          {% endfor %}
-        </div>
-      {% endif %}
-      {% if auth_provider.account_url_template %}
-        <button
-          onclick="location.href='{{ auth_provider.get_account_url(request, user.model_dump()) }}'"
-          class="account">
-          Account management
-        </button>
-      {% endif %}
-      <button onclick="location.href='{{ request.url_for("refresh") }}'" class="refresh">Refresh access token</button>
-      <button onclick="location.href='{{ request.url_for("logout") }}'" class="logout">Logout</button>
     </div>
   {% endif %}
   <hr>
   <div class="content">
-    {% if resources %}
+    <p>
-      <p>This application provides all these resources, eventually protected with scope or roles:</p>
+      These links should get different response codes depending on the authorization:
-      <div class="links-to-check">
+    </p>
-        {% for name, resource in resources.items() %}
+    <div id="links-to-check">
-          {% if resource.default_resource_id %}
+      <a href="public">Public</a>
-            <button resource-name="{{ name }}"
+      <a href="protected">Auth protected content</a>
-                    resource-id="{{ resource.default_resource_id }}"
+      <a href="protected-by-foorole">Auth + foorole protected content</a>
-                    onclick="get_resource('{{ name }}', '{{ access_token }}', '{{ auth_provider.id }}', '{{ resource.default_resource_id }}')"
+      <a href="protected-by-foorole-or-barrole">Auth + foorole or barrole protected content</a>
-                    >
+      <a href="protected-by-barrole">Auth + barrole protected content</a>
-              {{ resource.name }}
+      <a href="protected-by-foorole-and-barrole">Auth + foorole and barrole protected content</a>
-            </button>
+      <a href="fast_api_depends" class="hidden">Using FastAPI Depends</a>
-          {% else %}
+      <a href="other">Other</a>
-            <button resource-name="{{ name }}"
-                    onclick="get_resource('{{ name }}', '{{ access_token }}', '{{ auth_provider.id }}')"
-                    >
-              {{ resource.name }}
-            </button>
-          {% endif %}
-        {% endfor %}
-      </div>
-    {% endif %}
-    {% if auth_provider.resources %}
-      <p>{{ auth_provider.name }} is also defined as a provider for these resources:</p>
-      <div class="links-to-check">
-        {% for resource in auth_provider.resources %}
-          {% if resource.default_resource_id %}
-            <button resource-name="{{ resource.resource_name }}"
-                    resource-id="{{ resource.default_resource_id }}"
-                    onclick="get_resource('{{ resource.resource_name }}', '{{ access_token }}', '{{ auth_provider.id }}', '{{ resource.default_resource_id }}')"
-                    >
-              {{ resource.name }}
-            </button>
-          {% else %}
-            <button resource-name="{{ resource.resource_name }}"
-                    onclick="get_resource('{{ resource.resource_name }}', '{{ access_token }}', '{{ auth_provider.id }}')"
-                    >
-              {{ resource.name }}
-            </button>
-          {% endif %}
-        {% endfor %}
-      </div>
-    {% endif %}
-    {% if resource_providers %}
-      <p>{{ auth_provider.name }} allows this application to request resources from third party resource providers:</p>
-      {% for resource_provider in resource_providers %}
-        <div class="links-to-check">
-          {{ resource_provider.name }}
-          {% for resource in resource_provider.resources %}
-            <button resource-name="{{ resource.resource_name }}"
-                    resource-id="{{ resource.default_resource_id  }}"
-                    resource-provider-id="{{ resource_provider.id }}"
-                    onclick="get_resource('{{ resource.resource_name }}', '{{ access_token }}',
-                                          '{{ auth_provider.id }}', '{{ resource.default_resource_id }}',
-                                          '{{ resource_provider.id }}')"
-                    >
-              {{ resource.name }}
-            </button>
-          {% endfor %}
-        </div>
-      {% endfor %}
-    {% endif %}
-    <div class="resourceResult">
-      <div id="resource" class="resource"></div>
-      <div id="msg" class="msg error"></div>
     </div>
-  </div>
+  {% if user_info_details %}
-  {% if show_token and id_token_parsed %}
+    <div class="debug-auth">
-    <div class="token-info">
+      <p>User info</p>
-      <hr>
+      <ul>
-      <div>
+        {% for key, value in user_info_details.items() %}
-        <h2>id token</h2>
+        <li>
-        <div class="token">
+          <span class="key">{{ key }}</span>: {{ value }}
-        {% for key, value in id_token_parsed.items() %}
+        </li>
-          <div>
-            <div class="key">{{ key }}</div>
-            <div class="value">{{ value }}</div>
-          </div>
         {% endfor %}
-        </div>
+      </ul>
-        <h2>access token</h2>
-        <div class="token">
-        {% for key, value in access_token_parsed.items() %}
-          <div>
-            <div class="key">{{ key }}</div>
-            <div class="value">{{ value }}</div>
-          </div>
-        {% endfor %}
-        </div>
-        <h2>refresh token</h2>
-        <div class="token">
-        {% for key, value in refresh_token_parsed.items() %}
-          <div>
-            <div class="key">{{ key }}</div>
-            <div class="value">{{ value }}</div>
-          </div>
-        {% endfor %}
-        </div>
       </div>
+      <div>Now is: {{ now }}</div>
     </div>
   {% endif %}
 {% endblock %}

src/oidc_test/templates/non_compliant_logout.html

@@ -6,12 +6,12 @@
     authorisation to log in again without asking for credentials.
   </p>
   <p>
-    This is because {{ auth_provider.name }} does not provide "end_session_endpoint" in its metadata
+    This is because {{ provider.name }} does not provide "end_session_endpoint" in its metadata
-    (see: <a href="{{ auth_provider.authlib_client._server_metadata_url }}">{{ auth_provider.authlib_client._server_metadata_url }}</a>).
+    (see: <a href="{{ provider._server_metadata_url }}">{{ provider._server_metadata_url }}</a>).
   </p>
   <p>
     You can just also go back to the <a href="{{ home_url }}">application home page</a>, but
-    it recommended to go to the <a href="{{ auth_provider.authlib_client.server_metadata['issuer'] }}">OIDC provider's site</a>
+    it recommended to go to the <a href="{{ provider.server_metadata['issuer'] }}">provider's site</a>
     and log out explicitely from there.
   </p>
 {% endblock %}

uv.lock

@@ -1,5 +1,4 @@
 version = 1
-revision = 1
 requires-python = ">=3.13"
 
 [[package]]
@@ -207,18 +206,6 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/68/1b/e0a87d256e40e8c888847551b20a017a6b98139178505dc7ffb96f04e954/dnspython-2.7.0-py3-none-any.whl", hash = "sha256:b4c34b7d10b51bcc3a5071e7b8dee77939f1e878477eeecc965e9835f63c6c86", size = 313632 },
 ]
 
-[[package]]
-name = "dunamai"
-version = "1.23.0"
-source = { registry = "https://pypi.org/simple" }
-dependencies = [
-    { name = "packaging" },
-]
-sdist = { url = "https://files.pythonhosted.org/packages/06/4e/a5c8c337a1d9ac0384298ade02d322741fb5998041a5ea74d1cd2a4a1d47/dunamai-1.23.0.tar.gz", hash = "sha256:a163746de7ea5acb6dacdab3a6ad621ebc612ed1e528aaa8beedb8887fccd2c4", size = 44681 }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/21/4c/963169386309fec4f96fd61210ac0a0666887d0fb0a50205395674d20b71/dunamai-1.23.0-py3-none-any.whl", hash = "sha256:a0906d876e92441793c6a423e16a4802752e723e9c9a5aabdc5535df02dbe041", size = 26342 },
-]
-
 [[package]]
 name = "ecdsa"
 version = "0.19.0"
@@ -495,17 +482,15 @@ wheels = [
 
 [[package]]
 name = "oidc-fastapi-test"
+version = "0.0.0"
 source = { editable = "." }
 dependencies = [
     { name = "authlib" },
     { name = "cachetools" },
     { name = "fastapi", extra = ["standard"] },
-    { name = "httpx" },
     { name = "itsdangerous" },
     { name = "passlib", extra = ["bcrypt"] },
-    { name = "pkce" },
     { name = "pydantic-settings" },
-    { name = "pyjwt" },
     { name = "python-jose", extra = ["cryptography"] },
     { name = "requests" },
     { name = "sqlmodel" },
@@ -513,7 +498,6 @@ dependencies = [
 
 [package.dev-dependencies]
 dev = [
-    { name = "dunamai" },
     { name = "ipdb" },
     { name = "pytest" },
 ]
@@ -523,12 +507,9 @@ requires-dist = [
     { name = "authlib", specifier = ">=1.4.0" },
     { name = "cachetools", specifier = ">=5.5.0" },
     { name = "fastapi", extras = ["standard"], specifier = ">=0.115.6" },
-    { name = "httpx", specifier = ">=0.28.1" },
     { name = "itsdangerous", specifier = ">=2.2.0" },
     { name = "passlib", extras = ["bcrypt"], specifier = ">=1.7.4" },
-    { name = "pkce", specifier = ">=1.0.3" },
     { name = "pydantic-settings", specifier = ">=2.7.1" },
-    { name = "pyjwt", specifier = ">=2.10.1" },
     { name = "python-jose", extras = ["cryptography"], specifier = ">=3.3.0" },
     { name = "requests", specifier = ">=2.32.3" },
     { name = "sqlmodel", specifier = ">=0.0.22" },
@@ -536,7 +517,6 @@ requires-dist = [
 
 [package.metadata.requires-dev]
 dev = [
-    { name = "dunamai", specifier = ">=1.23.0" },
     { name = "ipdb", specifier = ">=0.13.13" },
     { name = "pytest", specifier = ">=8.3.4" },
 ]
@@ -585,15 +565,6 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/9e/c3/059298687310d527a58bb01f3b1965787ee3b40dce76752eda8b44e9a2c5/pexpect-4.9.0-py2.py3-none-any.whl", hash = "sha256:7236d1e080e4936be2dc3e326cec0af72acf9212a7e1d060210e70a47e253523", size = 63772 },
 ]
 
-[[package]]
-name = "pkce"
-version = "1.0.3"
-source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/29/ea/ddd845c2ec21bf1e8555c782b32dc39b82f0b12764feb9f73ccbb2470f13/pkce-1.0.3.tar.gz", hash = "sha256:9775fd76d8a743d39b87df38af1cd04a58c9b5a5242d5a6350ef343d06814ab6", size = 2757 }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/15/51/52c22ec0812d25f5bf297a01153604bfa7bfa59ed66f6cd8345beb3c2b2a/pkce-1.0.3-py3-none-any.whl", hash = "sha256:55927e24c7d403b2491ebe182b95d9dcb1807643243d47e3879fbda5aad4471d", size = 3200 },
-]
-
 [[package]]
 name = "pluggy"
 version = "1.5.0"
@@ -712,15 +683,6 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/f7/3f/01c8b82017c199075f8f788d0d906b9ffbbc5a47dc9918a945e13d5a2bda/pygments-2.18.0-py3-none-any.whl", hash = "sha256:b8e6aca0523f3ab76fee51799c488e38782ac06eafcf95e7ba832985c8e7b13a", size = 1205513 },
 ]
 
-[[package]]
-name = "pyjwt"
-version = "2.10.1"
-source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/e7/46/bd74733ff231675599650d3e47f361794b22ef3e3770998dda30d3b63726/pyjwt-2.10.1.tar.gz", hash = "sha256:3cc5772eb20009233caf06e9d8a0577824723b44e6648ee0a2aedb6cf9381953", size = 87785 }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/61/ad/689f02752eeec26aed679477e80e632ef1b682313be70793d798c1d5fc8f/PyJWT-2.10.1-py3-none-any.whl", hash = "sha256:dcdd193e30abefd5debf142f9adfcdd2b58004e644f25406ffaebd50bd98dacb", size = 22997 },
-]
-
 [[package]]
 name = "pytest"
 version = "8.3.4"