136 lines
4.2 KiB
Python
136 lines
4.2 KiB
Python
from functools import lru_cache
|
|
import json
|
|
from typing import Any
|
|
from pydantic import Field,PrivateAttr
|
|
from pydantic_settings import BaseSettings,SettingsConfigDict
|
|
from src.infrastructure.vault import VaultClient
|
|
|
|
|
|
class Settings(BaseSettings):
|
|
model_config = SettingsConfigDict(env_file='.env',extra='ignore')
|
|
_vault_beorg_secrets: dict[str,Any] = PrivateAttr(default_factory=dict)
|
|
_vault_database_secrets: dict[str,Any] = PrivateAttr(default_factory=dict)
|
|
|
|
DOCS_USERNAME: str = 'admin'
|
|
DOCS_PASSWORD: str = 'admin'
|
|
VAULT_ADDR: str = 'https://corp.vault.elcsa.ru'
|
|
VAULT_ROLE_ID: str = ''
|
|
VAULT_SECRET_ID: str = ''
|
|
VAULT_NAMESPACE: str | None = None
|
|
VAULT_MOUNT_POINT: str = 'dev-secrets'
|
|
VAULT_BEORG_SECRET_PATH: str = 'beorg'
|
|
VAULT_DATABASE_SECRET_PATH: str = 'database'
|
|
VAULT_JWT_SECRET_PATH: str = 'jwt'
|
|
VAULT_DOCS_SECRET_PATH: str = 'docs'
|
|
VAULT_JWT_KID_PATH: str = 'jwt/kid'
|
|
VAULT_JWT_KIDS_PREFIX: str = 'jwt/kids'
|
|
LOG_FORMAT: str = 'json'
|
|
LOG_LEVEL: str = 'INFO'
|
|
JWT_KEYS_REFRESH_SECONDS: int = 300
|
|
JWT_ALGORITHM: str = 'RS256'
|
|
JWT_AUDIENCE: str | None = None
|
|
JWT_ISSUER: str | None = None
|
|
DATABASE_POOL_SIZE: int = 5
|
|
DATABASE_MAX_OVERFLOW: int = 10
|
|
DATABASE_POOL_TIMEOUT: int = 30
|
|
DATABASE_POOL_RECYCLE: int = 1800
|
|
DATABASE_ECHO: bool = False
|
|
EXCLUDED_PATHS: tuple[str,...] = ('/docs','/redoc','/openapi.json','/ping')
|
|
BEORG_TIMEOUT: int = 15
|
|
BEORG_PROCESS_INFO: list[dict[str,Any]] = Field(default_factory=lambda: [
|
|
{
|
|
'key': 'SELFIE1',
|
|
'type': 'SELFIE',
|
|
'options': {
|
|
'stages': [
|
|
'biometry_liveness',
|
|
],
|
|
},
|
|
'attempts': 3,
|
|
},
|
|
{
|
|
'key': 'PASSPORT1',
|
|
'type': 'PASSPORT',
|
|
'options': {
|
|
'stages': [
|
|
'verification',
|
|
'biometry_match',
|
|
],
|
|
'relation': {
|
|
'biometry_match': 'SELFIE1',
|
|
},
|
|
},
|
|
'attempts': 3,
|
|
},
|
|
])
|
|
|
|
|
|
@property
|
|
def DATABASE_URL(self) -> str:
|
|
return self._get_vault_secret(self._vault_database_secrets,'DATABASE_URL','database_url')
|
|
|
|
|
|
@property
|
|
def BEORG_PROJECT_ID(self) -> str:
|
|
return self._get_beorg_secret('project_id','BEORG_PROJECT_ID')
|
|
|
|
|
|
@property
|
|
def BEORG_MACHINE_UID(self) -> str:
|
|
return self._get_beorg_secret('machine_uid','BEORG_MACHINE_UID')
|
|
|
|
|
|
@property
|
|
def BEORG_TOKEN(self) -> str:
|
|
return self._get_beorg_secret('token','BEORG_TOKEN')
|
|
|
|
|
|
def _get_beorg_secret(self,*keys: str) -> str:
|
|
return self._get_vault_secret(self._vault_beorg_secrets,*keys)
|
|
|
|
|
|
def _get_vault_secret(self,secrets: dict[str,Any],*keys: str) -> str:
|
|
for key in keys:
|
|
value = secrets.get(key)
|
|
if value is not None:
|
|
return str(value)
|
|
return ''
|
|
|
|
|
|
def model_post_init(self,__context: Any) -> None:
|
|
if not self.VAULT_ROLE_ID or not self.VAULT_SECRET_ID:
|
|
return
|
|
|
|
client = VaultClient(
|
|
addr=self.VAULT_ADDR,
|
|
role_id=self.VAULT_ROLE_ID,
|
|
secret_id=self.VAULT_SECRET_ID,
|
|
namespace=self.VAULT_NAMESPACE,
|
|
mount_point=self.VAULT_MOUNT_POINT,
|
|
)
|
|
object.__setattr__(self,'_vault_beorg_secrets',client.read_many(self.VAULT_BEORG_SECRET_PATH))
|
|
object.__setattr__(self,'_vault_database_secrets',client.read_many(self.VAULT_DATABASE_SECRET_PATH))
|
|
secrets = client.read_many(
|
|
self.VAULT_BEORG_SECRET_PATH,
|
|
self.VAULT_DATABASE_SECRET_PATH,
|
|
self.VAULT_JWT_SECRET_PATH,
|
|
self.VAULT_DOCS_SECRET_PATH,
|
|
)
|
|
for field in type(self).model_fields:
|
|
if field.startswith('VAULT_'):
|
|
continue
|
|
value = secrets.get(field,secrets.get(field.lower()))
|
|
if value is None:
|
|
continue
|
|
if field == 'BEORG_PROCESS_INFO' and isinstance(value,str):
|
|
value = json.loads(value)
|
|
object.__setattr__(self,field,value)
|
|
|
|
|
|
@lru_cache
|
|
def get_settings() -> Settings:
|
|
return Settings()
|
|
|
|
|
|
settings = get_settings()
|