| name | detecting-dnp3-protocol-anomalies |
| description | Detect anomalies in DNP3 communications used in SCADA/ICS systems by monitoring unauthorized control commands, firmware update attempts, protocol violations, and deviations from baseline traffic using deep packet inspection and machine learning approaches. Use when securing energy-sector or other OT/ICS networks, investigating suspicious DNP3 master/outstation activity, or building an anomaly-based IDS for industrial control traffic. |
| domain | cybersecurity |
| subdomain | ot-ics-security |
| tags | ["ot-security","ics","dnp3","scada","anomaly-detection","protocol-analysis","energy-sector","ids"] |
| version | 1.0 |
| author | mahipal |
| license | Apache-2.0 |
| atlas_techniques | ["AML.T0043","AML.T0018"] |
| nist_ai_rmf | ["MEASURE-2.7","MEASURE-2.5","MAP-5.1"] |
| nist_csf | ["PR.IR-01","DE.CM-01","ID.AM-05","GV.OC-02"] |
| mitre_attack | ["T1078","T1190","T1059","T0816","T0836"] |
Detecting DNP3 Protocol Anomalies
When to Use
- When monitoring SCADA systems in the energy sector where DNP3 is the primary protocol
- When building detection rules for DNP3-based attacks against RTUs and substations
- When investigating suspected unauthorized control commands sent via DNP3
- When deploying IDS with DNP3 deep packet inspection at utility substations
- When responding to alerts from OT monitoring platforms about DNP3 traffic anomalies
Do not use for non-DNP3 protocol monitoring (see detecting-modbus-command-injection-attacks for Modbus), for DNP3 Secure Authentication configuration (separate implementation), or for protocol-agnostic network anomaly detection.
Prerequisites
- Network TAP/SPAN on DNP3 communication segments (TCP port 20000 or serial)
- Baseline of normal DNP3 traffic patterns (masters, outstations, poll intervals, function codes)
- Suricata or Zeek with DNP3 protocol parser enabled
- Understanding of DNP3 function codes and object groups used in the environment
- DNP3 communication topology map (master-to-outstation relationships)
Workflow
Step 1: Analyze DNP3 Traffic for Anomalies
"""DNP3 Protocol Anomaly Detector.
Monitors DNP3 communications for unauthorized control commands,
protocol violations, and deviations from established baselines.
Supports both TCP and serial DNP3 deployments.
"""
import struct
import sys
import json
from collections import defaultdict
from datetime import datetime
from typing import Dict, List, Optional, Set
try:
from scapy.all import rdpcap, IP, TCP
except ImportError:
print("Install scapy: pip install scapy")
sys.exit(1)
DNP3_FUNCTIONS = {
: , : , : ,
: , : , : ,
: , : ,
: , : ,
: , : ,
: , : ,
: , : ,
: , : ,
: , : ,
: , : ,
: , : ,
: , : ,
: , : ,
: , : ,
: , : , : ,
}
DNP3_CRITICAL_FUNCTIONS = {
,
, , , ,
,
,
,
,
,
, , ,
}
:
():
.alerts = []
.sessions = defaultdict(: {
: ,
: defaultdict(),
: ,
: ,
: ,
})
.packet_count =
.dnp3_count =
.authorized_masters: [] = ()
.authorized_pairs: [, []] = defaultdict()
.baseline_functions: [, []] = defaultdict()
baseline_file:
.load_baseline(baseline_file)
():
(filepath, ) f:
baseline = json.load(f)
entry baseline.get(, []):
master = entry[]
outstation = entry[]
.authorized_masters.add(master)
.authorized_pairs[master].add(outstation)
.baseline_functions[] = (
entry.get(, [, ])
)
() -> []:
(payload) < :
start_bytes = struct.unpack(, payload[:])[]
start_bytes != :
length = payload[]
control = payload[]
dest_addr = struct.unpack(, payload[:])[]
source_addr = struct.unpack(, payload[:])[]
direction = (control & )
result = {
: length,
: control,
: direction,
: dest_addr,
: source_addr,
: (control & ),
}
(payload) >= :
transport_header = payload[]
(payload) >= :
app_control = payload[]
func_code = payload[]
result[] = func_code
result[] = DNP3_FUNCTIONS.get(
func_code,
)
result
():
.packet_count +=
pkt.haslayer(IP) pkt.haslayer(TCP):
tcp = pkt[TCP]
tcp.dport != tcp.sport != :
payload = (tcp.payload)
payload:
dnp3 = .parse_dnp3_header(payload)
dnp3:
.dnp3_count +=
src_ip = pkt[IP].src
dst_ip = pkt[IP].dst
session_key =
session = .sessions[session_key]
session[] +=
func_code = dnp3.get()
func_code :
session[][func_code] +=
dnp3.get() .authorized_masters:
src_ip .authorized_masters:
.alerts.append({
: ,
: ,
: src_ip, : dst_ip,
: dnp3.get(),
: ,
: ,
})
func_code (, ):
session[] +=
restart_type = func_code ==
.alerts.append({
: ,
: ,
: src_ip, : dst_ip,
: ,
: ,
: ,
})
func_code (, , , , , ):
session[] +=
.alerts.append({
: ,
: ,
: src_ip, : dst_ip,
: dnp3.get(),
: ,
: ,
})
func_code (, , , ):
session[] +=
session_key .baseline_functions:
func_code .baseline_functions[session_key]:
.alerts.append({
: ,
: ,
: src_ip, : dst_ip,
: dnp3.get(),
: ,
: ,
})
session_key .baseline_functions:
func_code .baseline_functions[session_key]:
func_code (, , ):
.alerts.append({
: ,
: ,
: src_ip, : dst_ip,
: dnp3.get(),
: ,
: ,
})
():
()
()
()
()
()
()
()
()
key, session .sessions.items():
()
()
funcs = [DNP3_FUNCTIONS.get(f, ) f session[]]
()
()
()
()
.alerts:
()
alert .alerts:
()
()
()
()
()
__name__ == :
detector = DNP3AnomalyDetector(
baseline_file=sys.argv[] (sys.argv) >
)
(sys.argv) >= :
()
packets = rdpcap(sys.argv[])
pkt packets:
detector.analyze_packet(pkt)
detector.generate_report()
:
()
