1  from abc import ABC
  2  from typing import ClassVar
  3  from typing import Dict
  4  
  5  import uuid6
  6  from litestar import Request
  7  from litestar.connection import ASGIConnection
  8  from litestar.di import Provide
  9  from litestar.handlers import BaseRouteHandler
 10  from litestar.types import ASGIApp
 11  from litestar.types import Receive
 12  from litestar.types import Scope
 13  from litestar.types import Send
 14  
 15  from evidently._pydantic_compat import SecretStr
 16  from evidently.pydantic_utils import register_type_alias
 17  from evidently.ui.service.components.base import Component
 18  from evidently.ui.service.components.base import ComponentContext
 19  from evidently.ui.service.errors import NotEnoughPermissions
 20  from evidently.ui.service.security.service import SecurityService
 21  from evidently.ui.service.storage.common import NoopAuthManager
 22  from evidently.ui.service.type_aliases import ZERO_UUID
 23  from evidently.ui.service.type_aliases import OrgID
 24  from evidently.ui.service.type_aliases import UserID
 25  
 26  
 27  class SecurityComponent(Component, ABC):
 28      add_security_middleware: ClassVar[bool] = True
 29  
 30      class Config:
 31          is_base_type = True
 32  
 33      def get_security(self) -> SecurityService:
 34          raise NotImplementedError
 35  
 36      def get_security_middleware(self, ctx: ComponentContext):
 37          security = self.get_security()
 38  
 39          def auth_middleware_factory(app: ASGIApp) -> ASGIApp:
 40              async def middleware(scope: Scope, receive: Receive, send: Send) -> None:
 41                  request: Request = Request(scope)
 42                  auth = security.authenticate(request)
 43                  if auth is None:
 44                      scope["auth"] = {
 45                          "authenticated": False,
 46                      }
 47                  else:
 48                      scope["auth"] = {
 49                          "user_id": auth.id,
 50                          "authenticated": True,
 51                      }
 52                  await app(scope, receive, send)
 53  
 54              return middleware
 55  
 56          return auth_middleware_factory
 57  
 58      def get_middlewares(self, ctx: ComponentContext):
 59          if self.add_security_middleware:
 60              return [self.get_security_middleware(ctx)]
 61          return []
 62  
 63      def get_auth_guard(self):
 64          def is_authenticated(connection: ASGIConnection, _: BaseRouteHandler) -> None:
 65              if not connection.scope["auth"]["authenticated"]:
 66                  raise NotEnoughPermissions()
 67  
 68          return is_authenticated
 69  
 70  
 71  register_type_alias(SecurityComponent, "evidently.ui.service.components.security.NoSecurityComponent", "none")
 72  register_type_alias(SecurityComponent, "evidently.ui.service.components.security.TokenSecurityComponent", "token")
 73  
 74  
 75  async def get_user_id() -> UserID:
 76      return ZERO_UUID
 77  
 78  
 79  class SimpleSecurity(SecurityComponent):
 80      def get_dependencies(self, ctx: ComponentContext) -> Dict[str, Provide]:
 81          return {
 82              "user_id": Provide(get_user_id),
 83              "security": Provide(self.get_security, sync_to_thread=False, use_cache=True),
 84              "security_config": Provide(lambda: self, sync_to_thread=False, use_cache=True),
 85              "auth_manager": Provide(lambda: NoopAuthManager(), sync_to_thread=False, use_cache=True),
 86          }
 87  
 88  
 89  class NoSecurityComponent(SimpleSecurity):
 90      class Config:
 91          type_alias = "none"
 92  
 93      dummy_user_id: UserID = uuid6.UUID(int=1, version=7)
 94      dummy_org_id: OrgID = uuid6.UUID(int=2, version=7)
 95  
 96      def get_security(self) -> SecurityService:
 97          from evidently.ui.service.security.no_security import NoSecurityService
 98  
 99          return NoSecurityService(self)
100  
101  
102  class TokenSecurityComponent(SimpleSecurity):
103      class Config:
104          type_alias = "token"
105  
106      token: SecretStr
107  
108      def get_security(self) -> SecurityService:
109          from evidently.ui.service.security.token import TokenSecurity
110  
111          return TokenSecurity(self)