Icmp exfiltration detector

[Bruno-Paese]

Overview

This pull request introduces an analytical mechanism designed to detect ICMP tunneling and potential data exfiltration within the Maltrail framework. I would like to formally acknowledge the significant contributions of @henriqueAmbrosi, whose collaboration and technical expertise were instrumental in the development and implementation of this functionality.

Theoretical Foundation

The proposed solution strictly adheres to Maltrail's fundamental paradigm of maintaining low computational overhead. Consequently, computationally exhaustive methodologies, such as data entropy analysis, were deliberately excluded. Instead, the detection algorithms are founded upon Inter-Packet Arrival Time (IPAT) metrics to establish a reliable baseline for normative network traffic.

Furthermore, to mitigate the statistical distortion caused by transient, high-volume operational events (e.g., ping sweeps or brief connectivity checks), the baseline traffic frequency is calculated utilizing a weighted average predicated on cumulative packet counts. This ensures that persistent flows guide the baseline, while isolated, noisy flows are appropriately attenuated.

Operational Methodology

The integration extends the core _process_packet routine within the sensor.py module to systematically analyze both ICMPv4 and ICMPv6 protocols. The system employs memory-efficient data structures—specifically period and size accumulators—to monitor historical packet data without storing individual packet timestamps.

The network traffic is continuously evaluated against three distinct heuristic criteria:

1. Payload Volumetric Analysis: Identifies datagrams whose payload size exceeds a dynamically established average or a predefined absolute volumetric threshold.

2. Frequency Anomaly Detection: Monitors transmission frequencies to identify statistically significant deviations, either targeting a singular destination IP address or localized strictly within a specific source-destination coordinate pair.

3. Distributed Exfiltration Identification: Detects coordinated architectural anomalies wherein multiple distinct source IP addresses transmit suspicious volumes of ICMP traffic to a centralized destination server.

Empirical Testing Procedures

To empirically validate the implementation, it is required to execute the sensor module with administrative privileges (sudo ./.venv/bin/python sensor.py) concurrently with the web interface service (./.venv/bin/python server.py) to ensure proper pcap-ng packet capture. Ensure that these processes are actively running before initiating the following test cases.

• Test Case 1: Baseline Traffic (Negative Control)

  • Objective: Verify that standard, normative ICMP traffic does not trigger false positive alerts.

  • Execution:

    • IPv4: ping 1.1.1.24

    • IPv6: ping ipv6.google.com -6

  • Expected Result: The web interface should not display any ICMP exfiltration alerts, as the transmission frequency and payload sizes remain below the calculated heuristic thresholds.

• Test Case 2: Frequency Anomaly Validation

  • Objective: Validate the detection of high-frequency datagram transmissions indicative of potential exfiltration.

  • Prerequisites: Configure ICMP_DESTINATION_TRAFFIC_AUTO_DETECT_BASELINE_TOLERANCE = 2 and ICMP_SRC_DEST_PAIR_EXFILTRATION_DETECTION_TOLERANCE = 2 within the Maltrail configuration file to establish stringent thresholds.

  • Execution:

    • IPv4: ping 1.1.1.24 -i 0.002

    • IPv6: ping ipv6.google.com -6 -i 0.002

  • Expected Result: The web interface must generate the alerts: "ICMPv4 exfiltration by anomalous traffic to destination (suspicious)" and "ICMPv4 exfiltration by src/dst ips pair (suspicious)" (along with their respective ICMPv6 counterparts).

• Test Case 3: Payload Volumetric Validation

  • Objective: Validate the detection of abnormally large ICMP payloads.

  • Execution:

    • IPv4: ping 1.1.1.24 -s 1000

    • IPv6: ping ipv6.google.com -6 -s 1000

  • Expected Result: The web interface must generate the alerts: "ICMPv4 large package size (suspicious)" and "ICMPv6 large package size (suspicious)". Note that alerts are expected for both Echo Request and Echo Reply datagrams.

• Test Case 4: Practical ICMP Tunnel Simulation

  • Objective: Verify the heuristic triggers under realistic data exfiltration conditions utilizing specialized open-source tunneling software (icmptunnel).

  • Execution:

    • Server instantiation: sudo ./icmptunnel -s

    • Client instantiation (on a distinct node): sudo ./icmptunnel -c <SERVER_IP>

  • Expected Result: Upon tunnel activation, the web interface must concurrently generate multiple alerts, specifically: "ICMPv4 exfiltration by anomalous traffic to destination (suspicious)", "ICMPv4 exfiltration by src/dst ips pair (suspicious)", and "ICMPv4 large package size (suspicious)".

Test Fixtures

To facilitate the validation of the distributed exfiltration heuristic (multiple sources to a singular destination), a proprietary Python script leveraging the scapy library has been developed.

This testing fixture circumvents the infrastructural complexities associated with configuring Docker containers or Linux network namespaces. By initiating a native operating system raw socket, the script programmatically synthesizes and transmits pre-compiled raw packets. This approach effectively simulates 200 distinct internal hosts executing spoofed ICMP echo requests towards a designated target, allowing for the rigorous testing of distributed anomaly detection.

import socket
import time
from scapy.all import IP, ICMP, Raw

# Change this to your Maltrail sensor's IP if it's on a different machine or bridge interface
target_ip = "19.168.0.1"
payload = b"hidden_exfil_data_chunk"

# 1. Pre-craft all packets before the loop begins
print("Pre-compiling packets into raw bytes...")
raw_packets = []

# Simulating 200 different internal hosts
for host_id in range(10, 210):
    spoofed_ip = f"192.168.0.{host_id}"
    # Craft the packet using Scapy
    packet = IP(src=spoofed_ip, dst=target_ip)/ICMP(type=8)/Raw(load=payload)
    # Cast the Scapy object to raw bytes for maximum transmission speed
    raw_packets.append(bytes(packet))

# 2. Open a native OS raw socket
try:
    # 255 is the standard IANA protocol number for Raw IP
    s = socket.socket(socket.AF_INET, socket.SOCK_RAW, 255)
except PermissionError:
    print("Error: Raw sockets require root privileges. Please run with sudo.")
    exit(1)

try:
    packet_count = 0
    start_time = time.time()
    
    while True:
        for raw_packet in raw_packets:
            s.sendto(raw_packet, (target_ip, 0))
            packet_count += 1
            
        current_time = time.time()
        if current_time - start_time >= 1.0:
            print(f"Speed: {packet_count} packets/second")
            packet_count = 0
            start_time = current_time

except KeyboardInterrupt:
    print("\nStress test stopped.")
finally:
    s.close()