JA3 and JA3S Fingerprinting

#!/usr/bin/env python3
"""
Compute JA3 hashes from PCAP using dpkt.
Based on: https://github.com/salesforce/ja3
"""
import dpkt, socket, hashlib, sys, struct
from collections import defaultdict

# GREASE values to filter out
GREASE = {0x0a0a,0x1a1a,0x2a2a,0x3a3a,0x4a4a,0x5a5a,
           0x6a6a,0x7a7a,0x8a8a,0x9a9a,0xaaaa,0xbaba,
           0xcaca,0xdada,0xeaea,0xfafa}

def ip_to_str(addr: bytes) -> str:
    return socket.inet_ntoa(addr)

def compute_ja3(client_hello_data: bytes) -> tuple[str, str]:
    """
    Parse TLS ClientHello and compute JA3.
    Returns (ja3_string, ja3_hash).
    """
    try:
        pos = 0
        # TLS record: type(1) + version(2) + length(2)
        if len(client_hello_data) < 5:
            return "", ""
        rec_type = client_hello_data[0]
        if rec_type != 22:  # 22 = Handshake
            return "", ""
        pos = 5

        # Handshake header: type(1) + length(3)
        if client_hello_data[pos] != 1:  # 1 = ClientHello
            return "", ""
        pos += 4

        # ClientHello: version(2) + random(32) + session_id_len(1)
        tls_version = struct.unpack_from(">H", client_hello_data, pos)[0]
        pos += 2 + 32  # skip version + random
        session_id_len = client_hello_data[pos]
        pos += 1 + session_id_len

        # Cipher suites
        cs_len = struct.unpack_from(">H", client_hello_data, pos)[0]
        pos += 2
        ciphers = []
        for i in range(0, cs_len, 2):
            cs = struct.unpack_from(">H", client_hello_data, pos + i)[0]
            if cs not in GREASE:
                ciphers.append(cs)
        pos += cs_len

        # Compression methods
        comp_len = client_hello_data[pos]
        pos += 1 + comp_len

        # Extensions
        if pos >= len(client_hello_data) - 2:
            # No extensions
            ja3_str = f"{tls_version},{','.join(map(str,ciphers))},,,"
            return ja3_str, hashlib.md5(ja3_str.encode()).hexdigest()

        ext_total_len = struct.unpack_from(">H", client_hello_data, pos)[0]
        pos += 2
        ext_end = pos + ext_total_len

        extensions = []
        curves = []
        curve_formats = []

        while pos < ext_end:
            ext_type = struct.unpack_from(">H", client_hello_data, pos)[0]
            ext_len = struct.unpack_from(">H", client_hello_data, pos + 2)[0]
            ext_data = client_hello_data[pos + 4: pos + 4 + ext_len]
            pos += 4 + ext_len

            if ext_type not in GREASE:
                extensions.append(ext_type)

            # supported_groups (elliptic curves) = type 10
            if ext_type == 10 and len(ext_data) >= 2:
                list_len = struct.unpack_from(">H", ext_data, 0)[0]
                for i in range(2, 2 + list_len, 2):
                    curve = struct.unpack_from(">H", ext_data, i)[0]
                    if curve not in GREASE:
                        curves.append(curve)

            # ec_point_formats = type 11
            if ext_type == 11 and ext_data:
                fmt_len = ext_data[0]
                curve_formats = list(ext_data[1:1 + fmt_len])

        ja3_str = (
            f"{tls_version},"
            f"{','.join(map(str, ciphers))}-"
            f"{','.join(map(str, extensions))}-"
            f"{','.join(map(str, curves))}-"
            f"{','.join(map(str, curve_formats))}"
        )
        return ja3_str, hashlib.md5(ja3_str.encode()).hexdigest()

    except Exception:
        return "", ""

# Known malicious JA3 hashes
KNOWN_BAD = {
    "51c64c77e60f3980eea90869b68c58a8": "Cobalt Strike default",
    "de9f2c7fd25e1b3afad3e85a0226a4c4": "Metasploit",
    "a0e9f5d64349fb13191bc781f81f42e1": "Trickbot",
    "72a589da586844d7f0818ce684948eea": "PlugX",
    "a5a95f09fe86e07ace7bb75124419d26": "Emotet",
}

# Analyze PCAP
pcap_file = sys.argv[1]
results = defaultdict(set)  # (src, dst, dport) → set of ja3 hashes

with open(pcap_file, "rb") as f:
    pcap = dpkt.pcap.Reader(f)
    for ts, buf in pcap:
        try:
            eth = dpkt.ethernet.Ethernet(buf)
            if not isinstance(eth.data, dpkt.ip.IP):
                continue
            ip = eth.data
            if not isinstance(ip.data, dpkt.tcp.TCP):
                continue
            tcp = ip.data
            payload = bytes(tcp.data)
            if len(payload) < 10:
                continue
            # Check for TLS ClientHello
            if payload[0] == 22 and payload[5] == 1:
                src = ip_to_str(ip.src)
                dst = ip_to_str(ip.dst)
                dport = tcp.dport
                ja3_str, ja3_hash = compute_ja3(payload)
                if ja3_hash:
                    results[(src, dst, dport)].add(ja3_hash)
        except Exception:
            pass

print(f"{'JA3 Hash':35}  {'Alert':25}  Connection")
print("-" * 90)
for (src, dst, port), hashes in sorted(results.items()):
    for h in hashes:
        alert = KNOWN_BAD.get(h, "")
        marker = " *** MALWARE ***" if alert else ""
        print(f"{h:35}  {alert:25}  {src}→{dst}:{port}{marker}")