|
1 | 1 | import typing |
2 | | -import fastapi |
3 | 2 |
|
| 3 | +import fastapi |
4 | 4 | import firebase_admin |
5 | 5 | import pydantic |
6 | 6 | from fastapi import Depends, Security |
|
12 | 12 |
|
13 | 13 | token = HTTPBearer( |
14 | 14 | scheme_name="firebaseIdToken", |
| 15 | + bearerFormat="JWT", |
| 16 | + description="The firebase Id token, provided by client SDK.", |
15 | 17 | ) |
16 | 18 | _failed_auth_headers = {"WWW-Authenticate": "Bearer"} |
17 | 19 |
|
18 | 20 |
|
19 | 21 | def validate_token( |
20 | | - credential: HTTPAuthorizationCredentials = Security(token), |
| 22 | + credential: typing.Optional[HTTPAuthorizationCredentials] = Security(token), |
21 | 23 | app: firebase_admin.App = Depends(firebase_app), |
22 | | -) -> typing.Dict[str, typing.Any]: |
| 24 | +) -> typing.Optional[typing.Dict[str, typing.Any]]: |
| 25 | + if credential is None: |
| 26 | + return None |
| 27 | + |
23 | 28 | try: |
24 | 29 | return auth.verify_id_token(credential.credentials, app) |
25 | | - except auth.InvalidIdTokenError: |
26 | | - raise fastapi.HTTPException(401, "Invalid token received.", _failed_auth_headers) |
27 | | - except auth.UserDisabledError: |
28 | | - raise fastapi.HTTPException(403, "The user has been disabled.") |
29 | 30 | except auth.RevokedIdTokenError: |
30 | 31 | raise fastapi.HTTPException(403, "The token has been revoked.") |
31 | 32 | except auth.ExpiredIdTokenError: |
32 | 33 | raise fastapi.HTTPException(403, "The token has expired.") |
| 34 | + except auth.InvalidIdTokenError: |
| 35 | + raise fastapi.HTTPException(401, "Invalid token received.", _failed_auth_headers) |
| 36 | + except auth.UserDisabledError: |
| 37 | + raise fastapi.HTTPException(403, "The user has been disabled.") |
| 38 | + |
33 | 39 |
|
| 40 | +def token_info(token: typing.Optional[typing.Dict[str, typing.Any]] = Depends(validate_token)): |
| 41 | + if token is None: |
| 42 | + return None |
34 | 43 |
|
35 | | -def token_info(token: typing.Dict[str, typing.Any] = Depends(validate_token)): |
36 | 44 | return pydantic.parse_obj_as(TokenData, token) |
| 45 | + |
| 46 | + |
| 47 | +def required_token_info(info: TokenData = fastapi.Depends(token_info)): |
| 48 | + if info is None: |
| 49 | + raise fastapi.HTTPException( |
| 50 | + status_code=fastapi.status.HTTP_401_UNAUTHORIZED, |
| 51 | + detail="Not authenticated", |
| 52 | + headers={"www-authenticate": "Bearer"}, |
| 53 | + ) |
| 54 | + |
| 55 | + return info |
0 commit comments