| name | detecting-modbus-command-injection-attacks |
| description | 通过使用ICS感知IDS和协议深度包检测，监控未授权写入操作、异常功能码、格式错误的帧以及与已建立通信基线的偏差，检测ICS环境中针对Modbus TCP/RTU协议的命令注入攻击。
|
| domain | cybersecurity |
| subdomain | ot-ics-security |
| tags | ["ot-security","ics","modbus","command-injection","protocol-analysis","ids","scada","threat-detection"] |
| version | 1.0 |
| author | mahipal |
| license | Apache-2.0 |
检测Modbus命令注入攻击
适用场景
- 为使用Modbus TCP（端口502）或Modbus RTU的环境部署入侵检测
- 调查对PLC寄存器或线圈的疑似未授权修改
- 为监控Modbus密集型环境的OT SOC构建检测分析
- 响应利用Modbus TCP造成操作影响的FrostyGoop式攻击
- 在疑似Modbus主站被入侵后进行基线验证
不适用于检测非Modbus协议的攻击（DNP3相关参见detecting-dnp3-protocol-anomalies）、一般IT网络入侵检测，或Modbus设备配置（参见performing-ot-vulnerability-scanning-safely）。
前置条件
- 承载Modbus TCP流量的网段的网络SPAN/TAP（通常为端口502）
- 正常Modbus通信模式基线（主站、从站、功能码、寄存器范围、轮询间隔）
- 部署了Modbus协议解析器的Suricata、Zeek或商业OT IDS
- 了解环境中使用的Modbus功能码（读取与写入操作）
- 访问PLC编程文档以验证预期寄存器范围
工作流程
步骤 1：构建Modbus通信基线
捕获并分析正常Modbus流量，确定合法通信模式的构成。
"""Modbus Command Injection Detector.
Monitors Modbus TCP traffic for unauthorized write operations, anomalous
function codes, and deviations from established communication baselines.
Detects attacks like FrostyGoop that use Modbus TCP for operational impact.
"""
import json
import struct
import sys
import time
from collections import defaultdict
from datetime import datetime
from typing import Dict, List, Optional, Set, Tuple
try:
from scapy.all import sniff, IP, TCP
except ImportError:
print("Install scapy: pip install scapy")
sys.exit(1)
MODBUS_READ_FUNCTIONS = {1, 2, 3, 4}
MODBUS_WRITE_FUNCTIONS = {5, 6, 15, 16}
MODBUS_DIAGNOSTIC_FUNCTIONS = {8, 17, 43}
MODBUS_FUNC_NAMES = {
1: "Read Coils", 2: "Read Discrete Inputs",
3: "Read Holding Registers", 4: "Read Input Registers",
5: "Write Single Coil", 6: ,
: , : ,
: , : ,
: , : ,
: ,
}
:
():
.timestamp = datetime.now().isoformat()
.severity = severity
.alert_type = alert_type
.src_ip = src_ip
.dst_ip = dst_ip
.unit_id = unit_id
.func_code = func_code
.func_name = MODBUS_FUNC_NAMES.get(func_code, )
.description = description
.mitre_technique = mitre_technique
():
(
)
:
():
.alerts: [ModbusAlert] = []
.packet_count =
.modbus_count =
.authorized_masters: [] = ()
.authorized_pairs: [[, ]] = ()
.allowed_write_sources: [] = ()
.allowed_function_codes: [, []] = defaultdict()
.allowed_register_ranges: [, [[, ]]] = defaultdict()
.polling_intervals: [, ] = {}
.last_seen: [, ] = {}
.write_counts: [, []] = defaultdict()
baseline_file:
.load_baseline(baseline_file)
():
(filepath, ) f:
baseline = json.load(f)
session_key, data baseline.get(, {}).items():
src, dst = session_key.split()
.authorized_pairs.add((src.strip(), dst.strip()))
.authorized_masters.add(src.strip())
fc_set = (data.get(, []))
.allowed_function_codes[session_key] = fc_set
fc_set & MODBUS_WRITE_FUNCTIONS:
.allowed_write_sources.add(src.strip())
reg_range data.get(, []):
.allowed_register_ranges[session_key].append(
(reg_range[], reg_range[])
)
data.get():
.polling_intervals[session_key] = data[]
(
)
() -> []:
(payload) < :
transaction_id = struct.unpack(, payload[:])[]
protocol_id = struct.unpack(, payload[:])[]
length = struct.unpack(, payload[:])[]
unit_id = payload[]
func_code = payload[]
protocol_id != :
result = {
: transaction_id,
: protocol_id,
: length,
: unit_id,
: func_code,
}
(payload) >= func_code (, , , , , , , ):
result[] = struct.unpack(, payload[:])[]
result[] = struct.unpack(, payload[:])[]
result
():
.packet_count +=
pkt.haslayer(IP) pkt.haslayer(TCP):
tcp = pkt[TCP]
tcp.dport != tcp.sport != :
payload = (tcp.payload)
payload:
modbus = .parse_modbus_mbap(payload)
modbus:
.modbus_count +=
src_ip = pkt[IP].src
dst_ip = pkt[IP].dst
session_key =
now = time.time()
.authorized_masters src_ip .authorized_masters:
tcp.dport == :
.alerts.append(ModbusAlert(
severity=,
alert_type=,
src_ip=src_ip, dst_ip=dst_ip,
unit_id=modbus[],
func_code=modbus[],
description=,
mitre_technique=,
))
modbus[] MODBUS_WRITE_FUNCTIONS:
.allowed_write_sources src_ip .allowed_write_sources:
.alerts.append(ModbusAlert(
severity=,
alert_type=,
src_ip=src_ip, dst_ip=dst_ip,
unit_id=modbus[],
func_code=modbus[],
description=,
mitre_technique=,
))
.write_counts[src_ip].append(now)
recent_writes = [t t .write_counts[src_ip] now - t < ]
.write_counts[src_ip] = recent_writes
(recent_writes) > :
.alerts.append(ModbusAlert(
severity=,
alert_type=,
src_ip=src_ip, dst_ip=dst_ip,
unit_id=modbus[],
func_code=modbus[],
description=,
mitre_technique=,
))
session_key .allowed_function_codes:
modbus[] .allowed_function_codes[session_key]:
.alerts.append(ModbusAlert(
severity=,
alert_type=,
src_ip=src_ip, dst_ip=dst_ip,
unit_id=modbus[],
func_code=modbus[],
description=(
),
mitre_technique=,
))
modbus[] == modbus[] MODBUS_WRITE_FUNCTIONS:
.alerts.append(ModbusAlert(
severity=,
alert_type=,
src_ip=src_ip, dst_ip=dst_ip,
unit_id=,
func_code=modbus[],
description=,
mitre_technique=,
))
modbus session_key .allowed_register_ranges:
addr = modbus[]
qty = modbus.get(, )
in_range = (
start <= addr addr + qty <= end
start, end .allowed_register_ranges[session_key]
)
in_range:
.alerts.append(ModbusAlert(
severity=,
alert_type=,
src_ip=src_ip, dst_ip=dst_ip,
unit_id=modbus[],
func_code=modbus[],
description=,
mitre_technique=,
))
modbus[] MODBUS_DIAGNOSTIC_FUNCTIONS:
.alerts.append(ModbusAlert(
severity=,
alert_type=,
src_ip=src_ip, dst_ip=dst_ip,
unit_id=modbus[],
func_code=modbus[],
description=,
mitre_technique=,
))
():
()
()
()
()
()
()
()
.alerts:
severity_counts = defaultdict()
alert .alerts:
severity_counts[alert.severity] +=
()
sev [, , , ]:
sev severity_counts:
()
()
alert .alerts:
()
()
()
()
()
()
alert.mitre_technique:
()
():
()
()
:
sniff(
iface=interface,
=,
prn=.analyze_packet,
timeout=duration duration > ,
)
KeyboardInterrupt:
.print_report()
__name__ == :
detector = ModbusInjectionDetector(
baseline_file=sys.argv[] (sys.argv) >
)
(sys.argv) >= :
sys.argv[].endswith() sys.argv[].endswith():
scapy. rdpcap
()
packets = rdpcap(sys.argv[])
pkt packets:
detector.analyze_packet(pkt)
detector.print_report()
:
detector.start_live_monitoring(sys.argv[])
:
()
()
()
