import os import sqlite3 import time import socket import urllib.request import urllib.error import ipaddress import json import hashlib DB_PATH = '/home/snuffleupagus/teamwork_projects/axiom_zero_audit/dev_control_center/threat_intel_cache.db' DATACENTER_ASN_PREFIXES = [ # AWS '3.0.0.0/8', '13.0.0.0/8', '15.0.0.0/8', '18.0.0.0/8', '34.0.0.0/8', '35.0.0.0/8', '52.0.0.0/8', '54.0.0.0/8', # GCP '35.190.0.0/16', '34.64.0.0/10', '104.196.0.0/14', '35.200.0.0/13', # Azure '20.0.0.0/8', '40.0.0.0/8', '51.0.0.0/8', '52.0.0.0/8', # DigitalOcean '104.131.0.0/18', '138.68.0.0/16', '159.65.0.0/16', # Hetzner '5.9.0.0/16', '78.46.0.0/15', '88.99.0.0/16', # Vultr '45.32.0.0/12', '45.63.0.0/18', # Linode '45.33.0.0/17', '45.56.0.0/21', '45.79.0.0/16', # Cloudflare / OVH / Oracle '104.16.0.0/13', '149.202.0.0/16', '129.146.0.0/15' ] DATACENTER_NETWORKS = [ipaddress.ip_network(p) for p in DATACENTER_ASN_PREFIXES] def init_db(): os.makedirs(os.path.dirname(DB_PATH), exist_ok=True) conn = sqlite3.connect(DB_PATH) conn.execute('PRAGMA journal_mode=WAL;') conn.execute(''' CREATE TABLE IF NOT EXISTS ip_cache ( ip TEXT PRIMARY KEY, abuseipdb_score INT, spamhaus_listed BOOL, spamhaus_list_type TEXT, is_tor BOOL, is_datacenter BOOL, cached_at TIMESTAMP ) ''') conn.commit() conn.close() init_db() class AbuseIPDB: API_URL = 'https://api.abuseipdb.com/api/v2/check' def __init__(self, api_key: str = None): self.api_key = api_key or os.environ.get('ABUSEIPDB_API_KEY') self.demo_mode = not bool(self.api_key) def check_ip(self, ip: str) -> dict: if self.demo_mode: # Synthetic score based on deterministic hash of IP for demo mode if ip.startswith('127.') or ip.startswith('192.168.') or ip.startswith('10.'): score = 0 elif ip == '198.51.100.1': # example spam ip score = 100 else: score = int(hashlib.md5(ip.encode('utf-8')).hexdigest(), 16) % 60 return { 'confidence_score': score, 'is_public': True, 'abuse_categories': [], 'total_reports': score // 10, 'last_reported': '2023-01-01T00:00:00Z' } else: req = urllib.request.Request( f"{self.API_URL}?ipAddress={ip}&maxAgeInDays=30", headers={ 'Accept': 'application/json', 'Key': self.api_key } ) try: with urllib.request.urlopen(req) as response: data = json.loads(response.read().decode('utf-8')) return { 'confidence_score': data['data']['abuseConfidenceScore'], 'is_public': data['data']['isPublic'], 'abuse_categories': data['data'].get('usageType', []), 'total_reports': data['data']['totalReports'], 'last_reported': data['data']['lastReportedAt'] } except urllib.error.URLError: return {'confidence_score': 0, 'is_public': True, 'abuse_categories': [], 'total_reports': 0, 'last_reported': ''} class SpamhausDBL: def check_ip(self, ip: str) -> dict: if not ip or ':' in ip: return {'listed': False, 'list_type': '', 'response_code': ''} try: reversed_ip = '.'.join(reversed(ip.split('.'))) query = f"{reversed_ip}.zen.spamhaus.org" answers = socket.getaddrinfo(query, None) for answer in answers: addr = answer[4][0] if addr.startswith('127.0.0.'): code = int(addr.split('.')[-1]) list_type = "UNKNOWN" if code == 2: list_type = "SBL" elif code == 3: list_type = "SBL CSS" elif 4 <= code <= 7: list_type = "XBL" elif 10 <= code <= 11: list_type = "PBL" return {'listed': True, 'list_type': list_type, 'response_code': addr} return {'listed': False, 'list_type': '', 'response_code': ''} except socket.gaierror: return {'listed': False, 'list_type': '', 'response_code': ''} class TorExitDetector: LIST_URL = 'https://check.torproject.org/torbulkexitlist' def __init__(self): self._exit_nodes: set = set() self._last_fetch: float = 0 self._fetch_interval = 6 * 3600 # 6 hours def _refresh_if_stale(self): if time.time() - self._last_fetch > self._fetch_interval: try: req = urllib.request.Request( self.LIST_URL, headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'} ) with urllib.request.urlopen(req) as response: content = response.read().decode('utf-8') new_nodes = set() for line in content.splitlines(): line = line.strip() if line and not line.startswith('#'): new_nodes.add(line) if new_nodes: self._exit_nodes = new_nodes self._last_fetch = time.time() except urllib.error.URLError as e: print(f"Warning: Failed to fetch Tor exit list: {e}") def is_tor_exit(self, ip: str) -> bool: self._refresh_if_stale() return ip in self._exit_nodes def is_datacenter_ip(ip: str) -> bool: try: ip_obj = ipaddress.ip_address(ip) for network in DATACENTER_NETWORKS: if ip_obj in network: return True return False except ValueError: return False def get_cached_result(ip: str): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row conn.execute('PRAGMA journal_mode=WAL;') cur = conn.cursor() cutoff = time.time() - 3600 # Enforce 1-hour TTL cur.execute('SELECT * FROM ip_cache WHERE ip = ? AND cached_at > ?', (ip, cutoff)) row = cur.fetchone() conn.close() if row: return dict(row) return None def save_cache(ip: str, abuseipdb_score: int, spamhaus_listed: bool, spamhaus_list_type: str, is_tor: bool, is_datacenter: bool): conn = sqlite3.connect(DB_PATH) conn.execute('PRAGMA journal_mode=WAL;') conn.execute(''' INSERT OR REPLACE INTO ip_cache (ip, abuseipdb_score, spamhaus_listed, spamhaus_list_type, is_tor, is_datacenter, cached_at) VALUES (?, ?, ?, ?, ?, ?, ?) ''', (ip, abuseipdb_score, spamhaus_listed, spamhaus_list_type, is_tor, is_datacenter, time.time())) conn.commit() conn.close() abuse_db = AbuseIPDB() spamhaus = SpamhausDBL() tor_detector = TorExitDetector() def enrich_ip(ip: str) -> dict: if not ip or not isinstance(ip, str): return { 'ip': str(ip) if ip is not None else '', 'abuseipdb_score': 0, 'spamhaus_listed': False, 'spamhaus_list_type': '', 'is_tor_exit': False, 'is_datacenter': False, 'overall_risk_score': 0.0, 'risk_level': 'INVALID', 'recommendation': 'BLOCK' } # Validate IP format try: socket.inet_aton(ip) except (socket.error, OSError, TypeError): try: ipaddress.ip_address(ip) except (ValueError, TypeError): return { 'ip': ip, 'abuseipdb_score': 0, 'spamhaus_listed': False, 'spamhaus_list_type': '', 'is_tor_exit': False, 'is_datacenter': False, 'overall_risk_score': 0.0, 'risk_level': 'INVALID', 'recommendation': 'BLOCK' } cached = get_cached_result(ip) if cached: abuseipdb_score = cached['abuseipdb_score'] spamhaus_listed = bool(cached['spamhaus_listed']) spamhaus_list_type = cached['spamhaus_list_type'] is_tor = bool(cached['is_tor']) is_datacenter = bool(cached['is_datacenter']) else: abuse_info = abuse_db.check_ip(ip) abuseipdb_score = abuse_info['confidence_score'] spam_info = spamhaus.check_ip(ip) spamhaus_listed = spam_info['listed'] spamhaus_list_type = spam_info['list_type'] is_tor = tor_detector.is_tor_exit(ip) is_datacenter = is_datacenter_ip(ip) save_cache(ip, abuseipdb_score, spamhaus_listed, spamhaus_list_type, is_tor, is_datacenter) overall_risk_score = (abuseipdb_score / 100.0 * 0.4) + (int(spamhaus_listed) * 0.3) + (int(is_tor) * 0.2) + (int(is_datacenter) * 0.1) if overall_risk_score >= 0.7: risk_level = 'CRITICAL' recommendation = 'BLOCK' elif overall_risk_score >= 0.5: risk_level = 'HIGH' recommendation = 'BLOCK' elif overall_risk_score >= 0.3: risk_level = 'MEDIUM' recommendation = 'CHALLENGE' elif overall_risk_score > 0: risk_level = 'LOW' recommendation = 'CHALLENGE' else: risk_level = 'NONE' recommendation = 'ALLOW' return { 'ip': ip, 'abuseipdb_score': abuseipdb_score, 'spamhaus_listed': spamhaus_listed, 'spamhaus_list_type': spamhaus_list_type, 'is_tor_exit': is_tor, 'is_datacenter': is_datacenter, 'overall_risk_score': round(overall_risk_score, 2), 'risk_level': risk_level, 'recommendation': recommendation } if __name__ == '__main__': tor_detector._refresh_if_stale() known_tor = next(iter(tor_detector._exit_nodes)) if tor_detector._exit_nodes else '103.208.220.122' test_ips = [ known_tor, # Tor exit node '127.0.0.2', # Known spam IP for Zen Spamhaus testing '8.8.8.8', # Clean residential / DNS '3.128.0.0', # AWS IP (Datacenter) '127.0.0.1', # Local IP 'invalid.ip.address', # Invalid hostname/IP test '2001:db8::1' # IPv6 address test ] print("Testing Threat Intel Pipeline...") for ip in test_ips: res = enrich_ip(ip) print(f"\n--- Results for {ip} ---") for k, v in res.items(): print(f" {k}: {v}")