detecting-modbus-protocol-anomalies

name	detecting-modbus-protocol-anomalies
description	Detect anomalies in Modbus/TCP and Modbus RTU industrial traffic via function code monitoring, register range validation, timing analysis, and deep packet inspection, using Zeek's Modbus analyzer, Suricata IDS with OT rules, and Python Markov chain models of normal transaction sequences. Use for deploying Modbus IDS in OT environments, baselining polling patterns, investigating suspicious Modbus traffic, or building function code allowlists.
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
nist_ai_rmf	["MEASURE-2.7","MAP-5.1","MANAGE-2.4"]
atlas_techniques	["AML.T0070","AML.T0066","AML.T0082"]
nist_csf	["PR.IR-01","DE.CM-01","ID.AM-05","GV.OC-02"]
mitre_attack	["T1078","T1190","T1059","T0816","T0836"]

Detecting Modbus Protocol Anomalies

When to Use

• When deploying Modbus-specific intrusion detection in an OT environment
• When building baseline models for deterministic Modbus polling patterns
• When investigating suspicious Modbus traffic flagged by OT monitoring tools
• When implementing function code allowlisting on industrial firewalls
• When detecting unauthorized Modbus write commands that could manipulate process setpoints

Do not use for securing Modbus communications end-to-end (Modbus has no native security; see implementing-network-segmentation-for-ot for firewall-based controls), for non-Modbus protocol monitoring (see detecting-anomalies-in-industrial-control-systems for multi-protocol), or for active fuzzing of Modbus implementations (see performing-plc-firmware-security-analysis).

Prerequisites

• Network SPAN/TAP access to monitor Modbus/TCP traffic (port 502)
• Zeek (formerly Bro) with Modbus protocol analyzer or Suricata with OT rulesets
• Python 3.9+ with scapy and pymodbus for custom analysis
• Baseline capture of normal Modbus traffic (minimum 1-2 weeks)
• Documentation of authorized Modbus clients, function codes, and register maps

Workflow

Step 1: Capture and Parse Modbus Traffic

Deploy passive monitoring to capture all Modbus/TCP traffic and parse it into structured records for analysis.

#!/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:
    ()
    sys.exit()


MODBUS_FUNCTION_CODES = {
    : (, ),
    : (, ),
    : (, ),
    : (, ),
    : (, ),
    : (, ),
    : (, ),
    : (, ),
    : (, ),
    : (, ),
    : (, ),
    : (, ),
    : (, ),
    : (, ),
    : (, ),
    : (, ),
}



 :
    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()
    :
        ()