openosint-ai-osint-framework

# For local LLM (no API key required)
pip install ollama
ollama pull llama3.2
 
# For Shodan API
pip install shodan
export SHODAN_API_KEY=your-shodan-key
 
# For PDF reports
pip install reportlab
 
# API keys for enhanced functionality
export HIBP_API_KEY=your-hibp-key           # HaveIBeenPwned
export IPINFO_TOKEN=your-ipinfo-token       # ipinfo.io
export VIRUSTOTAL_API_KEY=your-vt-key       # VirusTotal

# Launch interactive AI agent
openosint
 
# Or explicitly
openosint shell
 
# Use local Ollama instead of Anthropic
openosint --provider ollama

openosint ❯ investigate user@example.com
openosint ❯ find accounts for johndoe99
openosint ❯ check breaches for admin@company.com
openosint ❯ enumerate subdomains of example.com

# Email investigation
openosint email target@example.com
openosint email target@example.com -t 60  # 60 second timeout
 
# Username search
openosint username johndoe99
openosint username johndoe99 -t 120
 
# Shodan lookup
openosint shodan 8.8.8.8
openosint shodan "apache port:80 country:DE"
 
# VirusTotal lookup
openosint virustotal example.com
openosint virustotal 192.168.1.1
 
# Multi-target investigation
openosint multi "user@example.com johndoe99 example.com"
 
# Parallel execution (run multiple tools simultaneously)
openosint --parallel email user@example.com
openosint --parallel username johndoe99
 
# JSON output for scripting
openosint --json email user@example.com

import asyncio
from openosint.tools.email import search_email
from openosint.tools.username import search_username
from openosint.tools.breach import search_breach
from openosint.tools.whois import search_whois
from openosint.tools.ip import search_ip
from openosint.tools.domain import search_domain
from openosint.tools.dorks import generate_dorks
from openosint.tools.paste import search_paste
from openosint.tools.phone import search_phone
from openosint.tools.shodan import search_shodan
from openosint.tools.virustotal import search_virustotal
 
async def investigate_email(email: str):
    """Investigate an email address."""
    # Search social accounts
    accounts = await search_email(email, timeout=60)
    print(f"Social accounts: {accounts}")
    
    # Check data breaches
    breaches = await search_breach(email)
    print(f"Breaches: {breaches}")
    
    # Search paste dumps
    pastes = await search_paste(email)
    print(f"Paste dumps: {pastes}")
    
    # Generate search dorks
    dorks = await generate_dorks(email)
    print(f"Google dorks: {dorks}")
 
async def investigate_username(username: str):
    """Find username across platforms."""
    platforms = await search_username(username, timeout=120)
    print(f"Found on platforms: {platforms}")
 
async def investigate_domain(domain: str):
    """Investigate a domain."""
    # WHOIS lookup
    whois = await search_whois(domain)
    print(f"WHOIS: {whois}")
    
    # Subdomain enumeration
    subdomains = await search_domain(domain)
    print(f"Subdomains: {subdomains}")
    
    # VirusTotal check
    vt_result = await search_virustotal(domain)
    print(f"VirusTotal: {vt_result}")
 
async def investigate_ip(ip: str):
    """Investigate an IP address."""
    # IP geolocation
    ip_info = await search_ip(ip)
    print(f"IP info: {ip_info}")
    
    # Shodan lookup
    shodan_data = await search_shodan(ip)
    print(f"Shodan: {shodan_data}")
 
async def investigate_phone(phone: str):
    """Investigate a phone number (E.164 format)."""
    phone_info = await search_phone(phone)
    print(f"Phone info: {phone_info}")
 
# Run investigations
asyncio.run(investigate_email("target@example.com"))
asyncio.run(investigate_username("johndoe99"))
asyncio.run(investigate_domain("example.com"))
asyncio.run(investigate_ip("8.8.8.8"))
asyncio.run(investigate_phone("+14155552671"))

import asyncio
from openosint.agent import OpenOSINTAgent
 
async def ai_investigation():
    """Run AI-powered investigation."""
    agent = OpenOSINTAgent(
        api_key=None,  # Uses ANTHROPIC_API_KEY from env
        provider="anthropic",  # or "ollama"
        model="claude-3-5-sonnet-20241022"
    )
    
    # Natural language investigation
    response = await agent.investigate("investigate user@example.com")
    print(response)
    
    # Agent automatically chains tools based on findings
    response = await agent.investigate(
        "find all accounts for johndoe99 and check for breaches"
    )
    print(response)
    
    # Get conversation history
    history = agent.get_history()
    
    # Clear conversation
    agent.clear_history()
 
asyncio.run(ai_investigation())

import asyncio
from openosint.tools.email import search_email
from openosint.tools.breach import search_breach
from openosint.tools.paste import search_paste
 
async def parallel_email_investigation(email: str):
    """Run multiple tools in parallel."""
    results = await asyncio.gather(
        search_email(email, timeout=60),
        search_breach(email),
        search_paste(email),
        return_exceptions=True  # Don't fail if one tool errors
    )
    
    accounts, breaches, pastes = results
    
    report = {
        "email": email,
        "accounts": accounts if not isinstance(accounts, Exception) else str(accounts),
        "breaches": breaches if not isinstance(breaches, Exception) else str(breaches),
        "pastes": pastes if not isinstance(pastes, Exception) else str(pastes)
    }
    
    return report
 
# Run parallel investigation
result = asyncio.run(parallel_email_investigation("target@example.com"))
print(result)

async def comprehensive_email_scan(email: str):
    """Complete email OSINT scan."""
    from openosint.tools.email import search_email
    from openosint.tools.breach import search_breach
    from openosint.tools.paste import search_paste
    from openosint.tools.dorks import generate_dorks
    
    print(f"[*] Investigating {email}")
    
    # Find social accounts
    print("[*] Searching social accounts...")
    accounts = await search_email(email)
    
    # Check breaches (requires HIBP_API_KEY)
    print("[*] Checking data breaches...")
    breaches = await search_breach(email)
    
    # Search paste dumps
    print("[*] Searching paste sites...")
    pastes = await search_paste(email)
    
    # Generate Google dorks
    print("[*] Generating Google dorks...")
    dorks = await generate_dorks(email)
    
    return {
        "accounts": accounts,
        "breaches": breaches,
        "pastes": pastes,
        "dorks": dorks
    }

async def comprehensive_username_scan(username: str):
    """Complete username OSINT scan."""
    from openosint.tools.username import search_username
    from openosint.tools.paste import search_paste
    from openosint.tools.dorks import generate_dorks
    
    print(f"[*] Investigating {username}")
    
    # Search across 300+ platforms
    print("[*] Searching platforms (this may take 2+ minutes)...")
    platforms = await search_username(username, timeout=180)
    
    # Search paste dumps
    print("[*] Searching paste sites...")
    pastes = await search_paste(username)
    
    # Generate Google dorks
    print("[*] Generating Google dorks...")
    dorks = await generate_dorks(username)
    
    return {
        "platforms": platforms,
        "pastes": pastes,
        "dorks": dorks
    }

async def comprehensive_domain_scan(domain: str):
    """Complete domain OSINT scan."""
    from openosint.tools.whois import search_whois
    from openosint.tools.domain import search_domain
    from openosint.tools.virustotal import search_virustotal
    from openosint.tools.dorks import generate_dorks
    
    print(f"[*] Investigating {domain}")
    
    # WHOIS lookup
    print("[*] Running WHOIS...")
    whois = await search_whois(domain)
    
    # Subdomain enumeration
    print("[*] Enumerating subdomains...")
    subdomains = await search_domain(domain)
    
    # VirusTotal scan (requires VIRUSTOTAL_API_KEY)
    print("[*] Checking VirusTotal...")
    vt_result = await search_virustotal(domain)
    
    # Generate Google dorks
    print("[*] Generating Google dorks...")
    dorks = await generate_dorks(domain)
    
    return {
        "whois": whois,
        "subdomains": subdomains,
        "virustotal": vt_result,
        "dorks": dorks
    }

async def comprehensive_ip_scan(ip: str):
    """Complete IP OSINT scan."""
    from openosint.tools.ip import search_ip
    from openosint.tools.shodan import search_shodan
    from openosint.tools.virustotal import search_virustotal
    
    print(f"[*] Investigating {ip}")
    
    # IP geolocation
    print("[*] Geolocating IP...")
    ip_info = await search_ip(ip)
    
    # Shodan lookup (requires SHODAN_API_KEY)
    print("[*] Querying Shodan...")
    shodan_data = await search_shodan(ip)
    
    # VirusTotal scan (requires VIRUSTOTAL_API_KEY)
    print("[*] Checking VirusTotal...")
    vt_result = await search_virustotal(ip)
    
    return {
        "ip_info": ip_info,
        "shodan": shodan_data,
        "virustotal": vt_result
    }

async def shodan_queries():
    """Advanced Shodan search examples."""
    from openosint.tools.shodan import search_shodan
    
    # Host lookup (specific IP)
    host = await search_shodan("8.8.8.8")
    
    # Banner search
    apache = await search_shodan("apache port:80")
    
    # Country-specific search
    german_servers = await search_shodan("apache port:80 country:DE")
    
    # Vulnerable systems
    heartbleed = await search_shodan("vuln:CVE-2014-0160")
    
    # ICS/SCADA
    scada = await search_shodan("port:502")  # Modbus
    
    # Webcams
    cameras = await search_shodan("Server: SQ-WEBCAM")
    
    return {
        "host": host,
        "apache": apache,
        "german_servers": german_servers,
        "heartbleed": heartbleed,
        "scada": scada,
        "cameras": cameras
    }

{
  "mcpServers": {
    "openosint": {
      "command": "python",
      "args": ["-m", "openosint.mcp_server"],
      "env": {
        "ANTHROPIC_API_KEY": "your-key",
        "HIBP_API_KEY": "your-key",
        "SHODAN_API_KEY": "your-key",
        "VIRUSTOTAL_API_KEY": "your-key",
        "IPINFO_TOKEN": "your-token"
      }
    }
  }
}

"Investigate user@example.com for social accounts and data breaches"
"Find all platforms where johndoe99 has accounts"
"Enumerate subdomains for example.com"

# Required for AI agent (unless using --provider ollama)
export ANTHROPIC_API_KEY=sk-ant-your-key
 
# Optional API keys
export HIBP_API_KEY=your-hibp-key           # For search_breach
export SHODAN_API_KEY=your-shodan-key       # For search_shodan
export VIRUSTOTAL_API_KEY=your-vt-key       # For search_virustotal
export IPINFO_TOKEN=your-ipinfo-token       # For enhanced search_ip
 
# For local LLM
export OLLAMA_HOST=http://localhost:11434   # Default Ollama endpoint

# Default timeouts
await search_email(email, timeout=60)      # 60 seconds
await search_username(username, timeout=120)  # 120 seconds (sherlock is slow)
await search_domain(domain, timeout=90)    # 90 seconds
await search_phone(phone, timeout=30)      # 30 seconds
await search_shodan(query, timeout=30)     # 30 seconds
 
# Custom timeout
await search_email(email, timeout=180)     # 3 minutes

async def batch_email_investigation(emails: list[str]):
    """Investigate multiple emails in parallel."""
    from openosint.tools.email import search_email
    from openosint.tools.breach import search_breach
    
    async def investigate_one(email: str):
        accounts = await search_email(email)
        breaches = await search_breach(email)
        return {
            "email": email,
            "accounts": accounts,
            "breaches": breaches
        }
    
    results = await asyncio.gather(
        *[investigate_one(email) for email in emails],
        return_exceptions=True
    )
    
    return [r for r in results if not isinstance(r, Exception)]
 
emails = ["user1@example.com", "user2@example.com", "user3@example.com"]
results = asyncio.run(batch_email_investigation(emails))

async def investigate_and_export(target: str, output_path: str):
    """Run investigation and save report."""
    from openosint.agent import OpenOSINTAgent
    import json
    
    agent = OpenOSINTAgent()
    response = await agent.investigate(f"investigate {target}")
    
    # Save as JSON
    with open(output_path, "w") as f:
        json.dump({
            "target": target,
            "findings": response,
            "timestamp": datetime.now().isoformat()
        }, f, indent=2)
    
    return response
 
asyncio.run(investigate_and_export(
    "user@example.com",
    "reports/investigation.json"
))

async def safe_investigation(email: str):
    """Investigate with proper error handling."""
    from openosint.tools.email import search_email
    from openosint.tools.breach import search_breach
    
    results = {}
    
    # Each tool runs independently
    try:
        results["accounts"] = await search_email(email, timeout=60)
    except asyncio.TimeoutError:
        results["accounts"] = "ERROR: Timeout after 60 seconds"
    except Exception as e:
        results["accounts"] = f"ERROR: {str(e)}"
    
    try:
        results["breaches"] = await search_breach(email)
    except Exception as e:
        results["breaches"] = f"ERROR: {str(e)}"
    
    return results

# Install holehe
pip install holehe
 
# Verify installation
which holehe
holehe --help

# Install sherlock-project
pip install sherlock-project
 
# Verify installation
which sherlock
sherlock --help

# Download binary from GitHub releases
wget https://github.com/sundowndev/phoneinfoga/releases/download/v2.11.0/phoneinfoga_Linux_x86_64.tar.gz
tar -xzf phoneinfoga_Linux_x86_64.tar.gz
sudo mv phoneinfoga /usr/local/bin/
chmod +x /usr/local/bin/phoneinfoga
 
# Verify
phoneinfoga version

# Get API key from https://haveibeenpwned.com/API/Key
export HIBP_API_KEY=your-actual-key

# Get API key from https://account.shodan.io
export SHODAN_API_KEY=your-actual-key
 
# Verify
python -c "import shodan; api=shodan.Shodan('$SHODAN_API_KEY'); print(api.info())"

# Get API key from https://www.virustotal.com/gui/my-apikey
export VIRUSTOTAL_API_KEY=your-actual-key

# Install Ollama
curl -fsSL https://ollama.com/install.sh | sh
 
# Pull a model
ollama pull llama3.2
 
# Verify running
curl http://localhost:11434/api/tags
 
# Use with OpenOSINT
openosint --provider ollama

# Increase timeout for slow tools
await search_username(username, timeout=300)  # 5 minutes for sherlock

# Add delays between requests
import asyncio
 
async def rate_limited_batch(emails: list[str], delay: float = 2.0):
    """Investigate emails with rate limiting."""
    results = []
    for email in emails:
        result = await search_email(email)
        results.append(result)
        await asyncio.sleep(delay)  # 2 second delay
    return results

# Stream results instead of loading all at once
async def stream_investigation(emails: list[str]):
    """Process results one at a time."""
    for email in emails:
        result = await search_email(email)
        # Process immediately
        print(f"Results for {email}: {result}")
        # Don't accumulate in memory
        del result

async def check_tool_status():
    """Verify which tools are available."""
    from openosint.tools.email import search_email
    from openosint.tools.username import search_username
    from openosint.tools.phone import search_phone
    import shutil
    import os
    
    status = {
        "holehe": shutil.which("holehe") is not None,
        "sherlock": shutil.which("sherlock") is not None,
        "sublist3r": shutil.which("sublist3r") is not None,
        "phoneinfoga": shutil.which("phoneinfoga") is not None,
        "hibp_api_key": os.getenv("HIBP_API_KEY") is not None,
        "shodan_api_key": os.getenv("SHODAN_API_KEY") is not None,
        "virustotal_api_key": os.getenv("VIRUSTOTAL_API_KEY") is not None,
        "ipinfo_token": os.getenv("IPINFO_TOKEN") is not None,
    }
    
    for tool, available in status.items():
        symbol = "✓" if available else "✗"
        print(f"{symbol} {tool}")
    
    return status
 
asyncio.run(check_tool_status())