fix: auth port to joserfc

authlib.jose is deprecated and no longer works with other updated dependencies.
This commit is contained in:
Chris Milne 2026-05-07 12:54:55 +01:00
parent 02cfc5f7e7
commit 5f069285c2

View file

@ -6,16 +6,15 @@ Exports:
- authed_dependency - authed_dependency
""" """
import json import json
import requests
from typing import Annotated from typing import Annotated, Any
from authlib.jose import jwt from joserfc import jwt
from urllib.request import urlopen from urllib.request import urlopen
from fastapi import Depends, HTTPException from fastapi import Depends, HTTPException
from fastapi.security import OpenIdConnect from fastapi.security import OpenIdConnect
from authlib.jose.rfc7517.jwk import JsonWebKey from joserfc.jwk import KeySet
from authlib.jose.rfc7517.key_set import KeySet
from authlib.oauth2.rfc7523.validator import JWTBearerToken
from src.auth.config import auth_settings from src.auth.config import auth_settings
@ -24,12 +23,12 @@ oidc = OpenIdConnect(openIdConnectUrl=auth_settings.OIDC_CONFIG)
oidc_dependency = Annotated[str, Depends(oidc)] oidc_dependency = Annotated[str, Depends(oidc)]
async def get_current_user(oidc_auth_string: oidc_dependency) -> JWTBearerToken: async def get_current_user(oidc_auth_string: oidc_dependency) -> dict[str, Any]:
config_url = urlopen(auth_settings.OIDC_CONFIG) config_url = urlopen(auth_settings.OIDC_CONFIG)
config = json.loads(config_url.read()) config = json.loads(config_url.read())
jwks_uri = config["jwks_uri"] jwks_uri = config["jwks_uri"]
key_response = urlopen(jwks_uri) key_response = requests.get(jwks_uri)
jwk_keys: KeySet = JsonWebKey.import_key_set(json.loads(key_response.read())) jwk_keys = KeySet.import_key_set(key_response.json())
claims_options = { claims_options = {
"exp": {"essential": True}, "exp": {"essential": True},
@ -37,19 +36,19 @@ async def get_current_user(oidc_auth_string: oidc_dependency) -> JWTBearerToken:
"iss": {"essential": True, "value": auth_settings.OIDC_ISSUER}, "iss": {"essential": True, "value": auth_settings.OIDC_ISSUER},
} }
claims: JWTBearerToken = jwt.decode( token = jwt.decode(
oidc_auth_string.replace("Bearer ", ""), oidc_auth_string.replace("Bearer ", ""),
jwk_keys, jwk_keys
claims_options=claims_options,
claims_cls=JWTBearerToken,
) )
claims.validate() claims_requests = jwt.JWTClaimsRegistry(**claims_options)
return claims claims_requests.validate(token.claims)
return token.claims
claims_dependency = Annotated[JWTBearerToken, Depends(get_current_user)] claims_dependency = Annotated[dict[str, Any], Depends(get_current_user)]
async def is_authed_user(claims: claims_dependency) -> bool: async def is_authed_user(claims: claims_dependency) -> bool: