keycloakfastapi

Authenticate FastAPI with clientid/clientsecret


I'm trying to set up FastAPI (0.71.0) authentication with clientid-clientsecret.

I configured OAuth2AuthorizationCodeBearer and apparently from the swagger (/docs) endpoint it looks fine, it asks for client-id and client-secret for authentication.

auth = OAuth2PasswordBearer(
    authorizationUrl=AUTH_URL,
    tokenUrl=TOKEN_URL,
)

agent = FastAPI(
    description=_DESCRIPTION,
    version=VERSION,
    dependencies=[Depends(auth)],
    middleware=middlewares,
    root_path=os.getenv('BASEPATH', '/'),
    swagger_ui_init_oauth={
        'usePkceWithAuthorizationCodeGrant': True,
        'scopes': 'openid profile email'
    }
)

However calling the API directly I can access with any Bearer such as:

import requests

url = 'http://localhost:8080/test'
headers = {'Authorization': 'Bearer BADTOKEN', 'Content-Type': 'application/json', 'Accept': 'application/json'}
response = requests.get(url=url, params={}, headers=headers)

and response.ok = True, so I might be missing something in FastAPI settings but cannot see where.

Is this the correct authentication flow?

PS: What I want to achieve is service to service communication, so the API is not getting accessed by a regular user


Solution

  • After having a look at OAuth2AuthorizationCodeBearer implementation, I ended up adding a method to check the validity of the bearer token:

    public_key = requests.get(ISSUER_URL).json().get('public_key')
    key = '-----BEGIN PUBLIC KEY-----\n' + public_key + '\n-----END PUBLIC KEY-----'
    
    oauth = OAuth2AuthorizationCodeBearer(
            authorizationUrl=AUTH_URL,
            tokenUrl=TOKEN_URL,
    )
    async def auth(token: str | None = Depends(oauth)):
        try:
            jwt.decode(
                token,
                key=key,
                options={
                    "verify_signature": True,
                    "verify_aud": False,
                    "verify_iss": ISSUER_URL
                }
            )
        except JOSEError as e:  # catches any exception
            raise HTTPException(
                status_code=401,
                detail=str(e))