| name | detecting-dnp3-protocol-anomalies |
| description | Detect anomalies in DNP3 (Distributed Network Protocol 3) communications used in SCADA systems by monitoring for unauthorized control commands, firmware update attempts, protocol violations, and deviations from baseline traffic patterns using deep packet inspection and machine learning approaches.
|
| 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"] |
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.
Detection Gaps & Validation
- Serial DNP3 is invisible to a TCP-only sensor. Many substations run DNP3 over serial/radio to the RTU; a sensor watching only TCP/20000 misses outstation traffic entirely. Tap the serial-to-IP gateway or document the coverage gap explicitly.
- No auth means valid-looking control commands evade rules. Without DNP3 Secure Authentication (SAv5), a spoofed master can issue Select/Operate (FC 0x03/0x04), Cold/Warm Restart (0x0D/0x0E), or file-transfer/firmware objects (0x19-0x1E) that pass signature checks. Baseline allowed function codes and object groups per master->outstation pair.
- Unsolicited responses are easy to overlook. Manipulated unsolicited responses (0x82) can spoof event/class data to the master; baseline which outstations send unsolicited traffic and on which classes (0,1,2,3).
- Validate safely. Replay captured DNP3 pcaps with injected restart/file-transfer frames into the detector offline. Never send a live Operate or Cold Restart to a production outstation to "test" a rule — a cold restart is a real DoS. Confirm alerts distinguish attacks from scheduled integrity polls and authorized firmware updates in the change log.
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.
"""
struct
sys
json
collections defaultdict
datetime datetime
typing , , ,
:
scapy. rdpcap, IP, TCP
ImportError:
()
sys.exit()
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()
:
()
