| name | detecting-dnp3-protocol-anomalies |
| description | 通过监控未授权控制命令、固件更新尝试、协议违规以及与基线流量模式的偏差，使用深度包检测和机器学习方法检测SCADA系统中使用的DNP3（分布式网络协议3）通信中的异常。
|
| 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 |
检测DNP3协议异常
适用场景
- 在DNP3为主要协议的能源行业监控SCADA系统
- 为针对RTU和变电站的DNP3攻击构建检测规则
- 调查通过DNP3发送的疑似未授权控制命令
- 在电力公司变电站部署具有DNP3深度包检测的IDS
- 响应OT监控平台关于DNP3流量异常的告警
不适用于非DNP3协议监控（Modbus相关参见detecting-modbus-command-injection-attacks）、DNP3安全认证配置（独立实现），或与协议无关的网络异常检测。
前置条件
- DNP3通信段的网络TAP/SPAN（TCP端口20000或串行）
- 正常DNP3流量模式基线（主站、外站、轮询间隔、功能码）
- 启用了DNP3协议解析器的Suricata或Zeek
- 了解环境中使用的DNP3功能码和对象组
- DNP3通信拓扑图（主站对外站关系）
工作流程
步骤 1：分析DNP3流量中的异常
"""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 = {
0x00: "Confirm", 0x01: "Read", 0x02: "Write",
0x03: "Select", 0x04: "Operate", 0x05: "Direct Operate",
0x06: "Direct Operate No Ack", 0x07: "Immediate Freeze",
0x08: "Immediate Freeze No Ack", 0x09: "Freeze and Clear",
0x0A: "Freeze and Clear No Ack", 0x0B: "Freeze at Time",
0x0C: , : ,
: , : ,
: , : ,
: , : ,
: , : ,
: , : ,
: , : ,
: , : ,
: , : ,
: , : , : ,
}
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()
:
()
核心概念
| 术语 | 定义 |
|---|
| DNP3 | 分布式网络协议版本3，能源行业SCADA系统中主站与外站通信的主要协议 |
| 外站（Outstation） | DNP3从站设备（通常为RTU或IED），响应主站的轮询和命令 |
| 先选后操（Select-Before-Operate） | DNP3安全机制，要求在操作前先发送选择命令，防止意外控制操作 |
| 冷重启（Cold Restart，FC 0x0D） | 完全重启外站的DNP3命令，重置所有配置——高风险的拒绝服务操作 |
| DNP3安全认证（DNP3 Secure Authentication） | 可选的DNP3扩展（SA v5），添加基于HMAC的认证以防止命令欺骗 |
| PIPEDREAM | 具有DNP3功能的ICS攻击框架，用于操控外站和执行固件更新 |
输出格式
DNP3异常检测报告
================================
分析周期: [开始] 至 [结束]
监控点: [变电站/网段]
流量摘要:
DNP3数据包: [数量]
唯一主站-外站对: [数量]
控制命令: [数量]
文件操作: [数量]
告警:
[严重] 未授权DNP3主站 [IP]
[严重] 向外站 [地址] 发送冷重启命令
[高] 来自 [IP] 的意外控制命令
建议:
1. 部署DNP3安全认证（SA v5）
2. 在防火墙阻断未授权来源
3. 在工业防火墙启用DNP3深度包检测
