undefined

undefined

undefined

undefined

undefined

In REPL, type natural language queries:

REPL commands:
- `<target>` - Investigate email/username/domain/IP/phone
- `clear` - Reset conversation memory
- `save` - Save last report to reports/
- `tools` - List available tools and status
- `config` - Show current configuration
- `help` - Show all commands
- `exit` / Ctrl-D - Exit

undefined

undefined

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}")

undefined

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

undefined

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"

undefined

undefined

undefined

undefined

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

undefined

undefined

undefined

undefined

undefined

undefined

undefined

undefined

undefined

undefined

undefined

undefined

undefined

undefined

undefined

undefined

undefined

undefined

undefined

undefined

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())