from __future__ import annotations import os from functools import lru_cache from typing import List,Literal from dotenv import find_dotenv,load_dotenv from pydantic import AliasChoices,Field,field_validator,model_validator from pydantic_settings import BaseSettings,SettingsConfigDict from src.infrastructure.vault import create_hvac_client_from_approle, read_kv2_secret env_file = find_dotenv('.env') if env_file: load_dotenv(env_file) def normalize_vault_base_url(raw: str) -> str: u = raw.strip().rstrip('/') if not u: return raw.strip() if '://' not in u: return f'https://{u}' return u class Settings(BaseSettings): VAULT_ADDR: str = Field(default='http://localhost:8200') VAULT_ROLE_ID: str = Field(..., description='AppRole role_id') VAULT_SECRET_ID: str = Field( ..., description='AppRole secret_id', validation_alias=AliasChoices('VAULT_SECRET_ID', 'VAULT_SECRET_TOKEN'), ) VAULT_NAMESPACE: str | None = Field(default=None) VAULT_MOUNT_POINT: str = Field(default='secrets') VAULT_JWT_KID_PATH: str = "jwt/kid" VAULT_JWT_KIDS_PREFIX: str = "jwt/kids" JWT_KEYS_REFRESH_SECONDS: int = 3600 DATABASE_HOST: str DATABASE_PORT: int = Field(default=5432, ge=1, le=65535) DATABASE_NAME: str DATABASE_USER: str DATABASE_PASSWORD: str DATABASE_POOL_SIZE: int = 10 DATABASE_MAX_OVERFLOW: int = 20 DATABASE_POOL_TIMEOUT: int = 30 DATABASE_POOL_RECYCLE: int = 3600 DATABASE_ECHO: bool = False CSRF_SECRET_KEY: str = Field( default="change-me-change-me-change-me-change-me", min_length=32, ) CSRF_COOKIE_SECURE: bool = False CSRF_COOKIE_HTTPONLY: bool = True CSRF_COOKIE_SAMESITE: Literal["Lax", "Strict", "None"] = "Lax" CSRF_COOKIE_PATH: str = "/" CSRF_COOKIE_DOMAIN: str | None = None CORS_ALLOW_ORIGIN_REGEX: str = r'https?://([a-z0-9-]+\.)*elcsa\.ru(:\d+)?$' DOCS_USERNAME: str = "admin" DOCS_PASSWORD: str = "admin" JWT_ACCESS_TTL_SECONDS: int = 15 * 60 JWT_REFRESH_TTL_SECONDS: int = 30 * 24 * 60 * 60 JWT_ISSUER: str | None = None JWT_AUDIENCE: str | None = None JWT_ALGORITHM: str = "RS256" REDIS_HOST: str = "localhost" REDIS_PORT: int = 6379 REDIS_PASSWORD: str | None = None REDIS_DB: int = 0 KEYDB_REMOTE_HOST: str | None = None KEYDB_REMOTE_PORT: int | None = None KEYDB_REMOTE_PASSWORD: str | None = None KEYDB_REMOTE_DB: int | None = None KEYDB_HOST: str | None = Field(default=None,validation_alias=AliasChoices('KEYDB_HOST','KEYDB_CACHE_HOST')) KEYDB_PORT: int | None = Field(default=None,validation_alias=AliasChoices('KEYDB_PORT','KEYDB_CACHE_PORT')) KEYDB_PASSWORD: str | None = Field(default=None,validation_alias=AliasChoices('KEYDB_PASSWORD','KEYDB_CACHE_PASSWORD')) KEYDB_DB: int | None = Field(default=None,validation_alias=AliasChoices('KEYDB_DB','KEYDB_CACHE_DB')) RABBIT_HOST: str = "localhost" RABBIT_PORT: int = 5672 RABBIT_USER: str = "guest" RABBIT_PASSWORD: str = "guest" RABBIT_VHOST: str = "/" RABBIT_PUBLISH_PERSIST: bool = True RABBIT_CONNECT_TIMEOUT: int = 5 RABBIT_EMAIL_CODE_QUEUE: str = "email.verification_code" RABBIT_CRYPTO_TRANSFER_QUEUE: str = "crypto.transfer.requested" RABBIT_CRYPTO_TRANSFER_COMPLETED_QUEUE: str = "crypto.transfer.completed" RABBIT_SBP_WITHDRAWAL_USDT_REQUESTED_QUEUE: str = 'sbp.withdrawal.usdt.requested' RABBIT_SBP_WITHDRAWAL_WALLET_EVENTS_QUEUE: str = 'sbp.withdrawal.wallet.events' ITPAY_PUBLIC_ID: str ITPAY_API_SECRET: str MOZEN_API_BASE_URL: str = 'https://extgw.mozen.io' MOZEN_API_SECRET: str = '' MOZEN_CLIENT_ID: str = '' MOZEN_ENDPOINT_ID: str = '' MOZEN_MERCHANT_ID: str = '' MOZEN_PARK_ID: str = '' MOZEN_PARK_TOKEN: str = '' CRYPTO_USDT_CONTRACT_ADDRESS: str = '' CLOUD_KASSIR_PUBLIC_ID: str = '' CLOUD_KASSIR_API_SECRET: str = '' LOG_LEVEL: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO" LOG_FORMAT: Literal["JSON", "TEXT"] = "TEXT" model_config = SettingsConfigDict( env_file=".env", env_file_encoding="utf-8", case_sensitive=True, extra="ignore", populate_by_name=True, ) @field_validator('VAULT_ADDR', mode='before') @classmethod def vault_addr_scheme(cls, v): if v is None or not isinstance(v, str): return v return normalize_vault_base_url(v) @model_validator(mode='before') @classmethod def load_from_vault(cls, data: dict): if not isinstance(data, dict): return data addr_raw = data.get('VAULT_ADDR') or os.getenv('VAULT_ADDR') or 'http://localhost:8200' addr = normalize_vault_base_url(addr_raw) data['VAULT_ADDR'] = addr role_id = data.get('VAULT_ROLE_ID') or os.getenv('VAULT_ROLE_ID') secret_id = ( data.get('VAULT_SECRET_ID') or data.get('VAULT_SECRET_TOKEN') or os.getenv('VAULT_SECRET_ID') or os.getenv('VAULT_SECRET_TOKEN') ) namespace = data.get('VAULT_NAMESPACE') if namespace is None: namespace = os.getenv('VAULT_NAMESPACE') namespace = namespace if namespace else None mount = data.get('VAULT_MOUNT_POINT') or os.getenv('VAULT_MOUNT_POINT') or 'secrets' if not role_id or not secret_id: raise RuntimeError('VAULT_ROLE_ID and VAULT_SECRET_ID (or VAULT_SECRET_TOKEN) are required for Vault AppRole') data['VAULT_ROLE_ID'] = str(role_id).strip() data['VAULT_SECRET_ID'] = str(secret_id).strip() client = create_hvac_client_from_approle( url=addr, role_id=role_id, secret_id=secret_id, namespace=namespace, timeout=5, ) def read_secret(path: str) -> dict: return read_kv2_secret(client=client, mount_point=mount, path=path) def read_secret_optional(path: str) -> dict: try: return read_secret(path) except Exception: return {} database = read_secret('database') csrf = read_secret_optional('csrf') rabbitmq = read_secret_optional('rabbitmq') db_ci = {str(k).lower(): v for k, v in database.items()} def db_nonempty(key: str) -> bool: v = db_ci.get(key) if v is None: return False if isinstance(v, str) and not v.strip(): return False return True required_db = ['host', 'name', 'user', 'password', 'port'] missing_db = [k for k in required_db if not db_nonempty(k)] if missing_db: raise RuntimeError(f'Vault secret database missing non-empty keys: {missing_db}') data['DATABASE_HOST'] = str(db_ci['host']).strip() data['DATABASE_PORT'] = int(db_ci['port']) data['DATABASE_NAME'] = str(db_ci['name']).strip() data['DATABASE_USER'] = str(db_ci['user']).strip() data['DATABASE_PASSWORD'] = str(db_ci['password']).strip() if csrf: csrf_secret = None for entry_key, entry_val in csrf.items(): if str(entry_key).lower() == 'key' and entry_val is not None and str(entry_val).strip(): csrf_secret = str(entry_val).strip() break if csrf_secret: data['CSRF_SECRET_KEY'] = csrf_secret if rabbitmq: r_ci = {str(k).lower(): v for k, v in rabbitmq.items()} def rb_set(field: str, env_key: str, *, as_int: bool = False) -> None: v = r_ci.get(field) if v is None: return if isinstance(v, str) and not v.strip(): return data[env_key] = int(v) if as_int else str(v).strip() rb_set('host', 'RABBIT_HOST') rb_set('port', 'RABBIT_PORT', as_int=True) rb_set('user', 'RABBIT_USER') rb_set('password', 'RABBIT_PASSWORD') rb_set('vhost', 'RABBIT_VHOST') extgw = read_secret_optional('extgw') if extgw: extgw_ci = {str(k).lower(): v for k, v in extgw.items()} def extgw_set(field: str, env_key: str) -> None: v = extgw_ci.get(field) if v is None: return if isinstance(v, str) and not v.strip(): return data[env_key] = str(v).strip() extgw_set('api_base_url', 'MOZEN_API_BASE_URL') extgw_set('api_secret', 'MOZEN_API_SECRET') extgw_set('client_id', 'MOZEN_CLIENT_ID') extgw_set('endpoint_id', 'MOZEN_ENDPOINT_ID') extgw_set('merchant_id', 'MOZEN_MERCHANT_ID') extgw_set('park_id', 'MOZEN_PARK_ID') extgw_set('park_token', 'MOZEN_PARK_TOKEN') crypto = read_secret_optional('crypto') if crypto: crypto_ci = {str(k).lower(): v for k, v in crypto.items()} usdt_contract_address = crypto_ci.get('usdt_contract_address') if usdt_contract_address is not None and str(usdt_contract_address).strip(): data['CRYPTO_USDT_CONTRACT_ADDRESS'] = str(usdt_contract_address).strip() keydb = read_secret('keydb') k_ci = {str(k).lower(): v for k, v in keydb.items()} def keydb_nonempty(key: str) -> bool: v = k_ci.get(key) if v is None: return False if isinstance(v, str) and not v.strip(): return False return True missing_keydb = [] for req in ('host', 'port'): if not keydb_nonempty(req): missing_keydb.append(req) db_raw = k_ci.get('database') if db_raw is None: db_raw = k_ci.get('db') if db_raw is None or (isinstance(db_raw, str) and not str(db_raw).strip()): missing_keydb.append('database') if missing_keydb: raise RuntimeError( f'Vault secret keydb missing non-empty keys: {missing_keydb} (mount={mount},path=keydb)' ) data['KEYDB_REMOTE_HOST'] = str(k_ci['host']).strip() data['KEYDB_REMOTE_PORT'] = int(k_ci['port']) data['KEYDB_REMOTE_DB'] = int(db_raw) pw_raw = k_ci.get('password') if pw_raw is not None and str(pw_raw).strip(): data['KEYDB_REMOTE_PASSWORD'] = str(pw_raw).strip() else: data['KEYDB_REMOTE_PASSWORD'] = None itpay_public_id = data.get('ITPAY_PUBLIC_ID') or os.getenv('ITPAY_PUBLIC_ID') itpay_api_secret = data.get('ITPAY_API_SECRET') or os.getenv('ITPAY_API_SECRET') if itpay_public_id is not None and str(itpay_public_id).strip() and itpay_api_secret is not None and str(itpay_api_secret).strip(): data['ITPAY_PUBLIC_ID'] = str(itpay_public_id).strip() data['ITPAY_API_SECRET'] = str(itpay_api_secret).strip() else: itpay = read_secret('itpay') itpay_ci = {str(k).lower(): v for k, v in itpay.items()} public_id = itpay_ci.get('public_id') api_secret = itpay_ci.get('api_secret') if api_secret is None: api_secret = itpay_ci.get('secret') missing = [] if public_id is None or not str(public_id).strip(): missing.append('public_id') if api_secret is None or not str(api_secret).strip(): missing.append('api_secret') if missing: raise RuntimeError(f'Vault secret itpay missing non-empty keys: {missing} (mount={mount},path=itpay)') data['ITPAY_PUBLIC_ID'] = str(public_id).strip() data['ITPAY_API_SECRET'] = str(api_secret).strip() ck_public = data.get('CLOUD_KASSIR_PUBLIC_ID') or os.getenv('CLOUD_KASSIR_PUBLIC_ID') ck_secret = data.get('CLOUD_KASSIR_API_SECRET') or os.getenv('CLOUD_KASSIR_API_SECRET') if ck_public is not None and str(ck_public).strip() and ck_secret is not None and str(ck_secret).strip(): data['CLOUD_KASSIR_PUBLIC_ID'] = str(ck_public).strip() data['CLOUD_KASSIR_API_SECRET'] = str(ck_secret).strip() else: cloudkassir = read_secret_optional('cloudkassir') if cloudkassir: ck_ci = {str(k).lower(): v for k, v in cloudkassir.items()} public_id_ck = ck_ci.get('public_id') api_secret_ck = ck_ci.get('api_secret') if api_secret_ck is None: api_secret_ck = ck_ci.get('secret') if public_id_ck is not None and str(public_id_ck).strip(): data['CLOUD_KASSIR_PUBLIC_ID'] = str(public_id_ck).strip() if api_secret_ck is not None and str(api_secret_ck).strip(): data['CLOUD_KASSIR_API_SECRET'] = str(api_secret_ck).strip() return data @property def DATABASE_URL(self) -> str: return ( f"postgresql+asyncpg://{self.DATABASE_USER}:{self.DATABASE_PASSWORD}" f"@{self.DATABASE_HOST}:{self.DATABASE_PORT}/{self.DATABASE_NAME}" ) @property def REDIS_URL(self) -> str: return self.KEYDB_MARKET_URL @staticmethod def _redis_url(*, host: str, port: int, password: str | None, db: int) -> str: auth = f':{password}@' if password else '' return f'redis://{auth}{host}:{port}/{db}' @property def KEYDB_MARKET_URL(self) -> str: if self.KEYDB_REMOTE_HOST is None or self.KEYDB_REMOTE_PORT is None or self.KEYDB_REMOTE_DB is None: raise RuntimeError('Vault KeyDB settings are required for market data') return self._redis_url( host=self.KEYDB_REMOTE_HOST, port=int(self.KEYDB_REMOTE_PORT), password=self.KEYDB_REMOTE_PASSWORD, db=int(self.KEYDB_REMOTE_DB), ) @property def KEYDB_REMOTE_URL(self) -> str: return self.KEYDB_MARKET_URL @property def KEYDB_CACHE_URL(self) -> str | None: if self.KEYDB_HOST is None or not self.KEYDB_HOST.strip(): return None if self.KEYDB_PORT is None: return None db = int(self.KEYDB_DB) if self.KEYDB_DB is not None else 0 return self._redis_url( host=self.KEYDB_HOST.strip(), port=int(self.KEYDB_PORT), password=self.KEYDB_PASSWORD, db=db, ) @property def RABBIT_URL(self) -> str: vhost = "%2F" if self.RABBIT_VHOST == "/" else self.RABBIT_VHOST.lstrip("/") return f"amqp://{self.RABBIT_USER}:{self.RABBIT_PASSWORD}@{self.RABBIT_HOST}:{self.RABBIT_PORT}/{vhost}" @property def EXCLUDED_PATHS(self) -> List[str]: return ["/docs", "/redoc", "/openapi.json", "/ping", "/health"] @lru_cache(maxsize=1) def get_settings() -> Settings: return Settings() settings = get_settings()