shadowhunter_api.py

#!/usr/bin/env python3
"""
ShadowHunter - Dark Web Credential Intelligence Platform
Module: FastAPI Backend Server
Author: Fevra
Version: 0.1.0

This provides the REST API backend for the web dashboard.
"""

from fastapi import FastAPI, HTTPException, WebSocket, WebSocketDisconnect, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, Response
from pydantic import BaseModel
from typing import List, Dict, Optional, Any
from datetime import datetime, timedelta
import asyncio
import json
import logging
import os
import time

# Import ShadowHunter modules
# from shadowhunter_core import ShadowHunter
# from shadowhunter_tor import DarkWebIntelligencePlatform
# from shadowhunter_telegram import TelegramIntelligencePlatform
# from shadowhunter_neo4j import Neo4jManager, ThreatIntelligenceGraph

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('ShadowHunter.API')

# Initialize FastAPI
app = FastAPI(
    title="ShadowHunter API",
    description="Dark Web Threat Intelligence Platform API",
    version="0.1.0"
)

# CORS middleware for frontend
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # In production, specify your frontend URL
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


# ---------------------------------------------------------------------------
# Prometheus metrics middleware (optional)
# ---------------------------------------------------------------------------
try:
    from shadowhunter_metrics import init_metrics, get_metrics_output, record_request
    init_metrics()

    @app.middleware("http")
    async def prometheus_middleware(request, call_next):
        start = time.perf_counter()
        response = await call_next(request)
        duration = time.perf_counter() - start
        path = request.url.path or "/"
        if path != "/metrics":
            record_request(request.method, path, response.status_code, duration)
        return response
except ImportError:
    pass

# ============================================================================
# DATA MODELS
# ============================================================================

class Alert(BaseModel):
    id: int
    severity: str
    title: str
    domain: str
    source: str
    description: str
    timestamp: str
    metadata: Dict[str, Any] = {}

class Domain(BaseModel):
    domain: str
    organization: str = "Unknown"
    risk_score: int
    threat_count: int
    status: str

class ThreatActor(BaseModel):
    username: str
    platform: str
    activity_count: int
    reputation: Optional[str] = None

class StealerLog(BaseModel):
    log_id: str
    stealer_type: str
    credentials_count: int
    cookies_count: int
    price: Optional[float] = None
    channel: str
    timestamp: str

class SystemStatus(BaseModel):
    status: str
    active_modules: List[str]
    last_scan: str
    total_threats: int
    critical_alerts: int


class HuntRequest(BaseModel):
    """Request body for quick hunt: run hunters on one or more content snippets."""
    raw_content: Optional[str] = None
    query: Optional[str] = None
    # If both omitted, backend uses a minimal sample event.

# ============================================================================
# IN-MEMORY STORAGE (Replace with actual database in production)
# ============================================================================

class DataStore:
    def __init__(self):
        self.alerts: List[Alert] = []
        self.domains: List[Domain] = []
        self.threat_actors: List[ThreatActor] = []
        self.stealer_logs: List[StealerLog] = []
        self.system_status = SystemStatus(
            status="running",
            active_modules=["credential_monitor", "tor_scraper", "telegram_monitor"],
            last_scan=datetime.utcnow().isoformat(),
            total_threats=0,
            critical_alerts=0
        )
        
        # Initialize with sample data
        self._load_sample_data()
    
    def _load_sample_data(self):
        """Load sample data for demo purposes"""
        # Sample alerts
        self.alerts = [
            Alert(
                id=1,
                severity="CRITICAL",
                title="Credential Leak Detected",
                domain="acmecorp.com",
                source="Telegram",
                description="Admin credentials found in Lumma stealer log",
                timestamp=(datetime.utcnow() - timedelta(minutes=2)).isoformat(),
                metadata={"credential_count": 45, "has_passwords": True}
            ),
            Alert(
                id=2,
                severity="HIGH",
                title="Ransomware Victim Listing",
                domain="techcompany.com",
                source="Tor",
                description="Company listed on LockBit 4.0 leak site",
                timestamp=(datetime.utcnow() - timedelta(minutes=15)).isoformat(),
                metadata={"ransomware_group": "LockBit 4.0", "ransom_amount": 5000000}
            ),
            Alert(
                id=3,
                severity="HIGH",
                title="IAB Network Access Sale",
                domain="corporation.com",
                source="Dark Web",
                description="Domain admin access listed for $85,000",
                timestamp=(datetime.utcnow() - timedelta(hours=1)).isoformat(),
                metadata={"access_type": "Domain Admin", "price": 85000}
            )
        ]
        
        # Sample domains
        self.domains = [
            Domain(domain="acmecorp.com", organization="Acme Corp", risk_score=85, threat_count=12, status="Critical"),
            Domain(domain="techcompany.com", organization="Tech Company", risk_score=72, threat_count=8, status="High"),
            Domain(domain="corporation.com", organization="Corporation", risk_score=58, threat_count=5, status="Medium")
        ]

data_store = DataStore()

# WebSocket connections for real-time updates
active_connections: List[WebSocket] = []

# ============================================================================
# WEBSOCKET ENDPOINT FOR REAL-TIME UPDATES
# ============================================================================

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    active_connections.append(websocket)
    logger.info(f"WebSocket client connected. Total connections: {len(active_connections)}")
    
    try:
        while True:
            # Send periodic updates
            await asyncio.sleep(5)
            
            # Simulate new threat detection
            update = {
                "type": "threat_update",
                "data": {
                    "total_threats": data_store.system_status.total_threats,
                    "timestamp": datetime.utcnow().isoformat()
                }
            }
            await websocket.send_json(update)
            
    except WebSocketDisconnect:
        active_connections.remove(websocket)
        logger.info(f"WebSocket client disconnected. Total connections: {len(active_connections)}")

async def broadcast_alert(alert: Alert):
    """Broadcast new alert to all connected WebSocket clients"""
    message = {
        "type": "new_alert",
        "data": alert.dict()
    }
    
    for connection in active_connections:
        try:
            await connection.send_json(message)
        except:
            pass

# ============================================================================
# API ENDPOINTS
# ============================================================================

def _api_version() -> str:
    try:
        from shadowhunter._version import __version__
        return __version__
    except Exception:
        return "0.1.0"


@app.get("/")
async def root():
    """API root endpoint"""
    return {
        "name": "ShadowHunter API",
        "version": _api_version(),
        "status": "operational",
        "docs": "/docs"
    }

def _check_neo4j() -> dict:
    """Return { "status": "ok" | "unavailable", "detail": str }."""
    try:
        from config import get_config
        from neo4j import GraphDatabase
        c = get_config()
        uri = getattr(c.database, "neo4j_uri", "bolt://localhost:7687")
        user = getattr(c.database, "neo4j_user", "neo4j")
        password = getattr(c.database, "neo4j_password", "") or ""
        driver = GraphDatabase.driver(uri, auth=(user, password))
        driver.verify_connectivity()
        driver.close()
        return {"status": "ok", "detail": "connected"}
    except Exception as e:
        return {"status": "unavailable", "detail": str(e)}


def _check_redis() -> dict:
    """Return { "status": "ok" | "unavailable", "detail": str }."""
    try:
        import redis
        from config import get_config
        c = get_config()
        url = getattr(c.database, "redis_url", "redis://localhost:6379")
        r = redis.from_url(url)
        r.ping()
        return {"status": "ok", "detail": "connected"}
    except Exception as e:
        return {"status": "unavailable", "detail": str(e)}


def _check_hunters() -> bool:
    """True if Phase 3 hunters can be imported."""
    try:
        from shadowhunter.hunters import PasteHunter  # noqa: F401
        return True
    except ImportError:
        return False


@app.get("/api/health")
async def health_check():
    """Health check: Neo4j, Redis, and hunters availability."""
    neo4j = _check_neo4j()
    redis_status = _check_redis()
    hunters_available = _check_hunters()
    overall = "healthy" if neo4j["status"] == "ok" else "degraded"
    return {
        "status": overall,
        "timestamp": datetime.utcnow().isoformat(),
        "neo4j": neo4j,
        "redis": redis_status,
        "hunters_available": hunters_available,
        "modules": {
            "credential_monitor": True,
            "tor_scraper": True,
            "telegram_monitor": True,
        },
    }


@app.get("/metrics", include_in_schema=False)
async def metrics():
    """Prometheus metrics endpoint for monitoring (scrape at /metrics)."""
    try:
        from shadowhunter_metrics import get_metrics_output
        body, content_type = get_metrics_output()
        return Response(content=body, media_type=content_type)
    except ImportError:
        return Response(
            content=b"# Prometheus client not installed\n",
            media_type="text/plain; charset=utf-8",
        )

# ============================================================================
# OVERVIEW ENDPOINTS
# ============================================================================

@app.get("/api/overview")
async def get_overview():
    """Get dashboard overview statistics"""
    return {
        "total_threats": len(data_store.alerts),
        "critical_alerts": len([a for a in data_store.alerts if a.severity == "CRITICAL"]),
        "monitored_domains": len(data_store.domains),
        "active_sources": 8,
        "last_update": datetime.utcnow().isoformat(),
        "threat_trend": "+12%",
        "credentials_found": 3456,
        "stealer_logs": 127,
        "iab_listings": 18
    }

@app.get("/api/threat-trend")
async def get_threat_trend():
    """Get 7-day threat trend data"""
    # Generate sample trend data
    trend_data = []
    for i in range(7):
        date = datetime.utcnow() - timedelta(days=6-i)
        trend_data.append({
            "date": date.strftime("%a"),
            "threats": 150 + (i * 10) + (i % 2) * 20,
            "critical": 5 + (i % 3) * 2
        })
    return trend_data

@app.get("/api/threat-distribution")
async def get_threat_distribution():
    """Get threat type distribution"""
    return [
        {"name": "Stealer Logs", "value": 127, "color": "#ef4444"},
        {"name": "Breaches", "value": 34, "color": "#f97316"},
        {"name": "IAB Listings", "value": 18, "color": "#eab308"},
        {"name": "Ransomware", "value": 15, "color": "#8b5cf6"}
    ]

@app.get("/api/source-activity")
async def get_source_activity():
    """Get data source activity statistics"""
    return [
        {"source": "Telegram", "count": 543, "percentage": 65},
        {"source": "Tor Network", "count": 312, "percentage": 23},
        {"source": "Pastebin", "count": 156, "percentage": 8},
        {"source": "HIBP", "count": 78, "percentage": 4}
    ]

# ============================================================================
# ALERTS ENDPOINTS
# ============================================================================

@app.get("/api/alerts")
async def get_alerts(
    severity: Optional[str] = None,
    domain: Optional[str] = None,
    limit: int = 50,
    offset: int = 0,
):
    """Get list of alerts with optional filtering and pagination (limit/offset)."""
    alerts = data_store.alerts
    if severity and severity.upper() != "ALL":
        alerts = [a for a in alerts if a.severity == severity.upper()]
    if domain:
        alerts = [a for a in alerts if domain.lower() in a.domain.lower()]
    total = len(alerts)
    page = alerts[offset : offset + limit]
    return {"items": page, "total": total, "limit": limit, "offset": offset}

@app.get("/api/alerts/{alert_id}", response_model=Alert)
async def get_alert(alert_id: int):
    """Get specific alert by ID"""
    alert = next((a for a in data_store.alerts if a.id == alert_id), None)
    if not alert:
        raise HTTPException(status_code=404, detail="Alert not found")
    return alert

@app.post("/api/alerts")
async def create_alert(alert: Alert):
    """Create new alert (called by monitoring modules)"""
    data_store.alerts.insert(0, alert)
    data_store.system_status.total_threats += 1
    
    if alert.severity == "CRITICAL":
        data_store.system_status.critical_alerts += 1
    
    # Broadcast to WebSocket clients
    await broadcast_alert(alert)
    
    logger.info(f"New alert created: {alert.title} ({alert.severity})")
    return {"status": "success", "alert_id": alert.id}

# ============================================================================
# DOMAINS ENDPOINTS
# ============================================================================

@app.get("/api/domains", response_model=List[Domain])
async def get_domains():
    """Get list of monitored domains"""
    return data_store.domains

@app.get("/api/domains/{domain}")
async def get_domain_details(domain: str):
    """Get detailed information for a specific domain"""
    domain_obj = next((d for d in data_store.domains if d.domain == domain), None)
    if not domain_obj:
        raise HTTPException(status_code=404, detail="Domain not found")
    
    # Get alerts for this domain
    domain_alerts = [a for a in data_store.alerts if a.domain == domain]
    
    return {
        "domain": domain_obj.dict(),
        "alerts": [a.dict() for a in domain_alerts],
        "threat_timeline": [],  # Would include historical data
        "risk_factors": ["Credential leaks", "Ransomware target", "IAB listing"]
    }

@app.post("/api/domains")
async def add_monitored_domain(domain: str, organization: str = "Unknown"):
    """Add a new domain to monitoring"""
    # Check if already exists
    if any(d.domain == domain for d in data_store.domains):
        raise HTTPException(status_code=400, detail="Domain already monitored")
    
    new_domain = Domain(
        domain=domain,
        organization=organization,
        risk_score=0,
        threat_count=0,
        status="Low"
    )
    data_store.domains.append(new_domain)
    
    logger.info(f"Added new monitored domain: {domain}")
    return {"status": "success", "domain": domain}

# ============================================================================
# THREAT ACTORS ENDPOINTS
# ============================================================================

@app.get("/api/threat-actors", response_model=List[ThreatActor])
async def get_threat_actors(limit: int = 10):
    """Get list of top threat actors"""
    # Sample data - in production, query from Neo4j
    actors = [
        ThreatActor(username="seller_pro", platform="Telegram", activity_count=45, reputation="Verified"),
        ThreatActor(username="access_broker", platform="XSS.is", activity_count=38, reputation="High"),
        ThreatActor(username="elite_stealer", platform="Telegram", activity_count=32, reputation="Verified"),
        ThreatActor(username="ransom_admin", platform="Dark Web", activity_count=28, reputation="Medium"),
        ThreatActor(username="breach_seller", platform="Telegram", activity_count=24, reputation="Low")
    ]
    return actors[:limit]

@app.get("/api/threat-actors/{username}")
async def get_threat_actor_details(username: str, platform: str):
    """Get detailed information about a threat actor"""
    # In production, query Neo4j for:
    # - Activity history
    # - Posted listings
    # - Associated threats
    # - Network graph
    
    return {
        "username": username,
        "platform": platform,
        "activity_count": 45,
        "reputation": "Verified",
        "first_seen": (datetime.utcnow() - timedelta(days=90)).isoformat(),
        "last_seen": datetime.utcnow().isoformat(),
        "listings": [],
        "associated_threats": []
    }

# ============================================================================
# ANALYTICS ENDPOINTS
# ============================================================================

@app.get("/api/analytics/stealer-logs")
async def get_stealer_log_analytics():
    """Get stealer malware analytics"""
    return [
        {"type": "Lumma", "count": 45, "avgPrice": 65, "credentials": 8920},
        {"type": "Risepro", "count": 32, "avgPrice": 45, "credentials": 5670},
        {"type": "Raccoon", "count": 28, "avgPrice": 55, "credentials": 6340},
        {"type": "Redline", "count": 15, "avgPrice": 35, "credentials": 3210},
        {"type": "Vidar", "count": 7, "avgPrice": 50, "credentials": 1890}
    ]

@app.get("/api/analytics/breach-trends")
async def get_breach_trends():
    """Get data breach trend analytics"""
    trends = []
    for i in range(6):
        month = datetime.utcnow() - timedelta(days=30*i)
        trends.insert(0, {
            "month": month.strftime("%b"),
            "breaches": 15 + (i * 3),
            "records": (2500000 + (i * 500000))
        })
    return trends

@app.get("/api/analytics/risk-timeline/{domain}")
async def get_domain_risk_timeline(domain: str, days: int = 30):
    """Get risk score timeline for a domain"""
    timeline = []
    for i in range(days):
        date = datetime.utcnow() - timedelta(days=days-i)
        timeline.append({
            "date": date.strftime("%Y-%m-%d"),
            "risk_score": 50 + (i % 10) * 5
        })
    return timeline

# ============================================================================
# SEARCH ENDPOINTS
# ============================================================================

@app.get("/api/search")
async def search(query: str, type: Optional[str] = None):
    """Global search across all data"""
    results = {
        "alerts": [],
        "domains": [],
        "threat_actors": []
    }
    
    query_lower = query.lower()
    
    # Search alerts
    if not type or type == "alerts":
        results["alerts"] = [
            a.dict() for a in data_store.alerts
            if query_lower in a.title.lower() or query_lower in a.description.lower()
        ]
    
    # Search domains
    if not type or type == "domains":
        results["domains"] = [
            d.dict() for d in data_store.domains
            if query_lower in d.domain.lower() or query_lower in d.organization.lower()
        ]
    
    return results

# ============================================================================
# SCAN CONTROL ENDPOINTS
# ============================================================================

@app.post("/api/scan/start")
async def start_scan(module: Optional[str] = None):
    """Start a scan (all modules or specific module)"""
    # In production, this would trigger actual scans
    # asyncio.create_task(shadowhunter.run_scan())
    
    logger.info(f"Starting scan: {module or 'all modules'}")
    return {
        "status": "started",
        "module": module or "all",
        "timestamp": datetime.utcnow().isoformat()
    }

@app.get("/api/scan/status")
async def get_scan_status():
    """Get current scan status"""
    return {
        "running": True,
        "modules": {
            "credential_monitor": {"status": "active", "last_scan": "2 minutes ago"},
            "tor_scraper": {"status": "active", "last_scan": "15 minutes ago"},
            "telegram_monitor": {"status": "active", "last_scan": "real-time"},
            "neo4j": {"status": "connected", "nodes": 12456}
        }
    }

# ============================================================================
# EXPORT ENDPOINTS
# ============================================================================

@app.get("/api/export/alerts")
async def export_alerts(format: str = "json"):
    """Export alerts data"""
    if format == "json":
        return {"data": [a.dict() for a in data_store.alerts]}
    elif format == "csv":
        # Generate CSV
        return {"error": "CSV export not yet implemented"}
    else:
        raise HTTPException(status_code=400, detail="Unsupported format")

@app.get("/api/export/report/{domain}")
async def export_threat_report(domain: str, format: str = "json"):
    """Export comprehensive threat report for a domain"""
    domain_obj = next((d for d in data_store.domains if d.domain == domain), None)
    if not domain_obj:
        raise HTTPException(status_code=404, detail="Domain not found")
    
    report = {
        "domain": domain,
        "generated_at": datetime.utcnow().isoformat(),
        "risk_assessment": domain_obj.dict(),
        "alerts": [a.dict() for a in data_store.alerts if a.domain == domain],
        "summary": {
            "total_threats": domain_obj.threat_count,
            "risk_score": domain_obj.risk_score,
            "status": domain_obj.status
        }
    }
    
    return report

# ============================================================================
# QUICK HUNT ENDPOINT (multi-agent hunters)
# ============================================================================

@app.post("/api/hunt", response_model=Dict[str, Any])
async def run_quick_hunt(request: Request, body: HuntRequest):
    """
    Run the multi-agent hunter pipeline on the given content.
    Accepts raw_content (paste/text) or query (natural language); builds event stream and returns synthesis report.
    Requires Neo4j and Phase 3 hunters to be available. Rate-limited per IP (see HUNT_RATE_LIMIT_PER_MIN).
    """
    client_ip = request.client.host if request.client else "unknown"
    if not _check_hunt_rate_limit(client_ip):
        raise HTTPException(status_code=429, detail="Rate limit exceeded for /api/hunt. Try again later.")
    _audit_log("/api/hunt", "POST", client_ip, detail="quick_hunt")
    try:
        from config import get_config
        from neo4j import GraphDatabase
        from shadowhunter.orchestration import Orchestrator, EscapeHatchRegistry
        from shadowhunter.hunters import (
            PasteHunter, CryptoHunter, AttributionHunter, AgenticProfiler,
            DACOAnalyst, IdentityHunter, EntropyMonitor, PipelineDefender,
            SupplyChainHunter, AIThreatHunter, InsiderThreatHunter, SyntheticIDHunter,
            PhysicalSecHunter, ExploitMarketHunter, SocialMediaHunter, GitHubHunter,
            APTHunter, ThreatScoringEngine,
        )
    except ImportError as e:
        logger.warning("Hunt endpoint: hunters not available: %s", e)
        raise HTTPException(
            status_code=503,
            detail="Hunter system not available. Install dependencies and ensure Neo4j is configured.",
        ) from e

    config = get_config()
    uri = getattr(config.database, "neo4j_uri", "bolt://localhost:7687")
    user = getattr(config.database, "neo4j_user", "neo4j")
    password = getattr(config.database, "neo4j_password", "") or "password"
    try:
        db_driver = GraphDatabase.driver(uri, auth=(user, password))
        db_driver.verify_connectivity()
    except Exception as e:
        logger.warning("Hunt endpoint: Neo4j connection failed: %s", e)
        raise HTTPException(status_code=503, detail=f"Neo4j connection failed: {e}") from e

    hunter_config = getattr(config, "hunters", None) or {}
    config_by_hunter = {
        "paste_hunter": hunter_config.get("paste_hunter", {}),
        "crypto_hunter": hunter_config.get("crypto_hunter", {}),
        "attribution_hunter": hunter_config.get("attribution_hunter", {}),
        "agentic_profiler": hunter_config.get("agentic_profiler", {}),
        "daco_analyst": hunter_config.get("daco_analyst", {}),
        "identity_hunter": hunter_config.get("identity_hunter", {}),
        "entropy_monitor": hunter_config.get("entropy_monitor", {}),
        "pipeline_defender": hunter_config.get("pipeline_defender", {}),
        "supply_chain_hunter": hunter_config.get("supply_chain_hunter", {}),
        "ai_threat_hunter": hunter_config.get("ai_threat_hunter", {}),
        "insider_threat_hunter": hunter_config.get("insider_threat_hunter", {}),
        "synthetic_id_hunter": hunter_config.get("synthetic_id_hunter", {}),
        "physical_sec_hunter": hunter_config.get("physical_sec_hunter", {}),
        "exploit_market_hunter": hunter_config.get("exploit_market_hunter", {}),
        "social_media_hunter": hunter_config.get("social_media_hunter", {}),
        "github_hunter": hunter_config.get("github_hunter", {}),
        "apt_hunter": hunter_config.get("apt_hunter", {}),
        "threat_scoring_engine": hunter_config.get("threat_scoring", hunter_config.get("threat_scoring_engine", {})),
    }
    session = db_driver.session()
    hunters_list = [
        PasteHunter(session, config_by_hunter["paste_hunter"]),
        CryptoHunter(session, config_by_hunter["crypto_hunter"]),
        AttributionHunter(session, config_by_hunter["attribution_hunter"]),
        AgenticProfiler(session, config_by_hunter["agentic_profiler"]),
        DACOAnalyst(session, config_by_hunter["daco_analyst"]),
        IdentityHunter(session, config_by_hunter["identity_hunter"]),
        EntropyMonitor(session, config_by_hunter["entropy_monitor"]),
        PipelineDefender(session, config_by_hunter["pipeline_defender"]),
        SupplyChainHunter(session, config_by_hunter["supply_chain_hunter"]),
        AIThreatHunter(session, config_by_hunter["ai_threat_hunter"]),
        InsiderThreatHunter(session, config_by_hunter["insider_threat_hunter"]),
        SyntheticIDHunter(session, config_by_hunter["synthetic_id_hunter"]),
        PhysicalSecHunter(session, config_by_hunter["physical_sec_hunter"]),
        ExploitMarketHunter(session, config_by_hunter["exploit_market_hunter"]),
        SocialMediaHunter(session, config_by_hunter["social_media_hunter"]),
        GitHubHunter(session, config_by_hunter["github_hunter"]),
        APTHunter(session, config_by_hunter["apt_hunter"]),
        ThreatScoringEngine(session, config_by_hunter["threat_scoring_engine"]),
    ]
    escape_registry = EscapeHatchRegistry()
    orchestrator = Orchestrator(hunters_list, db_driver, escape_registry)

    raw = (body.raw_content or body.query or "").strip()
    if raw:
        event_stream = [{
            "source": "quick_hunt",
            "raw_content": raw,
            "stix_objects": [],
            "timestamp": datetime.utcnow().isoformat(),
            "metadata": {},
        }]
    else:
        event_stream = [{
            "source": "quick_hunt",
            "raw_content": "Sample paste with wallet 4AdkPJoxn7JCvAqP31V1EBhNj4pes2u3uT5r2W2R8f2N2abc and email test@example.com",
            "stix_objects": [],
            "timestamp": datetime.utcnow().isoformat(),
            "metadata": {},
        }]

    try:
        report = await orchestrator.run(event_stream)
        db_driver.close()
        return report
    except Exception as e:
        db_driver.close()
        logger.exception("Hunt run failed: %s", e)
        raise HTTPException(status_code=500, detail=str(e)) from e


# ============================================================================
# NL-TO-CYPHER (Ask the graph)
# ============================================================================

class NLToCypherRequest(BaseModel):
    """Natural language question for the threat graph."""
    query: str


@app.post("/api/nl-to-cypher")
async def nl_to_cypher_endpoint(body: NLToCypherRequest):
    """
    Translate natural language to read-only Cypher, execute, and return results.
    Requires Neo4j and optional OPENAI_API_KEY for LLM-backed translation.
    """
    try:
        from config import get_config
        from neo4j import GraphDatabase
        from shadowhunter.nl_to_cypher import translate_nl_to_cypher, validate_cypher_readonly
        from shadowhunter.nl_to_cypher.schema_context import load_schema
    except ImportError as e:
        raise HTTPException(status_code=503, detail=f"NL-to-Cypher not available: {e}") from e

    config = get_config()
    uri = getattr(config.database, "neo4j_uri", "bolt://localhost:7687")
    user = getattr(config.database, "neo4j_user", "neo4j")
    password = getattr(config.database, "neo4j_password", "") or "password"
    try:
        driver = GraphDatabase.driver(uri, auth=(user, password))
        driver.verify_connectivity()
    except Exception as e:
        raise HTTPException(status_code=503, detail=f"Neo4j connection failed: {e}") from e

    nl = (body.query or "").strip()
    if not nl:
        raise HTTPException(status_code=400, detail="query is required")

    # Prefer LLM translator if schema and API key available; else rule-based
    cypher = None
    try:
        from shadowhunter.nl_to_cypher.translator import LLMNLToCypherTranslator
        import asyncio
        schema_str = await load_schema(driver)
        translator = LLMNLToCypherTranslator(
            api_key=os.environ.get("OPENAI_API_KEY"),
            base_url=os.environ.get("OPENAI_BASE_URL"),
            model=os.environ.get("OPENAI_MODEL", "gpt-4o"),
        )
        translator.set_schema(schema_str)
        cypher = await translator.translate(nl)
    except Exception:
        cypher = translate_nl_to_cypher(nl)

    # If LLM returned no client or UNABLE_TO_QUERY, use rule-based fallback (no API key needed)
    if not cypher or cypher.strip().upper() == "UNABLE_TO_QUERY":
        cypher = translate_nl_to_cypher(nl)

    if not cypher or cypher.strip().upper() == "UNABLE_TO_QUERY":
        driver.close()
        return {"cypher": "", "results": [], "error": "Could not translate question to Cypher."}

    valid, err = validate_cypher_readonly(cypher, driver)
    if not valid:
        driver.close()
        return {"cypher": cypher, "results": [], "error": err or "Validation failed"}

    def _serialize(obj: Any) -> Any:
        """Make Neo4j Node/Relationship and other types JSON-serializable."""
        if obj is None or isinstance(obj, (bool, int, float, str)):
            return obj
        if hasattr(obj, "labels") and hasattr(obj, "items"):
            return {"_type": "node", "labels": list(obj.labels), "properties": {k: _serialize(v) for k, v in obj.items()}}
        if hasattr(obj, "type") and hasattr(obj, "start_node"):
            return {"_type": "relationship", "type": obj.type, "start_node": _serialize(obj.start_node), "end_node": _serialize(obj.end_node)}
        if isinstance(obj, dict):
            return {k: _serialize(v) for k, v in obj.items()}
        if isinstance(obj, (list, tuple)):
            return [_serialize(x) for x in obj]
        return str(obj)

    try:
        with driver.session() as session:
            result = session.run(cypher)
            records = list(result)
            out = []
            for r in records:
                out.append({k: _serialize(r[k]) for k in r.keys()})
        driver.close()
        return {"cypher": cypher, "results": out, "error": None}
    except Exception as e:
        driver.close()
        return {"cypher": cypher, "results": [], "error": str(e)}


# ============================================================================
# CONFIGURATION ENDPOINTS
# ============================================================================

@app.get("/api/config")
async def get_configuration():
    """Get current system configuration"""
    return {
        "monitored_domains": [d.domain for d in data_store.domains],
        "scan_intervals": {
            "telegram": 60,
            "tor": 21600,
            "pastebin": 300
        },
        "alert_thresholds": {
            "risk_score": 70,
            "credential_count": 10
        }
    }

@app.put("/api/config")
async def update_configuration(config: Dict[str, Any]):
    """Update system configuration"""
    # In production, update actual configuration
    logger.info(f"Configuration updated: {config}")
    return {"status": "success", "config": config}

# ============================================================================
# RATE LIMITING (in-memory; optional)
# ============================================================================

_hunt_rate: Dict[str, list] = {}  # ip -> [timestamps]
HUNT_RATE_LIMIT = int(os.environ.get("HUNT_RATE_LIMIT_PER_MIN", "10"))
HUNT_RATE_WINDOW = 60  # seconds


def _check_hunt_rate_limit(client_ip: str) -> bool:
    """True if request is within limit (HUNT_RATE_LIMIT per minute)."""
    now = time.time()
    if client_ip not in _hunt_rate:
        _hunt_rate[client_ip] = []
    times = _hunt_rate[client_ip]
    times[:] = [t for t in times if now - t < HUNT_RATE_WINDOW]
    if len(times) >= HUNT_RATE_LIMIT:
        return False
    times.append(now)
    return True


# ============================================================================
# AUDIT LOG (optional; set AUDIT_LOG_ENABLED=1 and optionally AUDIT_LOG_PATH)
# ============================================================================

def _audit_log(endpoint: str, method: str, client_ip: str, detail: str = ""):
    """Append a minimal audit line (timestamp, endpoint, method, ip, detail)."""
    if not os.environ.get("AUDIT_LOG_ENABLED", "").strip().lower() in ("1", "true", "yes"):
        return
    path = os.environ.get("AUDIT_LOG_PATH", "").strip() or os.path.join(os.getcwd(), "logs", "audit.log")
    try:
        os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
        with open(path, "a", encoding="utf-8") as f:
            f.write(
                f"{datetime.utcnow().isoformat()}Z {method} {endpoint} ip={client_ip} {detail}\n"
            )
    except Exception as e:
        logger.warning("Audit log write failed: %s", e)


# ============================================================================
# STARTUP AND SHUTDOWN EVENTS
# ============================================================================

@app.on_event("startup")
async def startup_event():
    """Initialize modules on startup; enforce JWT secret in production."""
    logger.info("ShadowHunter API starting up...")
    try:
        from config import get_config
        c = get_config()
        secret = getattr(c.api, "jwt_secret_key", "") or ""
        if secret == "change-this-in-production":
            if os.environ.get("ENV", "").lower() in ("production", "prod"):
                raise ValueError(
                    "JWT_SECRET_KEY must be set in production. Do not use default. Set JWT_SECRET_KEY in .env."
                )
            logger.warning("JWT_SECRET_KEY is default. Set JWT_SECRET_KEY in production.")
    except ValueError as e:
        logger.error(str(e))
        raise
    except Exception:
        pass
    logger.info("ShadowHunter API ready")

@app.on_event("shutdown")
async def shutdown_event():
    """Cleanup on shutdown"""
    logger.info("ShadowHunter API shutting down...")
    
    # Close all WebSocket connections
    for connection in active_connections:
        await connection.close()
    
    # In production, cleanup:
    # - Close Neo4j connection
    # - Disconnect Tor
    # - Stop Telegram client
    
    logger.info("ShadowHunter API stopped")

# ============================================================================
# RUN SERVER
# ============================================================================

if __name__ == "__main__":
    import uvicorn
    
    uvicorn.run(
        "shadowhunter_api:app",
        host="0.0.0.0",
        port=8000,
        reload=True,
        log_level="info"
    )