""" Application configuration with security improvements and validation """ import os import secrets from datetime import datetime from typing import List, Optional, Dict, Any from pathlib import Path from pydantic import validator, Field from pydantic_settings import BaseSettings, SettingsConfigDict from pydantic.networks import AnyHttpUrl, PostgresDsn, RedisDsn from typing import Literal import structlog logger = structlog.get_logger(__name__) # --- Added env aliases to accept existing .env variable names --- try: from pydantic_settings import BaseSettings, SettingsConfigDict except Exception: from pydantic import BaseSettings # fallback try: from pydantic import Field except Exception: def Field(default=None, **kwargs): return default # Map old env names to model fields if names differ ENV_FIELD_ALIASES = { "postgres_db": "POSTGRES_DB", "postgres_user": "POSTGRES_USER", "postgres_password": "POSTGRES_PASSWORD", "node_id": "NODE_ID", "node_type": "NODE_TYPE", "node_version": "NODE_VERSION", "network_mode": "NETWORK_MODE", "allow_incoming_connections": "ALLOW_INCOMING_CONNECTIONS", "uvicorn_host": "UVICORN_HOST", "uvicorn_port": "UVICORN_PORT", "docker_sock_path": "DOCKER_SOCK_PATH", "node_private_key_path": "NODE_PRIVATE_KEY_PATH", "node_public_key_path": "NODE_PUBLIC_KEY_PATH", "node_public_key_hex": "NODE_PUBLIC_KEY_HEX", "bootstrap_config": "BOOTSTRAP_CONFIG", "max_peer_connections": "MAX_PEER_CONNECTIONS", "sync_interval": "SYNC_INTERVAL", "convert_max_parallel": "CONVERT_MAX_PARALLEL", "convert_timeout": "CONVERT_TIMEOUT", } def _apply_env_aliases(cls): for field, env in ENV_FIELD_ALIASES.items(): if field in getattr(cls, "__annotations__", {}): # Prefer Field with validation extras preserved current = getattr(cls, field, None) try: setattr(cls, field, Field(default=current if current is not None else None, validation_alias=env, alias=env)) except Exception: setattr(cls, field, current) return cls # --- End aliases block --- @_apply_env_aliases class Settings(BaseSettings): """Application settings with validation""" # Accept unknown env vars and allow no prefix model_config = SettingsConfigDict(extra='allow', env_prefix='') # Application PROJECT_NAME: str = "My Uploader Bot" PROJECT_VERSION: str = "2.0.0" PROJECT_HOST: AnyHttpUrl = Field(default="http://127.0.0.1:15100") SANIC_PORT: int = Field(default=15100, ge=1000, le=65535) DEBUG: bool = Field(default=False) # Security SECRET_KEY: str = Field(default_factory=lambda: secrets.token_urlsafe(32)) JWT_SECRET_KEY: str = Field(default_factory=lambda: secrets.token_urlsafe(32)) JWT_EXPIRE_MINUTES: int = Field(default=60 * 24 * 7) # 7 days ENCRYPTION_KEY: Optional[str] = None # Rate Limiting RATE_LIMIT_REQUESTS: int = Field(default=100) RATE_LIMIT_WINDOW: int = Field(default=60) # seconds RATE_LIMIT_ENABLED: bool = Field(default=True) # Database # Legacy compose fields (optional). If all three are present, they will be used to build DATABASE_URL. POSTGRES_DB: Optional[str] = Field(default=None, validation_alias="POSTGRES_DB", alias="POSTGRES_DB") POSTGRES_USER: Optional[str] = Field(default=None, validation_alias="POSTGRES_USER", alias="POSTGRES_USER") POSTGRES_PASSWORD: Optional[str] = Field(default=None, validation_alias="POSTGRES_PASSWORD", alias="POSTGRES_PASSWORD") DATABASE_URL: str = Field( default="postgresql+asyncpg://user:password@localhost:5432/uploader_bot", validation_alias="DATABASE_URL", alias="DATABASE_URL" ) DATABASE_POOL_SIZE: int = Field(default=10, ge=1, le=100) DATABASE_MAX_OVERFLOW: int = Field(default=20, ge=0, le=100) DATABASE_ECHO: bool = Field(default=False) # Redis REDIS_URL: RedisDsn = Field(default="redis://localhost:6379/0", validation_alias="REDIS_URL", alias="REDIS_URL") REDIS_POOL_SIZE: int = Field(default=10, ge=1, le=100) REDIS_TTL_DEFAULT: int = Field(default=3600) # 1 hour REDIS_TTL_SHORT: int = Field(default=300) # 5 minutes REDIS_TTL_LONG: int = Field(default=86400) # 24 hours # File Storage UPLOADS_DIR: Path = Field(default=Path("/app/data"), validation_alias="UPLOADS_DIR", alias="UPLOADS_DIR") MAX_FILE_SIZE: int = Field(default=100 * 1024 * 1024) # 100MB ALLOWED_CONTENT_TYPES: List[str] = Field(default=[ 'image/jpeg', 'image/png', 'image/gif', 'image/webp', 'video/mp4', 'video/webm', 'video/ogg', 'video/quicktime', 'audio/mpeg', 'audio/ogg', 'audio/wav', 'audio/mp4', 'text/plain', 'application/json' ]) # Telegram TELEGRAM_API_KEY: Optional[str] = Field(default=None, validation_alias="TELEGRAM_API_KEY", alias="TELEGRAM_API_KEY") CLIENT_TELEGRAM_API_KEY: Optional[str] = Field(default=None, validation_alias="CLIENT_TELEGRAM_API_KEY", alias="CLIENT_TELEGRAM_API_KEY") TELEGRAM_WEBHOOK_ENABLED: bool = Field(default=False, validation_alias="TELEGRAM_WEBHOOK_ENABLED", alias="TELEGRAM_WEBHOOK_ENABLED") TELEGRAM_WEBHOOK_URL: Optional[str] = Field(default=None, validation_alias="TELEGRAM_WEBHOOK_URL", alias="TELEGRAM_WEBHOOK_URL") TELEGRAM_WEBHOOK_SECRET: str = Field(default_factory=lambda: secrets.token_urlsafe(32), validation_alias="TELEGRAM_WEBHOOK_SECRET", alias="TELEGRAM_WEBHOOK_SECRET") # TON Blockchain TESTNET: bool = Field(default=False, validation_alias="TESTNET", alias="TESTNET") TONCENTER_HOST: AnyHttpUrl = Field(default="https://toncenter.com/api/v2/", validation_alias="TONCENTER_HOST", alias="TONCENTER_HOST") TONCENTER_API_KEY: Optional[str] = Field(default=None, validation_alias="TONCENTER_API_KEY", alias="TONCENTER_API_KEY") TONCENTER_V3_HOST: AnyHttpUrl = Field(default="https://toncenter.com/api/v3/", validation_alias="TONCENTER_V3_HOST", alias="TONCENTER_V3_HOST") MY_PLATFORM_CONTRACT: str = Field(default="EQDmWp6hbJlYUrXZKb9N88sOrTit630ZuRijfYdXEHLtheMY", validation_alias="MY_PLATFORM_CONTRACT", alias="MY_PLATFORM_CONTRACT") MY_FUND_ADDRESS: str = Field(default="UQDarChHFMOI2On9IdHJNeEKttqepgo0AY4bG1trw8OAAwMY", validation_alias="MY_FUND_ADDRESS", alias="MY_FUND_ADDRESS") # Logging LOG_LEVEL: str = Field(default="INFO", pattern="^(DEBUG|INFO|WARNING|ERROR|CRITICAL)$", validation_alias="LOG_LEVEL", alias="LOG_LEVEL") LOG_DIR: Path = Field(default=Path("logs"), validation_alias="LOG_DIR", alias="LOG_DIR") LOG_FORMAT: str = Field(default="json", validation_alias="LOG_FORMAT", alias="LOG_FORMAT") LOG_ROTATION: str = Field(default="1 day", validation_alias="LOG_ROTATION", alias="LOG_ROTATION") LOG_RETENTION: str = Field(default="30 days", validation_alias="LOG_RETENTION", alias="LOG_RETENTION") # Monitoring METRICS_ENABLED: bool = Field(default=True, validation_alias="METRICS_ENABLED", alias="METRICS_ENABLED") METRICS_PORT: int = Field(default=9090, ge=1000, le=65535, validation_alias="METRICS_PORT", alias="METRICS_PORT") HEALTH_CHECK_ENABLED: bool = Field(default=True, validation_alias="HEALTH_CHECK_ENABLED", alias="HEALTH_CHECK_ENABLED") # --- Legacy/compose compatibility fields (env-driven) --- # Node identity/config NODE_ID: Optional[str] = Field(default=None, validation_alias="NODE_ID", alias="NODE_ID") NODE_TYPE: Optional[str] = Field(default=None, validation_alias="NODE_TYPE", alias="NODE_TYPE") NODE_VERSION: Optional[str] = Field(default=None, validation_alias="NODE_VERSION", alias="NODE_VERSION") NETWORK_MODE: Optional[str] = Field(default=None, validation_alias="NETWORK_MODE", alias="NETWORK_MODE") ALLOW_INCOMING_CONNECTIONS: Optional[bool] = Field(default=None, validation_alias="ALLOW_INCOMING_CONNECTIONS", alias="ALLOW_INCOMING_CONNECTIONS") # Uvicorn compatibility (compose overrides) UVICORN_HOST: Optional[str] = Field(default=None, validation_alias="UVICORN_HOST", alias="UVICORN_HOST") UVICORN_PORT: Optional[int] = Field(default=None, validation_alias="UVICORN_PORT", alias="UVICORN_PORT") # Docker socket path for converters DOCKER_SOCK_PATH: Optional[str] = Field(default=None, validation_alias="DOCKER_SOCK_PATH", alias="DOCKER_SOCK_PATH") # Keys and crypto paths NODE_PRIVATE_KEY_PATH: Optional[Path] = Field(default=None, validation_alias="NODE_PRIVATE_KEY_PATH", alias="NODE_PRIVATE_KEY_PATH") NODE_PUBLIC_KEY_PATH: Optional[Path] = Field(default=None, validation_alias="NODE_PUBLIC_KEY_PATH", alias="NODE_PUBLIC_KEY_PATH") NODE_PUBLIC_KEY_HEX: Optional[str] = Field(default=None, validation_alias="NODE_PUBLIC_KEY_HEX", alias="NODE_PUBLIC_KEY_HEX") # Bootstrap/runtime tuning BOOTSTRAP_CONFIG: Optional[str] = Field(default=None, validation_alias="BOOTSTRAP_CONFIG", alias="BOOTSTRAP_CONFIG") MAX_PEER_CONNECTIONS: Optional[int] = Field(default=None, validation_alias="MAX_PEER_CONNECTIONS", alias="MAX_PEER_CONNECTIONS") SYNC_INTERVAL: Optional[int] = Field(default=None, validation_alias="SYNC_INTERVAL", alias="SYNC_INTERVAL") CONVERT_MAX_PARALLEL: Optional[int] = Field(default=None, validation_alias="CONVERT_MAX_PARALLEL", alias="CONVERT_MAX_PARALLEL") CONVERT_TIMEOUT: Optional[int] = Field(default=None, validation_alias="CONVERT_TIMEOUT", alias="CONVERT_TIMEOUT") # --- Legacy/compose compatibility fields (env-driven) --- # Postgres (used by legacy compose; DATABASE_URL remains the primary DSN) postgres_db: Optional[str] = Field(default=None, validation_alias="POSTGRES_DB", alias="POSTGRES_DB") postgres_user: Optional[str] = Field(default=None, validation_alias="POSTGRES_USER", alias="POSTGRES_USER") postgres_password: Optional[str] = Field(default=None, validation_alias="POSTGRES_PASSWORD", alias="POSTGRES_PASSWORD") # Node identity/config node_id: Optional[str] = Field(default=None, validation_alias="NODE_ID", alias="NODE_ID") node_type: Optional[str] = Field(default=None, validation_alias="NODE_TYPE", alias="NODE_TYPE") node_version: Optional[str] = Field(default=None, validation_alias="NODE_VERSION", alias="NODE_VERSION") network_mode: Optional[str] = Field(default=None, validation_alias="NETWORK_MODE", alias="NETWORK_MODE") allow_incoming_connections: Optional[bool] = Field(default=None, validation_alias="ALLOW_INCOMING_CONNECTIONS", alias="ALLOW_INCOMING_CONNECTIONS") # Uvicorn compatibility (compose overrides) uvicorn_host: Optional[str] = Field(default=None, validation_alias="UVICORN_HOST", alias="UVICORN_HOST") uvicorn_port: Optional[int] = Field(default=None, validation_alias="UVICORN_PORT", alias="UVICORN_PORT") # Docker socket path for converters docker_sock_path: Optional[str] = Field(default=None, validation_alias="DOCKER_SOCK_PATH", alias="DOCKER_SOCK_PATH") # Keys and crypto paths node_private_key_path: Optional[Path] = Field(default=None, validation_alias="NODE_PRIVATE_KEY_PATH", alias="NODE_PRIVATE_KEY_PATH") node_public_key_path: Optional[Path] = Field(default=None, validation_alias="NODE_PUBLIC_KEY_PATH", alias="NODE_PUBLIC_KEY_PATH") node_public_key_hex: Optional[str] = Field(default=None, validation_alias="NODE_PUBLIC_KEY_HEX", alias="NODE_PUBLIC_KEY_HEX") # Bootstrap/runtime tuning bootstrap_config: Optional[str] = Field(default=None, validation_alias="BOOTSTRAP_CONFIG", alias="BOOTSTRAP_CONFIG") max_peer_connections: Optional[int] = Field(default=None, validation_alias="MAX_PEER_CONNECTIONS", alias="MAX_PEER_CONNECTIONS") sync_interval: Optional[int] = Field(default=None, validation_alias="SYNC_INTERVAL", alias="SYNC_INTERVAL") convert_max_parallel: Optional[int] = Field(default=None, validation_alias="CONVERT_MAX_PARALLEL", alias="CONVERT_MAX_PARALLEL") convert_timeout: Optional[int] = Field(default=None, validation_alias="CONVERT_TIMEOUT", alias="CONVERT_TIMEOUT") # Background Services INDEXER_ENABLED: bool = Field(default=True, validation_alias="INDEXER_ENABLED", alias="INDEXER_ENABLED") INDEXER_INTERVAL: int = Field(default=5, ge=1, le=3600, validation_alias="INDEXER_INTERVAL", alias="INDEXER_INTERVAL") TON_DAEMON_ENABLED: bool = Field(default=True, validation_alias="TON_DAEMON_ENABLED", alias="TON_DAEMON_ENABLED") TON_DAEMON_INTERVAL: int = Field(default=3, ge=1, le=3600, validation_alias="TON_DAEMON_INTERVAL", alias="TON_DAEMON_INTERVAL") LICENSE_SERVICE_ENABLED: bool = Field(default=True, validation_alias="LICENSE_SERVICE_ENABLED", alias="LICENSE_SERVICE_ENABLED") LICENSE_SERVICE_INTERVAL: int = Field(default=10, ge=1, le=3600, validation_alias="LICENSE_SERVICE_INTERVAL", alias="LICENSE_SERVICE_INTERVAL") CONVERT_SERVICE_ENABLED: bool = Field(default=True, validation_alias="CONVERT_SERVICE_ENABLED", alias="CONVERT_SERVICE_ENABLED") CONVERT_SERVICE_INTERVAL: int = Field(default=30, ge=1, le=3600, validation_alias="CONVERT_SERVICE_INTERVAL", alias="CONVERT_SERVICE_INTERVAL") # Web App URLs WEB_APP_URLS: Dict[str, str] = Field(default={ 'uploadContent': "https://web2-client.vercel.app/uploadContent" }, validation_alias="WEB_APP_URLS", alias="WEB_APP_URLS") # Maintenance MAINTENANCE_MODE: bool = Field(default=False, validation_alias="MAINTENANCE_MODE", alias="MAINTENANCE_MODE") MAINTENANCE_MESSAGE: str = Field(default="System is under maintenance", validation_alias="MAINTENANCE_MESSAGE", alias="MAINTENANCE_MESSAGE") # Development MOCK_EXTERNAL_SERVICES: bool = Field(default=False, validation_alias="MOCK_EXTERNAL_SERVICES", alias="MOCK_EXTERNAL_SERVICES") DISABLE_WEBHOOKS: bool = Field(default=False, validation_alias="DISABLE_WEBHOOKS", alias="DISABLE_WEBHOOKS") @validator('UPLOADS_DIR') def create_uploads_dir(cls, v): """Create uploads directory if it doesn't exist and is writable""" try: if not v.exists(): v.mkdir(parents=True, exist_ok=True) except (OSError, PermissionError) as e: # Handle read-only filesystem or permission errors logger.warning(f"Cannot create uploads directory {v}: {e}") # Use current directory as fallback fallback = Path("./data") try: fallback.mkdir(parents=True, exist_ok=True) return fallback except Exception: # Last fallback - current directory return Path(".") return v @validator('LOG_DIR') def create_log_dir(cls, v): """Create log directory if it doesn't exist and is writable""" try: if not v.exists(): v.mkdir(parents=True, exist_ok=True) except (OSError, PermissionError) as e: # Handle read-only filesystem or permission errors logger.warning(f"Cannot create log directory {v}: {e}") # Use current directory as fallback fallback = Path("./logs") try: fallback.mkdir(parents=True, exist_ok=True) return fallback except Exception: # Last fallback - current directory return Path(".") return v @validator('DATABASE_URL', pre=True, always=True) def build_database_url_from_parts(cls, v, values): """If DATABASE_URL is default and POSTGRES_* are provided, build DSN from parts.""" try: default_mark = "user:password@localhost:5432/uploader_bot" if (not v) or default_mark in str(v): db = values.get('POSTGRES_DB') or os.getenv('POSTGRES_DB') user = values.get('POSTGRES_USER') or os.getenv('POSTGRES_USER') pwd = values.get('POSTGRES_PASSWORD') or os.getenv('POSTGRES_PASSWORD') if db and user and pwd: return f"postgresql+asyncpg://{user}:{pwd}@postgres:5432/{db}" except Exception: pass return v @validator('DATABASE_URL') def validate_database_url(cls, v): """Validate database URL format - allow SQLite for testing""" v_str = str(v) if not (v_str.startswith('postgresql+asyncpg://') or v_str.startswith('sqlite+aiosqlite://')): logger.warning(f"Using non-standard database URL: {v_str}") return v @validator('TELEGRAM_API_KEY', 'CLIENT_TELEGRAM_API_KEY') def validate_telegram_keys(cls, v): """ Validate Telegram bot tokens format if provided. Empty/None values are allowed to run the app without Telegram bots. """ if v in (None, "", " "): return None v = v.strip() # Allow common dev-pattern tokens if v.startswith('1234567890:'): return v parts = v.split(':') if len(parts) != 2 or not parts[0].isdigit() or len(parts[1]) != 35: raise ValueError('Invalid Telegram bot token format') return v @validator('SECRET_KEY', 'JWT_SECRET_KEY') def validate_secret_keys(cls, v): """Validate secret keys length""" if len(v) < 32: raise ValueError('Secret keys must be at least 32 characters long') return v model_config = { "env_file": ".env", "case_sensitive": True, "validate_assignment": True, "extra": "allow" # Allow extra fields from environment } class SecurityConfig: """Security-related configurations""" # CORS settings CORS_ORIGINS = [ "https://web2-client.vercel.app", "https://t.me", "https://web.telegram.org" ] # Content Security Policy CSP_DIRECTIVES = { 'default-src': ["'self'"], 'script-src': ["'self'", "'unsafe-inline'", "https://cdn.jsdelivr.net"], 'style-src': ["'self'", "'unsafe-inline'", "https://cdn.jsdelivr.net"], 'img-src': ["'self'", "data:", "https:"], 'connect-src': ["'self'", "https://api.telegram.org"], 'frame-ancestors': ["'none'"], 'form-action': ["'self'"], 'base-uri': ["'self'"] } # Request size limits MAX_REQUEST_SIZE = 100 * 1024 * 1024 # 100MB MAX_JSON_SIZE = 10 * 1024 * 1024 # 10MB # Session settings SESSION_COOKIE_SECURE = True SESSION_COOKIE_HTTPONLY = True SESSION_COOKIE_SAMESITE = "Strict" # Rate limiting patterns RATE_LIMIT_PATTERNS = { "auth": {"requests": 5, "window": 300}, # 5 requests per 5 minutes "upload": {"requests": 10, "window": 3600}, # 10 uploads per hour "api": {"requests": 100, "window": 60}, # 100 API calls per minute "heavy": {"requests": 1, "window": 60} # 1 heavy operation per minute } # Create settings instance settings = Settings() # Expose commonly used settings DATABASE_URL = str(settings.DATABASE_URL) REDIS_URL = str(settings.REDIS_URL) DATABASE_POOL_SIZE = settings.DATABASE_POOL_SIZE DATABASE_MAX_OVERFLOW = settings.DATABASE_MAX_OVERFLOW REDIS_POOL_SIZE = settings.REDIS_POOL_SIZE TELEGRAM_API_KEY = settings.TELEGRAM_API_KEY or "" CLIENT_TELEGRAM_API_KEY = settings.CLIENT_TELEGRAM_API_KEY or "" PROJECT_HOST = str(settings.PROJECT_HOST) SANIC_PORT = settings.SANIC_PORT UPLOADS_DIR = settings.UPLOADS_DIR ALLOWED_CONTENT_TYPES = settings.ALLOWED_CONTENT_TYPES TESTNET = settings.TESTNET TONCENTER_HOST = str(settings.TONCENTER_HOST) TONCENTER_API_KEY = settings.TONCENTER_API_KEY TONCENTER_V3_HOST = str(settings.TONCENTER_V3_HOST) MY_PLATFORM_CONTRACT = settings.MY_PLATFORM_CONTRACT MY_FUND_ADDRESS = settings.MY_FUND_ADDRESS LOG_LEVEL = settings.LOG_LEVEL LOG_DIR = settings.LOG_DIR MAINTENANCE_MODE = settings.MAINTENANCE_MODE # Cache keys patterns CACHE_KEYS = { "user_session": "user:session:{user_id}", "user_data": "user:data:{user_id}", "content_metadata": "content:meta:{content_id}", "rate_limit": "rate_limit:{pattern}:{identifier}", "blockchain_task": "blockchain:task:{task_id}", "temp_upload": "upload:temp:{upload_id}", "wallet_connection": "wallet:conn:{wallet_address}", "ton_price": "ton:price:usd", "system_status": "system:status:{service}", } # Log current configuration (without secrets) def log_config(): """Log current configuration without sensitive data""" safe_config = { "project_name": settings.PROJECT_NAME, "project_version": settings.PROJECT_VERSION, "debug": settings.DEBUG, "sanic_port": settings.SANIC_PORT, "testnet": settings.TESTNET, "maintenance_mode": settings.MAINTENANCE_MODE, "metrics_enabled": settings.METRICS_ENABLED, "uploads_dir": str(settings.UPLOADS_DIR), "log_level": settings.LOG_LEVEL, } logger.info("Configuration loaded", **safe_config) # Initialize logging configuration log_config() # Функция для получения настроек (для совместимости с остальным кодом) def get_settings() -> Settings: """ Получить экземпляр настроек приложения. Returns: Settings: Конфигурация приложения """ return settings