detecting-modbus-command-injection-attacks

name	detecting-modbus-command-injection-attacks
description	Detect command injection attacks against Modbus TCP/RTU protocol in ICS environments by monitoring for unauthorized write operations, anomalous function codes, malformed frames, and deviations from established communication baselines using ICS-aware IDS and protocol deep packet inspection.
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
nist_csf	["PR.IR-01","DE.CM-01","ID.AM-05","GV.OC-02"]

Detecting Modbus Command Injection Attacks

When to Use

• When deploying intrusion detection for environments using Modbus TCP (port 502) or Modbus RTU
• When investigating suspected unauthorized modifications to PLC registers or coils
• When building detection analytics for OT SOC monitoring Modbus-heavy environments
• When responding to FrostyGoop-style attacks that leverage Modbus TCP for operational impact
• When performing baseline validation after a suspected compromise of a Modbus master

Do not use for detecting attacks on non-Modbus protocols (see detecting-dnp3-protocol-anomalies for DNP3), for general IT network intrusion detection, or for Modbus device configuration (see performing-ot-vulnerability-scanning-safely).

Detection Gaps & Validation

• Source IP is the only identity, and it spoofs. Modbus/TCP has no authentication, so a malicious write (FC5/FC6/FC15/FC16) from a spoofed master IP looks legitimate. Without ARP inspection or static MAC bindings, IP-based allowlisting alone produces false negatives.
• Writes inside the learned register range evade detection. FrostyGoop-style attacks reuse permitted function codes and registers; pair the FC allowlist with register value/setpoint bounds and write-rate limits, not just function-code matching.
• Broadcast and serial gaps. A unit ID 0 broadcast write hits every slave on the segment at once and needs its own rule; and a SPAN watching only TCP/502 misses Modbus RTU serial behind a gateway.
• Validate the rule safely. Replay a labeled pcap containing injected FC16 writes, a broadcast write, and an out-of-range register access through the detector offline. Never write to a live PLC register to test detection. Before escalating, check the source against the authorized-writer list and the change/maintenance log — legitimate SCADA and engineering writes use the same function codes.

Prerequisites

• Network SPAN/TAP on the segment carrying Modbus TCP traffic (typically port 502)
• Baseline of normal Modbus communication patterns (masters, slaves, function codes, register ranges, polling intervals)
• Suricata, Zeek, or commercial OT IDS deployed with Modbus protocol parsers enabled
• Understanding of Modbus function codes used in the environment (read vs write operations)
• Access to PLC programming documentation to validate expected register ranges

Workflow

Step 1: Build Modbus Communication Baseline

Capture and analyze normal Modbus traffic to establish what constitutes legitimate communication patterns.

 json
 struct
 sys
 time
 collections  defaultdict
 datetime  datetime
 typing  , , , , 

:
     scapy.  sniff, IP, TCP
 ImportError:
    ()
    sys.exit()



MODBUS_READ_FUNCTIONS = {, , , }
MODBUS_WRITE_FUNCTIONS = {, , , }
MODBUS_DIAGNOSTIC_FUNCTIONS = {, , }

MODBUS_FUNC_NAMES = {
    : , : ,
    : , : ,
    : , : ,
    : , : ,
    : , : ,
    : , : ,
    : ,
}


 :
    

     ():
        .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[])
    :
        ()
        ()
        ()