"""Tests for modules/stealth/ja3_spoofer — profile loading, cipher mapping, randomization validation. Validates JA3 fingerprint profile selection, cipher suite mapping, SSL context configuration, and database loading without requiring root, iptables, or NFQUEUE. """ import json import os import sqlite3 import ssl from unittest.mock import MagicMock, patch import pytest from modules.base import BaseModule from modules.stealth.ja3_spoofer import JA3Spoofer, BUILTIN_PROFILES # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture def spoofer(mock_bus, mock_state): """Return a JA3Spoofer instance with NFQUEUE disabled.""" config = {"use_nfqueue": False} return JA3Spoofer(mock_bus, mock_state, config) @pytest.fixture def ja3_db(tmp_path): """Create a test ja3_fingerprints.db with sample profiles.""" db_path = tmp_path / "ja3_fingerprints.db" conn = sqlite3.connect(str(db_path)) conn.execute(""" CREATE TABLE ja3_profiles ( name TEXT PRIMARY KEY, ja3_hash TEXT, description TEXT, cipher_suites TEXT, extensions TEXT, elliptic_curves TEXT, ec_point_formats TEXT ) """) # Insert test profiles profiles = [ ("safari_17_macos", "aabbccdd11223344", "Safari 17 on macOS", json.dumps([0x1301, 0x1302, 0xc02b, 0xc02f]), json.dumps([0, 23, 65281, 10, 11]), json.dumps([0x001d, 0x0017]), json.dumps([0])), ("curl_8_linux", "eeff00112233aabb", "curl 8.x on Linux", json.dumps([0x1301, 0x1303, 0xc02f, 0x009c]), json.dumps([0, 10, 11, 13]), json.dumps([0x0017, 0x0018]), json.dumps([0])), ] conn.executemany( "INSERT INTO ja3_profiles VALUES (?, ?, ?, ?, ?, ?, ?)", profiles ) conn.commit() conn.close() return str(db_path) # --------------------------------------------------------------------------- # Structure # --------------------------------------------------------------------------- class TestJA3SpooferStructure: def test_is_base_module(self): assert issubclass(JA3Spoofer, BaseModule) def test_module_attributes(self): assert JA3Spoofer.name == "ja3_spoofer" assert JA3Spoofer.module_type == "stealth" assert JA3Spoofer.requires_root is True # --------------------------------------------------------------------------- # Built-in profiles # --------------------------------------------------------------------------- class TestBuiltinProfiles: def test_builtin_profiles_exist(self): """At least 4 built-in profiles should be defined.""" assert len(BUILTIN_PROFILES) >= 4 def test_all_profiles_have_required_fields(self): """Every profile has ja3_hash, cipher_suites, extensions, elliptic_curves.""" required = {"ja3_hash", "cipher_suites", "extensions", "elliptic_curves"} for name, profile in BUILTIN_PROFILES.items(): missing = required - set(profile.keys()) assert not missing, f"Profile {name} missing fields: {missing}" def test_all_ja3_hashes_are_32_hex(self): """JA3 hashes should be 32-char hex strings (MD5).""" for name, profile in BUILTIN_PROFILES.items(): h = profile["ja3_hash"] assert len(h) == 32, f"{name}: hash length {len(h)} != 32" assert all(c in "0123456789abcdef" for c in h), f"{name}: non-hex chars in hash" def test_cipher_suites_are_valid_ints(self): """Cipher suite IDs should be positive integers.""" for name, profile in BUILTIN_PROFILES.items(): for cs in profile["cipher_suites"]: assert isinstance(cs, int) and cs > 0, f"{name}: invalid cipher {cs}" def test_profiles_have_tls13_ciphers(self): """Modern profiles should include TLS 1.3 cipher suites (0x1301-0x1303).""" tls13 = {0x1301, 0x1302, 0x1303} for name, profile in BUILTIN_PROFILES.items(): suites = set(profile["cipher_suites"]) assert suites & tls13, f"{name}: no TLS 1.3 ciphers" def test_chrome_and_firefox_profiles_differ(self): """Chrome and Firefox profiles should have different JA3 hashes.""" chrome = BUILTIN_PROFILES["chrome_120_win"]["ja3_hash"] firefox = BUILTIN_PROFILES["firefox_121_win"]["ja3_hash"] assert chrome != firefox # --------------------------------------------------------------------------- # Cipher ID to OpenSSL name mapping # --------------------------------------------------------------------------- class TestCipherMapping: def test_known_cipher_ids_resolve(self): """All TLS 1.3 and common TLS 1.2 ciphers map to OpenSSL names.""" known_ids = [0x1301, 0x1302, 0x1303, 0xc02b, 0xc02f, 0x009c, 0x009d] names = JA3Spoofer._cipher_ids_to_openssl_names(known_ids) assert len(names) == len(known_ids) assert "TLS_AES_128_GCM_SHA256" in names assert "TLS_AES_256_GCM_SHA384" in names def test_unknown_cipher_ids_skipped(self): """Unknown cipher IDs are silently skipped.""" ids = [0x1301, 0xFFFF, 0xDEAD] names = JA3Spoofer._cipher_ids_to_openssl_names(ids) assert len(names) == 1 assert names[0] == "TLS_AES_128_GCM_SHA256" def test_empty_input(self): """Empty cipher list returns empty names.""" assert JA3Spoofer._cipher_ids_to_openssl_names([]) == [] def test_all_builtin_profiles_resolve_ciphers(self): """Every builtin profile's cipher_suites should resolve to at least some names.""" for name, profile in BUILTIN_PROFILES.items(): names = JA3Spoofer._cipher_ids_to_openssl_names(profile["cipher_suites"]) assert len(names) > 0, f"{name}: no ciphers resolved" def test_cipher_ordering_preserved(self): """Output cipher name ordering matches input ID ordering.""" ids = [0xc02f, 0x1301, 0x009c] names = JA3Spoofer._cipher_ids_to_openssl_names(ids) assert names[0] == "ECDHE-RSA-AES128-GCM-SHA256" assert names[1] == "TLS_AES_128_GCM_SHA256" assert names[2] == "AES128-GCM-SHA256" # --------------------------------------------------------------------------- # External fingerprint database loading # --------------------------------------------------------------------------- class TestFingerprintDatabase: def test_load_from_sqlite(self, spoofer, ja3_db): """Profiles from SQLite database are loaded alongside builtins.""" initial_count = len(spoofer._profiles) with patch("os.path.isfile", return_value=True): with patch("modules.stealth.ja3_spoofer.os.path.isfile", return_value=True): # Temporarily point to our test DB original_load = spoofer._load_fingerprint_db def patched_load(): import sqlite3 as sq3 conn = sq3.connect(ja3_db) conn.row_factory = sq3.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: spoofer._profiles[row["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() patched_load() assert len(spoofer._profiles) == initial_count + 2 assert "safari_17_macos" in spoofer._profiles assert "curl_8_linux" in spoofer._profiles def test_db_profiles_have_correct_structure(self, spoofer, ja3_db): """Loaded DB profiles have the same structure as builtins.""" # Load profiles from DB import sqlite3 as sq3 conn = sq3.connect(ja3_db) conn.row_factory = sq3.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: spoofer._profiles[row["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() safari = spoofer._profiles["safari_17_macos"] assert isinstance(safari["cipher_suites"], list) assert isinstance(safari["extensions"], list) assert isinstance(safari["elliptic_curves"], list) assert safari["ja3_hash"] == "aabbccdd11223344" def test_missing_db_uses_builtins_only(self, spoofer): """When no external DB exists, only builtins are loaded.""" spoofer._profiles = dict(BUILTIN_PROFILES) spoofer._load_fingerprint_db() assert len(spoofer._profiles) == len(BUILTIN_PROFILES) # --------------------------------------------------------------------------- # Profile auto-selection # --------------------------------------------------------------------------- class TestAutoSelection: def test_default_is_chrome_windows(self, spoofer): """Default auto-selection returns chrome_120_win.""" profile = spoofer._auto_select_profile() assert profile == "chrome_120_win" def test_windows_heavy_network_selects_chrome_win(self, spoofer): """Windows-heavy network environment selects Windows Chrome.""" spoofer.state.get = MagicMock( return_value=json.dumps({"windows": 15, "linux": 3}) ) profile = spoofer._auto_select_profile() assert profile == "chrome_120_win" def test_linux_heavy_network_selects_chrome_linux(self, spoofer): """Linux-heavy network environment selects Linux Chrome.""" spoofer.state.get = MagicMock( return_value=json.dumps({"windows": 2, "linux": 10}) ) profile = spoofer._auto_select_profile() assert profile == "chrome_120_linux" # --------------------------------------------------------------------------- # SSL context configuration # --------------------------------------------------------------------------- class TestSSLContext: def test_get_ssl_context_returns_context(self, spoofer): """get_ssl_context returns a valid SSLContext.""" spoofer._active_profile = "chrome_120_win" ctx = spoofer.get_ssl_context() assert isinstance(ctx, ssl.SSLContext) assert ctx.minimum_version == ssl.TLSVersion.TLSv1_2 def test_get_ssl_context_with_invalid_profile(self, spoofer): """Invalid profile still returns a usable SSLContext (default ciphers).""" spoofer._active_profile = "nonexistent_profile" ctx = spoofer.get_ssl_context() assert isinstance(ctx, ssl.SSLContext) # --------------------------------------------------------------------------- # Status reporting # --------------------------------------------------------------------------- class TestStatus: def test_status_not_running(self, spoofer): s = spoofer.status() assert s["running"] is False assert s["packets_modified"] == 0 assert s["profiles_loaded"] == len(BUILTIN_PROFILES) def test_status_after_start(self, spoofer): """After start (with nfqueue disabled), status shows running.""" spoofer.start() s = spoofer.status() assert s["running"] is True assert s["active_profile"] is not None assert s["active_ja3_hash"] != "none" spoofer.stop() def test_configure_switches_profile(self, spoofer): """configure() with target_profile switches the active profile.""" spoofer._active_profile = "chrome_120_win" spoofer.configure({"target_profile": "firefox_121_win"}) assert spoofer._active_profile == "firefox_121_win" # --------------------------------------------------------------------------- # Randomization validation — ensure profiles produce distinct fingerprints # --------------------------------------------------------------------------- class TestRandomizationValidation: def test_all_profiles_produce_unique_ja3_hashes(self): """Every profile should have a unique JA3 hash.""" hashes = [p["ja3_hash"] for p in BUILTIN_PROFILES.values()] assert len(hashes) == len(set(hashes)), "Duplicate JA3 hashes found" def test_all_profiles_have_distinct_cipher_orderings(self): """Profiles should have at least 2 distinct cipher suite orderings (Chromium-based browsers share orderings, Firefox differs).""" orderings = [] for p in BUILTIN_PROFILES.values(): orderings.append(tuple(p["cipher_suites"])) unique = len(set(orderings)) assert unique >= 2, f"Only {unique} unique cipher orderings across {len(BUILTIN_PROFILES)} profiles" def test_profile_ciphers_produce_valid_ssl_context(self): """Each profile's cipher string should be accepted by OpenSSL.""" for name, profile in BUILTIN_PROFILES.items(): names = JA3Spoofer._cipher_ids_to_openssl_names(profile["cipher_suites"]) if not names: continue ctx = ssl.create_default_context() # Some ciphers might not be available on all OpenSSL builds try: ctx.set_ciphers(":".join(names)) except ssl.SSLError: # Acceptable — some cipher combos not supported on all platforms pass