#!/usr/bin/env python3 """JA3 fingerprint spoofing module: rewrite outbound TLS ClientHello to match common browser fingerprints, defeating JA3-based network detection.""" import json import logging import os import ssl import struct import subprocess import time import threading from pathlib import Path from typing import Optional from modules.base import BaseModule logger = logging.getLogger(__name__) # Built-in JA3 fingerprint profiles (fallback if data/ja3_fingerprints.db absent) # Format: {name: {ja3_hash, cipher_suites, extensions, description}} BUILTIN_PROFILES = { "chrome_120_win": { "ja3_hash": "cd08e31494f9531f560d64c695473da9", "description": "Chrome 120 on Windows 10/11", "cipher_suites": [ 0x1301, 0x1302, 0x1303, # TLS 1.3: AES_128_GCM, AES_256_GCM, CHACHA20 0xc02b, 0xc02f, 0xc02c, 0xc030, # ECDHE_ECDSA/RSA with AES-GCM 0xcca9, 0xcca8, # ECDHE with CHACHA20 0xc013, 0xc014, # ECDHE_RSA with AES-CBC 0x009c, 0x009d, # AES-GCM 0x002f, 0x0035, # AES-CBC ], "extensions": [0, 23, 65281, 10, 11, 35, 16, 5, 13, 18, 51, 45, 43, 27, 17513, 21], "elliptic_curves": [0x001d, 0x0017, 0x0018], # x25519, secp256r1, secp384r1 "ec_point_formats": [0], # uncompressed }, "firefox_121_win": { "ja3_hash": "579ccef312d18482fc42e2b822ca2430", "description": "Firefox 121 on Windows 10/11", "cipher_suites": [ 0x1301, 0x1303, 0x1302, 0xc02b, 0xc02f, 0xcca9, 0xcca8, 0xc02c, 0xc030, 0xc013, 0xc014, 0x009c, 0x009d, 0x002f, 0x0035, ], "extensions": [0, 23, 65281, 10, 11, 35, 16, 5, 34, 51, 43, 13, 45, 28, 21], "elliptic_curves": [0x001d, 0x0017, 0x0018, 0x0019], "ec_point_formats": [0], }, "edge_120_win": { "ja3_hash": "b32309a26951912be7dba376398abc3b", "description": "Edge 120 on Windows 10/11", "cipher_suites": [ 0x1301, 0x1302, 0x1303, 0xc02b, 0xc02f, 0xc02c, 0xc030, 0xcca9, 0xcca8, 0xc013, 0xc014, 0x009c, 0x009d, 0x002f, 0x0035, ], "extensions": [0, 23, 65281, 10, 11, 35, 16, 5, 13, 18, 51, 45, 43, 27, 17513, 21], "elliptic_curves": [0x001d, 0x0017, 0x0018], "ec_point_formats": [0], }, "chrome_120_linux": { "ja3_hash": "a17a3bfd385b62b1e15606dbd08c9f89", "description": "Chrome 120 on Linux", "cipher_suites": [ 0x1301, 0x1302, 0x1303, 0xc02b, 0xc02f, 0xc02c, 0xc030, 0xcca9, 0xcca8, 0xc013, 0xc014, 0x009c, 0x009d, 0x002f, 0x0035, ], "extensions": [0, 23, 65281, 10, 11, 35, 16, 5, 13, 18, 51, 45, 43, 27, 17513, 21], "elliptic_curves": [0x001d, 0x0017, 0x0018], "ec_point_formats": [0], }, } class JA3Spoofer(BaseModule): """Spoof JA3 TLS fingerprints on outbound HTTPS connections to match common browser profiles, evading JA3-based detection.""" name = "ja3_spoofer" module_type = "stealth" priority = -300 requires_root = True def __init__(self, bus, state, config, engine=None): super().__init__(bus, state, config, engine) self._profiles = dict(BUILTIN_PROFILES) self._active_profile: Optional[str] = config.get("target_profile", None) self._nfqueue_proc: Optional[subprocess.Popen] = None self._packets_modified = 0 self._use_nfqueue = config.get("use_nfqueue", True) self._nfqueue_num = config.get("nfqueue_num", 42) self._lock = threading.Lock() self._iptables_rules: list[list[str]] = [] # ------------------------------------------------------------------ # BaseModule interface # ------------------------------------------------------------------ def start(self) -> None: self._running = True self._start_time = time.time() self._pid = os.getpid() # Load external fingerprint database if available self._load_fingerprint_db() # Select target profile based on network environment if not self._active_profile: self._active_profile = self._auto_select_profile() profile = self._profiles.get(self._active_profile) if not profile: logger.warning("JA3 profile '%s' not found, using chrome_120_win", self._active_profile) self._active_profile = "chrome_120_win" profile = self._profiles["chrome_120_win"] # Apply cipher suite ordering to Python SSL contexts self._configure_ssl_context(profile) # Set up iptables NFQUEUE for non-Python TLS (bettercap, mitmproxy) if self._use_nfqueue: self._setup_nfqueue() self.state.set_module_status(self.name, "running", pid=self._pid) self.bus.emit("MODULE_STARTED", {"module": self.name}, source_module=self.name) logger.info( "JA3Spoofer active — profile: %s (%s)", self._active_profile, profile.get("description", "unknown"), ) def stop(self) -> None: self._running = False self._teardown_nfqueue() self.state.set_module_status(self.name, "stopped") self.bus.emit("MODULE_STOPPED", {"module": self.name}, source_module=self.name) logger.info("JA3Spoofer stopped (modified %d packets)", self._packets_modified) def status(self) -> dict: profile = self._profiles.get(self._active_profile, {}) return { "running": self._running, "pid": self._pid, "uptime": time.time() - self._start_time if self._start_time else 0, "active_ja3_hash": profile.get("ja3_hash", "none"), "target_browser": profile.get("description", "none"), "active_profile": self._active_profile, "packets_modified": self._packets_modified, "nfqueue_active": self._nfqueue_proc is not None and self._nfqueue_proc.poll() is None, "profiles_loaded": len(self._profiles), } def configure(self, config: dict) -> None: if "target_profile" in config: self._active_profile = config["target_profile"] profile = self._profiles.get(self._active_profile) if profile: self._configure_ssl_context(profile) logger.info("JA3 profile switched to: %s", self._active_profile) # ------------------------------------------------------------------ # SSL context configuration (Python requests/urllib) # ------------------------------------------------------------------ def _configure_ssl_context(self, profile: dict) -> None: """Configure the default Python SSL context with specific cipher ordering to match the target JA3 fingerprint.""" try: cipher_names = self._cipher_ids_to_openssl_names(profile.get("cipher_suites", [])) if not cipher_names: logger.warning("No cipher names resolved for profile") return cipher_string = ":".join(cipher_names) # Patch the default SSL context ctx = ssl.create_default_context() ctx.set_ciphers(cipher_string) ctx.minimum_version = ssl.TLSVersion.TLSv1_2 # Store for other modules to use self.state.set(self.name, "ssl_cipher_string", cipher_string) self.state.set(self.name, "active_ja3", profile.get("ja3_hash", "")) logger.debug("SSL context configured with %d ciphers", len(cipher_names)) except Exception: logger.exception("Failed to configure SSL context") @staticmethod def _cipher_ids_to_openssl_names(cipher_ids: list[int]) -> list[str]: """Map TLS cipher suite IDs to OpenSSL names.""" # Mapping of common cipher suite IDs to OpenSSL names id_to_name = { 0x1301: "TLS_AES_128_GCM_SHA256", 0x1302: "TLS_AES_256_GCM_SHA384", 0x1303: "TLS_CHACHA20_POLY1305_SHA256", 0xc02b: "ECDHE-ECDSA-AES128-GCM-SHA256", 0xc02f: "ECDHE-RSA-AES128-GCM-SHA256", 0xc02c: "ECDHE-ECDSA-AES256-GCM-SHA384", 0xc030: "ECDHE-RSA-AES256-GCM-SHA384", 0xcca9: "ECDHE-ECDSA-CHACHA20-POLY1305", 0xcca8: "ECDHE-RSA-CHACHA20-POLY1305", 0xc013: "ECDHE-RSA-AES128-SHA", 0xc014: "ECDHE-RSA-AES256-SHA", 0x009c: "AES128-GCM-SHA256", 0x009d: "AES256-GCM-SHA384", 0x002f: "AES128-SHA", 0x0035: "AES256-SHA", } names = [] for cid in cipher_ids: name = id_to_name.get(cid) if name: names.append(name) return names # ------------------------------------------------------------------ # NFQUEUE interception (for non-Python TLS) # ------------------------------------------------------------------ def _setup_nfqueue(self) -> None: """Set up iptables NFQUEUE rules to intercept outbound TLS ClientHello.""" try: # Add iptables rule to queue outbound TLS (port 443) to NFQUEUE rule = [ "iptables", "-I", "OUTPUT", "-p", "tcp", "--dport", "443", "-m", "u32", # Match TLS ClientHello: content type 0x16, handshake type 0x01 "--u32", "0>>22&0x3C@12>>26&0x3C@0=0x16030100:0x16030300", "-j", "NFQUEUE", "--queue-num", str(self._nfqueue_num), ] result = subprocess.run(rule, capture_output=True, timeout=10) if result.returncode == 0: self._iptables_rules.append(rule) logger.info("NFQUEUE iptables rule installed (queue %d)", self._nfqueue_num) else: # Fallback: simpler rule without u32 match rule_simple = [ "iptables", "-I", "OUTPUT", "-p", "tcp", "--dport", "443", "--syn", "-j", "NFQUEUE", "--queue-num", str(self._nfqueue_num), ] result = subprocess.run(rule_simple, capture_output=True, timeout=10) if result.returncode == 0: self._iptables_rules.append(rule_simple) logger.info("NFQUEUE simple iptables rule installed") else: logger.warning( "Failed to install NFQUEUE iptables rule: %s", result.stderr.decode(errors="replace"), ) except FileNotFoundError: logger.warning("iptables not found — NFQUEUE unavailable") except Exception: logger.exception("NFQUEUE setup failed") def _teardown_nfqueue(self) -> None: """Remove iptables NFQUEUE rules.""" for rule in self._iptables_rules: try: # Replace -I with -D to delete del_rule = list(rule) idx = del_rule.index("-I") del_rule[idx] = "-D" subprocess.run(del_rule, capture_output=True, timeout=10) except Exception: logger.exception("Failed to remove iptables rule") self._iptables_rules.clear() if self._nfqueue_proc and self._nfqueue_proc.poll() is None: self._nfqueue_proc.terminate() try: self._nfqueue_proc.wait(timeout=5) except subprocess.TimeoutExpired: self._nfqueue_proc.kill() self._nfqueue_proc = None # ------------------------------------------------------------------ # Profile selection # ------------------------------------------------------------------ def _auto_select_profile(self) -> str: """Auto-select JA3 profile based on observed network environment.""" # Check state for OS distribution data from host_discovery os_dist = self.state.get(self.name, "network_os_distribution") if os_dist: try: dist = json.loads(os_dist) if isinstance(os_dist, str) else os_dist # Windows-heavy network: use Chrome Windows if dist.get("windows", 0) > dist.get("linux", 0): return "chrome_120_win" else: return "chrome_120_linux" except (json.JSONDecodeError, AttributeError): pass # Default: Chrome on Windows (most common on corporate networks) return "chrome_120_win" # ------------------------------------------------------------------ # External fingerprint database # ------------------------------------------------------------------ def _load_fingerprint_db(self) -> None: """Load additional JA3 profiles from data/ja3_fingerprints.db if present.""" db_path = "data/ja3_fingerprints.db" if not os.path.isfile(db_path): logger.debug("No external JA3 database at %s — using builtins", db_path) return try: import sqlite3 conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row rows = conn.execute( "SELECT name, ja3_hash, description, cipher_suites, extensions, " "elliptic_curves, ec_point_formats FROM ja3_profiles" ).fetchall() for row in rows: name = row["name"] self._profiles[name] = { "ja3_hash": row["ja3_hash"], "description": row["description"], "cipher_suites": json.loads(row["cipher_suites"]), "extensions": json.loads(row["extensions"]), "elliptic_curves": json.loads(row["elliptic_curves"]), "ec_point_formats": json.loads(row["ec_point_formats"]), } conn.close() logger.info("Loaded %d JA3 profiles from database", len(rows)) except Exception: logger.exception("Failed to load JA3 fingerprint database") # ------------------------------------------------------------------ # Public: get SSL context for other modules # ------------------------------------------------------------------ def get_ssl_context(self) -> ssl.SSLContext: """Return an SSL context configured with the active JA3 profile's ciphers. Other modules (C2, exfil) should use this for outbound HTTPS.""" profile = self._profiles.get(self._active_profile, {}) ctx = ssl.create_default_context() cipher_names = self._cipher_ids_to_openssl_names(profile.get("cipher_suites", [])) if cipher_names: try: ctx.set_ciphers(":".join(cipher_names)) except ssl.SSLError: pass # Fall back to default ciphers ctx.minimum_version = ssl.TLSVersion.TLSv1_2 return ctx