server: implement user authentication (none by default)

server/reflector/auth/__init__.py

@@ -0,0 +1,13 @@
+from reflector.settings import settings
+from reflector.logger import logger
+import importlib
+
+logger.info(f"User authentication using {settings.AUTH_BACKEND}")
+module_name = f"reflector.auth.auth_{settings.AUTH_BACKEND}"
+auth_module = importlib.import_module(module_name)
+
+UserInfo = auth_module.UserInfo
+AccessTokenInfo = auth_module.AccessTokenInfo
+authenticated = auth_module.authenticated
+current_user = auth_module.current_user
+current_user_optional = auth_module.current_user_optional

server/reflector/auth/auth_fief.py

@@ -0,0 +1,25 @@
+from fastapi.security import OAuth2AuthorizationCodeBearer
+from fief_client import FiefAccessTokenInfo, FiefAsync, FiefUserInfo
+from fief_client.integrations.fastapi import FiefAuth
+from reflector.settings import settings
+
+fief = FiefAsync(
+    settings.AUTH_FIEF_URL,
+    settings.AUTH_FIEF_CLIENT_ID,
+    settings.AUTH_FIEF_CLIENT_SECRET,
+)
+
+scheme = OAuth2AuthorizationCodeBearer(
+    f"{settings.AUTH_FIEF_URL}/authorize",
+    f"{settings.AUTH_FIEF_URL}/api/token",
+    scopes={"openid": "openid", "offline_access": "offline_access"},
+    auto_error=False,
+)
+
+auth = FiefAuth(fief, scheme)
+
+UserInfo = FiefUserInfo
+AccessTokenInfo = FiefAccessTokenInfo
+authenticated = auth.authenticated()
+current_user = auth.current_user()
+current_user_optional = auth.current_user(optional=True)

server/reflector/auth/auth_none.py

@@ -0,0 +1,35 @@
+from pydantic import BaseModel
+from typing import Annotated
+from fastapi import Depends
+from fastapi.security import OAuth2PasswordBearer
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
+
+
+class UserInfo(BaseModel):
+    sub: str
+
+
+class AccessTokenInfo(BaseModel):
+    pass
+
+
+def authenticated(token: Annotated[str, Depends(oauth2_scheme)]):
+    def _authenticated():
+        return None
+
+    return _authenticated
+
+
+def current_user(token: Annotated[str, Depends(oauth2_scheme)]):
+    def _current_user():
+        return None
+
+    return _current_user
+
+
+def current_user_optional(token: Annotated[str, Depends(oauth2_scheme)]):
+    def _current_user_optional():
+        return None
+
+    return _current_user_optional