Building Automated IOC Extractors

#!/usr/bin/env python3
"""
Automated IOC extractor with threat intel enrichment.
Usage: python3 ioc-extractor.py   [--vt-key KEY] [--abuseipdb-key KEY]
"""
import subprocess, json, time, sys, os, glob, ipaddress, re
from pathlib import Path
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor, as_completed
import urllib.request, urllib.parse

# ── CONFIGURATION ─────────────────────────────────────────────────────
VT_API_KEY = os.getenv("VT_API_KEY", "")
ABUSE_API_KEY = os.getenv("ABUSEIPDB_API_KEY", "")

PRIVATE_NETS = [
    ipaddress.ip_network("10.0.0.0/8"),
    ipaddress.ip_network("172.16.0.0/12"),
    ipaddress.ip_network("192.168.0.0/16"),
    ipaddress.ip_network("127.0.0.0/8"),
    ipaddress.ip_network("169.254.0.0/16"),
    ipaddress.ip_network("224.0.0.0/4"),
    ipaddress.ip_network("240.0.0.0/4"),
]

ALLOWLIST_DOMAINS = {
    "microsoft.com", "windows.com", "windowsupdate.com",
    "google.com", "googleapis.com", "gstatic.com",
    "cloudflare.com", "akamai.com", "akamaiedge.net",
    "amazon.com", "amazonaws.com", "s3.amazonaws.com",
    "apple.com", "icloud.com",
}

def is_private(ip_str: str) -> bool:
    try:
        addr = ipaddress.ip_address(ip_str)
        return any(addr in net for net in PRIVATE_NETS)
    except ValueError:
        return True

def is_allowlisted(domain: str) -> bool:
    domain = domain.rstrip(".")
    for allowed in ALLOWLIST_DOMAINS:
        if domain == allowed or domain.endswith("." + allowed):
            return True
    return False

# ── EXTRACTION ─────────────────────────────────────────────────────────
def extract_from_pcap(pcap: str) -> dict:
    """Extract all IOC types from a single PCAP file."""
    def tshark(filter_str: str, fields: list) -> list:
        cmd = ["tshark", "-r", pcap, "-n", "-Y", filter_str, "-T", "fields",
               "-E", "separator=\t"] + [f for field in fields for f in ["-e", field]]
        try:
            result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
            return [line.split("\t") for line in result.stdout.splitlines() if line.strip()]
        except Exception:
            return []

    ips = set()
    domains = set()
    urls = set()
    sni = set()
    user_agents = set()

    # IPs from TCP SYN packets (external only)
    for row in tshark("tcp.flags.syn==1 and not tcp.flags.ack", ["ip.dst"]):
        ip = row[0].strip() if row else ""
        if ip and not is_private(ip):
            ips.add(ip)

    # DNS query names
    for row in tshark("dns.flags.response==0", ["dns.qry.name"]):
        name = row[0].strip() if row else ""
        if name and not is_allowlisted(name):
            domains.add(name)

    # HTTP hosts + URIs
    for row in tshark("http.request", ["http.host", "http.request.uri"]):
        host = row[0].strip() if len(row) > 0 else ""
        uri = row[1].strip() if len(row) > 1 else ""
        if host and not is_allowlisted(host):
            urls.add(f"http://{host}{uri}")

    # TLS SNI
    for row in tshark("tls.handshake.type==1", ["tls.handshake.extensions_server_name"]):
        name = row[0].strip() if row else ""
        if name and not is_allowlisted(name):
            sni.add(name)

    # User agents
    for row in tshark("http.user_agent", ["http.user_agent"]):
        ua = row[0].strip() if row else ""
        if ua:
            user_agents.add(ua)

    return {"ips": ips, "domains": domains, "urls": urls, "sni": sni, "user_agents": user_agents}

# ── ENRICHMENT ─────────────────────────────────────────────────────────
def vt_lookup(indicator: str, ioc_type: str) -> dict:
    if not VT_API_KEY:
        return {}
    endpoint = {"ip": "ip_addresses", "domain": "domains"}.get(ioc_type, "urls")
    url = f"https://www.virustotal.com/api/v3/{endpoint}/{urllib.parse.quote(indicator, safe='')}"
    try:
        req = urllib.request.Request(url, headers={"x-apikey": VT_API_KEY})
        with urllib.request.urlopen(req, timeout=10) as resp:
            data = json.loads(resp.read())
            stats = data.get("data", {}).get("attributes", {}).get("last_analysis_stats", {})
            return {"malicious": stats.get("malicious", 0), "suspicious": stats.get("suspicious", 0)}
    except Exception as e:
        return {"error": str(e)}

# ── MAIN ──────────────────────────────────────────────────────────────
def main():
    pcap_dir = Path(sys.argv[1])
    out_dir = Path(sys.argv[2])
    out_dir.mkdir(parents=True, exist_ok=True)

    pcap_files = sorted(glob.glob(str(pcap_dir / "*.pcap")))
    print(f"Processing {len(pcap_files)} PCAP files...")

    all_iocs = defaultdict(set)
    for pcap in pcap_files:
        print(f"  Extracting: {os.path.basename(pcap)}")
        result = extract_from_pcap(pcap)
        for key, values in result.items():
            all_iocs[key].update(values)

    print(f"\nExtracted:")
    for key, values in all_iocs.items():
        print(f"  {key}: {len(values)}")

    # Enrich IPs
    print(f"\nEnriching {len(all_iocs['ips'])} IPs against VirusTotal...")
    enriched = []
    for ip in sorted(all_iocs["ips"]):
        vt = vt_lookup(ip, "ip")
        score = vt.get("malicious", 0) * 3 + vt.get("suspicious", 0)
        enriched.append({"type": "ip", "value": ip, "vt_malicious": vt.get("malicious", 0),
                         "vt_suspicious": vt.get("suspicious", 0), "score": score})
        time.sleep(0.25)  # Rate limiting

    # Enrich domains
    print(f"Enriching {len(all_iocs['domains'])} domains...")
    for domain in sorted(all_iocs["domains"]):
        vt = vt_lookup(domain, "domain")
        score = vt.get("malicious", 0) * 3 + vt.get("suspicious", 0)
        enriched.append({"type": "domain", "value": domain, "vt_malicious": vt.get("malicious", 0),
                         "vt_suspicious": vt.get("suspicious", 0), "score": score})
        time.sleep(0.25)

    # Write outputs
    enriched.sort(key=lambda x: -x["score"])
    with open(out_dir / "enriched_iocs.json", "w") as f:
        json.dump(enriched, f, indent=2)

    print(f"\n=== Top 20 High-Score IOCs ===")
    for ioc in enriched[:20]:
        print(f"  score={ioc['score']:3d}  {ioc['type']:8}  {ioc['value']}")

    print(f"\nOutput: {out_dir}/enriched_iocs.json")

if __name__ == "__main__":
    main()