import base64 from functools import lru_cache from pathlib import Path from urllib.parse import urlparse from pydantic import Field from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): model_config = SettingsConfigDict(env_file=".env", extra="ignore") app_env: str = Field(default="development", pattern=r"^(development|test|production)$") app_base_url: str = "http://localhost:8080" app_secret: str | None = Field(default=None, min_length=32) app_secret_file: str | None = None owner_bootstrap_token: str | None = Field(default=None, min_length=32) owner_bootstrap_token_file: str | None = None database_url: str redis_url: str ai_service_url: str = "http://ai-agent:8001" ai_service_token: str | None = Field(default=None, min_length=32) ai_service_token_file: str | None = None computer_service_url: str = "http://computer-agent:8002" computer_service_token: str | None = Field(default=None, min_length=32) computer_service_token_file: str | None = None computer_capability_private_key: str | None = None computer_capability_private_key_file: str | None = None computer_capability_key_id: str = Field(default="computer-capability-v1", min_length=1, max_length=80) compute_provider: str = Field(default="unconfigured", pattern=r"^(unconfigured|libvirt_gateway|proxmox_gateway|cloud_gateway)$") compute_provider_url: str | None = None compute_provider_token: str | None = Field(default=None, min_length=32) compute_provider_token_file: str | None = None compute_provider_timeout_seconds: float = Field(default=30.0, ge=1, le=120) browser_profile_root: str = "/data/browser-profiles" data_encryption_key: str | None = None data_encryption_key_file: str | None = None data_encryption_key_version: int = Field(default=1, ge=1) session_idle_minutes: int = Field(default=30, ge=5, le=1440) password_reset_minutes: int = Field(default=15, ge=5, le=60) mfa_issuer: str = Field(default="Hayva.ai", min_length=1, max_length=80) maintenance_interval_seconds: int = Field(default=60, ge=10, le=3600) maintenance_batch_size: int = Field(default=100, ge=1, le=1000) log_level: str = Field(default="INFO", pattern=r"^(DEBUG|INFO|WARNING|ERROR|CRITICAL)$") def read_app_secret(self) -> str | None: value = self.app_secret if self.app_secret_file: try: value = Path(self.app_secret_file).read_text(encoding="utf-8").strip() except OSError: return None return value if value and len(value) >= 32 else None def require_app_secret(self) -> str: value = self.read_app_secret() if not value: raise RuntimeError("Application secret is unconfigured") return value def read_owner_bootstrap_token(self) -> str | None: if self.owner_bootstrap_token_file: try: token = Path(self.owner_bootstrap_token_file).read_text(encoding="utf-8").strip() except OSError: return None return token if len(token) >= 32 else None return self.owner_bootstrap_token def read_ai_service_token(self) -> str | None: if self.ai_service_token_file: try: token = Path(self.ai_service_token_file).read_text(encoding="utf-8").strip() except OSError: return None return token if len(token) >= 32 else None return self.ai_service_token def read_computer_service_token(self) -> str | None: if self.computer_service_token_file: try: token = Path(self.computer_service_token_file).read_text(encoding="utf-8").strip() except OSError: return None return token if len(token) >= 32 else None return self.computer_service_token def read_compute_provider_token(self) -> str | None: if self.compute_provider_token_file: try: token = Path(self.compute_provider_token_file).read_text( encoding="utf-8" ).strip() except OSError: return None return token if len(token) >= 32 else None return self.compute_provider_token def read_data_encryption_key(self) -> bytes | None: encoded = self.data_encryption_key if self.data_encryption_key_file: try: encoded = Path(self.data_encryption_key_file).read_text(encoding="utf-8").strip() except OSError: return None if not encoded: return None try: key = base64.urlsafe_b64decode(encoded + "=" * (-len(encoded) % 4)) except (ValueError, TypeError): return None return key if len(key) == 32 else None def read_computer_capability_private_key(self) -> bytes | None: encoded = self.computer_capability_private_key if self.computer_capability_private_key_file: try: encoded = Path(self.computer_capability_private_key_file).read_text( encoding="utf-8" ).strip() except OSError: return None if not encoded: return None try: key = base64.urlsafe_b64decode(encoded + "=" * (-len(encoded) % 4)) except (ValueError, TypeError): return None return key if len(key) == 32 else None def production_configuration_errors(self) -> list[str]: """Return safe, non-secret reasons production startup must remain blocked.""" if self.app_env != "production": return [] errors: list[str] = [] required_secret_files = { "APP_SECRET_FILE": (self.app_secret_file, self.read_app_secret()), "AI_SERVICE_TOKEN_FILE": ( self.ai_service_token_file, self.read_ai_service_token() ), "COMPUTER_SERVICE_TOKEN_FILE": ( self.computer_service_token_file, self.read_computer_service_token() ), "COMPUTER_CAPABILITY_PRIVATE_KEY_FILE": ( self.computer_capability_private_key_file, self.read_computer_capability_private_key(), ), "DATA_ENCRYPTION_KEY_FILE": ( self.data_encryption_key_file, self.read_data_encryption_key() ), "COMPUTE_PROVIDER_TOKEN_FILE": ( self.compute_provider_token_file, self.read_compute_provider_token() ), } for label, (path, value) in required_secret_files.items(): if not path or value is None: errors.append(f"{label} must reference valid mounted production material") database = urlparse(self.database_url) if ( database.scheme != "postgresql+asyncpg" or not database.hostname or not database.username or not database.password or _is_placeholder(database.username) or _is_placeholder(database.password) ): errors.append("DATABASE_URL must use authenticated postgresql+asyncpg") redis = urlparse(self.redis_url) if ( redis.scheme not in {"redis", "rediss"} or not redis.hostname or not redis.username or not redis.password or _is_placeholder(redis.username) or _is_placeholder(redis.password) ): errors.append("REDIS_URL must use an authenticated Redis identity") public_origin = urlparse(self.app_base_url) if ( public_origin.scheme != "https" or not public_origin.hostname or public_origin.username or public_origin.password or public_origin.query or public_origin.fragment or public_origin.path not in {"", "/"} or public_origin.hostname.lower() in {"localhost", "localhost.localdomain"} or public_origin.hostname.lower().endswith(".local") ): errors.append("APP_BASE_URL must be a public HTTPS origin") if self.compute_provider == "unconfigured": errors.append("COMPUTE_PROVIDER must select a real VM gateway") provider = urlparse(self.compute_provider_url or "") if provider.scheme != "https" or not provider.hostname or provider.username or provider.password: errors.append("COMPUTE_PROVIDER_URL must be an HTTPS URL without userinfo") return errors def require_production_configuration(self) -> None: errors = self.production_configuration_errors() if errors: raise RuntimeError("Production configuration is unsafe: " + "; ".join(errors)) def _is_placeholder(value: str) -> bool: normalized = value.strip().lower() return normalized in {"replace-me", "password", "changeme", "unconfigured"} or ( normalized.startswith("replace-") ) @lru_cache def get_settings() -> Settings: return Settings() # type: ignore[call-arg]