JA4T: TCP Client Fingerprinting

<ttl>_<window_size>_<options_hash>_<flags>

MSS=1460, SACK_PERM, Timestamp, NOP, Window_Scale=7
Types: 02,04,08,01,03
Hash: SHA256("02,04,08,01,03")[:12]

IP TTL: 128
TCP Window Size: 8192
TCP Options:
  - MSS: 1460 bytes
  - NOP
  - Window Scale: 8
  - NOP
  - NOP
  - SACK Permitted
  - Timestamp: 123456789, 0
TCP Flags: SYN

IP TTL: 64
TCP Window Size: 14600
TCP Options:
  - MSS: 1460 bytes
  - SACK Permitted
  - Timestamp: 987654321, 0
  - NOP
  - Window Scale: 7
TCP Flags: SYN

IP TTL: 64 (default Linux)
TCP Window Size: 1024 (unusual)
TCP Options:
  - MSS: 1460 bytes
TCP Flags: SYN

import hashlib
from scapy.all import *
from typing import Optional, Dict, List, Tuple
from dataclasses import dataclass
from collections import defaultdict
 
@dataclass
class JA4TFingerprint:
    """Represents a JA4T TCP fingerprint"""
    ttl: int
    window_size: int
    options_hash: str
    flags: str
    raw_fingerprint: str
    
    def __str__(self) -> str:
        return self.raw_fingerprint
 
class JA4TFingerprinter:
    """
    Production-ready JA4T TCP Client Fingerprinter
    
    Analyzes TCP SYN packets to generate fingerprints for OS detection,
    device classification, and anomaly detection.
    """
    
    # TCP Option Type Codes
    TCP_OPT_EOL = 0
    TCP_OPT_NOP = 1
    TCP_OPT_MSS = 2
    TCP_OPT_WSCALE = 3
    TCP_OPT_SACK_PERMITTED = 4
    TCP_OPT_SACK = 5
    TCP_OPT_TIMESTAMP = 8
    
    # Common OS TTL Values
    OS_TTL_SIGNATURES = {
        64: ["Linux", "macOS", "BSD", "Unix"],
        128: ["Windows"],
        255: ["Cisco IOS", "Solaris", "Embedded"],
        32: ["Windows 9x", "Legacy"]
    }
    
    # Common Window Sizes by OS
    OS_WINDOW_SIGNATURES = {
        8192: ["Windows 7", "Windows 8", "Windows 10"],
        16384: ["macOS"],
        14600: ["Linux 2.6+"],
        5840: ["Windows XP"],
        65535: ["Custom Stack", "Modified"]
    }
    
    def __init__(self):
        self.fingerprint_db: Dict[str, List[str]] = defaultdict(list)
        self.seen_fingerprints: Dict[str, int] = defaultdict(int)
    
    def extract_tcp_options(self, packet: Packet) -> List[int]:
        """Extract TCP option types in order"""
        if not packet.haslayer(TCP):
            return []
        
        tcp_layer = packet[TCP]
        options = []
        
        if hasattr(tcp_layer, 'options') and tcp_layer.options:
            for opt in tcp_layer.options:
                if isinstance(opt, tuple):
                    opt_type = opt[0]
                    # Map option names to codes
                    opt_code = self._option_name_to_code(opt_type)
                    if opt_code is not None:
                        options.append(opt_code)
                elif isinstance(opt, str):
                    opt_code = self._option_name_to_code(opt)
                    if opt_code is not None:
                        options.append(opt_code)
        
        return options
    
    def _option_name_to_code(self, opt_name) -> Optional[int]:
        """Convert option name to numeric code"""
        mapping = {
            'EOL': 0,
            'NOP': 1,
            'MSS': 2,
            'WScale': 3,
            'SAckOK': 4,
            'SAck': 5,
            'Timestamp': 8,
        }
        return mapping.get(opt_name, None)
    
    def compute_options_hash(self, options: List[int]) -> str:
        """Compute SHA-256 hash of option types (first 12 chars)"""
        if not options:
            return "000000000000"
        
        options_str = ",".join(str(opt) for opt in options)
        hash_obj = hashlib.sha256(options_str.encode())
        return hash_obj.hexdigest()[:12]
    
    def extract_ttl(self, packet: Packet) -> int:
        """Extract TTL from IP layer"""
        if packet.haslayer(IP):
            return packet[IP].ttl
        return 0
    
    def infer_original_ttl(self, current_ttl: int) -> int:
        """Infer original TTL from common values"""
        common_ttls = [32, 64, 128, 255]
        
        for ttl in common_ttls:
            if current_ttl <= ttl:
                return ttl
        
        return current_ttl
    
    def extract_window_size(self, packet: Packet) -> int:
        """Extract TCP window size"""
        if packet.haslayer(TCP):
            return packet[TCP].window
        return 0
    
    def extract_flags(self, packet: Packet) -> str:
        """Extract TCP flags as single character"""
        if not packet.haslayer(TCP):
            return "U"  # Unknown
        
        tcp = packet[TCP]
        
        # Check specific flag combinations for scan types
        if tcp.flags.S and not tcp.flags.A:
            return "S"  # SYN
        elif tcp.flags.A and not tcp.flags.S:
            return "A"  # ACK
        elif tcp.flags.F:
            return "F"  # FIN
        elif tcp.flags.R:
            return "R"  # RST
        elif tcp.flags == 0:
            return "N"  # NULL
        elif tcp.flags.F and tcp.flags.P and tcp.flags.U:
            return "X"  # Xmas
        else:
            return "M"  # Mixed/Other
    
    def generate_fingerprint(self, packet: Packet) -> Optional[JA4TFingerprint]:
        """Generate JA4T fingerprint from TCP SYN packet"""
        if not packet.haslayer(TCP) or not packet.haslayer(IP):
            return None
        
        tcp = packet[TCP]
        
        # Extract components
        ttl = self.infer_original_ttl(self.extract_ttl(packet))
        window_size = self.extract_window_size(packet)
        options = self.extract_tcp_options(packet)
        options_hash = self.compute_options_hash(options)
        flags = self.extract_flags(packet)
        
        # Construct fingerprint
        raw_fp = f"{ttl}_{window_size}_{options_hash}_{flags}"
        
        return JA4TFingerprint(
            ttl=ttl,
            window_size=window_size,
            options_hash=options_hash,
            flags=flags,
            raw_fingerprint=raw_fp
        )
    
    def identify_os(self, fingerprint: JA4TFingerprint) -> List[str]:
        """Identify likely operating systems from fingerprint"""
        candidates = set()
        
        # Check TTL signatures
        if fingerprint.ttl in self.OS_TTL_SIGNATURES:
            candidates.update(self.OS_TTL_SIGNATURES[fingerprint.ttl])
        
        # Refine with window size
        if fingerprint.window_size in self.OS_WINDOW_SIGNATURES:
            window_os = set(self.OS_WINDOW_SIGNATURES[fingerprint.window_size])
            if candidates:
                candidates = candidates.intersection(window_os)
            else:
                candidates = window_os
        
        return list(candidates) if candidates else ["Unknown"]
    
    def detect_scan_type(self, fingerprint: JA4TFingerprint) -> Optional[str]:
        """Detect if fingerprint indicates port scanning"""
        # Unusual window sizes often indicate scanners
        unusual_windows = [1024, 2048, 4096, 512]
        
        if fingerprint.window_size in unusual_windows:
            if fingerprint.flags == "S":
                return "SYN Scan (Nmap-like)"
            elif fingerprint.flags == "A":
                return "ACK Scan"
            elif fingerprint.flags == "F":
                return "FIN Scan"
            elif fingerprint.flags == "N":
                return "NULL Scan"
            elif fingerprint.flags == "X":
                return "Xmas Scan"
        
        # Minimal options with SYN
        if fingerprint.flags == "S" and fingerprint.options_hash == "5c6d7e8f9a0b":
            return "Minimal Options Scan"
        
        return None
    
    def analyze_packet(self, packet: Packet) -> Dict:
        """Comprehensive packet analysis"""
        fp = self.generate_fingerprint(packet)
        if not fp:
            return {"error": "Not a valid TCP packet"}
        
        # Track fingerprint frequency
        self.seen_fingerprints[fp.raw_fingerprint] += 1
        
        # Perform analysis
        os_candidates = self.identify_os(fp)
        scan_type = self.detect_scan_type(fp)
        
        return {
            "fingerprint": fp.raw_fingerprint,
            "components": {
                "ttl": fp.ttl,
                "window_size": fp.window_size,
                "options_hash": fp.options_hash,
                "flags": fp.flags
            },
            "os_candidates": os_candidates,
            "scan_detection": scan_type,
            "first_seen": self.seen_fingerprints[fp.raw_fingerprint] == 1
        }
    
    def analyze_pcap(self, pcap_file: str) -> Dict:
        """Analyze entire PCAP file"""
        packets = rdpcap(pcap_file)
        results = {
            "total_packets": len(packets),
            "fingerprints": defaultdict(int),
            "os_distribution": defaultdict(int),
            "scan_activities": []
        }
        
        for packet in packets:
            if packet.haslayer(TCP) and packet[TCP].flags.S:
                analysis = self.analyze_packet(packet)
                
                if "error" not in analysis:
                    fp = analysis["fingerprint"]
                    results["fingerprints"][fp] += 1
                    
                    for os in analysis["os_candidates"]:
                        results["os_distribution"][os] += 1
                    
                    if analysis["scan_detection"]:
                        results["scan_activities"].append({
                            "src": packet[IP].src,
                            "dst": packet[IP].dst,
                            "dport": packet[TCP].dport,
                            "scan_type": analysis["scan_detection"],
                            "fingerprint": fp
                        })
        
        return dict(results)
 
# Example Usage
if __name__ == "__main__":
    fingerprinter = JA4TFingerprinter()
    
    # Capture live traffic
    def packet_callback(packet):
        if packet.haslayer(TCP) and packet[TCP].flags.S:
            result = fingerprinter.analyze_packet(packet)
            print(f"\n[+] New TCP SYN:")
            print(f"    Source: {packet[IP].src}:{packet[TCP].sport}")
            print(f"    Destination: {packet[IP].dst}:{packet[TCP].dport}")
            print(f"    Fingerprint: {result['fingerprint']}")
            print(f"    OS: {', '.join(result['os_candidates'])}")
            if result['scan_detection']:
                print(f"    ⚠ SCAN DETECTED: {result['scan_detection']}")
    
    print("[*] Starting JA4T capture (Ctrl+C to stop)...")
    sniff(filter="tcp[tcpflags] & tcp-syn != 0", prn=packet_callback, store=0)

tcp.flags.syn == 1 && tcp.flags.ack == 0

# Capture SYN packets
sudo tcpdump -i eth0 'tcp[tcpflags] & tcp-syn != 0' -w syn_capture.pcap
 
# Display with details
sudo tcpdump -i eth0 -vvv 'tcp[tcpflags] == 2' -c 100
 
# Filter specific port
sudo tcpdump -i eth0 'tcp port 80 and tcp[tcpflags] == 2'

from scapy.all import *
 
def capture_syn_packets(interface="eth0", count=100):
    """Capture TCP SYN packets"""
    
    def process_packet(pkt):
        if pkt.haslayer(TCP) and pkt[TCP].flags.S and not pkt[TCP].flags.A:
            print(f"\n[SYN] {pkt[IP].src}:{pkt[TCP].sport} → "
                  f"{pkt[IP].dst}:{pkt[TCP].dport}")
            print(f"TTL: {pkt[IP].ttl}, Window: {pkt[TCP].window}")
            print(f"Options: {pkt[TCP].options}")
    
    sniff(iface=interface, 
          filter="tcp[tcpflags] & tcp-syn != 0",
          prn=process_packet,
          count=count)
 
capture_syn_packets()

# ja4t.zeek
@load base/protocols/conn

module JA4T;

export {
    redef enum Log::ID += { LOG };
    
    type Info: record {
        ts: time &log;
        uid: string &log;
        src: addr &log;
        dst: addr &log;
        dport: port &log;
        ttl: count &log;
        window: count &log;
        options: string &log;
        ja4t: string &log;
    };
}

event connection_established(c: connection) {
    # Extract TCP options from SYN packet
    # Generate JA4T fingerprint
    # Log to ja4t.log
}

class OSDetector:
    """Advanced OS detection using JA4T"""
    
    def __init__(self):
        self.fingerprinter = JA4TFingerprinter()
        self.os_database = self._load_os_signatures()
    
    def _load_os_signatures(self) -> Dict[str, str]:
        """Load known OS fingerprints"""
        return {
            # Windows 10
            "128_8192_c3a7b8d4e5f6_S": "Windows 10/11",
            # Ubuntu Linux
            "64_14600_7e8f9a0b1c2d_S": "Ubuntu 20.04+",
            # macOS
            "64_16384_9a0b1c2d3e4f_S": "macOS 12+",
            # Windows 7
            "128_8192_a1b2c3d4e5f6_S": "Windows 7",
        }
    
    def detect_os(self, packet: Packet) -> Dict:
        """Detect OS from packet"""
        fp = self.fingerprinter.generate_fingerprint(packet)
        if not fp:
            return {"error": "Invalid packet"}
        
        # Check exact match
        if fp.raw_fingerprint in self.os_database:
            return {
                "os": self.os_database[fp.raw_fingerprint],
                "confidence": "high",
                "method": "exact_match"
            }
        
        # Fuzzy matching on TTL and window
        candidates = self.fingerprinter.identify_os(fp)
        
        return {
            "os": candidates,
            "confidence": "medium",
            "method": "heuristic"
        }

class PortScanDetector:
    """Detect port scanning activity using JA4T"""
    
    def __init__(self, threshold=10, time_window=60):
        self.fingerprinter = JA4TFingerprinter()
        self.threshold = threshold
        self.time_window = time_window
        self.scan_tracker = defaultdict(lambda: {
            "ports": set(),
            "first_seen": None,
            "last_seen": None,
            "fingerprints": set()
        })
    
    def analyze_packet(self, packet: Packet):
        """Analyze packet for scan indicators"""
        if not packet.haslayer(TCP) or not packet.haslayer(IP):
            return None
        
        src_ip = packet[IP].src
        dst_port = packet[TCP].dport
        timestamp = packet.time
        
        fp = self.fingerprinter.generate_fingerprint(packet)
        if not fp:
            return None
        
        # Track per source IP
        tracker = self.scan_tracker[src_ip]
        tracker["ports"].add(dst_port)
        tracker["fingerprints"].add(fp.raw_fingerprint)
        
        if tracker["first_seen"] is None:
            tracker["first_seen"] = timestamp
        tracker["last_seen"] = timestamp
        
        # Check for scan characteristics
        time_diff = tracker["last_seen"] - tracker["first_seen"]
        
        if (len(tracker["ports"]) >= self.threshold and 
            time_diff <= self.time_window):
            
            scan_type = self.fingerprinter.detect_scan_type(fp)
            
            return {
                "alert": "Port Scan Detected",
                "source": src_ip,
                "ports_scanned": len(tracker["ports"]),
                "time_window": time_diff,
                "scan_type": scan_type or "Unknown",
                "fingerprint": fp.raw_fingerprint
            }
        
        return None

class NATDetector:
    """Detect NAT/proxy using TTL and fingerprint analysis"""
    
    def __init__(self):
        self.fingerprinter = JA4TFingerprinter()
        self.ip_fingerprints = defaultdict(set)
    
    def analyze_connection(self, packet: Packet):
        """Detect NAT by analyzing fingerprint diversity"""
        if not packet.haslayer(IP):
            return None
        
        src_ip = packet[IP].src
        fp = self.fingerprinter.generate_fingerprint(packet)
        
        if fp:
            self.ip_fingerprints[src_ip].add(fp.raw_fingerprint)
            
            # Multiple fingerprints from same IP = likely NAT
            if len(self.ip_fingerprints[src_ip]) > 2:
                return {
                    "alert": "Possible NAT/Proxy Detected",
                    "ip": src_ip,
                    "unique_fingerprints": len(self.ip_fingerprints[src_ip]),
                    "fingerprints": list(self.ip_fingerprints[src_ip])
                }
        
        return None

# ja4t-detection.zeek
module JA4T;

export {
    global ja4t_fingerprints: table[string] of string = table();
    
    # Alert on suspicious fingerprints
    redef enum Notice::Type += {
        Suspicious_TCP_Fingerprint,
        Port_Scan_Detected
    };
}

event connection_state_remove(c: connection) {
    local ja4t = generate_ja4t(c);
    
    if (ja4t in suspicious_fingerprints) {
        NOTICE([$note=Suspicious_TCP_Fingerprint,
                $msg=fmt("Suspicious JA4T: %s", ja4t),
                $conn=c]);
    }
}

# suricata-ja4t.rules
alert tcp any any -> any any (msg:"Nmap SYN Scan Detected"; \
    tcp.flags:S; tcp.window:1024; \
    threshold:type threshold, track by_src, count 10, seconds 60; \
    classtype:attempted-recon; sid:1000001; rev:1;)
 
alert tcp any any -> any any (msg:"Unusual TCP Window Size"; \
    tcp.window:<512; tcp.flags:S; \
    classtype:suspicious; sid:1000002; rev:1;)

# Splunk query for JA4T analysis
index=network sourcetype=ja4t
| stats count by ja4t_fingerprint, src_ip
| where count > 100
| eval risk_score=case(
    like(ja4t_fingerprint, "%_1024_%"), 90,
    like(ja4t_fingerprint, "128_%"), 30,
    like(ja4t_fingerprint, "64_%"), 20,
    1==1, 10
  )
| where risk_score > 50
| table src_ip, ja4t_fingerprint, count, risk_score

def calculate_ttl_distance(current_ttl: int, original_ttl: int) -> int:
    """Calculate network distance based on TTL"""
    return original_ttl - current_ttl
 
def detect_ttl_manipulation(packet: Packet) -> bool:
    """Detect TTL manipulation (spoofing indicator)"""
    if not packet.haslayer(IP):
        return False
    
    ttl = packet[IP].ttl
    
    # TTL > common maximums indicates manipulation
    if ttl > 255:
        return True
    
    # TTL that doesn't match standard values
    standard_ttls = [32, 64, 128, 255]
    
    # If TTL is exactly a standard value, likely not manipulated
    if ttl in standard_ttls:
        return False
    
    # Check if TTL is suspiciously close to upper limits
    for std_ttl in standard_ttls:
        if abs(ttl - std_ttl) < 3:
            return False
    
    return True

def detect_tcp_option_fuzzing(options: List[int]) -> bool:
    """Detect unusual or malformed TCP options"""
    
    # Check for invalid option types
    valid_options = [0, 1, 2, 3, 4, 5, 8, 14, 15, 19, 28, 29]
    
    for opt in options:
        if opt not in valid_options:
            return True
    
    # Check for unusual option ordering
    # Standard: MSS, SACK, Timestamp, NOP, WScale
    if len(options) > 5:
        # Too many options
        return True
    
    # Multiple timestamps or MSS (invalid)
    if options.count(2) > 1 or options.count(8) > 1:
        return True
    
    return False

def analyze_window_size(window: int, window_scale: int = 0) -> Dict:
    """Analyze TCP window size for anomalies"""
    
    effective_window = window * (2 ** window_scale) if window_scale else window
    
    analysis = {
        "raw_window": window,
        "window_scale": window_scale,
        "effective_window": effective_window,
        "anomaly": False,
        "notes": []
    }
    
    # Zero window (suspicious for SYN)
    if window == 0:
        analysis["anomaly"] = True
        analysis["notes"].append("Zero window in SYN packet")
    
    # Window size is power of 2 (common for scanners)
    if window & (window - 1) == 0 and window >= 512:
        analysis["notes"].append("Power-of-2 window (possible scanner)")
    
    # Unusually small window
    if window < 1024:
        analysis["anomaly"] = True
        analysis["notes"].append("Unusually small window")
    
    # Maximum window without scaling
    if window == 65535 and window_scale == 0:
        analysis["notes"].append("Maximum window (possible tuned stack)")
    
    return analysis

def build_baseline(pcap_file: str, duration_days: int = 7):
    """Build fingerprint baseline for normal traffic"""
    fingerprinter = JA4TFingerprinter()
    baseline = {
        "fingerprints": defaultdict(int),
        "os_distribution": defaultdict(int),
        "window_sizes": defaultdict(int)
    }
    
    packets = rdpcap(pcap_file)
    
    for packet in packets:
        if packet.haslayer(TCP) and packet[TCP].flags.S:
            fp = fingerprinter.generate_fingerprint(packet)
            if fp:
                baseline["fingerprints"][fp.raw_fingerprint] += 1
                baseline["window_sizes"][fp.window_size] += 1
    
    return baseline

# Windows 10/11
128_8192_c3a7b8d4e5f6_S
Components: TTL=128, Win=8192, Options=MSS,NOP,WS,NOP,NOP,SACK,TS

# Windows Server 2019
128_8192_a1b2c3d4e5f6_S
Components: TTL=128, Win=8192, Options=MSS,NOP,NOP,SACK,TS,NOP,WS

# Ubuntu 20.04+
64_14600_7e8f9a0b1c2d_S
Components: TTL=64, Win=14600, Options=MSS,SACK,TS,NOP,WS

# CentOS 8
64_29200_3e4f5a6b7c8d_S
Components: TTL=64, Win=29200, Options=MSS,SACK,TS,NOP,WS

# macOS 12+
64_16384_9a0b1c2d3e4f_S
Components: TTL=64, Win=16384, Options=MSS,NOP,WS,NOP,NOP,TS,SACK,EOL

# Nmap default SYN scan
64_1024_5c6d7e8f9a0b_S
Components: TTL=64, Win=1024, Options=MSS

# Masscan
64_1024_0123456789ab_S
Components: TTL=64, Win=1024, Options=MSS,SACK,TS,NOP,WS

def normalize_ttl(ttl: int, tolerance: int = 5) -> int:
    """Normalize TTL to standard values with tolerance"""
    standard = [32, 64, 128, 255]
    
    for std_ttl in standard:
        if abs(ttl - std_ttl) <= tolerance:
            return std_ttl
    
    return ttl

# Adjust based on environment
PORT_SCAN_THRESHOLD = 20  # Increase for busy networks
TIME_WINDOW = 120  # Increase time window