File size: 2,313 Bytes
3763483
 
c14732f
2bdcb14
c14732f
48449c0
3763483
c14732f
168aee7
 
 
 
 
 
3763483
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
2bdcb14
 
168aee7
2bdcb14
 
 
 
 
 
 
 
2226ee3
2bdcb14
 
 
 
168aee7
 
c14732f
168aee7
 
 
48449c0
 
 
 
168aee7
 
 
3763483
 
c14732f
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
"""Authentication and ACL configuration."""

from typing import Optional

from fastapi import Request
from pydantic import BaseModel, ValidationError

from .config import env


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] = None
  access: UserAccess
  auth_enabled: bool


def get_session_user(request: Request) -> Optional[UserInfo]:
  """Get the user from the session."""
  if not env('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 = env('LILAC_AUTH_ENABLED')
  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))