"""Explicit startup validation; importing this module never requires secrets. Errors name settings, never their values. This is intentionally independent of application imports, database access, and optional integration clients. """ import base64 import binascii import math import posixpath import re from collections.abc import Mapping from urllib.parse import unquote, urlsplit class ConfigurationError(ValueError): """One or more configuration settings are unsafe or malformed.""" def validate_configuration(env: Mapping[str, str], *, production: bool = True) -> None: errors: list[str] = [] def value(name: str) -> str: return env.get(name, "").strip() def problem(name: str, reason: str) -> None: errors.append(f"{name}: {reason}") def placeholder(text: str) -> bool: text = text.lower() return text in {"secret_key", "changeme", "password", "raggr_dev_password"} or ( text.startswith(("your-", "replace-", "replace_", "example-")) ) def secret(name: str, *, required: bool = False, minimum: int = 1) -> None: text = value(name) if not text: if required: problem(name, "required for production startup") elif production and (placeholder(text) or len(text) < minimum): problem( name, f"use a non-placeholder secret of at least {minimum} characters" ) def url(name: str, *, https: bool = False) -> None: text = value(name) if not text: return try: parsed = urlsplit(text) valid = ( parsed.scheme in ({"https"} if https else {"http", "https"}) and parsed.hostname and not parsed.username and not parsed.password and not parsed.fragment and not any(c.isspace() for c in text) ) # Accessing port validates malformed/out-of-range ports too. if parsed.port == 0: valid = False except ValueError: valid = False if not valid: problem( name, "must be a valid HTTPS URL" if https else "must be a valid HTTP(S) URL", ) secret("JWT_SECRET_KEY", required=production, minimum=32) secret("OPENAI_API_KEY", required=production) secret("FERNET_KEY", required=production) if value("FERNET_KEY"): try: if not re.fullmatch(r"[A-Za-z0-9_-]{43}=", value("FERNET_KEY")): raise ValueError decoded = base64.b64decode( value("FERNET_KEY"), altchars=b"-_", validate=True ) if len(decoded) != 32: raise ValueError except (ValueError, binascii.Error): problem("FERNET_KEY", "must encode exactly 32 bytes in URL-safe base64") database = value("DATABASE_URL") if production or database: try: parsed = urlsplit(database) valid = ( parsed.scheme == "postgres" and parsed.hostname and parsed.username and parsed.password and parsed.path not in {"", "/"} and not parsed.fragment and not any(c.isspace() for c in database) ) if parsed.port == 0: valid = False if production and placeholder(unquote(parsed.password or "")): valid = False except ValueError: valid = False if not valid: problem( "DATABASE_URL", "use postgres:// with host, user, non-placeholder password and database", ) integers = { "CONTEXT_WINDOW_TOKENS": 32768, "PROMPT_BUDGET_TOKENS": 16000, "RESPONSE_RESERVE_TOKENS": 4096, "YNAB_SYNC_INTERVAL": 900, "OBSIDIAN_SYNC_INTERVAL": 60, "EMAIL_RATE_LIMIT_MAX": 5, "EMAIL_RATE_LIMIT_WINDOW": 300, "WHATSAPP_RATE_LIMIT_MAX": 10, "WHATSAPP_RATE_LIMIT_WINDOW": 60, "IMESSAGE_RATE_LIMIT_MAX": 10, "IMESSAGE_RATE_LIMIT_WINDOW": 60, } numbers: dict[str, int] = {} for name, default in integers.items(): try: number = int(env.get(name, str(default))) if number <= 0: raise ValueError numbers[name] = number except ValueError: problem(name, "must be a positive integer") if all( name in numbers for name in ( "CONTEXT_WINDOW_TOKENS", "PROMPT_BUDGET_TOKENS", "RESPONSE_RESERVE_TOKENS", ) ): if ( numbers["PROMPT_BUDGET_TOKENS"] + numbers["RESPONSE_RESERVE_TOKENS"] > numbers["CONTEXT_WINDOW_TOKENS"] ): problem( "PROMPT_BUDGET_TOKENS", "prompt budget plus response reserve exceeds context window", ) for name, default in { "TTFT_TIMEOUT_SECONDS": "30", "LLAMA_TIMEOUT_SECONDS": "120", }.items(): try: number = float(env.get(name, default)) if not math.isfinite(number) or number <= 0: raise ValueError except ValueError: problem(name, "must be a finite positive number") for name in ( "OIDC_USE_DISCOVERY", "OBSIDIAN_CONTINUOUS_SYNC", "TWILIO_SIGNATURE_VALIDATION", "MAILGUN_SIGNATURE_VALIDATION", "SENDBLUE_SIGNATURE_VALIDATION", ): if name in env and env[name] not in {"true", "false"}: problem(name, "must be true or false") for name in ( "LLAMA_SERVER_URL", "EMBEDDING_SERVER_URL", "MEALIE_BASE_URL", "HOMEBOX_URL", "BOOKSTACK_BASE_URL", "S3_ENDPOINT_URL", "SIMBAKIT_URL", "CRONICLE_URL", "OLLAMA_URL", "OLLAMA_HOST", "TWILIO_WEBHOOK_URL", ): url(name) for name in ("LLAMA_MODEL_NAME", "EMBEDDING_MODEL_NAME"): if name in env and not value(name): problem(name, "must not be empty when supplied") # OIDC is optional; a partially configured login must not pass validation. oidc = ("OIDC_ISSUER", "OIDC_CLIENT_ID", "OIDC_CLIENT_SECRET", "OIDC_REDIRECT_URI") if any(value(name) for name in oidc): for name in oidc: if not value(name): problem(name, "required when OIDC is configured") secret("OIDC_CLIENT_SECRET") url("OIDC_ISSUER", https=production) url("OIDC_REDIRECT_URI", https=production) if env.get("OIDC_USE_DISCOVERY", "true") == "false": for name in ( "OIDC_AUTHORIZATION_ENDPOINT", "OIDC_TOKEN_ENDPOINT", "OIDC_USERINFO_ENDPOINT", "OIDC_JWKS_URI", ): if not value(name): problem(name, "required when OIDC discovery is disabled") url(name, https=production) for trigger, signing_key, flag in ( ("TWILIO_ACCOUNT_SID", "TWILIO_AUTH_TOKEN", "TWILIO_SIGNATURE_VALIDATION"), ( "MAILGUN_DOMAIN", "MAILGUN_WEBHOOK_SIGNING_KEY", "MAILGUN_SIGNATURE_VALIDATION", ), ( "SENDBLUE_API_KEY", "SENDBLUE_WEBHOOK_SECRET", "SENDBLUE_SIGNATURE_VALIDATION", ), ): if production: if value(trigger) or value(signing_key): secret(signing_key, required=True) # Routes are registered even when outbound integrations are unused. # Missing keys fail closed; disabling verification would bypass that. if env.get(flag, "true") != "true": problem(flag, "signature validation must remain enabled in production") for name in ("METRICS_TOKEN", "CRON_SECRET", "EMAIL_HMAC_SECRET"): secret(name, minimum=32) if env.get("OBSIDIAN_CONTINUOUS_SYNC") == "true": for name in ("OBSIDIAN_EMAIL", "OBSIDIAN_PASSWORD", "OBSIDIAN_VAULT_ID"): if not value(name): problem(name, "required when continuous sync is enabled") paths = { "CHECKPOINT_DB_PATH": env.get( "CHECKPOINT_DB_PATH", "/app/checkpoints/checkpoints.db" ), "CONVERSATION_SUMMARY_DB_PATH": env.get( "CONVERSATION_SUMMARY_DB_PATH", env.get("CHECKPOINT_DB_PATH", "/app/checkpoints/checkpoints.db") + ".summaries.db", ), } for name, path in paths.items(): if ( not path or path == ":memory:" or "\x00" in path or (production and not path.startswith("/")) or path.endswith("/") ): problem(name, "must be a persistent file path (absolute in production)") if posixpath.normpath(paths["CHECKPOINT_DB_PATH"]) == posixpath.normpath( paths["CONVERSATION_SUMMARY_DB_PATH"] ): problem( "CONVERSATION_SUMMARY_DB_PATH", "must differ from checkpoint database path" ) if errors: raise ConfigurationError( "Invalid configuration:\n" + "\n".join(f"- {error}" for error in errors) )