JA4TS: TCP Server Response Fingerprinting

Client → Server: SYN (JA4T captures this)
Server → Client: SYN-ACK (JA4TS captures this)
Client → Server: ACK

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

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

IP TTL: 64
TCP Source Port: 443 (HTTPS)
TCP Flags: SYN+ACK
TCP Window Size: 5792
TCP Options:
  - MSS: 1460 bytes
  - SACK Permitted
  - Timestamp: 3456789012, 123456789
  - NOP
  - Window Scale: 7

IP TTL: 128
TCP Source Port: 443
TCP Flags: SYN+ACK
TCP Window Size: 8192
TCP Options:
  - MSS: 1460 bytes
  - NOP
  - Window Scale: 8
  - NOP
  - NOP
  - SACK Permitted
  - Timestamp: 9876543210, 234567890

IP TTL: 64
TCP Flags: RST+ACK
TCP Window Size: 0
TCP Options: None

import hashlib
from scapy.all import *
from typing import Optional, Dict, List, Tuple
from dataclasses import dataclass
from collections import defaultdict
import time
 
@dataclass
class JA4TSFingerprint:
    """Represents a JA4TS TCP Server fingerprint"""
    ttl: int
    window_size: int
    options_hash: str
    flags: str
    raw_fingerprint: str
    server_ip: str
    server_port: int
    
    def __str__(self) -> str:
        return self.raw_fingerprint
 
class JA4TSFingerprinter:
    """
    Production-ready JA4TS TCP Server Fingerprinter
    
    Analyzes TCP SYN-ACK packets to generate server fingerprints for
    OS detection, service identification, and infrastructure mapping.
    """
    
    # 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
    
    # Server OS TTL Signatures
    SERVER_OS_TTL = {
        64: ["Linux", "BSD", "Unix", "nginx", "Apache"],
        128: ["Windows Server", "IIS"],
        255: ["Cisco", "Network Device"]
    }
    
    # Server OS Window Signatures
    SERVER_OS_WINDOW = {
        5792: ["Linux 2.6+", "nginx", "Apache"],
        8192: ["Windows Server", "IIS"],
        14600: ["Linux with scaling"],
        16384: ["BSD variants"],
        65535: ["Tuned stack", "Custom"]
    }
    
    def __init__(self):
        self.server_db: Dict[str, Dict] = {}
        self.fingerprint_counts: Dict[str, int] = defaultdict(int)
    
    def extract_tcp_options(self, packet: Packet) -> List[int]:
        """Extract TCP option types in order from SYN-ACK"""
        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_code = self._option_name_to_code(opt[0])
                    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 server 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 from SYN-ACK"""
        if not packet.haslayer(TCP):
            return "UN"
        
        tcp = packet[TCP]
        
        # Check for SYN+ACK (normal server response)
        if tcp.flags.S and tcp.flags.A:
            return "SA"
        # RST+ACK (port closed/filtered)
        elif tcp.flags.R and tcp.flags.A:
            return "RA"
        # ACK only
        elif tcp.flags.A and not tcp.flags.S:
            return "A"
        # Other combinations
        else:
            return "X"  # Unknown/Other
    
    def generate_fingerprint(self, packet: Packet) -> Optional[JA4TSFingerprint]:
        """Generate JA4TS fingerprint from TCP SYN-ACK packet"""
        if not packet.haslayer(TCP) or not packet.haslayer(IP):
            return None
        
        tcp = packet[TCP]
        ip = packet[IP]
        
        # Verify this is a server response (SYN-ACK or RST-ACK)
        if not (tcp.flags.A):
            return None
        
        # 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 JA4TSFingerprint(
            ttl=ttl,
            window_size=window_size,
            options_hash=options_hash,
            flags=flags,
            raw_fingerprint=raw_fp,
            server_ip=ip.src,
            server_port=tcp.sport
        )
    
    def identify_server_os(self, fingerprint: JA4TSFingerprint) -> List[str]:
        """Identify likely server operating systems"""
        candidates = set()
        
        # Check TTL signatures
        if fingerprint.ttl in self.SERVER_OS_TTL:
            candidates.update(self.SERVER_OS_TTL[fingerprint.ttl])
        
        # Refine with window size
        if fingerprint.window_size in self.SERVER_OS_WINDOW:
            window_os = set(self.SERVER_OS_WINDOW[fingerprint.window_size])
            if candidates:
                candidates = candidates.intersection(window_os)
            else:
                candidates = window_os
        
        return list(candidates) if candidates else ["Unknown"]
    
    def identify_service(self, fingerprint: JA4TSFingerprint) -> str:
        """Identify service type based on port and fingerprint"""
        common_ports = {
            80: "HTTP",
            443: "HTTPS",
            22: "SSH",
            21: "FTP",
            25: "SMTP",
            3389: "RDP",
            3306: "MySQL",
            5432: "PostgreSQL",
            6379: "Redis",
            27017: "MongoDB"
        }
        
        return common_ports.get(fingerprint.server_port, f"Port {fingerprint.server_port}")
    
    def detect_load_balancer(self, fingerprint: JA4TSFingerprint) -> bool:
        """Detect if server is behind load balancer"""
        # Load balancers often normalize TCP options
        # Look for minimal options or unusual configurations
        
        if fingerprint.options_hash == "000000000000":
            return True
        
        # Very generic window sizes
        if fingerprint.window_size in [8192, 16384, 32768]:
            return True
        
        return False
    
    def detect_honeypot(self, fingerprint: JA4TSFingerprint) -> bool:
        """Detect potential honeypot based on unusual characteristics"""
        # Honeypots may have unusual TCP stack configurations
        
        # Window size of 0 on SYN-ACK is unusual
        if fingerprint.window_size == 0 and fingerprint.flags == "SA":
            return True
        
        # Unusual TTL values
        if fingerprint.ttl not in [32, 64, 128, 255]:
            return True
        
        # No TCP options on modern system (suspicious)
        if fingerprint.options_hash == "000000000000" and fingerprint.flags == "SA":
            return True
        
        return False
    
    def analyze_packet(self, packet: Packet) -> Dict:
        """Comprehensive server packet analysis"""
        fp = self.generate_fingerprint(packet)
        if not fp:
            return {"error": "Not a valid TCP server response"}
        
        # Track fingerprint
        self.fingerprint_counts[fp.raw_fingerprint] += 1
        
        # Store server information
        server_key = f"{fp.server_ip}:{fp.server_port}"
        if server_key not in self.server_db:
            self.server_db[server_key] = {
                "fingerprints": set(),
                "first_seen": time.time(),
                "packet_count": 0
            }
        
        self.server_db[server_key]["fingerprints"].add(fp.raw_fingerprint)
        self.server_db[server_key]["last_seen"] = time.time()
        self.server_db[server_key]["packet_count"] += 1
        
        # Perform analysis
        os_candidates = self.identify_server_os(fp)
        service = self.identify_service(fp)
        is_load_balanced = self.detect_load_balancer(fp)
        is_honeypot = self.detect_honeypot(fp)
        
        return {
            "fingerprint": fp.raw_fingerprint,
            "server": f"{fp.server_ip}:{fp.server_port}",
            "components": {
                "ttl": fp.ttl,
                "window_size": fp.window_size,
                "options_hash": fp.options_hash,
                "flags": fp.flags
            },
            "os_candidates": os_candidates,
            "service": service,
            "load_balanced": is_load_balanced,
            "possible_honeypot": is_honeypot,
            "first_seen": self.fingerprint_counts[fp.raw_fingerprint] == 1
        }
    
    def track_server_changes(self, server_ip: str, server_port: int) -> Dict:
        """Track fingerprint changes for a specific server"""
        server_key = f"{server_ip}:{server_port}"
        
        if server_key not in self.server_db:
            return {"error": "Server not found"}
        
        server_info = self.server_db[server_key]
        fingerprints = list(server_info["fingerprints"])
        
        # Multiple fingerprints indicate changes or load balancing
        if len(fingerprints) > 1:
            return {
                "server": server_key,
                "alert": "Multiple fingerprints detected",
                "fingerprints": fingerprints,
                "possible_causes": [
                    "Load balancer with multiple backend servers",
                    "Server configuration changed",
                    "Server compromised",
                    "Failover to different system"
                ]
            }
        
        return {
            "server": server_key,
            "status": "stable",
            "fingerprint": fingerprints[0],
            "duration": time.time() - server_info["first_seen"]
        }
    
    def analyze_pcap(self, pcap_file: str) -> Dict:
        """Analyze entire PCAP file for server fingerprints"""
        packets = rdpcap(pcap_file)
        results = {
            "total_packets": len(packets),
            "servers": {},
            "os_distribution": defaultdict(int),
            "service_distribution": defaultdict(int),
            "honeypot_alerts": []
        }
        
        for packet in packets:
            if packet.haslayer(TCP) and packet.haslayer(IP):
                tcp = packet[TCP]
                # Look for SYN-ACK or RST-ACK (server responses)
                if tcp.flags.A:
                    analysis = self.analyze_packet(packet)
                    
                    if "error" not in analysis:
                        server = analysis["server"]
                        
                        if server not in results["servers"]:
                            results["servers"][server] = {
                                "fingerprints": set(),
                                "os": set(),
                                "service": analysis["service"]
                            }
                        
                        results["servers"][server]["fingerprints"].add(
                            analysis["fingerprint"]
                        )
                        results["servers"][server]["os"].update(
                            analysis["os_candidates"]
                        )
                        
                        for os in analysis["os_candidates"]:
                            results["os_distribution"][os] += 1
                        
                        results["service_distribution"][analysis["service"]] += 1
                        
                        if analysis["possible_honeypot"]:
                            results["honeypot_alerts"].append({
                                "server": server,
                                "fingerprint": analysis["fingerprint"],
                                "reason": "Unusual TCP characteristics"
                            })
        
        # Convert sets to lists for JSON serialization
        for server in results["servers"]:
            results["servers"][server]["fingerprints"] = list(
                results["servers"][server]["fingerprints"]
            )
            results["servers"][server]["os"] = list(
                results["servers"][server]["os"]
            )
        
        return results
 
# Example Usage
if __name__ == "__main__":
    fingerprinter = JA4TSFingerprinter()
    
    # Capture live server responses
    def packet_callback(packet):
        if packet.haslayer(TCP) and packet.haslayer(IP):
            tcp = packet[TCP]
            # Only process server responses (ACK flag set)
            if tcp.flags.A:
                result = fingerprinter.analyze_packet(packet)
                
                if "error" not in result:
                    print(f"\n[+] Server Response:")
                    print(f"    Server: {result['server']}")
                    print(f"    Service: {result['service']}")
                    print(f"    Fingerprint: {result['fingerprint']}")
                    print(f"    OS: {', '.join(result['os_candidates'])}")
                    
                    if result['load_balanced']:
                        print(f"    ⚠ Possible Load Balancer")
                    
                    if result['possible_honeypot']:
                        print(f"    🚨 POSSIBLE HONEYPOT DETECTED")
    
    print("[*] Starting JA4TS capture (Ctrl+C to stop)...")
    print("[*] Listening for TCP SYN-ACK packets...")
    sniff(filter="tcp[tcpflags] & tcp-ack != 0", prn=packet_callback, store=0)

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

tcp.flags.ack == 1

# Capture SYN-ACK responses
sudo tcpdump -i eth0 'tcp[tcpflags] & (tcp-syn|tcp-ack) == (tcp-syn|tcp-ack)' -w server_responses.pcap
 
# Display with verbose output
sudo tcpdump -i eth0 -vvv 'tcp[tcpflags] == 18' -c 100
 
# Filter specific server
sudo tcpdump -i eth0 'src host 192.168.1.100 and tcp[tcpflags] == 18'

from scapy.all import *
 
def probe_server(target_ip: str, target_port: int = 80):
    """Actively probe server to get JA4TS fingerprint"""
    
    # Send SYN packet
    syn = IP(dst=target_ip)/TCP(dport=target_port, flags='S')
    
    # Wait for SYN-ACK
    response = sr1(syn, timeout=2, verbose=0)
    
    if response and response.haslayer(TCP):
        if response[TCP].flags.S and response[TCP].flags.A:
            # Got SYN-ACK, analyze it
            print(f"[+] Server responded:")
            print(f"    TTL: {response[IP].ttl}")
            print(f"    Window: {response[TCP].window}")
            print(f"    Options: {response[TCP].options}")
            
            # Send RST to close connection cleanly
            rst = IP(dst=target_ip)/TCP(dport=target_port, 
                                         sport=response[TCP].dport,
                                         flags='R',
                                         seq=response[TCP].ack)
            send(rst, verbose=0)
            
            return response
        elif response[TCP].flags.R:
            print(f"[-] Port closed (RST received)")
    else:
        print(f"[-] No response (filtered/timeout)")
    
    return None
 
# Probe multiple servers
servers = [
    ("google.com", 443),
    ("github.com", 443),
    ("192.168.1.1", 80)
]
 
for host, port in servers:
    print(f"\n[*] Probing {host}:{port}")
    probe_server(host, port)

# ja4ts.zeek
@load base/protocols/conn

module JA4TS;

export {
    redef enum Log::ID += { LOG };
    
    type Info: record {
        ts: time &log;
        uid: string &log;
        server_ip: addr &log;
        server_port: port &log;
        ttl: count &log;
        window: count &log;
        options: string &log;
        ja4ts: string &log;
        service: string &log;
    };
}

event connection_established(c: connection) {
    # Extract SYN-ACK characteristics
    # Generate JA4TS fingerprint
    # Log to ja4ts.log
}

class InfrastructureMapper:
    """Map attacker infrastructure using JA4TS"""
    
    def __init__(self):
        self.fingerprinter = JA4TSFingerprinter()
        self.infrastructure = defaultdict(lambda: {
            "servers": set(),
            "fingerprints": set(),
            "first_seen": None,
            "last_seen": None
        })
    
    def track_server(self, packet: Packet, threat_group: str):
        """Track server as part of threat infrastructure"""
        fp = self.fingerprinter.generate_fingerprint(packet)
        if not fp:
            return
        
        group_data = self.infrastructure[threat_group]
        group_data["servers"].add(f"{fp.server_ip}:{fp.server_port}")
        group_data["fingerprints"].add(fp.raw_fingerprint)
        
        current_time = time.time()
        if group_data["first_seen"] is None:
            group_data["first_seen"] = current_time
        group_data["last_seen"] = current_time
    
    def find_related_infrastructure(self, fingerprint: str) -> List[str]:
        """Find threat groups using same server fingerprint"""
        related_groups = []
        
        for group, data in self.infrastructure.items():
            if fingerprint in data["fingerprints"]:
                related_groups.append(group)
        
        return related_groups
    
    def generate_report(self) -> Dict:
        """Generate infrastructure mapping report"""
        report = {}
        
        for group, data in self.infrastructure.items():
            report[group] = {
                "server_count": len(data["servers"]),
                "unique_fingerprints": len(data["fingerprints"]),
                "servers": list(data["servers"]),
                "fingerprints": list(data["fingerprints"]),
                "active_duration": data["last_seen"] - data["first_seen"]
                    if data["first_seen"] else 0
            }
        
        return report

class CompromisedServerDetector:
    """Detect servers with changed fingerprints (potential compromise)"""
    
    def __init__(self, baseline_period: int = 86400):  # 24 hours
        self.fingerprinter = JA4TSFingerprinter()
        self.baseline_period = baseline_period
        self.server_baselines = {}
    
    def establish_baseline(self, pcap_file: str):
        """Establish baseline fingerprints for servers"""
        packets = rdpcap(pcap_file)
        
        for packet in packets:
            if packet.haslayer(TCP) and packet[TCP].flags.A:
                fp = self.fingerprinter.generate_fingerprint(packet)
                if fp:
                    server_key = f"{fp.server_ip}:{fp.server_port}"
                    
                    if server_key not in self.server_baselines:
                        self.server_baselines[server_key] = {
                            "fingerprints": set(),
                            "established": time.time()
                        }
                    
                    self.server_baselines[server_key]["fingerprints"].add(
                        fp.raw_fingerprint
                    )
    
    def check_server(self, packet: Packet) -> Optional[Dict]:
        """Check if server fingerprint deviates from baseline"""
        fp = self.fingerprinter.generate_fingerprint(packet)
        if not fp:
            return None
        
        server_key = f"{fp.server_ip}:{fp.server_port}"
        
        if server_key not in self.server_baselines:
            return {
                "alert": "New server detected",
                "server": server_key,
                "fingerprint": fp.raw_fingerprint
            }
        
        baseline = self.server_baselines[server_key]
        
        if fp.raw_fingerprint not in baseline["fingerprints"]:
            return {
                "alert": "Server fingerprint changed",
                "server": server_key,
                "new_fingerprint": fp.raw_fingerprint,
                "baseline_fingerprints": list(baseline["fingerprints"]),
                "possible_causes": [
                    "Server OS updated/patched",
                    "Load balancer configuration changed",
                    "Server compromised",
                    "Different backend server"
                ],
                "severity": "high"
            }
        
        return None

class CloudProviderDetector:
    """Identify cloud providers based on server fingerprints"""
    
    CLOUD_SIGNATURES = {
        # AWS patterns
        "aws": {
            "fingerprints": ["64_5792_7e8f9a0b1c2d_SA"],
            "ttl": [64],
            "notes": "AWS typically uses Linux with standard configs"
        },
        # Azure patterns
        "azure": {
            "fingerprints": ["128_8192_c3a7b8d4e5f6_SA"],
            "ttl": [128],
            "notes": "Azure often uses Windows Server"
        },
        # GCP patterns
        "gcp": {
            "fingerprints": ["64_5792_7e8f9a0b1c2d_SA"],
            "ttl": [64],
            "notes": "GCP uses custom Linux configurations"
        }
    }
    
    def __init__(self):
        self.fingerprinter = JA4TSFingerprinter()
    
    def identify_cloud_provider(self, packet: Packet) -> Optional[str]:
        """Identify cloud provider from server response"""
        fp = self.fingerprinter.generate_fingerprint(packet)
        if not fp:
            return None
        
        # Check against known cloud signatures
        for provider, signatures in self.CLOUD_SIGNATURES.items():
            if fp.raw_fingerprint in signatures["fingerprints"]:
                return provider
            if fp.ttl in signatures["ttl"]:
                # Possible match based on TTL
                return f"Possible {provider.upper()}"
        
        return "Unknown/On-Premise"

# ja4ts-detection.zeek
module JA4TS;

global known_servers: set[addr, port] = {};
global server_fingerprints: table[addr, port] of string = {};

event connection_established(c: connection) {
    local server_addr = c$id$resp_h;
    local server_port = c$id$resp_p;
    local ja4ts = generate_ja4ts_fingerprint(c);
    
    # Check if server known
    if ([server_addr, server_port] !in known_servers) {
        known_servers += [server_addr, server_port];
        server_fingerprints[server_addr, server_port] = ja4ts;
        
        NOTICE([$note=New_Server_Detected,
                $msg=fmt("New server: %s:%s, JA4TS: %s", 
                        server_addr, server_port, ja4ts),
                $conn=c]);
    }
    else {
        # Check if fingerprint changed
        local known_fp = server_fingerprints[server_addr, server_port];
        if (ja4ts != known_fp) {
            NOTICE([$note=Server_Fingerprint_Changed,
                    $msg=fmt("Server %s:%s fingerprint changed: %s → %s",
                            server_addr, server_port, known_fp, ja4ts),
                    $conn=c]);
        }
    }
}

# suricata-ja4ts.rules
alert tcp any any -> $HOME_NET any (msg:"Suspicious Server Response - Possible Honeypot"; \
    flow:established,from_server; \
    tcp.window:0; tcp.flags:SA; \
    classtype:suspicious-traffic; sid:2000001; rev:1;)
 
alert tcp any any -> $HOME_NET any (msg:"Server Fingerprint Anomaly"; \
    flow:established,from_server; \
    tcp.ttl:<32; \
    classtype:suspicious-traffic; sid:2000002; rev:1;)

# Track server fingerprint changes
index=network sourcetype=ja4ts
| stats values(ja4ts_fingerprint) as fingerprints, count by server_ip, server_port
| where mvcount(fingerprints) > 1
| eval risk_score=case(
    mvcount(fingerprints) > 3, 90,
    mvcount(fingerprints) == 2, 50,
    1==1, 10
  )
| where risk_score > 40
| table server_ip, server_port, fingerprints, count, risk_score

def analyze_server_timestamp(packet: Packet) -> Dict:
    """Analyze TCP timestamp option for additional insights"""
    if not packet.haslayer(TCP):
        return {}
    
    tcp = packet[TCP]
    
    for opt_name, opt_value in tcp.options:
        if opt_name == 'Timestamp':
            ts_val, ts_echo = opt_value
            
            # Analyze timestamp value
            analysis = {
                "timestamp_value": ts_val,
                "timestamp_echo": ts_echo,
                "timestamp_hz": None,
                "uptime_estimate": None
            }
            
            # Estimate timestamp frequency (typically 100Hz or 1000Hz)
            if ts_val < 1000000:
                analysis["timestamp_hz"] = 100
            else:
                analysis["timestamp_hz"] = 1000
            
            # Estimate server uptime
            if analysis["timestamp_hz"]:
                uptime_seconds = ts_val / analysis["timestamp_hz"]
                analysis["uptime_estimate"] = uptime_seconds
            
            return analysis
    
    return {}

def extract_window_scaling(packet: Packet) -> Dict:
    """Extract and analyze window scaling factor"""
    if not packet.haslayer(TCP):
        return {}
    
    tcp = packet[TCP]
    
    for opt_name, opt_value in tcp.options:
        if opt_name == 'WScale':
            scale_factor = opt_value
            raw_window = tcp.window
            effective_window = raw_window * (2 ** scale_factor)
            
            return {
                "raw_window": raw_window,
                "scale_factor": scale_factor,
                "effective_window": effective_window,
                "max_bandwidth_estimate": effective_window * 8 / 1024  # Kbps estimate
            }
    
    return {"raw_window": tcp.window, "scale_factor": 0}

class FingerprintEvolutionTracker:
    """Track how server fingerprints change over time"""
    
    def __init__(self):
        self.fingerprinter = JA4TSFingerprinter()
        self.timeline = defaultdict(list)
    
    def record_fingerprint(self, packet: Packet):
        """Record fingerprint with timestamp"""
        fp = self.fingerprinter.generate_fingerprint(packet)
        if fp:
            server_key = f"{fp.server_ip}:{fp.server_port}"
            self.timeline[server_key].append({
                "timestamp": time.time(),
                "fingerprint": fp.raw_fingerprint
            })
    
    def detect_changes(self, server_key: str) -> List[Dict]:
        """Detect fingerprint changes for a server"""
        if server_key not in self.timeline:
            return []
        
        timeline = self.timeline[server_key]
        changes = []
        
        for i in range(1, len(timeline)):
            if timeline[i]["fingerprint"] != timeline[i-1]["fingerprint"]:
                changes.append({
                    "timestamp": timeline[i]["timestamp"],
                    "old": timeline[i-1]["fingerprint"],
                    "new": timeline[i]["fingerprint"],
                    "time_diff": timeline[i]["timestamp"] - timeline[i-1]["timestamp"]
                })
        
        return changes

def build_server_baseline(pcap_file: str) -> Dict:
    """Build baseline of expected server fingerprints"""
    fingerprinter = JA4TSFingerprinter()
    baseline = defaultdict(lambda: {
        "fingerprints": set(),
        "sample_count": 0
    })
    
    packets = rdpcap(pcap_file)
    
    for packet in packets:
        if packet.haslayer(TCP) and packet[TCP].flags.A:
            fp = fingerprinter.generate_fingerprint(packet)
            if fp:
                server_key = f"{fp.server_ip}:{fp.server_port}"
                baseline[server_key]["fingerprints"].add(fp.raw_fingerprint)
                baseline[server_key]["sample_count"] += 1
    
    return dict(baseline)

# nginx on Ubuntu
64_5792_7e8f9a0b1c2d_SA
Components: TTL=64, Win=5792, Options=MSS,SACK,TS,NOP,WS

# Apache on CentOS
64_5792_7e8f9a0b1c2d_SA
Similar to nginx, common Linux configuration

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

# Windows Server 2016
128_8192_a1b2c3d4e5f6_SA
Slightly different option ordering

# Cisco Router
255_4096_1234567890ab_SA
Components: TTL=255, Win=4096, Options=MSS

# F5 Load Balancer
64_8192_b2c3d4e5f6g7_SA
Normalized TCP options

# Linux RST-ACK
64_0_000000000000_RA
Components: TTL=64, Win=0, No options

# Windows RST-ACK
128_0_000000000000_RA
Components: TTL=128, Win=0, No options

def analyze_fingerprint_distribution(server_key: str, fingerprints: List[str]):
    """Analyze why server has multiple fingerprints"""
    if len(set(fingerprints)) == 1:
        return "Stable single server"
    elif len(set(fingerprints)) <= 3:
        return "Likely load balanced (few backends)"
    else:
        return "Complex infrastructure or configuration changes"