detecting-modbus-command-injection-attacks

name	detecting-modbus-command-injection-attacks
description	Detect command injection against Modbus TCP/RTU in ICS/SCADA environments by monitoring unauthorized writes, anomalous function codes, malformed frames, and deviations from communication baselines using ICS-aware IDS and deep packet inspection. Use when deploying IDS for Modbus OT networks, investigating unauthorized PLC register/coil changes, or responding to FrostyGoop-style Modbus attacks.
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"]
mitre_attack	["T1078","T1190","T1059","T1055","T0816"]

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

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.

#!/usr/bin/env python3
"""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  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[])
    :
        ()
        ()
        ()