import os import secrets import sys import logging from urllib.parse import urlsplit, urlunsplit from pydantic_settings import BaseSettings, SettingsConfigDict _PLACEHOLDER_SECRETS = frozenset({ "scenarist-v6-secret-change-me", "scenarist-v7-secret-change-me", "scenarist123", "changeme", "change-me", "secret", "password", }) logger = logging.getLogger(__name__) _GENERATED_DEV_JWT_SECRET = secrets.token_urlsafe(48) class Settings(BaseSettings): model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8") app_name: str = "Orsync Scenarist Backend" environment: str = "development" # development | staging | production port: int = 7860 # Default to HF Spaces port log_level: str = "INFO" log_request_bodies: bool = True log_request_body_max_chars: int = 1200 # ── Infrastructure ──────────────────────────────────────────────── redis_url: str = "redis://localhost:6379/0" rabbitmq_url: str = "amqp://guest:guest@localhost:5672/" neo4j_uri: str = "bolt://localhost:7687" neo4j_username: str = "neo4j" neo4j_password: str = "" chroma_url: str = "" chroma_host: str = "localhost" chroma_port: int = 8100 outbox_transport: str = "redis_streams" strategy_rejection_distance_threshold: float = 3.0 # MVP rate limits for expensive LLM/vectorization endpoints. Values are # intentionally permissive for local demos and can be tightened by env. rate_limit_enabled: bool = True rate_limit_strategy_full_evaluate_per_minute: int = 30 rate_limit_strategy_vectorize_per_minute: int = 60 rate_limit_strategy_optimize_per_minute: int = 20 rate_limit_simulation_turn_per_minute: int = 60 rate_limit_mohp_evaluate_per_minute: int = 40 # ── LLM (Ollama Cloud) ─────────────────────────────────────────── ollama_host: str = "https://ollama.com" ollama_api_key: str = "" ollama_model: str = "gemma4:31b-cloud" # ── Embedding ───────────────────────────────────────────────────── # "onnx-minilm" → built-in ONNX all-MiniLM-L6-v2 (384-dim, fast, local) # "ollama:" → use Ollama embed API (e.g. "ollama:nomic-embed-text") embedding_model: str = "onnx-minilm" # ── JWT / Auth ──────────────────────────────────────────────────── jwt_secret_key: str = "" jwt_algorithm: str = "HS256" jwt_access_token_expire_minutes: int = 60 # ── External APIs ───────────────────────────────────────────────── hume_api_key: str = "" did_api_key: str = "" # ── CORS ────────────────────────────────────────────────────────── # Comma-separated list of allowed origins. In production this MUST # be set to your actual frontend domains. cors_allowed_origins: str = "http://localhost:3000,http://localhost:3001,http://127.0.0.1:3000,http://127.0.0.1:3001" enable_admin_mutations: bool = False admin_token: str = "" # ── Neural Projection Bridge ────────────────────────────────────── projection_weights_path: str = "projection_weights.pt.npz" @property def is_production(self) -> bool: return self.environment.lower() == "production" @property def effective_jwt_secret_key(self) -> str: """Use env-provided JWT secret, or a random dev-only secret per process.""" if self.jwt_secret_key and self.jwt_secret_key.strip().lower() not in _PLACEHOLDER_SECRETS: return self.jwt_secret_key if self.is_production: return self.jwt_secret_key logger.warning("JWT_SECRET_KEY is not configured; using an ephemeral development secret for this process.") return _GENERATED_DEV_JWT_SECRET @property def resolved_cors_allowed_origins(self) -> list[str]: origins = [o.strip() for o in self.cors_allowed_origins.split(",") if o.strip()] if "*" in origins and not self.is_production: logger.warning("CORS_ALLOWED_ORIGINS='*' replaced with local development origins.") return [ "http://localhost:3000", "http://localhost:3001", "http://127.0.0.1:3000", "http://127.0.0.1:3001", ] return origins @staticmethod def _redact_url(url: str) -> str: if not url: return "" parsed = urlsplit(url) if not parsed.netloc: return url host = parsed.hostname or "" if parsed.port: host = f"{host}:{parsed.port}" if parsed.username: host = f"{parsed.username}:***@{host}" elif parsed.password: host = f"***@{host}" return urlunsplit((parsed.scheme, host, parsed.path, parsed.query, parsed.fragment)) @property def safe_redis_url(self) -> str: return self._redact_url(self.redis_url) @property def safe_chroma_endpoint(self) -> str: if self.chroma_url: return self._redact_url(self.chroma_url) return f"http://{self.chroma_host}:{self.chroma_port}" @staticmethod def _is_bind_all_host(url: str) -> bool: parsed = urlsplit(url if "://" in url else f"//{url}") return (parsed.hostname or url).strip("[]") in {"0.0.0.0", "::"} @property def resolved_ollama_host(self) -> str: """Client-safe Ollama endpoint. ``OLLAMA_HOST=0.0.0.0`` is commonly used to bind a local Ollama server, but it is not a valid remote client target. When a cloud API key is present, prefer Ollama Cloud rather than crashing against the bind address; without a key, fall back to the normal local client endpoint. """ configured = (self.ollama_host or "").strip().rstrip("/") if not configured: return "https://ollama.com" if self.ollama_api_key else "http://127.0.0.1:11434" if self._is_bind_all_host(configured): if self.ollama_api_key: logger.warning("OLLAMA_HOST is a bind address; using Ollama Cloud for client calls.") return "https://ollama.com" return "http://127.0.0.1:11434" if "://" not in configured: if configured in {"localhost", "127.0.0.1"}: return f"http://{configured}:11434" if configured.startswith("localhost:") or configured.startswith("127.0.0.1:"): return f"http://{configured}" return f"https://{configured}" return configured @property def chroma_connection_mode(self) -> str: if self.chroma_url: return "external" if self.chroma_host in {"localhost", "127.0.0.1", "chroma", "chromadb"}: return "local" return "external" @property def redis_connection_mode(self) -> str: parsed = urlsplit(self.redis_url) host = parsed.hostname or "" return "local" if host in {"localhost", "127.0.0.1", "redis"} else "external" @property def neo4j_connection_mode(self) -> str: parsed = urlsplit(self.neo4j_uri) host = parsed.hostname or "" return "local" if host in {"localhost", "127.0.0.1", "neo4j"} else "external" def infrastructure_diagnostics(self) -> dict[str, str]: """Safe-to-log database connectivity summary with credentials removed.""" return { "redis_mode": self.redis_connection_mode, "redis_endpoint": self.safe_redis_url, "neo4j_mode": self.neo4j_connection_mode, "neo4j_uri": self.neo4j_uri, "chroma_mode": self.chroma_connection_mode, "chroma_endpoint": self.safe_chroma_endpoint, } def validate_production_secrets(self) -> None: """Fail-fast: prevent boot with placeholder passwords in production.""" if not self.is_production: return violations: list[str] = [] checks = { "neo4j_password": self.neo4j_password, "jwt_secret_key": self.effective_jwt_secret_key, } for field_name, value in checks.items(): if not value or value.strip().lower() in _PLACEHOLDER_SECRETS: violations.append(field_name) if not self.ollama_api_key: violations.append("ollama_api_key (empty)") if violations: msg = ( "FATAL: Production environment detected with insecure / placeholder " f"secrets: {', '.join(violations)}. " "Set real values via environment variables before starting." ) print(msg, file=sys.stderr) sys.exit(1) # Validate CORS is not wide-open in production raw_origins = [o.strip() for o in self.cors_allowed_origins.split(",") if o.strip()] if "*" in raw_origins: print( "FATAL: CORS allow_origins='*' is forbidden in production. " "Set CORS_ALLOWED_ORIGINS to your frontend domains.", file=sys.stderr, ) sys.exit(1) settings = Settings()