"""DocDoe production preflight checks. Run before deploying to verify the production environment is safe: python backend/scripts/production_preflight.py Exit codes: 0 — no FAIL (WARN entries allowed) 1 — at least one FAIL The script never prints secret values. Sensitive env vars are masked to first 4 + last 4 characters with the middle replaced by ``****``. """ from __future__ import annotations import ast import os import shutil import sys from dataclasses import dataclass from datetime import date from pathlib import Path from typing import Callable, Iterable # Ensure the script can be run from anywhere — fall back to repo-relative imports. HERE = Path(__file__).resolve().parent BACKEND_DIR = HERE.parent REPO_ROOT = BACKEND_DIR.parent if str(BACKEND_DIR) not in sys.path: sys.path.insert(0, str(BACKEND_DIR)) # ── Result types ────────────────────────────────────────────────────────────── Status = str # "PASS" | "WARN" | "FAIL" @dataclass(frozen=True) class Check: section: str name: str status: Status detail: str def _pass(section: str, name: str, detail: str = "") -> Check: return Check(section, name, "PASS", detail) def _warn(section: str, name: str, detail: str) -> Check: return Check(section, name, "WARN", detail) def _fail(section: str, name: str, detail: str) -> Check: return Check(section, name, "FAIL", detail) # ── Secret masking ──────────────────────────────────────────────────────────── def mask_secret(value: str | None) -> str: if not value: return "(missing)" if len(value) <= 8: return "****" return f"{value[:4]}****{value[-4:]}" # ── Helpers to read env ─────────────────────────────────────────────────────── def env(name: str, default: str | None = None) -> str | None: raw = os.environ.get(name) if raw is None: return default stripped = raw.strip() return stripped if stripped else default def env_bool(name: str, default: bool = False) -> bool: raw = env(name) if raw is None: return default return raw.lower() in {"1", "true", "yes", "on"} # ── Section 1 — Environment ─────────────────────────────────────────────────── DEFAULT_JWT_SECRET = "change-this-local-dev-secret" LOCAL_HOST_PREFIXES = ("http://localhost", "http://127.0.0.1", "http://0.0.0.0") def check_environment(is_prod: bool) -> Iterable[Check]: sec = "Environment" yield _pass(sec, "ENVIRONMENT", f"set to '{env('ENVIRONMENT') or 'development'}'") db_url = env("DATABASE_URL") if not db_url: yield _fail(sec, "DATABASE_URL", "missing") elif is_prod and db_url.startswith("sqlite"): yield _fail(sec, "DATABASE_URL", "SQLite not allowed in production — use PostgreSQL") elif db_url.startswith("postgresql"): yield _pass(sec, "DATABASE_URL", "PostgreSQL configured") else: yield _warn(sec, "DATABASE_URL", f"non-PostgreSQL driver ({db_url.split(':', 1)[0]})") frontend = env("FRONTEND_BASE_URL") if is_prod and frontend: if frontend.startswith(LOCAL_HOST_PREFIXES): yield _fail(sec, "FRONTEND_BASE_URL", "points at localhost in production") elif not frontend.startswith("https://"): yield _warn(sec, "FRONTEND_BASE_URL", "should use https in production") else: yield _pass(sec, "FRONTEND_BASE_URL", frontend) elif frontend: yield _pass(sec, "FRONTEND_BASE_URL", frontend) else: yield _warn(sec, "FRONTEND_BASE_URL", "not set") cors = env("CORS_ORIGINS") if is_prod: if not cors: yield _fail(sec, "CORS_ORIGINS", "missing in production") else: origins = [o.strip() for o in cors.split(",") if o.strip()] bad = [o for o in origins if o == "*" or o.startswith(LOCAL_HOST_PREFIXES)] if bad: yield _fail(sec, "CORS_ORIGINS", f"unsafe entries: {bad}") else: yield _pass(sec, "CORS_ORIGINS", f"{len(origins)} origin(s) configured") else: yield _pass(sec, "CORS_ORIGINS", cors or "(default)") jwt = env("JWT_SECRET_KEY") if not jwt or jwt == DEFAULT_JWT_SECRET: if is_prod: yield _fail(sec, "JWT_SECRET_KEY", "missing or default value — generate with `openssl rand -hex 32`") else: yield _warn(sec, "JWT_SECRET_KEY", "using dev default") elif len(jwt) < 32: yield _fail(sec, "JWT_SECRET_KEY", f"too short ({len(jwt)} chars, need >=32)") else: yield _pass(sec, "JWT_SECRET_KEY", f"strong ({len(jwt)} chars, {mask_secret(jwt)})") auth_enabled = env_bool("AUTH_ENABLED") if is_prod and not auth_enabled: yield _fail(sec, "AUTH_ENABLED", "must be true in production") else: yield _pass(sec, "AUTH_ENABLED", str(auth_enabled)) auth_provider = (env("AUTH_PROVIDER") or "jwt").lower() if auth_provider not in {"jwt", "supabase"}: yield _fail(sec, "AUTH_PROVIDER", "must be 'jwt' or 'supabase' for real users") else: yield _pass(sec, "AUTH_PROVIDER", auth_provider) if auth_provider == "supabase": supabase_url = env("SUPABASE_URL") supabase_jwt = env("SUPABASE_JWT_SECRET") service_key = env("SUPABASE_SERVICE_ROLE_KEY") if not supabase_url: yield _fail(sec, "SUPABASE_URL", "required when AUTH_PROVIDER=supabase") else: yield _pass(sec, "SUPABASE_URL", supabase_url) if not supabase_jwt: yield _fail(sec, "SUPABASE_JWT_SECRET", "required to verify student sessions") elif len(supabase_jwt) < 32: yield _fail(sec, "SUPABASE_JWT_SECRET", "too short (need >=32 characters)") else: yield _pass(sec, "SUPABASE_JWT_SECRET", mask_secret(supabase_jwt)) if not service_key: yield _fail( sec, "SUPABASE_SERVICE_ROLE_KEY", "required for authenticated account deletion; backend only", ) else: yield _pass(sec, "SUPABASE_SERVICE_ROLE_KEY", mask_secret(service_key)) rate_limit = env_bool("RATE_LIMIT_ENABLED") if is_prod and not rate_limit: yield _warn(sec, "RATE_LIMIT_ENABLED", "should be true in production") else: yield _pass(sec, "RATE_LIMIT_ENABLED", str(rate_limit)) # ── Section 1b — Database recovery ────────────────────────────────────── def check_database_recovery(is_prod: bool) -> Iterable[Check]: sec = "Database recovery" strategy = (env("DATABASE_BACKUP_STRATEGY") or "").lower() if not strategy: if is_prod: yield _fail( sec, "DATABASE_BACKUP_STRATEGY", "missing; use 'managed' for provider backups or 'pg_dump' for the bundled backup job", ) else: yield _pass(sec, "DATABASE_BACKUP_STRATEGY", "not required in development") return if strategy not in {"managed", "pg_dump"}: yield _fail(sec, "DATABASE_BACKUP_STRATEGY", "must be 'managed' or 'pg_dump'") return yield _pass(sec, "DATABASE_BACKUP_STRATEGY", strategy) retention_raw = env("DATABASE_BACKUP_RETENTION_DAYS") or "0" try: retention_days = int(retention_raw) except ValueError: retention_days = 0 if retention_days < 7: yield _fail(sec, "DATABASE_BACKUP_RETENTION_DAYS", "must be at least 7 days") else: yield _pass(sec, "DATABASE_BACKUP_RETENTION_DAYS", f"{retention_days} days") restore_tested_raw = env("DATABASE_RESTORE_TESTED_AT") if not restore_tested_raw: yield _fail(sec, "DATABASE_RESTORE_TESTED_AT", "missing; record the latest successful restore drill date") else: try: restore_date = date.fromisoformat(restore_tested_raw) age_days = (date.today() - restore_date).days if age_days < -1: yield _fail(sec, "DATABASE_RESTORE_TESTED_AT", "cannot be a future date") elif age_days > 90: yield _fail(sec, "DATABASE_RESTORE_TESTED_AT", f"restore drill is stale ({age_days} days old)") else: yield _pass(sec, "DATABASE_RESTORE_TESTED_AT", f"{restore_tested_raw} ({max(age_days, 0)} days ago)") except ValueError: yield _fail(sec, "DATABASE_RESTORE_TESTED_AT", "must be an ISO date in YYYY-MM-DD format") if strategy == "pg_dump": backup_dir = env("DATABASE_BACKUP_DIR") if not backup_dir: yield _fail(sec, "DATABASE_BACKUP_DIR", "required when DATABASE_BACKUP_STRATEGY=pg_dump") else: yield _pass(sec, "DATABASE_BACKUP_DIR", "configured") for binary in ("pg_dump", "pg_restore"): if shutil.which(binary): yield _pass(sec, binary, "available") else: yield _fail(sec, binary, "not available on PATH") # ── Section 2 — Storage ─────────────────────────────────────────────────────── def check_storage(is_prod: bool) -> Iterable[Check]: sec = "Storage" provider = (env("STORAGE_PROVIDER") or "local").lower() if provider not in {"local", "r2", "s3", "cloudinary"}: yield _fail(sec, "STORAGE_PROVIDER", f"invalid value '{provider}' — must be local/r2/s3/cloudinary") return yield _pass(sec, "STORAGE_PROVIDER", provider) if provider == "cloudinary": cloud_name = env("CLOUDINARY_CLOUD_NAME") api_key = env("CLOUDINARY_API_KEY") api_secret = env("CLOUDINARY_API_SECRET") if not cloud_name: yield _fail(sec, "CLOUDINARY_CLOUD_NAME", "missing") else: yield _pass(sec, "CLOUDINARY_CLOUD_NAME", cloud_name) if not api_key: yield _fail(sec, "CLOUDINARY_API_KEY", "missing") else: yield _pass(sec, "CLOUDINARY_API_KEY", mask_secret(api_key)) if not api_secret: yield _fail(sec, "CLOUDINARY_API_SECRET", "missing — required for uploads") else: yield _pass(sec, "CLOUDINARY_API_SECRET", mask_secret(api_secret)) try: import cloudinary # noqa: F401 yield _pass(sec, "cloudinary package", "installed") except ImportError: yield _fail(sec, "cloudinary package", "not installed — add 'cloudinary' to backend/requirements.txt") return if provider == "local": if is_prod: yield _warn(sec, "Local storage in production", "generated videos lost on redeploy — see backend/STORAGE_SETUP.md") else: yield _pass(sec, "Local storage", "OK for development") return # R2 / S3 required vars bucket = env("STORAGE_BUCKET") access_key = env("STORAGE_ACCESS_KEY_ID") secret_key = env("STORAGE_SECRET_ACCESS_KEY") endpoint = env("STORAGE_ENDPOINT_URL") public_base = env("STORAGE_PUBLIC_BASE_URL") region = env("STORAGE_REGION") if not bucket: yield _fail(sec, "STORAGE_BUCKET", "missing") else: yield _pass(sec, "STORAGE_BUCKET", bucket) if not access_key: yield _fail(sec, "STORAGE_ACCESS_KEY_ID", "missing") else: yield _pass(sec, "STORAGE_ACCESS_KEY_ID", mask_secret(access_key)) if not secret_key: yield _fail(sec, "STORAGE_SECRET_ACCESS_KEY", "missing") else: yield _pass(sec, "STORAGE_SECRET_ACCESS_KEY", mask_secret(secret_key)) if provider == "r2": if not endpoint: yield _fail(sec, "STORAGE_ENDPOINT_URL", "required for R2") elif "r2.cloudflarestorage.com" not in endpoint and "cloudflare" not in endpoint: yield _warn(sec, "STORAGE_ENDPOINT_URL", "does not look like an R2 endpoint") else: yield _pass(sec, "STORAGE_ENDPOINT_URL", endpoint) elif provider == "s3": if not region: yield _warn(sec, "STORAGE_REGION", "not set — boto3 will use AWS default") if is_prod and public_base: if public_base.startswith(LOCAL_HOST_PREFIXES): yield _fail(sec, "STORAGE_PUBLIC_BASE_URL", "points at localhost in production") elif not public_base.startswith("https://"): yield _warn(sec, "STORAGE_PUBLIC_BASE_URL", "should use https in production") else: yield _pass(sec, "STORAGE_PUBLIC_BASE_URL", public_base) elif public_base: yield _pass(sec, "STORAGE_PUBLIC_BASE_URL", public_base) else: yield _warn(sec, "STORAGE_PUBLIC_BASE_URL", "not set — public URLs may fall back to provider default") # Optional: try to instantiate the boto3 provider without uploading if bucket and access_key and secret_key: try: import boto3 # noqa: F401 yield _pass(sec, "boto3", "installed") except ImportError: yield _fail(sec, "boto3", "not installed — run `pip install boto3`") # ── Section 3 — Video render dependencies ──────────────────────────────────── def _has_binary(name: str) -> bool: return shutil.which(name) is not None def check_video_deps() -> Iterable[Check]: sec = "Video render deps" for binary in ("ffmpeg", "ffprobe"): if _has_binary(binary): yield _pass(sec, binary, f"available at {shutil.which(binary)}") else: yield _fail(sec, binary, f"{binary} not found in PATH — video render/validation will fail") for binary in ("node", "npm"): if _has_binary(binary): yield _pass(sec, binary, f"available at {shutil.which(binary)}") else: yield _fail(sec, binary, f"{binary} not found in PATH — render CLI will fail") package_json = REPO_ROOT / "package.json" if not package_json.exists(): yield _fail(sec, "package.json", "not found at repo root") return try: import json as _json data = _json.loads(package_json.read_text(encoding="utf-8")) except Exception as exc: yield _fail(sec, "package.json", f"could not parse: {exc.__class__.__name__}") return scripts = data.get("scripts", {}) if "render:from-json" in scripts: yield _pass(sec, "render:from-json script", "defined in package.json") else: yield _fail(sec, "render:from-json script", "missing from package.json — backend invokes this for render") deps = {**data.get("dependencies", {}), **data.get("devDependencies", {})} if "remotion" in deps or "@remotion/cli" in deps: version = deps.get("remotion") or deps.get("@remotion/cli", "unknown") yield _pass(sec, "remotion package", f"declared ({version})") else: yield _fail(sec, "remotion package", "not declared in package.json") # ── Section 4 — AI / TTS config ────────────────────────────────────────────── def check_ai_config(is_prod: bool) -> Iterable[Check]: sec = "AI / TTS" ai_provider = (env("AI_PROVIDER") or "").lower() if not ai_provider: yield _fail(sec, "AI_PROVIDER", "not set") else: yield _pass(sec, "AI_PROVIDER", ai_provider) if is_prod and env_bool("AI_FALLBACK_TO_MOCK"): yield _fail(sec, "AI_FALLBACK_TO_MOCK", "must be false in production — mock responses must not reach students") else: yield _pass(sec, "AI_FALLBACK_TO_MOCK", str(env_bool("AI_FALLBACK_TO_MOCK"))) tts_provider = (env("TTS_PROVIDER") or "").lower() if tts_provider: yield _pass(sec, "TTS_PROVIDER", tts_provider) else: yield _warn(sec, "TTS_PROVIDER", "not set — falling back to default") video_tts_provider = (env("VIDEO_TTS_PROVIDER") or "ai4bharat").lower() if video_tts_provider: yield _pass(sec, "VIDEO_TTS_PROVIDER", video_tts_provider) else: yield _warn(sec, "VIDEO_TTS_PROVIDER", "not set - study video voice defaults to ai4bharat") if ai_provider == "openai": key = env("OPENAI_API_KEY") if not key: yield _fail(sec, "OPENAI_API_KEY", "required when AI_PROVIDER=openai") else: yield _pass(sec, "OPENAI_API_KEY", mask_secret(key)) needs_sarvam = ai_provider == "sarvam" or tts_provider == "sarvam" or video_tts_provider == "sarvam" sarvam_key = env("SARVAM_API_KEY") if needs_sarvam: if not sarvam_key: yield _fail(sec, "SARVAM_API_KEY", "required when AI_PROVIDER or TTS_PROVIDER is sarvam") else: yield _pass(sec, "SARVAM_API_KEY", mask_secret(sarvam_key)) ai4bharat_names = {"ai4bharat", "indic_parler", "indic_parler_tts"} needs_ai4bharat = tts_provider in ai4bharat_names or video_tts_provider in ai4bharat_names if needs_ai4bharat: hf_key = env("HUGGINGFACE_API_KEY") or env("HF_TOKEN") if not hf_key: yield _fail( sec, "HUGGINGFACE_API_KEY", "required for the initial gated AI4Bharat model download", ) else: yield _pass(sec, "HUGGINGFACE_API_KEY", mask_secret(hf_key)) if ai_provider == "openrouter": key = env("OPENROUTER_API_KEY") if not key: yield _fail(sec, "OPENROUTER_API_KEY", "required when AI_PROVIDER=openrouter") else: yield _pass(sec, "OPENROUTER_API_KEY", mask_secret(key)) if ai_provider == "gemini": key = env("GEMINI_API_KEY") if not key: yield _fail(sec, "GEMINI_API_KEY", "required when AI_PROVIDER=gemini") else: yield _pass(sec, "GEMINI_API_KEY", mask_secret(key)) if ai_provider == "auto": configured = [ name for name, key in ( ("openai", env("OPENAI_API_KEY")), ("sarvam", sarvam_key), ("openrouter", env("OPENROUTER_API_KEY")), ("gemini", env("GEMINI_API_KEY")), ) if key ] if configured: yield _pass(sec, "AI_PROVIDER_AUTO_KEYS", ",".join(configured)) else: yield _fail(sec, "AI_PROVIDER_AUTO_KEYS", "AI_PROVIDER=auto requires at least one real AI key") # Expanded TTS provider coverage tts_providers = { "TTS_PROVIDER": env("TTS_PROVIDER"), "VIDEO_TTS_PROVIDER": env("VIDEO_TTS_PROVIDER") or "ai4bharat", } for k, v in tts_providers.items(): if v: yield _pass(sec, f"{k}_set", v) else: yield _warn(sec, f"{k}_set", "falls back to default (ai4bharat)") # More providers sanity (edge, sarvam, mock, openai-tts etc recognized in code) known_tts = { "ai4bharat", "indic_parler", "indic_parler_tts", "edge", "sarvam", "mock", "openai", "gcp", "azure", "kokoro", "hybrid", } for k, v in tts_providers.items(): if v and v.lower() not in known_tts: yield _warn(sec, f"{k}_unknown", f"'{v}' not in common known list; ensure provider impl exists") # ── Section 5b — Billing / Stripe (if wired) ───────────────────────────────── def check_billing_stripe(is_prod: bool) -> Iterable[Check]: """Fail closed when Stripe is partially configured. A fully absent Stripe configuration is a valid free-only deployment. Once any Stripe setting is supplied, Checkout, signed webhook activation, and both student-facing recurring Prices must all be ready together. """ sec = "Billing / Stripe" stripe_key = env("STRIPE_SECRET_KEY") webhook_secret = env("STRIPE_WEBHOOK_SECRET") price_ids = env("STRIPE_PRICE_IDS") any_billing_config = bool(stripe_key or webhook_secret or price_ids) if not any_billing_config: detail = "not configured; paid checkout is safely disabled" if is_prod: yield _warn(sec, "Stripe billing", detail) else: yield _pass(sec, "Stripe billing", detail) return if not stripe_key: yield _fail(sec, "STRIPE_SECRET_KEY", "missing while other Stripe settings are present") elif is_prod and stripe_key.startswith(("sk_test_", "rk_test_")): yield _fail(sec, "STRIPE_SECRET_KEY", "test-mode key configured in production") elif is_prod and stripe_key.startswith("sk_live_"): yield _warn( sec, "STRIPE_SECRET_KEY", f"{mask_secret(stripe_key)}; prefer a restricted rk_live_ key", ) else: yield _pass(sec, "STRIPE_SECRET_KEY", mask_secret(stripe_key)) if not webhook_secret: yield _fail( sec, "STRIPE_WEBHOOK_SECRET", "missing; checkout remains disabled because entitlements cannot be activated safely", ) elif not webhook_secret.startswith("whsec_"): yield _fail(sec, "STRIPE_WEBHOOK_SECRET", "must be a Stripe endpoint signing secret") else: yield _pass(sec, "STRIPE_WEBHOOK_SECRET", mask_secret(webhook_secret)) mappings: dict[str, str] = {} for pair in (price_ids or "").split(","): if "=" not in pair: continue key, value = pair.split("=", 1) mappings[key.strip().lower()] = value.strip() required_plans = {"popular_299", "premium_599"} missing_plans = sorted( plan for plan in required_plans if not mappings.get(plan, "").startswith("price_") ) if missing_plans: yield _fail( sec, "STRIPE_PRICE_IDS", f"missing valid recurring Price mapping(s): {', '.join(missing_plans)}", ) else: yield _pass(sec, "STRIPE_PRICE_IDS", "Popular and Premium Prices configured") # ── Section 5c — Diagram logic coverage (generalized video) — static for env robustness ────────────────── def check_diagram_logic() -> Iterable[Check]: sec = "Diagram logic" vg_path = BACKEND_DIR / "app" / "routes" / "video_generator.py" if not vg_path.exists(): yield _fail(sec, "diagram module import", "video_generator.py source not found") return try: with open(vg_path, encoding="utf-8") as f: tree = ast.parse(f.read(), filename=str(vg_path)) # Collect defined function names (static, no import/runtime deps needed) func_names = { node.name for node in ast.walk(tree) if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) } # Find _DIAGRAM_KEYWORDS assign and approximate length (static tuple/list) kw_len = 0 for node in ast.walk(tree): if isinstance(node, ast.Assign): for t in node.targets: if isinstance(t, ast.Name) and t.id == "_DIAGRAM_KEYWORDS": if isinstance(node.value, (ast.Tuple, ast.List)): kw_len = len(node.value.elts) elif isinstance(node.value, ast.Constant) and isinstance(node.value.value, (tuple, list)): kw_len = len(node.value.value) break has_needs = "_needs_diagram" in func_names has_ensure = "_ensure_diagram_scene" in func_names has_generic = "_generic_diagram_scene" in func_names has_keywords = kw_len > 5 if has_needs and has_ensure and has_generic and has_keywords: yield _pass(sec, "_ensure_diagram_scene generalized", "present + _needs + _generic + keywords (static)") # Behavioral smoke skipped (would require full runtime env + deps); presence + keyword count is sufficient for preflight yield _pass(sec, "diagram keyword trigger", f"generalized keywords present (len={kw_len})") else: missing = [] if not has_needs: missing.append("_needs_diagram") if not has_ensure: missing.append("_ensure_diagram_scene") if not has_generic: missing.append("_generic_diagram_scene") if not has_keywords: missing.append("_DIAGRAM_KEYWORDS (len<=5)") yield _fail(sec, "_ensure_diagram_scene generalized", f"missing or incomplete: {', '.join(missing)}") except SyntaxError as exc: yield _fail(sec, "diagram module import", f"ast parse error in video_generator.py: {exc}") except Exception as exc: # noqa: BLE001 yield _fail(sec, "diagram module import", f"could not statically inspect video_generator.py: {exc.__class__.__name__}") # ── Section 5d — New 10kx features (spaced rep + streaming Ask) — static presence ────────────────── def check_new_features() -> Iterable[Check]: sec = "New features (10kx)" # Spaced repetition fc_path = BACKEND_DIR / "app" / "routes" / "flashcards.py" try: fc_src = fc_path.read_text(encoding="utf-8") if fc_path.exists() else "" has_sm2 = "_sm2_update" in fc_src or "SM-2" in fc_src has_due = "/due" in fc_src or "get_due_cards" in fc_src has_review = "/review" in fc_src or "record_flashcard_review" in fc_src if has_sm2 and has_due and has_review: yield _pass(sec, "spaced repetition (SM-2)", "due/review endpoints + scheduler helpers present") else: yield _warn(sec, "spaced repetition (SM-2)", "helpers/endpoints may be incomplete") except Exception: yield _warn(sec, "spaced repetition (SM-2)", "could not inspect flashcards.py") # Streaming Ask / StudyCast ai_path = BACKEND_DIR / "app" / "services" / "ai_provider.py" ask_path = BACKEND_DIR / "app" / "routes" / "ask.py" try: ai_src = ai_path.read_text(encoding="utf-8") if ai_path.exists() else "" ask_src = ask_path.read_text(encoding="utf-8") if ask_path.exists() else "" has_gen_stream = "def generate_streaming" in ai_src or "generate_streaming" in ai_src has_ask_stream = "/ask/stream" in ask_src or "ask_stream" in ask_src if has_gen_stream and has_ask_stream: yield _pass(sec, "streaming Ask/StudyCast", "generate_streaming + /ask/stream present") else: yield _warn(sec, "streaming Ask/StudyCast", "streaming support may be partial") except Exception: yield _warn(sec, "streaming Ask/StudyCast", "could not inspect ai_provider/ask sources") # ── Section 5 — URL safety ─────────────────────────────────────────────────── def check_url_safety(is_prod: bool) -> Iterable[Check]: if not is_prod: yield _pass("URL safety", "skipped", "not in production mode") return sec = "URL safety" candidates = { "FRONTEND_BASE_URL": env("FRONTEND_BASE_URL"), "STORAGE_PUBLIC_BASE_URL": env("STORAGE_PUBLIC_BASE_URL"), "GOOGLE_OAUTH_REDIRECT_URI": env("GOOGLE_OAUTH_REDIRECT_URI"), } for name, value in candidates.items(): if not value: continue if value.startswith(LOCAL_HOST_PREFIXES): yield _fail(sec, name, f"localhost URL in production: {value}") elif value.startswith("http://"): yield _warn(sec, name, f"not https: {value}") else: yield _pass(sec, name, value) # ── Section 6 — Deployment target ───────────────────────────────────────────── def check_deploy_target(is_prod: bool) -> Iterable[Check]: sec = "Deployment target" target = (env("DEPLOY_TARGET") or "").lower() if not target: if is_prod: yield _warn(sec, "DEPLOY_TARGET", "not set — recommend setting to huggingface/railway/flyio/docker") else: yield _pass(sec, "DEPLOY_TARGET", "not set (dev)") return if target not in {"huggingface", "railway", "flyio", "docker", "local"}: yield _warn(sec, "DEPLOY_TARGET", f"unrecognized value '{target}'") return yield _pass(sec, "DEPLOY_TARGET", target) if target == "huggingface": # HF Spaces sets PORT=7860 and expects 0.0.0.0 bind. port = env("PORT") or "7860" if port != "7860": yield _warn(sec, "PORT", f"HF Spaces expects 7860; got {port} (HF auto-sets via $PORT)") else: yield _pass(sec, "PORT", "7860 (HF Spaces default)") # Frontend on Vercel — backend CORS should include vercel.app or custom domain. cors = env("CORS_ORIGINS") or "" if cors and not any( "vercel.app" in o or ".docdoe.ai" in o or "docdoe.ai" in o for o in [origin.strip() for origin in cors.split(",")] ): yield _warn(sec, "Vercel origin in CORS", "no vercel.app or docdoe.ai origin in CORS_ORIGINS — frontend may not reach backend") frontend = env("FRONTEND_BASE_URL") or "" if frontend and "vercel.app" not in frontend and "docdoe.ai" not in frontend: yield _warn(sec, "Vercel frontend URL", f"FRONTEND_BASE_URL '{frontend}' does not look like Vercel/docdoe.ai") # ── Runner ──────────────────────────────────────────────────────────────────── def collect_all_checks() -> list[Check]: is_prod = (env("ENVIRONMENT") or "development").lower() == "production" sections: list[Callable[[], Iterable[Check]]] = [ lambda: check_environment(is_prod), lambda: check_database_recovery(is_prod), lambda: check_storage(is_prod), lambda: check_video_deps(), lambda: check_ai_config(is_prod), lambda: check_billing_stripe(is_prod), lambda: check_diagram_logic(), lambda: check_new_features(), lambda: check_url_safety(is_prod), lambda: check_deploy_target(is_prod), ] out: list[Check] = [] for fn in sections: out.extend(fn()) return out def render_table(checks: list[Check]) -> str: lines = [] current_section = None width_status = 6 width_name = max((len(c.name) for c in checks), default=24) width_name = min(max(width_name, 24), 48) for c in checks: if c.section != current_section: lines.append("") lines.append(f"-- {c.section} ".ljust(80, "-")) current_section = c.section icon = {"PASS": "[+]", "WARN": "[!]", "FAIL": "[x]"}[c.status] lines.append( f" {icon} {c.status:<{width_status}} {c.name:<{width_name}} {c.detail}" ) return "\n".join(lines) def main(argv: list[str] | None = None) -> int: # Allow load from .env file if present (dev convenience). # Skip in test mode (PREFLIGHT_SKIP_ENV_FILE=1) so tests aren't polluted. env_file = REPO_ROOT / "backend" / ".env" if env_file.exists() and not os.environ.get("PREFLIGHT_SKIP_ENV_FILE"): try: for raw_line in env_file.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() if not line or line.startswith("#") or "=" not in line: continue key, _, value = line.partition("=") key = key.strip() value = value.strip().strip('"').strip("'") os.environ.setdefault(key, value) except Exception: pass # don't fail preflight on .env parse errors checks = collect_all_checks() print("DocDoe Production Preflight") print("=" * 80) print(render_table(checks)) print() fail_count = sum(1 for c in checks if c.status == "FAIL") warn_count = sum(1 for c in checks if c.status == "WARN") pass_count = sum(1 for c in checks if c.status == "PASS") print(f"Summary: {pass_count} PASS · {warn_count} WARN · {fail_count} FAIL") if fail_count: print("\nFAILED — fix the items above before deploying.") return 1 if warn_count: print("\nReady to deploy with WARNINGS. Review them above.") else: print("\nAll checks passed.") return 0 if __name__ == "__main__": sys.exit(main())