I'm trying to use fastapi to return some basic ML models to users.
Currently, I secure user details with firebase auth. I want to use the JWT's users have when using the basic application to authenticate their request for the ML model.
With fastapi, there doesn't seem to be a straightforward answer to doing this.
I've followed two main threads as ways to work out how to do this, but am a bit lost as to how I can simply take the JWT from the header of a request and check it against firebase admin or whathave you?
Following this tutorial and using this package, I end up with something like this, https://github.com/tokusumi/fastapi-cloudauth . This doesn't really do anything - it doesn't authenticate the JWT for me, bit confused as to if this package is actually worthwhile?
from fastapi import FastAPI, HTTPException, Header,Depends
from fastapi.middleware.cors import CORSMiddleware
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from fastapi_cloudauth.firebase import FirebaseCurrentUser, FirebaseClaims
app = FastAPI()
security = HTTPBearer()
origins = [
xxxx
]
app.add_middleware(
xxxx
)
get_current_user = FirebaseCurrentUser(
project_id=os.environ["PROJECT_ID"]
)
@app.get("/user/")
def secure_user(current_user: FirebaseClaims = Depends(get_current_user)):
# ID token is valid and getting user info from ID token
return f"Hello, {current_user.user_id}"
Alternatively, looking at this,
https://github.com/tiangolo/fastapi/issues/4768
It seems like something like this would work,
security = HTTPBearer()
api = FastAPI()
security = HTTPBearer()
firebase_client = FirebaseClient(
firebase_admin_credentials_url=firebase_test_admin_credentials_url
# ...
)
user_roles = [test_role]
async def firebase_authentication(token: HTTPAuthorizationCredentials = Depends(security)) -> dict:
user = firebase_client.verify_token(token.credentials)
return user
async def firebase_authorization(user: dict = Depends(firebase_authentication)):
roles = firebase_client.get_user_roles(user)
for role in roles:
if role in user_roles:
return user
raise HTTPException(detail="User does not have the required roles", status_code=HTTPStatus.FORBIDDEN)
@api.get("/")
async def root(uid: str = Depends(firebase_authorization)):
return {"message": "Successfully authenticated & authorized!"}
But honestly I'm a bit confused about how I would set up the firebase environment variables, what packages I would need (firebaseadmin?)
Would love some helpers, thanks!
I hope this will help:
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi import Depends, HTTPException, status, Response
from firebase_admin import auth, credentials, initialize_app
credential = credentials.Certificate('./key.json')
initialize_app(credential)
def get_user_token(res: Response, credential: HTTPAuthorizationCredentials=Depends(HTTPBearer(auto_error=False))):
if cred is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Bearer authentication is needed",
headers={'WWW-Authenticate': 'Bearer realm="auth_required"'},
)
try:
decoded_token = auth.verify_id_token(credential.credentials)
except Exception as err:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=f"Invalid authentication from Firebase. {err}",
headers={'WWW-Authenticate': 'Bearer error="invalid_token"'},
)
res.headers['WWW-Authenticate'] = 'Bearer realm="auth_required"'
return decoded_token
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi import Depends, HTTPException, status, Response, FastAPI, Depends
from firebase_admin import auth, credentials, initialize_app
credential = credentials.Certificate('./key.json')
initialize_app(credential)
def get_user_token(res: Response, credential: HTTPAuthorizationCredentials=Depends(HTTPBearer(auto_error=False))):
if cred is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Bearer authentication is needed",
headers={'WWW-Authenticate': 'Bearer realm="auth_required"'},
)
try:
decoded_token = auth.verify_id_token(credential.credentials)
except Exception as err:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=f"Invalid authentication from Firebase. {err}",
headers={'WWW-Authenticate': 'Bearer error="invalid_token"'},
)
res.headers['WWW-Authenticate'] = 'Bearer realm="auth_required"'
return decoded_token
app = FastAPI()
@app.get("/api/")
async def hello():
return {"msg":"Hello, this is API server"}
@app.get("/api/user_token")
async def hello_user(user = Depends(get_user_token)):
return {"msg":"Hello, user","uid":user['uid']}
P.S. Do not forget to install requirements:
pip3 install firebase_admin
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With