JAD: DHCP and DNS Client Fingerprinting

Client → Broadcast: DHCP DISCOVER (JAD fingerprints this)
Server → Client: DHCP OFFER
Client → Server: DHCP REQUEST (Also fingerprinted)
Server → Client: DHCP ACK

DHCP:<dhcp_hash>_DNS:<dns_hash>_PATTERN:<pattern_code>

Options: 55=[1,3,6,15,119,252], 60="MSFT 5.0", 57=1500
String: "55:1,3,6,15,119,252;60:MSFT 5.0;57:1500"
Hash: SHA256(string)[:12] = a1b2c3d4e5f6

EDNS: 4096, Types: [A, AAAA], Flags: RD, Simultaneous: 2
String: "EDNS:4096;TYPES:A,AAAA;FLAGS:RD;SIM:2"
Hash: SHA256(string)[:12] = g7h8i9j0k1l2

Option 55: [1, 15, 3, 6, 44, 46, 47, 31, 33, 121, 249, 252]
Option 60: "MSFT 5.0"
Option 57: 1500
Option 61: 01:aa:bb:cc:dd:ee:ff

Query: www.microsoft.com (Type A)
EDNS Buffer: 4096
Flags: RD (Recursion Desired)

Option 55: [1, 121, 3, 6, 15, 119, 252, 95, 44, 46]
Option 60: "dhcpcd-5.5.6"
Option 57: 1500

Query: apple.com (Type A and AAAA simultaneously)
EDNS Buffer: 4096
Flags: RD

Queries: 
  - aGVsbG8ud29ybGQ.evil.com (Type TXT)
  - ZGF0YS5leGZpbHRyYXRpb24.evil.com (Type NULL)
Unusual characteristics:
  - Long subdomain names (base64-like)
  - High query frequency
  - Non-standard record types (TXT, NULL)
  - Single domain with random subdomains

import hashlib
from scapy.all import *
from typing import Optional, Dict, List, Set
from dataclasses import dataclass
from collections import defaultdict
import statistics
import base64
import re
 
@dataclass
class JADFingerprint:
    """Represents a JAD DHCP/DNS fingerprint"""
    dhcp_hash: str
    dns_hash: str
    pattern: str
    raw_fingerprint: str
    device_type: str
    confidence: float
 
class JADFingerprinter:
    """
    Production-ready JAD DHCP/DNS Fingerprinter
    
    Identifies devices and detects malicious activities through
    DHCP and DNS protocol analysis.
    """
    
    # Known DHCP Option 55 signatures
    DHCP_SIGNATURES = {
        "1,15,3,6,44,46,47,31,33,121,249,252": "Windows 10/11",
        "1,3,6,15,119,252": "macOS",
        "1,121,3,6,15,119,252,95,44,46": "iOS",
        "1,28,2,3,15,6,119,12,44,47,26,121,42": "Linux (dhclient)",
        "1,3,6,12,15,28,42": "Android",
    }
    
    # DNS tunneling indicators
    TUNNEL_INDICATORS = {
        "long_subdomains": 50,  # Character threshold
        "high_entropy": 4.5,    # Shannon entropy
        "base64_pattern": True,
        "txt_null_queries": True
    }
    
    def __init__(self):
        self.dhcp_cache: Dict[str, Dict] = {}
        self.dns_cache: Dict[str, List] = defaultdict(list)
        self.device_fingerprints: Dict[str, JADFingerprint] = {}
    
    def analyze_dhcp(self, packet: Packet) -> Optional[Dict]:
        """Analyze DHCP DISCOVER/REQUEST packet"""
        if not packet.haslayer(DHCP):
            return None
        
        dhcp = packet[DHCP]
        
        # Extract DHCP options
        options = {}
        for opt in dhcp.options:
            if isinstance(opt, tuple):
                opt_code, opt_value = opt[0], opt[1]
                options[opt_code] = opt_value
        
        # Extract client MAC
        if packet.haslayer(Ether):
            client_mac = packet[Ether].src
        else:
            client_mac = "unknown"
        
        # Extract key options
        param_request_list = options.get('param_req_list', [])
        vendor_class = options.get('vendor_class_id', b'').decode('utf-8', errors='ignore')
        max_dhcp_size = options.get('max_dhcp_size', 0)
        requested_addr = options.get('requested_addr', '')
        hostname = options.get('hostname', b'').decode('utf-8', errors='ignore')
        
        # Generate DHCP hash
        dhcp_string = self._build_dhcp_string(
            param_request_list, vendor_class, max_dhcp_size
        )
        dhcp_hash = hashlib.sha256(dhcp_string.encode()).hexdigest()[:12]
        
        # Identify device type
        device_type = self._identify_device_from_dhcp(param_request_list)
        
        # Cache DHCP info
        self.dhcp_cache[client_mac] = {
            "dhcp_hash": dhcp_hash,
            "device_type": device_type,
            "hostname": hostname,
            "vendor": vendor_class,
            "param_list": param_request_list
        }
        
        return {
            "client_mac": client_mac,
            "dhcp_hash": dhcp_hash,
            "device_type": device_type,
            "hostname": hostname,
            "vendor": vendor_class
        }
    
    def analyze_dns(self, packet: Packet, client_ip: str) -> Optional[Dict]:
        """Analyze DNS query packet"""
        if not packet.haslayer(DNS):
            return None
        
        dns = packet[DNS]
        
        # Only analyze queries
        if dns.qr != 0:  # qr=1 means response
            return None
        
        # Extract query information
        query_name = dns.qd.qname.decode('utf-8', errors='ignore').rstrip('.')
        query_type = dns.qd.qtype
        
        # Check for EDNS
        edns_size = 0
        if packet.haslayer(DNSRROPT):
            edns_size = packet[DNSRROPT].rclass
        
        # Record DNS query
        self.dns_cache[client_ip].append({
            "query": query_name,
            "type": query_type,
            "edns": edns_size,
            "timestamp": time.time()
        })
        
        # Analyze for malicious patterns
        pattern = self._analyze_dns_pattern(client_ip)
        
        # Generate DNS hash
        dns_string = self._build_dns_string(client_ip)
        dns_hash = hashlib.sha256(dns_string.encode()).hexdigest()[:12]
        
        return {
            "client_ip": client_ip,
            "query": query_name,
            "query_type": query_type,
            "dns_hash": dns_hash,
            "pattern": pattern,
            "malicious": pattern not in ["NM"]
        }
    
    def generate_jad_fingerprint(self, client_id: str) -> Optional[JADFingerprint]:
        """Generate complete JAD fingerprint for a client"""
        # Get DHCP info
        dhcp_info = self.dhcp_cache.get(client_id)
        if not dhcp_info:
            return None
        
        # Get DNS info
        dns_queries = self.dns_cache.get(client_id, [])
        
        if dns_queries:
            dns_string = self._build_dns_string(client_id)
            dns_hash = hashlib.sha256(dns_string.encode()).hexdigest()[:12]
            pattern = self._analyze_dns_pattern(client_id)
        else:
            dns_hash = "000000000000"
            pattern = "NM"
        
        # Construct fingerprint
        raw_fp = f"DHCP:{dhcp_info['dhcp_hash']}_DNS:{dns_hash}_PATTERN:{pattern}"
        
        fingerprint = JADFingerprint(
            dhcp_hash=dhcp_info['dhcp_hash'],
            dns_hash=dns_hash,
            pattern=pattern,
            raw_fingerprint=raw_fp,
            device_type=dhcp_info['device_type'],
            confidence=0.85 if dns_queries else 0.60
        )
        
        self.device_fingerprints[client_id] = fingerprint
        
        return fingerprint
    
    def _build_dhcp_string(self, param_list: List, vendor: str, max_size: int) -> str:
        """Build string for DHCP hashing"""
        param_str = ",".join(str(p) for p in param_list)
        return f"55:{param_str};60:{vendor};57:{max_size}"
    
    def _build_dns_string(self, client_ip: str) -> str:
        """Build string for DNS hashing"""
        queries = self.dns_cache.get(client_ip, [])
        if not queries:
            return "NONE"
        
        # Get characteristics
        edns_sizes = [q['edns'] for q in queries if q['edns'] > 0]
        avg_edns = int(statistics.mean(edns_sizes)) if edns_sizes else 0
        
        query_types = set(q['type'] for q in queries)
        type_str = ",".join(str(t) for t in sorted(query_types))
        
        return f"EDNS:{avg_edns};TYPES:{type_str};COUNT:{len(queries)}"
    
    def _identify_device_from_dhcp(self, param_list: List) -> str:
        """Identify device type from DHCP parameter list"""
        param_str = ",".join(str(p) for p in param_list)
        
        return self.DHCP_SIGNATURES.get(param_str, "Unknown")
    
    def _analyze_dns_pattern(self, client_ip: str) -> str:
        """Analyze DNS queries for malicious patterns"""
        queries = self.dns_cache.get(client_ip, [])
        if not queries:
            return "NM"
        
        # Check for DNS tunneling
        if self._detect_dns_tunneling(queries):
            return "T"
        
        # Check for DGA
        if self._detect_dga(queries):
            return "DG"
        
        # Check for C2
        if self._detect_c2_pattern(queries):
            return "BR"
        
        # Check for excessive TXT queries (potential exfiltration)
        txt_queries = [q for q in queries if q['type'] == 16]  # TXT = 16
        if len(txt_queries) / len(queries) > 0.5 and len(queries) > 10:
            return "EX"
        
        return "NM"
    
    def _detect_dns_tunneling(self, queries: List[Dict]) -> bool:
        """Detect DNS tunneling patterns"""
        if len(queries) < 10:
            return False
        
        # Check for long subdomains
        long_subdomain_count = 0
        for query in queries:
            domain = query['query']
            parts = domain.split('.')
            if len(parts) > 2:
                subdomain = parts[0]
                if len(subdomain) > self.TUNNEL_INDICATORS['long_subdomains']:
                    long_subdomain_count += 1
                    
                    # Check entropy (base64-like patterns have high entropy)
                    if self._calculate_entropy(subdomain) > self.TUNNEL_INDICATORS['high_entropy']:
                        return True
        
        # High proportion of long subdomains
        if long_subdomain_count / len(queries) > 0.3:
            return True
        
        # Check for unusual record types
        unusual_types = [q for q in queries if q['type'] in [16, 10]]  # TXT, NULL
        if len(unusual_types) / len(queries) > 0.3:
            return True
        
        return False
    
    def _detect_dga(self, queries: List[Dict]) -> bool:
        """Detect Domain Generation Algorithm patterns"""
        if len(queries) < 20:
            return False
        
        # DGA characteristics: many unique domains, high entropy, no resolution
        unique_domains = set(q['query'] for q in queries)
        
        # Check if querying many similar-looking random domains
        if len(unique_domains) > 15:
            # Calculate average entropy
            entropies = [self._calculate_entropy(d.split('.')[0]) 
                        for d in unique_domains]
            avg_entropy = statistics.mean(entropies)
            
            # DGA domains typically have high entropy
            if avg_entropy > 4.0:
                return True
        
        return False
    
    def _detect_c2_pattern(self, queries: List[Dict]) -> bool:
        """Detect C2 communication patterns"""
        if len(queries) < 5:
            return False
        
        # Beaconing: regular interval queries to same domain
        domain_queries = defaultdict(list)
        for query in queries:
            domain_queries[query['query']].append(query['timestamp'])
        
        for domain, timestamps in domain_queries.items():
            if len(timestamps) >= 5:
                # Check for regular intervals (beaconing)
                intervals = [timestamps[i+1] - timestamps[i] 
                           for i in range(len(timestamps)-1)]
                
                if len(intervals) > 1:
                    std_dev = statistics.stdev(intervals)
                    mean_interval = statistics.mean(intervals)
                    
                    # Regular intervals (low std dev relative to mean)
                    if std_dev / mean_interval < 0.2:
                        return True
        
        return False
    
    def _calculate_entropy(self, string: str) -> float:
        """Calculate Shannon entropy of a string"""
        if not string:
            return 0.0
        
        # Calculate frequency of each character
        freq = {}
        for char in string:
            freq[char] = freq.get(char, 0) + 1
        
        # Calculate entropy
        entropy = 0.0
        length = len(string)
        
        for count in freq.values():
            probability = count / length
            entropy -= probability * (probability and (probability * 
                                     (1 / probability)).bit_length() or 0)
        
        return entropy
    
    def detect_rogue_dhcp(self, packet: Packet) -> Optional[Dict]:
        """Detect rogue DHCP server"""
        if not packet.haslayer(DHCP):
            return None
        
        dhcp = packet[DHCP]
        
        # Check if this is a DHCP OFFER or ACK (server response)
        for opt in dhcp.options:
            if isinstance(opt, tuple) and opt[0] == 'message-type':
                if opt[1] in [2, 5]:  # OFFER or ACK
                    server_ip = packet[IP].src if packet.haslayer(IP) else "unknown"
                    
                    # Check against known legitimate DHCP servers
                    if not self._is_legitimate_dhcp_server(server_ip):
                        return {
                            "alert": "Rogue DHCP Server Detected",
                            "server_ip": server_ip,
                            "message_type": "OFFER" if opt[1] == 2 else "ACK",
                            "severity": "HIGH"
                        }
        
        return None
    
    def _is_legitimate_dhcp_server(self, server_ip: str) -> bool:
        """Check if DHCP server is legitimate (configure for your network)"""
        # TODO: Configure legitimate DHCP server IPs
        LEGITIMATE_SERVERS = [
            "192.168.1.1",
            "10.0.0.1"
        ]
        
        return server_ip in LEGITIMATE_SERVERS
    
    def generate_report(self) -> Dict:
        """Generate comprehensive JAD analysis report"""
        report = {
            "total_devices": len(self.dhcp_cache),
            "device_types": defaultdict(int),
            "malicious_patterns": defaultdict(int),
            "devices": []
        }
        
        for client_id, dhcp_info in self.dhcp_cache.items():
            fp = self.generate_jad_fingerprint(client_id)
            
            if fp:
                report["device_types"][fp.device_type] += 1
                report["malicious_patterns"][fp.pattern] += 1
                
                report["devices"].append({
                    "client": client_id,
                    "fingerprint": fp.raw_fingerprint,
                    "device_type": fp.device_type,
                    "pattern": fp.pattern,
                    "malicious": fp.pattern not in ["NM"],
                    "confidence": fp.confidence,
                    "hostname": dhcp_info.get("hostname", "N/A")
                })
        
        return report
 
# Example Usage
if __name__ == "__main__":
    fingerprinter = JADFingerprinter()
    
    def packet_handler(packet):
        # Analyze DHCP
        if packet.haslayer(DHCP):
            dhcp_result = fingerprinter.analyze_dhcp(packet)
            if dhcp_result:
                print(f"\n[+] DHCP Activity:")
                print(f"    Client: {dhcp_result['client_mac']}")
                print(f"    Device: {dhcp_result['device_type']}")
                print(f"    Hostname: {dhcp_result['hostname']}")
                print(f"    DHCP Hash: {dhcp_result['dhcp_hash']}")
            
            # Check for rogue DHCP
            rogue = fingerprinter.detect_rogue_dhcp(packet)
            if rogue:
                print(f"\n[!] {rogue['alert']}")
                print(f"    Server: {rogue['server_ip']}")
        
        # Analyze DNS
        if packet.haslayer(DNS) and packet.haslayer(IP):
            client_ip = packet[IP].src
            dns_result = fingerprinter.analyze_dns(packet, client_ip)
            
            if dns_result and dns_result.get('malicious'):
                print(f"\n[!] MALICIOUS DNS ACTIVITY:")
                print(f"    Client: {dns_result['client_ip']}")
                print(f"    Query: {dns_result['query']}")
                print(f"    Pattern: {dns_result['pattern']}")
    
    print("[*] Starting JAD monitoring...")
    sniff(prn=packet_handler, store=0)

class IoTDiscovery:
    """Discover and classify IoT devices"""
    
    IOT_DHCP_SIGNATURES = {
        "1,3,6,12,15,28,42": "Generic Android IoT",
        "1,3,6,15": "Simple IoT Device",
        "1,3,6,12,15": "Smart TV",
    }
    
    def __init__(self):
        self.fingerprinter = JADFingerprinter()
        self.iot_devices = {}
    
    def classify_device(self, packet: Packet) -> Optional[str]:
        """Classify IoT device from DHCP"""
        result = self.fingerprinter.analyze_dhcp(packet)
        if result:
            device_type = result['device_type']
            if device_type in self.IOT_DHCP_SIGNATURES.values():
                return device_type
        return None

class DNSTunnelDetector:
    """Real-time DNS tunneling detection"""
    
    def __init__(self, threshold=0.7):
        self.fingerprinter = JADFingerprinter()
        self.threshold = threshold
        self.alerts = []
    
    def monitor_client(self, client_ip: str):
        """Monitor specific client for tunneling"""
        queries = self.fingerprinter.dns_cache.get(client_ip, [])
        
        if len(queries) >= 10:
            pattern = self.fingerprinter._analyze_dns_pattern(client_ip)
            
            if pattern == "T":
                self.alerts.append({
                    "client": client_ip,
                    "alert": "DNS Tunneling Detected",
                    "query_count": len(queries),
                    "queries": [q['query'] for q in queries[-5:]]
                })
                return True
        
        return False

class RogueDHCPDetector:
    """Detect unauthorized DHCP servers"""
    
    def __init__(self, legitimate_servers: List[str]):
        self.legitimate_servers = set(legitimate_servers)
        self.detected_servers = set()
    
    def check_dhcp_response(self, packet: Packet) -> Optional[Dict]:
        """Check if DHCP response is from authorized server"""
        if packet.haslayer(DHCP) and packet.haslayer(IP):
            dhcp = packet[DHCP]
            
            for opt in dhcp.options:
                if isinstance(opt, tuple) and opt[0] == 'message-type':
                    if opt[1] in [2, 5]:  # OFFER or ACK
                        server_ip = packet[IP].src
                        
                        if server_ip not in self.legitimate_servers:
                            if server_ip not in self.detected_servers:
                                self.detected_servers.add(server_ip)
                                return {
                                    "alert": "ROGUE DHCP SERVER",
                                    "server_ip": server_ip,
                                    "server_mac": packet[Ether].src if packet.haslayer(Ether) else "N/A",
                                    "action_required": "Investigate and isolate"
                                }
        return None

# jad.zeek
module JAD;

export {
    redef enum Log::ID += { DHCP_LOG, DNS_LOG };
    
    type DhcpInfo: record {
        ts: time &log;
        client_mac: string &log;
        hostname: string &log;
        vendor: string &log;
        jad_hash: string &log;
        device_type: string &log;
    };
    
    redef enum Notice::Type += {
        Rogue_DHCP_Server,
        DNS_Tunneling_Detected,
        DGA_Activity
    };
}

event dhcp_discover(c: connection, msg: dhcp_msg) {
    # Extract DHCP options
    # Generate JAD DHCP hash
    # Log to dhcp.log
}

event dns_request(c: connection, msg: dns_msg) {
    # Analyze DNS query
    # Check for tunneling patterns
    # Generate alerts
}

# suricata-jad.rules
alert dhcp any any -> any any (msg:"Rogue DHCP Server Detected"; \
    dhcp.type:offer; \
    classtype:network-scan; sid:4000001; rev:1;)
 
alert dns any any -> any any (msg:"Possible DNS Tunneling - Long Subdomain"; \
    dns.query; content:"."; pcre:"/^[a-zA-Z0-9]{50,}\./"; \
    classtype:trojan-activity; sid:4000002; rev:1;)
 
alert dns any any -> any any (msg:"Excessive TXT Queries - Possible Exfiltration"; \
    dns.query; dns.query.type:TXT; \
    threshold:type threshold, track by_src, count 20, seconds 60; \
    classtype:trojan-activity; sid:4000003; rev:1;)

# Detect DNS tunneling
index=network sourcetype=dns
| eval subdomain=mvindex(split(query,"."),0)
| eval subdomain_len=len(subdomain)
| where subdomain_len > 50
| stats count by src_ip, query
| where count > 10
| table src_ip, query, count

def build_device_baseline():
    """Build baseline of expected devices"""
    baseline = {
        "known_devices": {},
        "allowed_types": ["Windows 10/11", "macOS", "iOS", "Linux"],
        "dhcp_servers": ["192.168.1.1"]
    }
    return baseline

WHITELIST_DOMAINS = [
    "update.microsoft.com",
    "apple.com",
    "google.com"
]

# Windows 10
DHCP:c3a7b8d4e5f6_DNS:7e8f9a0b1c2d_PATTERN:NM

# macOS
DHCP:9a0b1c2d3e4f_DNS:5a6b7c8d9e0f_PATTERN:NM

# iOS
DHCP:a1b2c3d4e5f6_DNS:6b7c8d9e0f1a_PATTERN:NM

# DNS Tunneling
DHCP:c3a7b8d4e5f6_DNS:a1a2a3a4a5a6_PATTERN:T

# DGA Malware
DHCP:c3a7b8d4e5f6_DNS:b2b3b4b5b6b7_PATTERN:DG

# Botnet C2
DHCP:c3a7b8d4e5f6_DNS:c3c4c5c6c7c8_PATTERN:BR