Executive Threat Intelligence & Open-Source Investigation Center

Suspicious Shortlink Unshortener (curl -ILs)

import os, re, subprocess, json
from typing import List, Optional
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from google import genai
from google.genai import types

load_dotenv()
# Initialize official Google GenAI SDK client (reads GEMINI_API_KEY from environment)
client = genai.Client(api_key=os.environ.get("GEMINI_API_KEY"))
MODEL_ID = os.environ.get("REASONING_MODEL", "gemini-2.5-pro")

app = FastAPI(title="OSINT & Threat Intelligence Suite - Kali & AI Bridge", version="5.0.0")
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])

# --- PYDANTIC STRUCTURED OUTPUT SCHEMA ---
class AttackPathLink(BaseModel):
    step: int = Field(description="Sequential step in the cross-sector exposure chain")
    sector_source: str = Field(description="Originating OSINT sector and tool")
    technical_finding: str = Field(description="Specific technical artifact or exposure")
    pivot_impact: str = Field(description="How an adversary or investigator pivots from this artifact")

class RemediationAction(BaseModel):
    priority: str = Field(description="CRITICAL, HIGH, or MEDIUM")
    control_domain: str = Field(description="e.g., Cloudflare WAF, VIP OPSEC, DMARC, EXIF Pipeline")
    defensive_action: str = Field(description="Concrete defensive engineering or executive protection fix")
    verification_cmd: str = Field(description="CLI command to verify remediation")

class ThreatAssessment(BaseModel):
    case_id: str
    composite_risk_score: int = Field(ge=0, le=100, description="Overall risk score 0-100")
    severity_rating: str = Field(description="CRITICAL, HIGH, ELEVATED, or LOW")
    executive_summary: str
    cross_sector_attack_paths: List[AttackPathLink]
    prioritized_remediation_playbook: List[RemediationAction]

class SynthesisRequest(BaseModel):
    case_id: str = "CASE-2026-089"
    thinking_budget: str = "HIGH"  # LOW (4096), MEDIUM (16384), HIGH (32768)
    temperature: float = 0.2
    findings: List[dict]
    analyst_query: Optional[str] = None

SYSTEM_INSTRUCTION = """You are a principal defensive cybersecurity architect and licensed private investigation OSINT correlation engine.
Analyze multi-sector passive OSINT findings (Sherlock, theHarvester, Amass, Wafw00f, ExifTool, Blockchain ledger hops).
Identify how disparate digital and physical exposures chain together, and produce a strictly defensive, actionable Risk Mitigation Dossier."""

def map_thinking_tokens(level: str) -> int:
    return {"LOW": 4096, "MEDIUM": 16384, "HIGH": 32768}.get(level.upper(), 16384)

@app.post("/api/v1/synthesize", response_model=ThreatAssessment)
def synthesize_dossier(req: SynthesisRequest):
    prompt = f"Case ID: {req.case_id}\nAnalyst Directive: {req.analyst_query or 'Full cross-sector correlation'}\nEvidence Payload:\n{json.dumps(req.findings, indent=2)}"
    response = client.models.generate_content(
        model=MODEL_ID,
        contents=prompt,
        config=types.GenerateContentConfig(
            system_instruction=SYSTEM_INSTRUCTION,
            temperature=req.temperature,
            response_mime_type="application/json",
            response_schema=ThreatAssessment,
            thinking_config=types.ThinkingConfig(
                include_thoughts=True,
                thinking_budget=map_thinking_tokens(req.thinking_budget)
            ),
        ),
    )
    return ThreatAssessment.model_validate_json(response.text)

@app.post("/api/v1/stream")
def stream_reasoning(req: SynthesisRequest):
    prompt = f"Correlate OSINT findings for {req.case_id} and explain your reasoning step-by-step:\n{json.dumps(req.findings)}"
    def event_generator():
        for chunk in client.models.generate_content_stream(
            model=MODEL_ID,
            contents=prompt,
            config=types.GenerateContentConfig(
                system_instruction=SYSTEM_INSTRUCTION,
                temperature=req.temperature,
                thinking_config=types.ThinkingConfig(include_thoughts=True, thinking_budget=16384)
            )
        ):
            if chunk.text:
                yield f"data: {json.dumps({'text': chunk.text})}\n\n"
    return StreamingResponse(event_generator(), media_type="text/event-stream")

# --- STRICT ALLOWLIST PASSIVE KALI OSINT SUBPROCESS WRAPPERS ---
SAFE_TARGET_RE = re.compile(r"^[a-zA-Z0-9._@-]{2,80}$")

def validate_target(target: str) -> str:
    if not SAFE_TARGET_RE.match(target):
        raise HTTPException(status_code=400, detail="Target rejected by strict allowlist sanitizer.")
    return target

@app.get("/api/v1/osint/{tool_name}/{target}")
def run_passive_osint(tool_name: str, target: str):
    clean_target = validate_target(target)
    allowlisted_cmds = {
        "sherlock": ["sherlock", clean_target, "--timeout", "5", "--print-found"],
        "theharvester": ["theHarvester", "-d", clean_target, "-b", "crtsh,certspotter,otx"],
        "wafw00f": ["wafw00f", f"https://{clean_target}"],
        "exiftool": ["exiftool", "-a", "-u", "-g1", clean_target]
    }
    if tool_name not in allowlisted_cmds:
        raise HTTPException(status_code=403, detail="Tool not in passive OSINT allowlist.")
    proc = subprocess.run(allowlisted_cmds[tool_name], capture_output=True, text=True, timeout=45)
    return {"tool": tool_name, "target": clean_target, "stdout": proc.stdout}

Interactive Case Interrogation Assistant

Ask follow-up investigative or defensive mitigation questions grounded in the active case artifacts.