#!/usr/bin/env python3 """Enrich IOCs using public threat intelligence APIs. Supports VirusTotal, OTX AlienVault, AbuseIPDB. Handles API keys via environment variables, implements rate limiting, and falls back gracefully when APIs are unavailable. """ from __future__ import annotations import argparse import json import os import sys import time from datetime import datetime from pathlib import Path try: import requests HAS_REQUESTS = True except ImportError: HAS_REQUESTS = False class ThreatEnricher: """Enrich IOCs with external threat intelligence.""" def __init__(self): self.vt_api_key = os.environ.get("VT_API_KEY") self.otx_api_key = os.environ.get("OTX_API_KEY") self.abuseipdb_key = os.environ.get("ABUSEIPDB_API_KEY") self.rate_limit_delay = 1.0 # seconds between API calls def enrich_hash(self, file_hash) -> dict: """Enrich a file hash with VirusTotal data.""" result = {"hash": file_hash, "sources": {}} if self.vt_api_key: try: headers = {"x-apikey": self.vt_api_key} resp = requests.get( f"https://www.virustotal.com/api/v3/files/{file_hash}", headers=headers, timeout=15 ) time.sleep(self.rate_limit_delay) if resp.status_code == 200: data = resp.json().get("data", {}).get("attributes", {}) stats = data.get("last_analysis_stats", {}) result["sources"]["virustotal"] = { "detected": stats.get("malicious", 0), "total": sum(stats.values()) if stats else 0, "type_description": data.get("type_description"), "popular_threat_classification": data.get("popular_threat_classification"), "names": data.get("names", [])[:5], "tags": data.get("tags", [])[:10], } elif resp.status_code == 404: result["sources"]["virustotal"] = {"status": "not_found"} else: result["sources"]["virustotal"] = {"error": f"HTTP {resp.status_code}"} except requests.RequestException as e: result["sources"]["virustotal"] = {"error": str(e)} else: result["sources"]["virustotal"] = {"status": "no_api_key"} return result def enrich_ip(self, ip_address) -> dict: """Enrich an IP address with reputation data.""" result = {"ip": ip_address, "sources": {}} # AbuseIPDB if self.abuseipdb_key: try: headers = {"Key": self.abuseipdb_key, "Accept": "application/json"} resp = requests.get( "https://api.abuseipdb.com/api/v2/check", params={"ipAddress": ip_address, "maxAgeInDays": 90}, headers=headers, timeout=15 ) time.sleep(self.rate_limit_delay) if resp.status_code == 200: data = resp.json().get("data", {}) result["sources"]["abuseipdb"] = { "abuse_confidence_score": data.get("abuseConfidenceScore"), "country": data.get("countryCode"), "isp": data.get("isp"), "total_reports": data.get("totalReports"), "is_tor": data.get("isTor"), } except requests.RequestException as e: result["sources"]["abuseipdb"] = {"error": str(e)} # VirusTotal IP lookup if self.vt_api_key: try: headers = {"x-apikey": self.vt_api_key} resp = requests.get( f"https://www.virustotal.com/api/v3/ip_addresses/{ip_address}", headers=headers, timeout=15 ) time.sleep(self.rate_limit_delay) if resp.status_code == 200: data = resp.json().get("data", {}).get("attributes", {}) stats = data.get("last_analysis_stats", {}) result["sources"]["virustotal"] = { "malicious": stats.get("malicious", 0), "suspicious": stats.get("suspicious", 0), "country": data.get("country"), "as_owner": data.get("as_owner"), } except requests.RequestException as e: result["sources"]["virustotal"] = {"error": str(e)} return result def enrich_domain(self, domain) -> dict: """Enrich a domain with reputation data.""" result = {"domain": domain, "sources": {}} # OTX AlienVault if self.otx_api_key: try: headers = {"X-OTX-API-KEY": self.otx_api_key} resp = requests.get( f"https://otx.alienvault.com/api/v1/indicators/domain/{domain}/general", headers=headers, timeout=15 ) time.sleep(self.rate_limit_delay) if resp.status_code == 200: data = resp.json() result["sources"]["otx"] = { "pulse_count": data.get("pulse_info", {}).get("count", 0), "alexa_rank": data.get("alexa"), "whois": data.get("whois"), } except requests.RequestException as e: result["sources"]["otx"] = {"error": str(e)} # VirusTotal domain lookup if self.vt_api_key: try: headers = {"x-apikey": self.vt_api_key} resp = requests.get( f"https://www.virustotal.com/api/v3/domains/{domain}", headers=headers, timeout=15 ) time.sleep(self.rate_limit_delay) if resp.status_code == 200: data = resp.json().get("data", {}).get("attributes", {}) stats = data.get("last_analysis_stats", {}) result["sources"]["virustotal"] = { "malicious": stats.get("malicious", 0), "suspicious": stats.get("suspicious", 0), "categories": data.get("categories", {}), "registrar": data.get("registrar"), } except requests.RequestException as e: result["sources"]["virustotal"] = {"error": str(e)} return result def enrich_iocs(self, iocs) -> dict: """Enrich a list of IOCs (auto-detect type).""" results = [] import re for ioc in iocs: ioc = ioc.strip() if not ioc: continue # Detect IOC type if re.match(r'^[a-fA-F0-9]{32,64}$', ioc): results.append(self.enrich_hash(ioc)) elif re.match(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$', ioc): results.append(self.enrich_ip(ioc)) elif re.match(r'^[a-zA-Z0-9]([a-zA-Z0-9-]*\.)+[a-zA-Z]{2,}$', ioc): results.append(self.enrich_domain(ioc)) else: results.append({"ioc": ioc, "type": "unknown", "error": "Could not determine IOC type"}) return results def main() -> None: parser = argparse.ArgumentParser( description="Enrich IOCs with threat intelligence" ) group = parser.add_mutually_exclusive_group(required=True) group.add_argument("--input", "--iocs", dest="iocs", help="JSON file with IOC list, or comma-separated IOCs") group.add_argument("--hash", help="Single file hash to enrich") group.add_argument("--ip", help="Single IP address to enrich") group.add_argument("--domain", help="Single domain to enrich") parser.add_argument("--output", "-o", help="Output file (JSON)") parser.add_argument("--format", choices=["json", "text"], default="text") args = parser.parse_args() if not HAS_REQUESTS: print("[!] requests library required. Install: pip install requests", file=sys.stderr) sys.exit(1) enricher = ThreatEnricher() # Check available APIs apis = [] if enricher.vt_api_key: apis.append("VirusTotal") if enricher.otx_api_key: apis.append("OTX") if enricher.abuseipdb_key: apis.append("AbuseIPDB") if not apis: print("[!] No API keys configured. Set VT_API_KEY, OTX_API_KEY, or ABUSEIPDB_API_KEY") print("[!] Running in offline mode - no enrichment available") print(f"[*] Available APIs: {', '.join(apis) if apis else 'None'}") # Process IOCs if args.hash: results = [enricher.enrich_hash(args.hash)] elif args.ip: results = [enricher.enrich_ip(args.ip)] elif args.domain: results = [enricher.enrich_domain(args.domain)] else: ioc_path = Path(args.iocs) if ioc_path.exists(): data = json.loads(ioc_path.read_text()) iocs = data if isinstance(data, list) else data.get("iocs", []) else: iocs = [i.strip() for i in args.iocs.split(",")] results = enricher.enrich_iocs(iocs) report = { "tool": "threat_enrichment", "timestamp": datetime.now().isoformat(), "apis_available": apis, "iocs_processed": len(results), "results": results, } if args.format == "json": output = json.dumps(report, indent=2) else: output = f"=== Threat Intelligence Enrichment ===\n" output += f"APIs: {', '.join(apis) if apis else 'None'}\n" output += f"IOCs processed: {len(results)}\n\n" for r in results: ioc_value = r.get("hash") or r.get("ip") or r.get("domain") or r.get("ioc", "unknown") output += f"--- {ioc_value} ---\n" for source, data in r.get("sources", {}).items(): output += f" [{source}] {json.dumps(data)}\n" output += "\n" if args.output: Path(args.output).write_text(output) print(f"[+] Report saved to {args.output}") else: print(output) if __name__ == "__main__": main()