我正在尝试使用将一些基本的ML模型返回给用户。
目前,我使用firebase保护用户详细信息。我希望在使用基本应用程序验证他们对ML模型的请求时,使用JWT的用户拥有的。
对于fastapi来说,这样做似乎没有一个直接的答案。
我遵循了两个主要线程,作为解决这一问题的方法,但对于如何简单地从请求的标题中提取JWT,并检查它是否与防火墙管理员或您的目标无关,我有点不知所措。
遵循本教程并使用此包,我将得到类似的内容,https://github.com/tokusumi/fastapi-cloudauth。这并没有真正做什么-它没有为我验证JWT,对于这个包是否真的值得有点困惑吗?
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}"
或者,看看这个,
https://github.com/tiangolo/fastapi/issues/4768
就像这样会有用的,
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!"}
但老实说,我对如何设置firebase环境变量感到有点困惑,我需要哪些包(firebase a?)
会喜欢一些帮手的,谢谢!
发布于 2022-05-20 20:11:09
我希望这会有帮助:
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']}
请不要忘记安装要求:pip3 install firebase_admin
https://stackoverflow.com/questions/72200552
复制相似问题