import os, sys sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) import threading from datetime import datetime from fastapi import APIRouter, Depends, BackgroundTasks, HTTPException, Header from loguru import logger from api.dependencies import get_db router = APIRouter() _pipeline_status = { "running": False, "last_run": None, "last_summary": None, "last_error": None, } # NEW-A6 FIX: lock prevents two concurrent POST /admin/pipeline requests from # both passing the running=False check before either sets it to True _pipeline_lock = threading.Lock() def _generate_and_persist_admin_secret() -> str: """Generate a random per-install admin secret and persist it. Desktop groundwork (Phase D1), upgraded in Phase D6: tries the real OS credential store (Windows Credential Manager / macOS Keychain / Linux Secret Service) first, via config.credentials.set_credential(). Falls back to the legacy config.json mechanism (config.settings's set_local_setting) only if the keychain genuinely isn't available -- e.g. no backend installed, a headless Linux box with no Secret Service running. Either way this is a real, cryptographically random secret (secrets.token_urlsafe, not a placeholder); D6 only changes WHERE it's persisted, not how it's generated. set_local_setting() already injects into os.environ itself; the keychain path does not (that's load_credentials_into_environ()'s job, called once at process startup), so it's done explicitly here too -- otherwise a secret generated via the keychain path would be correct on disk/in-keychain but invisible to the REST of this same running process until next restart. """ import secrets as _secrets_module from config.credentials import set_credential from config.settings import set_local_setting new_secret = _secrets_module.token_urlsafe(32) if set_credential("ADMIN_SECRET", new_secret): os.environ["ADMIN_SECRET"] = new_secret logger.info("[Admin] Generated new per-install ADMIN_SECRET (OS keychain)") else: set_local_setting("ADMIN_SECRET", new_secret) logger.info( "[Admin] Generated new per-install ADMIN_SECRET " "(config.json fallback -- OS keychain unavailable)" ) return new_secret def _require_admin(x_admin_secret: str = Header(default="")): """ H-04 FIX: even in DEBUG_MODE, require a non-empty X-Admin-Secret header if ADMIN_SECRET is set. Previously DEBUG_MODE=true bypassed ALL auth, allowing anyone to POST /admin/pipeline (database wipe/reseed) on the production HuggingFace server if DEBUG_MODE was accidentally left on. Phase D1 FIX: desktop builds (BHARATGRAPH_DESKTOP_MODE=true, a flag the not-yet-built Tauri shell's backend-spawning code will set -- see PHASE_ROADMAP.md Phase D4) skip the DEBUG_MODE path entirely -- per SECURITY.md's Desktop Application Threat Model: "The web deployment's DEBUG_MODE auth-bypass path... is not present in desktop builds at all -- there is no bypass path to accidentally leave enabled." Instead, a missing secret is generated and persisted on the spot. Since BHARATGRAPH_DESKTOP_MODE is never set by any current deployment (nothing sets it until Phase D4 exists), this branch is unreachable today and the web deployment's behavior below is completely unchanged. """ # Importing config.settings (even though nothing below references it # by name) is not decorative -- it's what actually loads the local # config file's persisted values into os.environ. api/main.py, the # real application entry point, does the same import for the same # reason (defense in depth: this function's correctness must not # silently depend on something else having imported it first, since # this file can be imported and tested in isolation, as it was # during verification -- see the D1 commit message). import config.settings # noqa: F401 is_desktop_mode = os.getenv("BHARATGRAPH_DESKTOP_MODE", "").lower() in ("1", "true", "yes") if is_desktop_mode: secret = os.getenv("ADMIN_SECRET", "") if not secret or not secret.strip(): secret = _generate_and_persist_admin_secret() if x_admin_secret != secret: raise HTTPException( status_code=403, detail="Forbidden: invalid or missing X-Admin-Secret header", ) return # Web deployment: existing behavior, completely unchanged. secret = os.getenv("ADMIN_SECRET", "") is_debug = os.getenv("DEBUG_MODE", "").lower() in ("1", "true", "yes") if not secret or not secret.strip(): # SEC-6 FIX: empty string is not valid if not is_debug: raise HTTPException( status_code=503, detail="Admin endpoints disabled: set ADMIN_SECRET env var", ) # DEBUG_MODE without ADMIN_SECRET: warn but allow (local dev only) logger.warning("[Admin] No ADMIN_SECRET set -- auth bypassed (DEBUG_MODE)") return # ADMIN_SECRET is set: always enforce it regardless of DEBUG_MODE if x_admin_secret != secret: raise HTTPException( status_code=403, detail="Forbidden: invalid or missing X-Admin-Secret header", ) @router.post("/admin/seed") def seed_database( driver=Depends(get_db), _auth=Depends(_require_admin), ): """Load sample nodes AND relationships into Neo4j for demonstration.""" from graph.seed import ( SAMPLE_POLITICIANS, SAMPLE_COMPANIES, SAMPLE_CONTRACTS, SAMPLE_AUDIT_REPORTS, SAMPLE_DIRECTOR_LINKS, ) from graph.loader import GraphLoader loader = GraphLoader(driver=driver) p = loader.load_politicians(SAMPLE_POLITICIANS) c = loader.load_companies(SAMPLE_COMPANIES) k = loader.load_contracts(SAMPLE_CONTRACTS) a = loader.load_audit_reports(SAMPLE_AUDIT_REPORTS) d = loader.load_politician_company_links(SAMPLE_DIRECTOR_LINKS) return { "status": "seeded", "politicians": p, "companies": c, "contracts": k, "audit_reports": a, "director_links":d, "try_searching": [ "Modi", "Gandhi", "Adani", "Tata", "Infosys", "Amit Shah", "Anurag Thakur", "road construction", "audit Maharashtra" ], "note": ( "Sample data loaded with DIRECTOR_OF and WON_CONTRACT relationships. " "Search any name above to see the investigation graph." ), "next": "POST /admin/pipeline to ingest all 21 live government sources", } @router.post("/admin/pipeline") def trigger_pipeline( background_tasks: BackgroundTasks, scrapers: str = "all", driver=Depends(get_db), _auth=Depends(_require_admin), ): """Trigger full pipeline in background. scrapers: comma-separated or 'all'.""" # NEW-A6 FIX: lock prevents two simultaneous requests both passing the # running=False check before either sets running=True with _pipeline_lock: if _pipeline_status["running"]: return { "status": "already_running", "started": _pipeline_status["last_run"], "message": "Pipeline already running. Check /admin/pipeline/status", } _pipeline_status["running"] = True _pipeline_status["last_run"] = datetime.now().isoformat() scraper_list = None if scrapers == "all" else scrapers.split(",") background_tasks.add_task(_run_pipeline_background, scraper_list, driver) return { "status": "started", "scrapers": scrapers, "message": "Pipeline running in background. Check /admin/pipeline/status", "note": "Full run takes 3-8 minutes depending on network.", } @router.get("/admin/pipeline/status") def pipeline_status(): return { "running": _pipeline_status["running"], "last_run": _pipeline_status["last_run"], "last_summary": _pipeline_status["last_summary"], "last_error": _pipeline_status["last_error"], } def _run_pipeline_background(scraper_list, driver): """Background task -- runs pipeline then loads ALL scraped data into Neo4j.""" try: logger.info("[Admin] Background pipeline started") from processing.pipeline import BharatGraphPipeline from graph.loader import GraphLoader results = BharatGraphPipeline().run(scrapers=scraper_list, parallel=True) summary = results["summary"] raw = results.get("raw", {}) logger.info("[Admin] Pipeline done -- loading into Neo4j...") loader = GraphLoader(driver=driver) SCRAPER_TO_LOADER = { "myneta": "load_politicians", "mca": "load_companies", "gem": "load_contracts", "cag": "load_audit_reports", "pib": "load_press_releases", "sebi": "load_regulatory_orders", "ed": "load_enforcement_actions", "electoral_bond": "load_electoral_bonds", "ibbi": "load_insolvency_orders", "ngo_darpan": "load_ngos", "cppp": "load_tenders", "loksabha": "load_parliament_questions", "cvc": "load_vigilance_circulars", "icij": "load_icij_entities", "opensanctions": "load_sanctioned_entities", "njdg": "load_court_cases", "lgd": "load_local_bodies", "ncrb": "load_crime_reports", "wikidata": "load_wikidata_enrichments", "datagov": "load_datagov_documents", } loaded = {} for scraper_key, method_name in SCRAPER_TO_LOADER.items(): records = raw.get(scraper_key, []) if not records: logger.debug(f"[Admin] No records for {scraper_key}, skipping") continue method = getattr(loader, method_name, None) if method is None: logger.warning(f"[Admin] loader.{method_name}() not found -- skipping {scraper_key}") continue try: n = method(records) loaded[scraper_key] = n logger.info(f"[Admin] Loaded {n} {scraper_key} records via {method_name}()") except Exception as load_err: logger.error(f"[Admin] Failed loading {scraper_key}: {load_err}") loaded[scraper_key] = 0 total_loaded = sum(loaded.values()) _pipeline_status["last_summary"] = { **summary, "loaded": loaded, "total_loaded": total_loaded, "completed_at": datetime.now().isoformat(), } logger.success( f"[Admin] Pipeline complete -- " f"{total_loaded} total records loaded across {len(loaded)} datasets" ) except Exception as e: logger.error(f"[Admin] Pipeline failed: {e}") _pipeline_status["last_error"] = str(e) finally: _pipeline_status["running"] = False