2022-11-06 18:42:33 +08:00
|
|
|
import base64
|
|
|
|
import binascii
|
|
|
|
|
2023-04-10 20:23:31 +08:00
|
|
|
from api.gql.deps import get_current_active_user
|
2022-11-06 20:09:44 +08:00
|
|
|
from starlette.authentication import AuthCredentials # UnauthenticatedUser,
|
2023-03-19 23:21:32 +08:00
|
|
|
from starlette.authentication import (
|
|
|
|
AuthenticationBackend,
|
|
|
|
AuthenticationError,
|
|
|
|
SimpleUser,
|
|
|
|
)
|
|
|
|
|
2022-11-06 18:42:33 +08:00
|
|
|
|
|
|
|
class FelicityAuthBackend(AuthenticationBackend):
|
|
|
|
async def authenticate(self, request):
|
|
|
|
if "Authorization" not in request.headers:
|
|
|
|
return
|
|
|
|
|
|
|
|
auth = request.headers["Authorization"]
|
|
|
|
try:
|
|
|
|
scheme, credentials = auth.split()
|
|
|
|
if scheme.lower() == "basic":
|
|
|
|
decoded = base64.b64decode(credentials).decode("ascii")
|
|
|
|
username, _, password = decoded.partition(":")
|
|
|
|
# TODO: You'd want to verify the username and password here if needed
|
|
|
|
elif scheme.lower() == "bearer":
|
2023-03-19 23:21:32 +08:00
|
|
|
""" "get is active user from token"""
|
2022-11-06 18:42:33 +08:00
|
|
|
user = await get_current_active_user(credentials)
|
|
|
|
username, _, password = user.auth.user_name, None, None
|
|
|
|
else:
|
|
|
|
raise AuthenticationError(
|
|
|
|
f"UnKnown Authentication Backend: {scheme.lower()}"
|
|
|
|
)
|
|
|
|
|
|
|
|
return AuthCredentials(["authenticated"]), SimpleUser(username)
|
|
|
|
|
|
|
|
except (ValueError, UnicodeDecodeError, binascii.Error) as exc:
|
|
|
|
raise AuthenticationError(f"Invalid auth credentials: {exc}")
|