detecting-modbus-protocol-anomalies

name	detecting-modbus-protocol-anomalies
description	This skill covers detecting anomalies in Modbus/TCP and Modbus RTU communications in industrial control systems. It addresses function code monitoring, register range validation, timing analysis, unauthorized client detection, and deep packet inspection for malformed Modbus frames. The skill leverages Zeek with Modbus protocol analyzers, Suricata IDS with OT rules, and custom Python-based detection using Markov chain models for normal Modbus transaction sequences.
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"]

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).

Detection Gaps & Validation

• Legit-looking writes slip past sequence models. Modbus has no authentication; an attacker replaying the authorized client IP with permitted writes (FC5/FC6/FC15/FC16) inside the baseline register map produces transactions a Markov/sequence model rates as normal. Add register value and setpoint bounds, not just function-code allowlisting.
• Per-pair, per-mode baselines or you drown in false positives. Polling intervals and function-code mixes differ by master->slave pair and operating mode; a global baseline flags every shift change and batch transition. Exclude maintenance windows explicitly.
• Coverage gaps. TCP/502 SPAN monitoring misses Modbus RTU serial behind gateways; unit ID 0 broadcast writes affect all slaves simultaneously and deserve a dedicated rule.
• Validate safely. Run the detector over a recorded pcap with injected anomalies (unauthorized FC, broadcast write, non-zero protocol-ID violation, off-interval poll) offline. Never fuzz or write to a live device. Confirm each alert separates an attack from an authorized engineering change recorded in the change log.

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


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

:
     scapy.  sniff, rdpcap, IP, TCP
 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()
    :
        ()