import hashlib import sqlite3 import threading import time from typing import List, Dict, Any, Tuple # Known Fingerprint Database KNOWN_FINGERPRINTS = { # Example hardcoded JA4 hashes based on the spec "t13d1512h2_8f2923298c4b_e6fb1c499691": {"name": "Chrome 131 Linux", "ja4": "t13d1512h2_8f2923298c4b_e6fb1c499691", "type": "BROWSER"}, "t13d1411h2_183350117822_2e5927342938": {"name": "Firefox 121 Linux", "ja4": "t13d1411h2_183350117822_2e5927342938", "type": "BROWSER"}, "t13d040800_209d84639e4a_c89b21f3014a": {"name": "Python requests", "ja4": "t13d040800_209d84639e4a_c89b21f3014a", "type": "SCRIPTED"}, "t13d070900_e58bc283f5c7_b6ac78df109a": {"name": "cURL", "ja4": "t13d070900_e58bc283f5c7_b6ac78df109a", "type": "SCRIPTED"}, "t13d1513h2_8f2923298c4b_d1ae53628e93": {"name": "Playwright Chromium", "ja4": "t13d1513h2_8f2923298c4b_d1ae53628e93", "type": "AUTOMATION"}, } # Standard TLS GREASE values (RFC 8701: 0x0A0A, 0x1A1A, ..., 0xFAFA) GREASE_TABLE = { 0x0A0A, 0x1A1A, 0x2A2A, 0x3A3A, 0x4A4A, 0x5A5A, 0x6A6A, 0x7A7A, 0x8A8A, 0x9A9A, 0xAAAA, 0xBABA, 0xCACA, 0xDADA, 0xEAEA, 0xFAFA } def is_grease(val: int) -> bool: """Filter out GREASE values (0x?A?A pattern where both nibbles of both bytes are 0xA).""" return val in GREASE_TABLE def compute_ja4( tls_version: int, ciphers: List[int], extensions: List[int], alpn: str, sni_present: bool, sig_algs: List[int] ) -> str: """ Computes JA4 Hash based on the given TLS fields. """ # TLS Version Map version_map = { 0x0304: "13", 0x0303: "12", 0x0302: "11", 0x0301: "10" } tls_ver = version_map.get(tls_version, "00") sni_flag = "d" if sni_present else "i" # Filter GREASE before counting/hashing clean_ciphers = [c for c in ciphers if not is_grease(c)] clean_exts = [e for e in extensions if not is_grease(e)] num_ciphers = len(clean_ciphers) num_exts = len(clean_exts) alpn_first2 = "00" if alpn: alpn_first2 = (alpn[:2] + "00")[:2] # Sort cipher suites and extensions alphabetically before hashing sorted_ciphers = sorted([f"{c:04x}" for c in clean_ciphers]) sorted_exts = sorted([f"{e:04x}" for e in clean_exts]) sorted_sig_algs = sorted([f"{s:04x}" for s in sig_algs]) # Cipher hash = first 12 chars of SHA256(sorted_ciphers_hex_joined) ciphers_joined = ",".join(sorted_ciphers) if not ciphers_joined: cipher_hash = "000000000000" else: cipher_hash = hashlib.sha256(ciphers_joined.encode()).hexdigest()[:12] # Ext hash = first 12 chars of SHA256(sorted_exts_hex_joined_sig_algs_appended) exts_joined = ",".join(sorted_exts) if sorted_sig_algs: exts_joined += "_" + ",".join(sorted_sig_algs) if not exts_joined.strip("_"): ext_hash = "000000000000" else: ext_hash = hashlib.sha256(exts_joined.encode()).hexdigest()[:12] return f"t{tls_ver}{sni_flag}{num_ciphers:02d}{num_exts:02d}{alpn_first2}_{cipher_hash}_{ext_hash}" def check_mismatch(user_agent: str, ja4_hash: str) -> dict: """ Checks if the User-Agent claims match the JA4 TLS fingerprint. Defensive against None/empty strings/invalid types. """ if not user_agent or not isinstance(user_agent, str): user_agent = "" if not ja4_hash or not isinstance(ja4_hash, str): return { "match": False, "risk_level": "UNKNOWN", "ua_claimed": "UNKNOWN", "ja4_actual": "UNKNOWN", "inferred_tool": "UNKNOWN" } ua_lower = user_agent.lower() claimed_browser = "UNKNOWN" if "chrome" in ua_lower: claimed_browser = "Chrome" elif "firefox" in ua_lower: claimed_browser = "Firefox" elif "safari" in ua_lower and "chrome" not in ua_lower: claimed_browser = "Safari" actual_tool = KNOWN_FINGERPRINTS.get(ja4_hash, {}).get("name", "UNKNOWN") tool_type = KNOWN_FINGERPRINTS.get(ja4_hash, {}).get("type", "UNKNOWN") match = True risk_level = "NONE" if actual_tool != "UNKNOWN": if tool_type in ("SCRIPTED", "AUTOMATION"): match = False risk_level = "CRITICAL" if tool_type == "SCRIPTED" else "HIGH" elif claimed_browser != "UNKNOWN" and claimed_browser.lower() not in actual_tool.lower(): match = False risk_level = "HIGH" else: # Unknown JA4, we can't be sure, but we might flag it as LOW risk risk_level = "LOW" return { "match": match, "risk_level": risk_level, "ua_claimed": claimed_browser, "ja4_actual": actual_tool, "inferred_tool": actual_tool if actual_tool != "UNKNOWN" else "UNKNOWN" } class JA4Cache: """ SQLite cache (WAL mode) for storing seen JA4 hashes. Thread-safe access using threading.Lock for ALL reads and writes. """ def __init__(self, db_path: str): self.db_path = db_path self.lock = threading.Lock() # Initialize DB with self.lock: with sqlite3.connect(self.db_path) as conn: conn.execute("PRAGMA journal_mode=WAL;") conn.execute(''' CREATE TABLE IF NOT EXISTS ja4_cache ( ja4_hash TEXT PRIMARY KEY, first_seen REAL, last_seen REAL, count INTEGER, associated_ua TEXT ) ''') conn.commit() def get_seen(self, ja4_hash: str) -> dict: """Read method with lock and WAL mode PRAGMA.""" with self.lock: with sqlite3.connect(self.db_path) as conn: conn.execute("PRAGMA journal_mode=WAL;") cursor = conn.cursor() cursor.execute('SELECT ja4_hash, first_seen, last_seen, count, associated_ua FROM ja4_cache WHERE ja4_hash = ?', (ja4_hash,)) row = cursor.fetchone() if row: return { "ja4_hash": row[0], "first_seen": row[1], "last_seen": row[2], "count": row[3], "associated_ua": row[4] } return None def update_seen(self, ja4_hash: str, user_agent: str): """Write method with lock and WAL mode PRAGMA.""" now = time.time() with self.lock: with sqlite3.connect(self.db_path) as conn: conn.execute("PRAGMA journal_mode=WAL;") cursor = conn.cursor() cursor.execute('SELECT count FROM ja4_cache WHERE ja4_hash = ?', (ja4_hash,)) row = cursor.fetchone() if row: cursor.execute(''' UPDATE ja4_cache SET last_seen = ?, count = count + 1, associated_ua = ? WHERE ja4_hash = ? ''', (now, user_agent, ja4_hash)) else: cursor.execute(''' INSERT INTO ja4_cache (ja4_hash, first_seen, last_seen, count, associated_ua) VALUES (?, ?, ?, 1, ?) ''', (ja4_hash, now, now, user_agent)) conn.commit() if __name__ == '__main__': # Test harness simulating 5 different client hello scenarios scenarios = [ { "name": "Legitimate Chrome 131", "ua": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36", "tls": { "tls_version": 0x0304, "ciphers": [0x1301, 0x1302, 0x1303, 0xc02b, 0xc02f, 0xcca9, 0xccaa, 0xc02c, 0xc030, 0x009e, 0x009c, 0x0a0a, 0x002f, 0x0035], "extensions": [0x0000, 0x0017, 0x0041, 0x000a, 0x000b, 0x0023, 0x0010, 0x0005, 0x000d, 0x0012, 0x0033, 0x002d, 0x002b, 0x001b, 0x0a0a], "alpn": "h2", "sni_present": True, "sig_algs": [0x0403, 0x0804, 0x0401, 0x0503, 0x0805, 0x0501, 0x0806, 0x0601] } }, { "name": "Python Requests pretending to be Chrome", "ua": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36", "tls": { "tls_version": 0x0304, "ciphers": [0x1301, 0x1302, 0x1303, 0xc02b], "extensions": [0x0000, 0x000b, 0x000a, 0x0023, 0x0016, 0x0017, 0x000d, 0x002b], "alpn": "", "sni_present": True, "sig_algs": [0x0403, 0x0804, 0x0401] } }, { "name": "Legitimate Firefox 121", "ua": "Mozilla/5.0 (X11; Linux x86_64; rv:121.0) Gecko/20100101 Firefox/121.0", "tls": { "tls_version": 0x0304, "ciphers": [0x1301, 0x1302, 0x1303, 0xc02b, 0xc02f, 0xcca9, 0xccaa, 0xc02c, 0xc030, 0x009e, 0x009c, 0x002f, 0x0035, 0x000a], "extensions": [0x0000, 0x0017, 0x000a, 0x000b, 0x0023, 0x0010, 0x0005, 0x000d, 0x0012, 0x0033, 0x002d, 0x002b, 0x001b], "alpn": "h2", "sni_present": True, "sig_algs": [0x0403, 0x0804, 0x0401, 0x0503, 0x0805, 0x0501, 0x0806, 0x0601] } }, { "name": "cURL Bot", "ua": "curl/7.81.0", "tls": { "tls_version": 0x0304, "ciphers": [0x1301, 0x1302, 0x1303, 0xc02b, 0xc02f, 0xcca9, 0xccaa], "extensions": [0x0000, 0x000b, 0x000a, 0x0023, 0x0016, 0x0017, 0x000d, 0x002b, 0x0010], "alpn": "00", "sni_present": True, "sig_algs": [0x0403, 0x0804, 0x0401, 0x0503] } }, { "name": "Playwright Chromium headless", "ua": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) HeadlessChrome/131.0.0.0 Safari/537.36", "tls": { "tls_version": 0x0304, "ciphers": [0x1301, 0x1302, 0x1303, 0xc02b, 0xc02f, 0xcca9, 0xccaa, 0xc02c, 0xc030, 0x009e, 0x009c, 0x0a0a, 0x002f, 0x0035], "extensions": [0x0000, 0x0017, 0x0041, 0x000a, 0x000b, 0x0023, 0x0010, 0x0005, 0x000d, 0x0012, 0x0033, 0x002d, 0x002b, 0x001b, 0x0015, 0x0a0a], "alpn": "h2", "sni_present": True, "sig_algs": [0x0403, 0x0804, 0x0401, 0x0503, 0x0805, 0x0501, 0x0806, 0x0601] } }, ] # Pre-populate known dictionary with actual computed hashes from the examples above # to make the test harness functional for demonstration. print("--- Pre-computing dummy JA4 hashes to populate KNOWN_FINGERPRINTS for testing ---") for s in scenarios: ja4 = compute_ja4(**s["tls"]) if "Chrome 131" in s["name"] and "Python" not in s["name"]: KNOWN_FINGERPRINTS[ja4] = {"name": "Chrome 131 Linux", "ja4": ja4, "type": "BROWSER"} elif "Firefox 121" in s["name"]: KNOWN_FINGERPRINTS[ja4] = {"name": "Firefox 121 Linux", "ja4": ja4, "type": "BROWSER"} elif "Python" in s["name"]: KNOWN_FINGERPRINTS[ja4] = {"name": "Python requests", "ja4": ja4, "type": "SCRIPTED"} elif "cURL" in s["name"]: KNOWN_FINGERPRINTS[ja4] = {"name": "cURL", "ja4": ja4, "type": "SCRIPTED"} elif "Playwright" in s["name"]: KNOWN_FINGERPRINTS[ja4] = {"name": "Playwright Chromium", "ja4": ja4, "type": "AUTOMATION"} db_path = "/home/snuffleupagus/teamwork_projects/axiom_zero_audit/dev_control_center/ja4_cache.db" import os os.makedirs(os.path.dirname(db_path), exist_ok=True) cache = JA4Cache(db_path) print("\n--- Running 5 Simulated Scenarios ---") for i, s in enumerate(scenarios, 1): print(f"\nScenario {i}: {s['name']}") print(f"User Agent: {s['ua']}") # 1. Compute JA4 ja4 = compute_ja4(**s["tls"]) print(f"Computed JA4: {ja4}") # 2. Check Mismatch mismatch_result = check_mismatch(s["ua"], ja4) print(f"Analysis: {mismatch_result}") # 3. Store in DB cache.update_seen(ja4, s["ua"]) cached_entry = cache.get_seen(ja4) print(f"Logged to JA4 cache DB. Read back: count={cached_entry['count']}") print("\n--- Testing Defensive Edge Cases in check_mismatch ---") print("None UA & None Hash:", check_mismatch(None, None)) print("Empty String Hash:", check_mismatch("Mozilla/5.0", ""))