fix(python): ja4x crashed on a lone certificate with 10+ extensions

python/ja4.py

@@ -146,6 +146,21 @@ def hops(x):
         initial_ttl = 255
     return (initial_ttl - x)
 
+def first_last_alpn(alpn):
+    # Keep the first and last character of the ALPN value, which is all the
+    # fingerprint needs. Matches the rust implementation: a non-ascii
+    # character is replaced with '9', a single character is followed by '0',
+    # and an empty value (or a missing ALPN extension) becomes '00'.
+    if isinstance(alpn, list):
+        alpn = alpn[0] if alpn else ''
+    if not alpn:
+        return '00'
+    first = alpn[0] if ord(alpn[0]) < 128 else '9'
+    if len(alpn) == 1:
+        return f"{first}0"
+    last = alpn[-1] if ord(alpn[-1]) < 128 else '9'
+    return f"{first}{last}"
+
 def calculate_ja4_latency(x, ptype, STREAM):
     try:
         cache = get_cache(x)
@@ -195,16 +210,7 @@ def to_ja4s(x, debug_stream):
         x['version'] = get_supported_version(x['supported_versions'])
     version = TLS_MAPPER[x['version']] if x['version'] in TLS_MAPPER else '00'
 
-    alpn = '00' 
-    if 'alpn_list' in x:
-        if isinstance(x['alpn_list'], list):
-            alpn = x['alpn_list'][0]
-        else:
-            alpn = x['alpn_list']
-    if len(alpn) > 2:
-        alpn = f"{alpn[0]}{alpn[-1]}"
-    if ord(alpn[0]) > 127:
-        alpn = '99'
+    alpn = first_last_alpn(x['alpn_list'] if 'alpn_list' in x else '')
 
     x['JA4S'] = f"{ptype}{version}{ext_len}{alpn}_{x['ciphers']}_{extensions}"
     x['JA4S_r'] = f"{ptype}{version}{ext_len}{alpn}_{x['ciphers']}_{','.join(x['extensions'])}"
@@ -266,18 +272,7 @@ def to_ja4(x, debug_stream):
         x['version'] = get_supported_version(x['supported_versions'])
     version = TLS_MAPPER[x['version']] if x['version'] in TLS_MAPPER else '00'
 
-    alpn = '00' 
-    if 'alpn_list' in x:
-        if isinstance(x['alpn_list'], list):
-            alpn = x['alpn_list'][0]
-        else:
-            alpn = x['alpn_list']
-
-    if len(alpn) > 2:
-        alpn = f"{alpn[0]}{alpn[-1]}"
-
-    if ord(alpn[0]) > 127:
-        alpn = '99'
+    alpn = first_last_alpn(x['alpn_list'] if 'alpn_list' in x else '')
 
     entry = get_cache(x)[x['stream']]
     if not entry.get('count'):

python/ja4x.py

@@ -37,7 +37,15 @@ def to_ja4x(x, debug_stream=-1):
     # we need to convert them into hex codes and then use sha256
     if 'extension_lengths' not in x:
         return
-
+
+    # tshark reports a field with a single value as a plain string. The
+    # certificate counts are iterated below, and a string would be walked
+    # one digit at a time: a certificate with 10 or more extensions ('12')
+    # was read as two certificates with 1 and 2 extensions.
+    for field in ('extension_lengths', 'issuer_sequence', 'subject_sequence'):
+        if field in x and not isinstance(x[field], list):
+            x[field] = [x[field]]
+
     x['issuers'] = []
     x['subjects'] = []
     x['issuer_hashes'] = []

python/test/test_ja4_unit.py

@@ -3,9 +3,10 @@
 
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 
-from common import epoch_diff  # noqa: E402
-from ja4 import hops  # noqa: E402
+from common import cache_update, epoch_diff  # noqa: E402
+from ja4 import first_last_alpn, hops, to_ja4s  # noqa: E402
 from ja4h import to_ja4h  # noqa: E402
+from ja4x import to_ja4x  # noqa: E402
 
 
 def test_epoch_diff_spans_seconds():
@@ -47,3 +48,64 @@ def test_ja4h_referer_flag_requires_exact_header():
 
 def test_ja4h_referer_flag_ignores_lookalike_headers():
     assert _ja4h_referer_flag(["X-Referer: https://example.com/"]) == "n"
+
+
+def test_first_last_alpn_matches_the_rust_implementation():
+    assert first_last_alpn("h2") == "h2"
+    assert first_last_alpn("http/1.1") == "h1"
+    assert first_last_alpn("x") == "x0"
+    # a non-ascii character becomes '9', per character
+    assert first_last_alpn("\u00e9x") == "9x"
+    assert first_last_alpn("x\u00e9") == "x9"
+    # an empty ALPN value behaves like a missing one
+    assert first_last_alpn("") == "00"
+    assert first_last_alpn(None) == "00"
+    # several offered protocols: the first one counts
+    assert first_last_alpn(["h2", "h3"]) == "h2"
+
+
+def test_to_ja4s_survives_an_empty_alpn_value():
+    x = {
+        "hl": "tls",
+        "stream": 7,
+        "quic": False,
+        "version": "0x0303",
+        "ciphers": ["0x1301"],
+        "extensions": ["0x0016"],
+        "alpn_list": "",
+    }
+    cache_update(x, "stream", 7, -1)
+    to_ja4s(x, debug_stream=-1)
+    # JA4S layout: transport(1) version(2) extension count(2) alpn(2)
+    assert x["JA4S"][5:7] == "00"
+
+
+def _ja4x_input(extension_lengths, nr_oids):
+    return {
+        "hl": "x509af",
+        "stream": 3,
+        "extension_lengths": extension_lengths,
+        "cert_extensions": [f"2.5.29.{i}" for i in range(10, 10 + nr_oids)],
+        "issuer_sequence": ["1"],
+        "subject_sequence": ["1"],
+        "rdn_oids": ["2.5.4.3", "2.5.4.3"],
+    }
+
+
+def test_to_ja4x_two_digit_count_is_one_certificate():
+    # tshark reports a lone certificate's counts as plain strings
+    x = _ja4x_input("10", nr_oids=10)
+    cache_update(x, "stream", 3, -1)
+    to_ja4x(x, debug_stream=-1)
+    assert "JA4X.1" in x
+    assert "JA4X.2" not in x
+
+
+def test_to_ja4x_string_and_list_counts_agree():
+    as_string = _ja4x_input("3", nr_oids=3)
+    as_list = _ja4x_input(["3"], nr_oids=3)
+    cache_update(as_string, "stream", 4, -1)
+    cache_update(as_list, "stream", 5, -1)
+    to_ja4x(as_string, debug_stream=-1)
+    to_ja4x(as_list, debug_stream=-1)
+    assert as_string["JA4X.1"] == as_list["JA4X.1"]