mirror of
https://github.com/OpenHands/OpenHands.git
synced 2025-12-26 05:48:36 +08:00
81 lines
2.8 KiB
Python
81 lines
2.8 KiB
Python
from dataclasses import dataclass
|
|
|
|
from fastapi import Request
|
|
from pydantic import SecretStr
|
|
|
|
from openhands.integrations.provider import PROVIDER_TOKEN_TYPE
|
|
from openhands.server import shared
|
|
from openhands.server.settings import Settings
|
|
from openhands.server.user_auth.user_auth import UserAuth
|
|
from openhands.storage.data_models.user_secrets import UserSecrets
|
|
from openhands.storage.settings.secret_store import SecretsStore
|
|
from openhands.storage.settings.settings_store import SettingsStore
|
|
|
|
|
|
@dataclass
|
|
class DefaultUserAuth(UserAuth):
|
|
"""Default user authentication mechanism"""
|
|
|
|
_settings: Settings | None = None
|
|
_settings_store: SettingsStore | None = None
|
|
_secrets_store: SecretsStore | None = None
|
|
_user_secrets: UserSecrets | None = None
|
|
|
|
async def get_user_id(self) -> str | None:
|
|
"""The default implementation does not support multi tenancy, so user_id is always None"""
|
|
return None
|
|
|
|
async def get_access_token(self) -> SecretStr | None:
|
|
"""The default implementation does not support multi tenancy, so access_token is always None"""
|
|
return None
|
|
|
|
async def get_user_settings_store(self):
|
|
settings_store = self._settings_store
|
|
if settings_store:
|
|
return settings_store
|
|
user_id = await self.get_user_id()
|
|
settings_store = await shared.SettingsStoreImpl.get_instance(
|
|
shared.config, user_id
|
|
)
|
|
self._settings_store = settings_store
|
|
return settings_store
|
|
|
|
async def get_user_settings(self) -> Settings | None:
|
|
settings = self._settings
|
|
if settings:
|
|
return settings
|
|
settings_store = await self.get_user_settings_store()
|
|
settings = await settings_store.load()
|
|
self._settings = settings
|
|
return settings
|
|
|
|
async def get_secrets_store(self):
|
|
secrets_store = self._secrets_store
|
|
if secrets_store:
|
|
return secrets_store
|
|
user_id = await self.get_user_id()
|
|
secret_store = await shared.SecretsStoreImpl.get_instance(
|
|
shared.config, user_id
|
|
)
|
|
self._secrets_store = secret_store
|
|
return secret_store
|
|
|
|
async def get_user_secrets(self) -> UserSecrets | None:
|
|
user_secrets = self._user_secrets
|
|
if user_secrets:
|
|
return user_secrets
|
|
secrets_store = await self.get_secrets_store()
|
|
user_secrets = await secrets_store.load()
|
|
self._user_secrets = user_secrets
|
|
return user_secrets
|
|
|
|
async def get_provider_tokens(self) -> PROVIDER_TOKEN_TYPE | None:
|
|
secrets_store = await self.get_user_secrets()
|
|
provider_tokens = getattr(secrets_store, 'provider_tokens', None)
|
|
return provider_tokens
|
|
|
|
@classmethod
|
|
async def get_instance(cls, request: Request) -> UserAuth:
|
|
user_auth = DefaultUserAuth()
|
|
return user_auth
|