"""Developer-only Beta Launch Doctor. GET /dev/beta-health Runs a structured pre-launch checklist and returns safe diagnostic data. No secret values are ever included in the response. Access rules: - In development (ENVIRONMENT=development): unauthenticated access allowed. - In production/staging: requires authenticated admin user (role="admin"). """ from __future__ import annotations import shutil import subprocess from dataclasses import dataclass, field from pathlib import Path from typing import Any from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy import inspect as sa_inspect, text from sqlalchemy.orm import Session from app.core.auth import get_current_user_optional from app.core.config import get_settings from app.core.database import get_db from app.models.user import User router = APIRouter() # ── Data model ──────────────────────────────────────────────────────────────── STATUS_OK = "ok" STATUS_WARN = "warning" STATUS_FAIL = "failed" @dataclass class Check: name: str category: str status: str message: str detail: str | None = None @dataclass class BetaHealthReport: status: str checks: list[Check] = field(default_factory=list) def add(self, check: Check) -> None: self.checks.append(check) def finalize(self) -> None: """Set overall status: failed > warning > ok.""" statuses = {c.status for c in self.checks} if STATUS_FAIL in statuses: self.status = STATUS_FAIL elif STATUS_WARN in statuses: self.status = STATUS_WARN else: self.status = STATUS_OK def to_dict(self) -> dict[str, Any]: ok = sum(1 for c in self.checks if c.status == STATUS_OK) warn = sum(1 for c in self.checks if c.status == STATUS_WARN) fail = sum(1 for c in self.checks if c.status == STATUS_FAIL) return { "status": self.status, "summary": {"ok": ok, "warning": warn, "failed": fail, "total": len(self.checks)}, "checks": [ { "name": c.name, "category": c.category, "status": c.status, "message": c.message, **({"detail": c.detail} if c.detail else {}), } for c in self.checks ], } # ── Individual check helpers ────────────────────────────────────────────────── def _check_db_connection(db: Session) -> Check: try: db.execute(text("SELECT 1")) return Check("db_connection", "database", STATUS_OK, "Database connection healthy.") except Exception as exc: # noqa: BLE001 return Check("db_connection", "database", STATUS_FAIL, "Cannot connect to database.", str(exc)) def _check_required_columns(db: Session) -> list[Check]: """Verify schema columns required for Phase 3 features exist.""" required: list[tuple[str, str, bool]] = [ # (table, column, is_critical) ("documents", "material_type", True), ("previous_papers", "verification_status", True), ("previous_papers", "official_source", True), ("video_render_jobs", "evidence_label", True), ("video_render_jobs", "source_document_id", True), ] optional_tables = ["provider_usage_logs", "generation_cache"] checks: list[Check] = [] try: insp = sa_inspect(db.bind) # type: ignore[arg-type] existing_tables = set(insp.get_table_names()) for table, column, critical in required: if table not in existing_tables: sev = STATUS_FAIL if critical else STATUS_WARN checks.append(Check( f"column_{table}_{column}", "database", sev, f"Table '{table}' does not exist.", )) continue cols = {c["name"] for c in insp.get_columns(table)} if column in cols: checks.append(Check( f"column_{table}_{column}", "database", STATUS_OK, f"{table}.{column} — present.", )) else: sev = STATUS_FAIL if critical else STATUS_WARN checks.append(Check( f"column_{table}_{column}", "database", sev, f"{table}.{column} — MISSING. Run migrations.", )) for table in optional_tables: if table in existing_tables: checks.append(Check( f"table_{table}", "database", STATUS_OK, f"Table '{table}' exists.", )) else: checks.append(Check( f"table_{table}", "database", STATUS_WARN, f"Optional table '{table}' not found — feature may be disabled.", )) except Exception as exc: # noqa: BLE001 checks.append(Check( "schema_inspect", "database", STATUS_FAIL, "Schema inspection failed.", str(exc), )) return checks def _check_storage(settings: Any) -> list[Check]: checks: list[Check] = [] dirs_to_check: list[tuple[str, Any, bool]] = [ ("upload_dir", settings.resolved_upload_dir, True), ("tts_output_dir", settings.resolved_tts_output_dir, False), ("video_output_dir", Path(settings.generated_video_output_dir), False), ("video_jobs_dir", Path(settings.generated_video_jobs_dir), False), ] for name, raw_path, critical in dirs_to_check: try: path = Path(raw_path) if raw_path else None if path is None: sev = STATUS_FAIL if critical else STATUS_WARN checks.append(Check(f"storage_{name}", "storage", sev, f"{name} not configured.")) continue if not path.exists(): try: path.mkdir(parents=True, exist_ok=True) checks.append(Check(f"storage_{name}", "storage", STATUS_WARN, f"{name}: directory created (was missing).")) continue except OSError as exc: sev = STATUS_FAIL if critical else STATUS_WARN checks.append(Check(f"storage_{name}", "storage", sev, f"{name}: cannot create directory.", str(exc))) continue # Write-test probe = path / ".beta_health_probe" try: probe.write_text("probe") probe.unlink() checks.append(Check(f"storage_{name}", "storage", STATUS_OK, f"{name}: exists and writable.")) except OSError as exc: sev = STATUS_FAIL if critical else STATUS_WARN checks.append(Check(f"storage_{name}", "storage", sev, f"{name}: directory not writable.", str(exc))) except Exception as exc: # noqa: BLE001 checks.append(Check(f"storage_{name}", "storage", STATUS_WARN, f"{name}: check error.", str(exc))) return checks def _check_provider_config(settings: Any) -> list[Check]: checks: list[Check] = [] # AI provider — never show key, only configured/missing provider = settings.ai_provider.strip().lower() if provider == "sarvam": has_key = bool(settings.sarvam_api_key) elif provider == "openrouter": has_key = bool(settings.openrouter_api_key) else: has_key = False key_status = STATUS_OK if has_key else STATUS_WARN checks.append(Check( "ai_provider_key", "provider", key_status if provider != "mock" else STATUS_WARN, f"AI provider: {provider} — key {'configured' if has_key else 'NOT configured'}." if provider != "mock" else "AI provider: mock — no real AI, fallback only.", )) # AI router checks.append(Check( "ai_router", "provider", STATUS_OK, f"AI_ROUTER_ENABLED={settings.ai_router_enabled}.", )) # TTS tts = getattr(settings, "tts_provider", None) or "ai4bharat" if tts in {"ai4bharat", "indic_parler", "indic_parler_tts"}: has_tts_key = bool(settings.huggingface_api_key) elif tts == "sarvam": has_tts_key = bool(settings.sarvam_api_key) elif tts == "indic_tts": has_tts_key = True # local else: has_tts_key = True # local/Edge providers need no paid API key tts_sev = STATUS_OK if has_tts_key else STATUS_WARN checks.append(Check( "tts_provider", "provider", tts_sev, f"TTS provider: {tts} — {'key configured' if has_tts_key else 'key NOT configured'}.", )) # PYQ discovery pyq_enabled = getattr(settings, "pyq_discovery_enabled", False) pyq_key = bool(getattr(settings, "pyq_search_api_key", None)) if pyq_enabled and not pyq_key: checks.append(Check( "pyq_discovery", "provider", STATUS_WARN, "PYQ_DISCOVERY_ENABLED=true but PYQ_SEARCH_API_KEY not set.", )) elif pyq_enabled: checks.append(Check( "pyq_discovery", "provider", STATUS_OK, "PYQ discovery enabled and key configured.", )) else: checks.append(Check( "pyq_discovery", "provider", STATUS_OK, "PYQ discovery disabled (safe default).", )) return checks def _check_service_imports() -> list[Check]: """Verify critical service modules import without error.""" services = [ ("source_guard", "app.services.source_guard", "Source Reality Guard"), ("evidence_contract", "app.services.evidence_contract", "Evidence Contract"), ("pyq_discovery", "app.services.pyq_discovery", "PYQ Discovery service"), ("video_study_planner", "app.services.video_study_planner", "Video Study Planner"), ("video_study_preview_renderer", "app.services.video_study_preview_renderer", "Video Preview Renderer"), ] checks: list[Check] = [] for key, module, label in services: try: __import__(module) checks.append(Check(f"import_{key}", "service", STATUS_OK, f"{label} — import OK.")) except ImportError as exc: checks.append(Check(f"import_{key}", "service", STATUS_FAIL, f"{label} — import FAILED.", str(exc))) except Exception as exc: # noqa: BLE001 checks.append(Check(f"import_{key}", "service", STATUS_WARN, f"{label} — import raised unexpected error.", str(exc))) return checks def _check_video_pipeline(settings: Any) -> list[Check]: checks: list[Check] = [] project_root = Path(__file__).resolve().parents[3] # Remotion render script render_script = project_root / "scripts" / "render-video-from-json.mjs" if render_script.exists(): checks.append(Check("render_script", "video", STATUS_OK, "Render script found at scripts/render-video-from-json.mjs.")) else: checks.append(Check("render_script", "video", STATUS_WARN, "Render script missing — video rendering will fail.", str(render_script))) # Node.js node = shutil.which("node") if node: try: result = subprocess.run([node, "--version"], capture_output=True, text=True, timeout=5) checks.append(Check("node_js", "video", STATUS_OK, f"Node.js available: {result.stdout.strip()}.")) except Exception: # noqa: BLE001 checks.append(Check("node_js", "video", STATUS_WARN, "Node.js found but version check failed.")) else: checks.append(Check("node_js", "video", STATUS_WARN, "Node.js not found — Remotion rendering unavailable.")) # ffmpeg / ffprobe for binary in ("ffmpeg", "ffprobe"): path = shutil.which(binary) if path: checks.append(Check(binary, "video", STATUS_OK, f"{binary} available.")) else: checks.append(Check(binary, "video", STATUS_WARN, f"{binary} not found — audio/video processing limited.")) # Generated video output dir writable video_out = Path(settings.generated_video_output_dir) if not video_out.is_absolute(): video_out = project_root / video_out probe = video_out / ".beta_health_probe" try: video_out.mkdir(parents=True, exist_ok=True) probe.write_text("probe") probe.unlink() checks.append(Check("video_output_dir", "video", STATUS_OK, "Video output directory writable.")) except OSError as exc: checks.append(Check("video_output_dir", "video", STATUS_WARN, "Video output directory not writable.", str(exc))) return checks def _check_security(settings: Any) -> list[Check]: checks: list[Check] = [] project_root = Path(__file__).resolve().parents[3] # .env not tracked in git for env_rel in (".env", "backend/.env"): env_path = project_root / env_rel try: result = subprocess.run( ["git", "ls-files", "--error-unmatch", str(env_path)], capture_output=True, cwd=str(project_root), timeout=10, ) if result.returncode == 0: checks.append(Check( f"env_not_tracked_{env_rel.replace('/', '_').replace('.', '_')}", "security", STATUS_FAIL, f"{env_rel} IS tracked in git — secrets may be exposed.", )) else: checks.append(Check( f"env_not_tracked_{env_rel.replace('/', '_').replace('.', '_')}", "security", STATUS_OK, f"{env_rel} not tracked in git.", )) except (FileNotFoundError, subprocess.TimeoutExpired): checks.append(Check( f"env_not_tracked_{env_rel.replace('/', '_').replace('.', '_')}", "security", STATUS_WARN, f"Could not verify git tracking for {env_rel} (git not available).", )) # Default JWT secret in use default_secret = "change-this-local-dev-secret" if settings.jwt_secret_key == default_secret and settings.environment != "development": checks.append(Check( "jwt_secret_not_default", "security", STATUS_FAIL, "JWT_SECRET_KEY is set to the default value — insecure in non-development.", )) elif settings.jwt_secret_key == default_secret: checks.append(Check( "jwt_secret_not_default", "security", STATUS_WARN, "JWT_SECRET_KEY is the default dev value — set a strong key for production.", )) else: checks.append(Check( "jwt_secret_not_default", "security", STATUS_OK, "JWT secret key is non-default.", )) # PYQ discovery safe-default pyq_enabled = getattr(settings, "pyq_discovery_enabled", False) if not pyq_enabled: checks.append(Check( "pyq_discovery_disabled_default", "security", STATUS_OK, "PYQ discovery disabled (safe default).", )) else: checks.append(Check( "pyq_discovery_disabled_default", "security", STATUS_WARN, "PYQ discovery is ENABLED — ensure this is intentional before launch.", )) # Quick scan for obvious secret patterns in tracked Python/env files try: result = subprocess.run( [ "git", "grep", "--count", "-E", r"(sk-[A-Za-z0-9]{20,}|AIza[A-Za-z0-9_-]{35}|ghp_[A-Za-z0-9]{36})", "--", "*.py", "*.env", ], capture_output=True, text=True, cwd=str(project_root), timeout=15, ) if result.stdout.strip(): checks.append(Check( "no_secrets_in_tracked_files", "security", STATUS_WARN, "Possible API key patterns found in tracked files — review before deploying.", result.stdout.strip()[:200], )) else: checks.append(Check( "no_secrets_in_tracked_files", "security", STATUS_OK, "No obvious API key patterns found in tracked files.", )) except (FileNotFoundError, subprocess.TimeoutExpired): checks.append(Check( "no_secrets_in_tracked_files", "security", STATUS_WARN, "Could not scan tracked files for secrets (git not available).", )) return checks # ── Endpoint ────────────────────────────────────────────────────────────────── @router.get( "/beta-health", summary="Beta Launch Doctor — pre-launch diagnostic checklist", description=( "Developer-only endpoint. Returns a structured checklist of system health checks " "required before closed-beta launch. Never exposes secret values. " "Accessible without auth in development; requires admin role in production." ), tags=["Dev"], ) def beta_health( db: Session = Depends(get_db), current_user: User | None = Depends(get_current_user_optional), ) -> dict[str, Any]: settings = get_settings() # Access control: allow in dev, require admin in production is_dev = settings.environment == "development" if not is_dev: if current_user is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Authentication required for beta-health outside development.", ) if current_user.role != "admin": raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Admin role required for beta-health endpoint.", ) report = BetaHealthReport(status=STATUS_OK) # 1. Database connection report.add(_check_db_connection(db)) # 2. Required schema columns for check in _check_required_columns(db): report.add(check) # 3. Storage directories for check in _check_storage(settings): report.add(check) # 4. Provider config (keys masked) for check in _check_provider_config(settings): report.add(check) # 5. Service imports (Source Reality Guard, Evidence Contract, etc.) for check in _check_service_imports(): report.add(check) # 6. Video pipeline for check in _check_video_pipeline(settings): report.add(check) # 7. Security for check in _check_security(settings): report.add(check) report.finalize() return report.to_dict()