JA4TSCAN: Active TCP Scanning Detection and Fingerprinting

Attacker → Target: SYN
Target → Attacker: SYN-ACK (port open) or RST (port closed)
Attacker → Target: RST (never completes handshake)

Attacker → Target: ACK
Target → Attacker: RST (if port accessible) or No response (filtered)

Attacker → Target: FIN
Target → Attacker: No response (open) or RST (closed)

Attacker → Target: (no flags)
Target → Attacker: No response (open) or RST (closed)

Attacker → Target: FIN+PSH+URG
Target → Attacker: No response (open) or RST (closed)

Attacker → Target: SYN
Target → Attacker: SYN-ACK
Attacker → Target: ACK
Attacker → Target: RST or FIN (closes connection)

<scan_type>_<rate>_<pattern>_<ttl>_<window>_<options_hash>

Scan Type: SYN
Rate: ~1000 pps
Pattern: Top 1000 ports (sequential within top ports)
TTL: 64 (Linux default)
Window: 1024 (Nmap default)
Options: MSS only (minimal)

Scan Type: SYN
Rate: 10,000+ pps
Pattern: Random
TTL: 64
Window: 1024
Options: MSS, SACK, Timestamp, Window Scale

Scan Type: FIN
Rate: 10 pps (slow and stealthy)
Pattern: Sequential
TTL: 64
Window: 4096
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
import statistics
 
@dataclass
class ScanActivity:
    """Represents detected scan activity"""
    source_ip: str
    scan_type: str
    rate: str
    pattern: str
    ttl: int
    window: int
    options_hash: str
    first_seen: float
    last_seen: float
    ports_scanned: set
    packet_count: int
    fingerprint: str
 
class JA4TSCANDetector:
    """
    Production-ready JA4TSCAN Detection System
    
    Detects, classifies, and fingerprints active TCP scanning activities.
    Identifies scanning tools and techniques.
    """
    
    # Rate thresholds (packets per second)
    RATE_LOW = 10
    RATE_MEDIUM = 100
    RATE_HIGH = 1000
    RATE_VHIGH = 10000
    
    # Scan detection thresholds
    MIN_PORTS_FOR_SCAN = 5
    SCAN_WINDOW = 60  # seconds
    
    # Known scanner signatures
    SCANNER_SIGNATURES = {
        "SYN_HIGH_TOP_64_1024_5c6d7e8f9a0b": "Nmap Default",
        "SYN_EXTREME_RAND_64_1024_a1b2c3d4e5f6": "Masscan",
        "SYN_EXTREME_RAND_64_1024_0123456789ab": "Zmap",
        "CONNECT_MEDIUM_SEQ_64_29200_7e8f9a0b1c2d": "Nikto",
        "SYN_LOW_SEQ_64_4096_000000000000": "Stealth Scan",
    }
    
    def __init__(self):
        self.tracked_sources: Dict[str, Dict] = defaultdict(self._init_tracker)
        self.detected_scans: List[ScanActivity] = []
    
    def _init_tracker(self) -> Dict:
        """Initialize tracking structure for a source IP"""
        return {
            "packets": [],
            "ports": set(),
            "first_seen": None,
            "last_seen": None,
            "ttl": [],
            "window": [],
            "tcp_flags": defaultdict(int),
            "options": []
        }
    
    def analyze_packet(self, packet: Packet) -> Optional[Dict]:
        """Analyze packet for scan characteristics"""
        if not packet.haslayer(TCP) or not packet.haslayer(IP):
            return None
        
        ip = packet[IP]
        tcp = packet[TCP]
        
        src_ip = ip.src
        dst_port = tcp.dport
        timestamp = packet.time if hasattr(packet, 'time') else time.time()
        
        # Update tracker
        tracker = self.tracked_sources[src_ip]
        
        if tracker["first_seen"] is None:
            tracker["first_seen"] = timestamp
        
        tracker["last_seen"] = timestamp
        tracker["packets"].append({
            "timestamp": timestamp,
            "dst_port": dst_port,
            "flags": tcp.flags
        })
        tracker["ports"].add(dst_port)
        tracker["ttl"].append(ip.ttl)
        tracker["window"].append(tcp.window)
        tracker["tcp_flags"][str(tcp.flags)] += 1
        
        if hasattr(tcp, 'options') and tcp.options:
            options = [opt[0] if isinstance(opt, tuple) else opt 
                      for opt in tcp.options]
            tracker["options"].append(tuple(options))
        
        # Check if this constitutes a scan
        if len(tracker["ports"]) >= self.MIN_PORTS_FOR_SCAN:
            time_span = tracker["last_seen"] - tracker["first_seen"]
            
            if time_span <= self.SCAN_WINDOW:
                return self._classify_scan(src_ip, tracker)
        
        return None
    
    def _classify_scan(self, source_ip: str, tracker: Dict) -> Dict:
        """Classify scan type and generate fingerprint"""
        # Determine scan type from TCP flags
        most_common_flags = max(tracker["tcp_flags"].items(), 
                               key=lambda x: x[1])[0]
        scan_type = self._identify_scan_type(most_common_flags)
        
        # Calculate rate
        time_span = tracker["last_seen"] - tracker["first_seen"]
        if time_span > 0:
            rate_pps = len(tracker["packets"]) / time_span
            rate = self._classify_rate(rate_pps)
        else:
            rate = "UNKNOWN"
        
        # Determine port pattern
        pattern = self._analyze_port_pattern(
            [p["dst_port"] for p in tracker["packets"]]
        )
        
        # Get average TTL and window
        avg_ttl = int(statistics.mean(tracker["ttl"])) if tracker["ttl"] else 0
        avg_window = int(statistics.mean(tracker["window"])) if tracker["window"] else 0
        
        # Infer original TTL
        original_ttl = self._infer_original_ttl(avg_ttl)
        
        # Compute options hash
        if tracker["options"]:
            most_common_opts = max(set(tracker["options"]), 
                                  key=tracker["options"].count)
            options_hash = self._compute_options_hash(most_common_opts)
        else:
            options_hash = "000000000000"
        
        # Generate fingerprint
        fingerprint = (f"{scan_type}_{rate}_{pattern}_"
                      f"{original_ttl}_{avg_window}_{options_hash}")
        
        # Create scan activity record
        scan = ScanActivity(
            source_ip=source_ip,
            scan_type=scan_type,
            rate=rate,
            pattern=pattern,
            ttl=original_ttl,
            window=avg_window,
            options_hash=options_hash,
            first_seen=tracker["first_seen"],
            last_seen=tracker["last_seen"],
            ports_scanned=tracker["ports"].copy(),
            packet_count=len(tracker["packets"]),
            fingerprint=fingerprint
        )
        
        self.detected_scans.append(scan)
        
        # Identify scanner tool
        tool = self._identify_scanner(fingerprint)
        
        return {
            "alert": "Port Scan Detected",
            "source_ip": source_ip,
            "scan_type": scan_type,
            "rate": rate,
            "rate_pps": len(tracker["packets"]) / (tracker["last_seen"] - tracker["first_seen"]),
            "pattern": pattern,
            "ports_count": len(tracker["ports"]),
            "ports": sorted(list(tracker["ports"])),
            "fingerprint": fingerprint,
            "scanner_tool": tool,
            "duration": tracker["last_seen"] - tracker["first_seen"],
            "severity": self._calculate_severity(scan)
        }
    
    def _identify_scan_type(self, flags_str: str) -> str:
        """Identify scan type from TCP flags"""
        # Parse flags string (e.g., "S", "SA", "F", etc.)
        if 'S' in flags_str and 'A' not in flags_str:
            return "SYN"
        elif 'A' in flags_str and 'S' not in flags_str:
            return "ACK"
        elif 'F' in flags_str:
            return "FIN"
        elif 'R' in flags_str:
            return "RST"
        elif flags_str == "" or flags_str == "0":
            return "NULL"
        elif 'F' in flags_str and 'P' in flags_str and 'U' in flags_str:
            return "XMAS"
        elif 'S' in flags_str and 'A' in flags_str:
            return "CONNECT"
        else:
            return "CUSTOM"
    
    def _classify_rate(self, pps: float) -> str:
        """Classify scan rate"""
        if pps < self.RATE_LOW:
            return "LOW"
        elif pps < self.RATE_MEDIUM:
            return "MEDIUM"
        elif pps < self.RATE_HIGH:
            return "HIGH"
        elif pps < self.RATE_VHIGH:
            return "VHIGH"
        else:
            return "EXTREME"
    
    def _analyze_port_pattern(self, ports: List[int]) -> str:
        """Analyze port scanning pattern"""
        if len(ports) < 2:
            return "SINGLE"
        
        # Check if sequential
        sorted_ports = sorted(ports)
        is_sequential = all(
            sorted_ports[i+1] - sorted_ports[i] <= 5 
            for i in range(len(sorted_ports)-1)
        )
        
        if is_sequential:
            return "SEQ"
        
        # Check if top ports
        common_ports = {
            21, 22, 23, 25, 53, 80, 110, 111, 135, 139,
            143, 443, 445, 993, 995, 1723, 3306, 3389, 
            5900, 8080
        }
        
        if len(set(ports) & common_ports) / len(ports) > 0.5:
            return "TOP"
        
        return "RAND"
    
    def _infer_original_ttl(self, current_ttl: int) -> int:
        """Infer original TTL"""
        common_ttls = [32, 64, 128, 255]
        for ttl in common_ttls:
            if current_ttl <= ttl:
                return ttl
        return current_ttl
    
    def _compute_options_hash(self, options: Tuple) -> str:
        """Compute hash of TCP options"""
        if not options:
            return "000000000000"
        
        opt_string = ",".join(str(opt) for opt in options)
        hash_obj = hashlib.sha256(opt_string.encode())
        return hash_obj.hexdigest()[:12]
    
    def _identify_scanner(self, fingerprint: str) -> str:
        """Identify scanning tool from fingerprint"""
        if fingerprint in self.SCANNER_SIGNATURES:
            return self.SCANNER_SIGNATURES[fingerprint]
        
        # Partial matching
        if "EXTREME" in fingerprint and "RAND" in fingerprint:
            return "Masscan/Zmap-like"
        elif "HIGH" in fingerprint and "TOP" in fingerprint:
            return "Nmap-like"
        elif "LOW" in fingerprint:
            return "Stealth Scanner"
        
        return "Unknown Scanner"
    
    def _calculate_severity(self, scan: ScanActivity) -> str:
        """Calculate threat severity"""
        score = 0
        
        # Rate contributes to severity
        rate_scores = {
            "LOW": 1, "MEDIUM": 2, "HIGH": 3, 
            "VHIGH": 4, "EXTREME": 5
        }
        score += rate_scores.get(scan.rate, 0)
        
        # Port count
        if len(scan.ports_scanned) > 100:
            score += 3
        elif len(scan.ports_scanned) > 20:
            score += 2
        else:
            score += 1
        
        # Stealth techniques
        if scan.scan_type in ["FIN", "NULL", "XMAS"]:
            score += 2
        
        # Classify
        if score >= 7:
            return "CRITICAL"
        elif score >= 5:
            return "HIGH"
        elif score >= 3:
            return "MEDIUM"
        else:
            return "LOW"
    
    def generate_report(self) -> Dict:
        """Generate comprehensive scan detection report"""
        report = {
            "total_scans": len(self.detected_scans),
            "scans_by_type": defaultdict(int),
            "scans_by_rate": defaultdict(int),
            "scans_by_tool": defaultdict(int),
            "unique_sources": set(),
            "severity_distribution": defaultdict(int),
            "scan_details": []
        }
        
        for scan in self.detected_scans:
            report["scans_by_type"][scan.scan_type] += 1
            report["scans_by_rate"][scan.rate] += 1
            report["unique_sources"].add(scan.source_ip)
            
            tool = self._identify_scanner(scan.fingerprint)
            report["scans_by_tool"][tool] += 1
            
            severity = self._calculate_severity(scan)
            report["severity_distribution"][severity] += 1
            
            report["scan_details"].append({
                "source": scan.source_ip,
                "type": scan.scan_type,
                "fingerprint": scan.fingerprint,
                "ports_scanned": len(scan.ports_scanned),
                "duration": scan.last_seen - scan.first_seen,
                "severity": severity
            })
        
        report["unique_sources"] = list(report["unique_sources"])
        
        return report
    
    def monitor_live(self, interface="eth0", duration=None):
        """Monitor live traffic for scans"""
        def packet_handler(packet):
            result = self.analyze_packet(packet)
            if result:
                print(f"\n[!] SCAN DETECTED:")
                print(f"    Source: {result['source_ip']}")
                print(f"    Type: {result['scan_type']} scan")
                print(f"    Rate: {result['rate']} ({result['rate_pps']:.2f} pps)")
                print(f"    Ports: {result['ports_count']} ports scanned")
                print(f"    Pattern: {result['pattern']}")
                print(f"    Scanner: {result['scanner_tool']}")
                print(f"    Severity: {result['severity']}")
                print(f"    Fingerprint: {result['fingerprint']}")
        
        print(f"[*] Monitoring {interface} for scan activity...")
        if duration:
            print(f"[*] Will run for {duration} seconds")
            sniff(iface=interface, prn=packet_handler, 
                  timeout=duration, store=0)
        else:
            print("[*] Press Ctrl+C to stop")
            sniff(iface=interface, prn=packet_handler, store=0)
 
# Example Usage
if __name__ == "__main__":
    detector = JA4TSCANDetector()
    
    # Monitor live traffic
    try:
        detector.monitor_live(interface="eth0", duration=300)
    except KeyboardInterrupt:
        print("\n[*] Stopping monitoring...")
    
    # Generate report
    report = detector.generate_report()
    print(f"\n{'='*60}")
    print("SCAN DETECTION REPORT")
    print(f"{'='*60}")
    print(f"Total Scans Detected: {report['total_scans']}")
    print(f"Unique Sources: {len(report['unique_sources'])}")
    print(f"\nScans by Type:")
    for scan_type, count in report['scans_by_type'].items():
        print(f"  {scan_type}: {count}")
    print(f"\nScans by Tool:")
    for tool, count in report['scans_by_tool'].items():
        print(f"  {tool}: {count}")
    print(f"\nSeverity Distribution:")
    for severity, count in report['severity_distribution'].items():
        print(f"  {severity}: {count}")

class RealTimeScanDetector:
    """Real-time scan detection with alerting"""
    
    def __init__(self, alert_callback=None):
        self.detector = JA4TSCANDetector()
        self.alert_callback = alert_callback
    
    def start_monitoring(self, interface="eth0"):
        """Start real-time monitoring"""
        def packet_handler(packet):
            result = self.detector.analyze_packet(packet)
            if result and result.get("alert"):
                self._send_alert(result)
        
        sniff(iface=interface, prn=packet_handler, store=0)
    
    def _send_alert(self, result: Dict):
        """Send alert for detected scan"""
        if self.alert_callback:
            self.alert_callback(result)
        else:
            self._default_alert(result)
    
    def _default_alert(self, result: Dict):
        """Default console alert"""
        print(f"\n{'!'*60}")
        print(f"ALERT: {result['alert']}")
        print(f"Source: {result['source_ip']}")
        print(f"Scanner: {result['scanner_tool']}")
        print(f"Severity: {result['severity']}")
        print(f"{'!'*60}\n")

def analyze_pcap_for_scans(pcap_file: str) -> Dict:
    """Analyze PCAP file for scanning activity"""
    detector = JA4TSCANDetector()
    packets = rdpcap(pcap_file)
    
    print(f"[*] Analyzing {len(packets)} packets...")
    
    for packet in packets:
        detector.analyze_packet(packet)
    
    return detector.generate_report()

class EvasionDetector:
    """Detect firewall evasion techniques in scans"""
    
    def __init__(self):
        self.evasion_patterns = []
    
    def detect_fragmentation(self, packet: Packet) -> bool:
        """Detect fragmented scan packets"""
        if packet.haslayer(IP):
            return packet[IP].flags.MF or packet[IP].frag > 0
        return False
    
    def detect_decoy_scan(self, packets: List[Packet]) -> Dict:
        """Detect Nmap-style decoy scans"""
        # Multiple source IPs targeting same ports
        sources = defaultdict(set)
        
        for packet in packets:
            if packet.haslayer(IP) and packet.haslayer(TCP):
                sources[packet[IP].src].add(packet[TCP].dport)
        
        # Check if multiple sources scan same ports
        port_sets = list(sources.values())
        if len(port_sets) > 1:
            common_ports = set.intersection(*port_sets)
            if len(common_ports) > 5:
                return {
                    "evasion": "Decoy Scan",
                    "sources": list(sources.keys()),
                    "common_ports": len(common_ports)
                }
        
        return {}
    
    def detect_timing_evasion(self, timestamps: List[float]) -> str:
        """Detect timing-based evasion (slow scans)"""
        if len(timestamps) < 2:
            return "None"
        
        intervals = [timestamps[i+1] - timestamps[i] 
                    for i in range(len(timestamps)-1)]
        avg_interval = statistics.mean(intervals)
        
        if avg_interval > 300:  # 5 minutes between packets
            return "Paranoid Timing (Nmap -T0)"
        elif avg_interval > 15:
            return "Sneaky Timing (Nmap -T1)"
        elif avg_interval > 0.4:
            return "Polite Timing (Nmap -T2)"
        
        return "Normal or Aggressive"

def passive_monitoring():
    """Monitor and log without response"""
    detector = JA4TSCANDetector()
    
    def log_scan(packet):
        result = detector.analyze_packet(packet)
        if result:
            # Log to SIEM
            log_to_siem(result)
            # Update threat intel
            update_threat_intel(result['source_ip'], result['fingerprint'])
    
    sniff(prn=log_scan, store=0)

def tarpit_scanner(scanner_ip: str, port: int):
    """Slow down scanner with tarpit technique"""
    # WARNING: Use only on systems you own
    
    def respond_slowly(packet):
        if packet.haslayer(TCP) and packet[IP].src == scanner_ip:
            # Respond with very small window size
            response = IP(dst=scanner_ip)/TCP(
                dport=packet[TCP].sport,
                sport=port,
                flags='SA',
                window=1,  # Minimal window
                ack=packet[TCP].seq + 1
            )
            send(response, verbose=0)
            # Add random delay
            time.sleep(random.uniform(1, 5))
    
    sniff(prn=respond_slowly, store=0)

def integrate_honeypot(scan_result: Dict):
    """Redirect scanner to honeypot"""
    scanner_ip = scan_result['source_ip']
    
    # Add firewall rule to redirect to honeypot
    honeypot_ip = "192.168.1.100"
    
    # Using iptables (Linux)
    rule = f"iptables -t nat -A PREROUTING -s {scanner_ip} -j DNAT --to-destination {honeypot_ip}"
    
    # Log the redirection
    print(f"[*] Redirecting {scanner_ip} to honeypot {honeypot_ip}")

# ja4tscan.zeek
module JA4TSCAN;

export {
    global scan_threshold = 10;
    global scan_window = 60sec;
    
    redef enum Notice::Type += {
        Port_Scan_Detected,
        High_Speed_Scan,
        Stealth_Scan_Attempt
    };
}

event new_connection(c: connection) {
    # Track connection attempts per source
    # Detect scan patterns
    # Generate JA4TSCAN fingerprint
}

# suricata-ja4tscan.rules
alert tcp any any -> $HOME_NET any (msg:"Nmap SYN Scan Detected"; \
    flags:S,12; tcp.window:1024; \
    threshold:type threshold, track by_src, count 10, seconds 60; \
    classtype:attempted-recon; sid:3000001; rev:1;)
 
alert tcp any any -> $HOME_NET any (msg:"High-Speed Scan (Masscan-like)"; \
    flags:S,12; \
    threshold:type threshold, track by_src, count 100, seconds 10; \
    classtype:attempted-recon; sid:3000002; rev:1;)
 
alert tcp any any -> $HOME_NET any (msg:"Stealth FIN Scan"; \
    flags:F,12; \
    threshold:type threshold, track by_src, count 5, seconds 60; \
    classtype:attempted-recon; sid:3000003; rev:1;)

# Detect port scans
index=network sourcetype=firewall action=allow
| stats dc(dest_port) as unique_ports, count by src_ip
| where unique_ports > 20 AND count > 50
| eval scan_rate=count/60
| eval severity=case(
    scan_rate > 100, "CRITICAL",
    scan_rate > 50, "HIGH",
    scan_rate > 10, "MEDIUM",
    1==1, "LOW"
  )
| table src_ip, unique_ports, count, scan_rate, severity

# Adjust based on your environment
MIN_PORTS_FOR_SCAN = 10  # Increase in noisy networks
SCAN_WINDOW = 120  # Increase for slow scan detection

WHITELIST = [
    "192.168.1.10",  # Internal vulnerability scanner
    "10.0.0.50",      # Security team scanner
]
 
def is_whitelisted(ip: str) -> bool:
    return ip in WHITELIST

def check_threat_intel(ip: str) -> Dict:
    """Check IP against threat intelligence feeds"""
    # Query threat intel APIs
    # Check reputation services
    # Return threat level and context
    pass

def apply_rate_limiting(scanner_ip: str):
    """Apply rate limiting to suspected scanner"""
    # Linux iptables
    rule = f"iptables -A INPUT -s {scanner_ip} -m limit --limit 1/sec -j ACCEPT"
    os.system(rule)

# Default SYN scan
SYN_HIGH_TOP_64_1024_5c6d7e8f9a0b

# Aggressive scan (-T4)
SYN_VHIGH_TOP_64_1024_5c6d7e8f9a0b

# Stealth scan (-T0/-T1)
SYN_LOW_SEQ_64_1024_5c6d7e8f9a0b

# FIN scan
FIN_MEDIUM_SEQ_64_4096_000000000000

# Ultra-fast scan
SYN_EXTREME_RAND_64_1024_a1b2c3d4e5f6

# Rate-limited
SYN_VHIGH_RAND_64_1024_a1b2c3d4e5f6

# Internet-wide scan
SYN_EXTREME_RAND_64_1024_0123456789ab

# Low and slow APT scan
SYN_LOW_CUST_64_8192_b2c3d4e5f6g7

# Botnet scanning
SYN_MEDIUM_RAND_128_8192_c3a7b8d4e5f6

# Whitelist known internal sources
# Increase MIN_PORTS_FOR_SCAN threshold
# Consider destination ports (don't alert on single service)

# Increase SCAN_WINDOW to 300+ seconds
# Implement long-term tracking
# Use statistical analysis for anomaly detection

# Build custom signatures
# Focus on behavioral patterns
# Correlate with other indicators