Spaces:
Runtime error
Runtime error
File size: 2,834 Bytes
3763483 168aee7 2bdcb14 168aee7 48449c0 3763483 168aee7 3763483 2bdcb14 168aee7 2bdcb14 168aee7 48449c0 168aee7 3763483 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 |
"""Authentication and ACL configuration."""
from typing import Any, Optional
from fastapi import HTTPException, Request, Response
from pydantic import BaseModel, ValidationError
from .config import CONFIG
LILAC_AUTH_ENABLED = CONFIG.get('LILAC_AUTH_ENABLED', False)
class AuthenticationMiddleware:
"""Middleware that catches the py authentication errors and returns a 401."""
async def __call__(self, request: Request, call_next: Any) -> Response:
"""Call the middleware."""
try:
response = await call_next(request)
except ConceptAuthorizationException as e:
raise HTTPException(status_code=401, detail='User does not have access.') from e
return response
class ConceptAuthorizationException(Exception):
"""Authorization exceptions thrown by the concept database."""
pass
class DatasetUserAccess(BaseModel):
"""User access for datasets."""
# Whether the user can compute a signal.
compute_signals: bool
# Whether the user can delete a dataset.
delete_dataset: bool
# Whether the user can delete a signal.
delete_signals: bool
# Whether the user can update settings.
update_settings: bool
class ConceptUserAccess(BaseModel):
"""User access for concepts."""
# Whether the user can delete any concept (not their own).
delete_any_concept: bool
class UserAccess(BaseModel):
"""User access."""
create_dataset: bool
# TODO(nsthorat): Make this keyed to each dataset and concept.
dataset: DatasetUserAccess
concept: ConceptUserAccess
class UserInfo(BaseModel):
"""User information."""
id: str
email: str
name: str
given_name: str
family_name: str
class AuthenticationInfo(BaseModel):
"""Authentication information for the user."""
user: Optional[UserInfo]
access: UserAccess
auth_enabled: bool
def get_session_user(request: Request) -> Optional[UserInfo]:
"""Get the user from the session."""
if not LILAC_AUTH_ENABLED:
return None
user_info_dict = request.session.get('user', None)
if user_info_dict:
try:
return UserInfo.parse_obj(user_info_dict)
except ValidationError:
return None
return None
def get_user_access() -> UserAccess:
"""Get the user access."""
auth_enabled = CONFIG.get('LILAC_AUTH_ENABLED', False)
if isinstance(auth_enabled, str):
auth_enabled = auth_enabled.lower() == 'true'
if auth_enabled:
return UserAccess(
create_dataset=False,
dataset=DatasetUserAccess(
compute_signals=False, delete_dataset=False, delete_signals=False, update_settings=False),
concept=ConceptUserAccess(delete_any_concept=False))
return UserAccess(
create_dataset=True,
dataset=DatasetUserAccess(
compute_signals=True, delete_dataset=True, delete_signals=True, update_settings=True),
concept=ConceptUserAccess(delete_any_concept=True))
|