detecting-modbus-protocol-anomalies

name	detecting-modbus-protocol-anomalies
description	本技能涵盖检测工业控制系统中Modbus/TCP和Modbus RTU通信中的异常。内容涉及功能码监控、寄存器范围验证、时序分析、未授权客户端检测以及针对格式错误Modbus帧的深度包检测。该技能利用带Modbus协议分析器的Zeek、带OT规则的Suricata IDS以及使用马尔可夫链模型分析正常Modbus事务序列的自定义Python检测。
domain	cybersecurity
subdomain	ot-ics-security
tags	["ot-security","ics","scada","industrial-control","iec62443","modbus","protocol-anomaly"]
version	1.0.0
author	mahipal
license	Apache-2.0

检测Modbus协议异常

适用场景

• 在OT环境中部署Modbus专用入侵检测
• 为确定性Modbus轮询模式构建基线模型
• 调查OT监控工具标记的可疑Modbus流量
• 在工业防火墙上实现功能码白名单
• 检测可能操控过程设定点的未授权Modbus写入命令

不适用于端到端保护Modbus通信（Modbus没有原生安全性；防火墙控制参见implementing-network-segmentation-for-ot）、非Modbus协议监控（多协议参见detecting-anomalies-in-industrial-control-systems），或对Modbus实现进行主动模糊测试（参见performing-plc-firmware-security-analysis）。

前置条件

• 监控Modbus/TCP流量的网络SPAN/TAP访问（端口502）
• 带Modbus协议分析器的Zeek（原Bro）或带OT规则集的Suricata
• Python 3.9+，包含用于自定义分析的scapy和pymodbus
• 正常Modbus流量的基线捕获（最少1-2周）
• 授权Modbus客户端、功能码和寄存器映射文档

工作流程

步骤 1：捕获和解析Modbus流量

部署被动监控以捕获所有Modbus/TCP流量，并将其解析为结构化记录用于分析。

#!/usr/bin/env python3
"""Modbus Protocol Anomaly Detection System.

Monitors Modbus/TCP traffic for anomalies including unauthorized
function codes, unusual register access, timing deviations,
and rogue client devices.
"""

import json
import struct
import sys
import time
from collections import defaultdict, deque
from dataclasses import dataclass, field
from datetime import datetime
from statistics import mean, stdev

try:
    from scapy.all import sniff, rdpcap, IP, TCP
except ImportError:
    print("Install scapy: pip install scapy")
    sys.exit(1)


MODBUS_FUNCTION_CODES = {
    1: ("Read Coils", "read"),
    2: ("Read Discrete Inputs", "read"),
    3: ("Read Holding Registers", "read"),
    4: ("Read Input Registers", "read"),
    5: ("Write Single Coil", "write"),
    6: ("Write Single Register", "write"),
    7: ("Read Exception Status", "diagnostic"),
    8: ("Diagnostics", "diagnostic"),
    11: (, ),
    : (, ),
    : (, ),
    : (, ),
    : (, ),
    : (, ),
    : (, ),
    : (, ),
}



 :
    timestamp: 
    anomaly_type: 
    severity: 
    src_ip: 
    dst_ip: 
    unit_id: 
    func_code: 
    detail: 
    mitre_technique:  = 



 :
    
    src_ip: 
    dst_ip: 
    func_codes_seen:  = field(default_factory=: defaultdict())
    register_ranges:  = field(default_factory=)
    intervals:  = field(default_factory=: deque(maxlen=))
    last_timestamp:  = 
    request_count:  = 
    write_count:  = 


 :
    

     ():
        .sessions = {}
        .baseline_sessions = {}
        .anomalies = []
        .authorized_clients = ()
        .authorized_func_codes = {}  
        .packet_count = 

     ():
        
        .authorized_clients = (clients)

     ():
        
        .authorized_func_codes[session_key] = (func_codes)

     ():
        
         (baseline_file)  f:
            baseline = json.load(f)
         key, data  baseline.get(, {}).items():
            .baseline_sessions[key] = data
            .authorized_func_codes[key] = (data.get(, []))
        ()

     ():
        
          pkt.haslayer(TCP)   pkt.haslayer(IP):
            

        
         pkt[TCP].dport !=   pkt[TCP].sport != :
            

        payload = (pkt[TCP].payload)
         (payload) < :
            

        .packet_count += 
        timestamp = (pkt.time)
        ts_str = datetime.fromtimestamp(timestamp).isoformat()

        
        :
            trans_id = struct.unpack(, payload[:])[]
            proto_id = struct.unpack(, payload[:])[]
            length = struct.unpack(, payload[:])[]
            unit_id = payload[]
            func_code = payload[]
         (IndexError, struct.error):
            

        
         pkt[TCP].dport == :
            src_ip = pkt[IP].src
            dst_ip = pkt[IP].dst
            is_request = 
        :
            src_ip = pkt[IP].dst
            dst_ip = pkt[IP].src
            is_request = 

          is_request:
              

        session_key = 

        
         session_key   .sessions:
            .sessions[session_key] = ModbusSession(src_ip=src_ip, dst_ip=dst_ip)

        session = .sessions[session_key]
        session.request_count += 
        session.func_codes_seen[func_code] += 

        

        
         .authorized_clients  src_ip   .authorized_clients:
            .anomalies.append(ModbusAnomaly(
                timestamp=ts_str,
                anomaly_type=,
                severity=,
                src_ip=src_ip, dst_ip=dst_ip,
                unit_id=unit_id, func_code=func_code,
                detail=,
                mitre_technique=,
            ))

        
        allowed_fcs = .authorized_func_codes.get(session_key)
         allowed_fcs  func_code   allowed_fcs:
            fc_info = MODBUS_FUNCTION_CODES.get(func_code, (, ))
            severity =   fc_info[] ==   
            .anomalies.append(ModbusAnomaly(
                timestamp=ts_str,
                anomaly_type=,
                severity=severity,
                src_ip=src_ip, dst_ip=dst_ip,
                unit_id=unit_id, func_code=func_code,
                detail=,
                mitre_technique=,
            ))

        
         func_code  (, , , , , ):
            session.write_count += 
            fc_name = MODBUS_FUNCTION_CODES.get(func_code, (, ))[]

            
             (payload) >= :
                register_addr = struct.unpack(, payload[:])[]
                session.register_ranges.add((func_code, register_addr))

                .anomalies.append(ModbusAnomaly(
                    timestamp=ts_str,
                    anomaly_type=,
                    severity=,
                    src_ip=src_ip, dst_ip=dst_ip,
                    unit_id=unit_id, func_code=func_code,
                    detail=,
                    mitre_technique=,
                ))

        
         session.last_timestamp > :
            interval = (timestamp - session.last_timestamp) *   
            session.intervals.append(interval)

            baseline = .baseline_sessions.get(session_key)
             baseline  (session.intervals) > :
                expected_interval = baseline.get(, ) * 
                expected_std = baseline.get(, ) * 

                 expected_std > :
                    z_score = (interval - expected_interval) / expected_std
                     z_score > :
                        .anomalies.append(ModbusAnomaly(
                            timestamp=ts_str,
                            anomaly_type=,
                            severity=,
                            src_ip=src_ip, dst_ip=dst_ip,
                            unit_id=unit_id, func_code=func_code,
                            detail=(
                                
                                
                            ),
                            mitre_technique=,
                        ))

        
         proto_id != :
            .anomalies.append(ModbusAnomaly(
                timestamp=ts_str,
                anomaly_type=,
                severity=,
                src_ip=src_ip, dst_ip=dst_ip,
                unit_id=unit_id, func_code=func_code,
                detail=,
                mitre_technique=,
            ))

        
         unit_id ==   func_code  (, , , ):
            .anomalies.append(ModbusAnomaly(
                timestamp=ts_str,
                anomaly_type=,
                severity=,
                src_ip=src_ip, dst_ip=dst_ip,
                unit_id=unit_id, func_code=func_code,
                detail=,
                mitre_technique=,
            ))

        session.last_timestamp = timestamp

     ():
        
        ()
        packets = rdpcap(pcap_file)
         pkt  packets:
            .process_packet(pkt)
        ()

     ():
        
        ()
        ()
        ()
        ()
        ()
        ()

        severity_counts = defaultdict()
        type_counts = defaultdict()
         a  .anomalies:
            severity_counts[a.severity] += 
            type_counts[a.anomaly_type] += 

        ()
         sev  [, , , ]:
             severity_counts[sev]:
                ()

        ()
         atype, count  (type_counts.items(), key= x: -x[]):
            ()

        ()
         a  .anomalies[:]:
            ()


 __name__ == :
    detector = ModbusAnomalyDetector()

     (sys.argv) > :
        
         (sys.argv) > :
            detector.load_baseline(sys.argv[])
        detector.analyze_pcap(sys.argv[])
        detector.generate_report()
    :
        ()

核心概念

术语	定义
Modbus/TCP	运行在TCP端口502上的工业协议，由MBAP报头和包含功能码和数据的PDU组成
功能码（Function Code）	Modbus命令标识符（FC1-4：读取，FC5-6/15-16：写入，FC8：诊断），决定操作类型
MBAP报头（MBAP Header）	Modbus应用协议报头，包含事务ID、协议ID（0x0000）、长度和单元ID
单元ID（Unit ID）	标识目标从站设备的Modbus地址（0-247）；单元ID 0是对所有从站的广播
寄存器映射（Register Map）	Modbus寄存器地址到过程变量的供应商特定映射（如寄存器40001=反应器温度）
功能码白名单（Function Code Allowlist）	定义每个源IP到每个目标设备允许使用哪些Modbus功能码的安全策略

工具与系统

• Zeek Modbus Analyzer：带内置Modbus/TCP协议分析和日志记录的网络安全监控工具
• Suricata with ET Open ICS rules：带有Modbus命令注入和异常检测规则的IDS/IPS
• Wireshark Modbus Dissector：具有完整Modbus/TCP和Modbus RTU解码的协议分析器
• PyModbus：用于构建自定义监控和测试工具的Python Modbus库

输出格式

Modbus协议异常检测报告
==========================================
捕获周期: YYYY-MM-DD 至 YYYY-MM-DD
分析的数据包数: [N]
会话数: [N]

异常数: [N]
  UNAUTHORIZED_CLIENT: [N]
  UNAUTHORIZED_FUNCTION_CODE: [N]
  WRITE_OPERATION: [N]
  TIMING_ANOMALY: [N]
  BROADCAST_WRITE: [N]