CODEWITHSUNDEEP
keycloakfastapi
fernandezcuesta
fernandezcuesta

Reputation: 2448

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

Upvotes: 5

Views: 9780

Answers (2)

mahesh goud shivannagari
mahesh goud shivannagari

Reputation: 1

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2AuthorizationCodeBearer
import uvicorn
from jose import JOSEError,jwt
import requests

ISSUER_URL="http://localhost:8083/realms/realmname"
app = FastAPI()


oauth_2_scheme = OAuth2AuthorizationCodeBearer(
    tokenUrl="http://localhost:8083/realms/realmname/protocol/openid-connect/token",
    authorizationUrl="http://localhost:8083/realms/realmname/protocol/openid-connect/auth")

public_key = requests.get(ISSUER_URL).json().get('public_key')
key = '-----BEGIN PUBLIC KEY-----\n' + public_key + '\n-----END PUBLIC KEY-----'

def valid_access_token(token: str | None = Depends(oauth_2_scheme)):
    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))

@app.get("/public")
def get_public():
    return {"message": "This endpoint is public"}


@app.get("/private", dependencies=[Depends(valid_access_token)])
def get_private():
    return {"message": "This endpoint is private"}


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8034)

Upvotes: 0

fernandezcuesta
fernandezcuesta

Reputation: 2448

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))

Upvotes: 2

Related Questions