Detecção de Ataques de Injeção de Comandos Modbus

Quando Usar

• Ao implantar detecção de intrusão para ambientes que usam Modbus TCP (porta 502) ou Modbus RTU
• Ao investigar modificações não autorizadas suspeitas em registradores ou coils de CLPs
• Ao construir análises de detecção para monitoramento de SOC de OT em ambientes com uso intensivo de Modbus
• Ao responder a ataques do tipo FrostyGoop que utilizam Modbus TCP para impacto operacional
• Ao realizar validação de linha de base após uma suspeita de comprometimento de um mestre Modbus

Não use para detectar ataques em protocolos não Modbus (veja detecting-dnp3-protocol-anomalies para DNP3), para detecção geral de intrusão em rede de TI, ou para configuração de dispositivos Modbus (veja performing-ot-vulnerability-scanning-safely).

Pré-requisitos

• SPAN/TAP de rede no segmento que carrega o tráfego Modbus TCP (tipicamente porta 502)
• Linha de base de padrões normais de comunicação Modbus (mestres, escravos, códigos de função, faixas de registradores, intervalos de polling)
• Suricata, Zeek ou IDS de OT comercial implantado com parsers de protocolo Modbus ativados
• Compreensão dos códigos de função Modbus usados no ambiente (operações de leitura vs. escrita)
• Acesso à documentação de programação do CLP para validar as faixas de registradores esperadas

Fluxo de Trabalho

Etapa 1: Construir a Linha de Base da Comunicação Modbus

Capture e analise o tráfego Modbus normal para estabelecer o que constitui padrões de comunicação legítimos.

#!/usr/bin/env python3
"""Detector de Injeção Modbus.

Monitora o tráfego Modbus TCP em busca de operações de escrita não autorizadas,
códigos de função anômalos e desvios de linhas de base de comunicação estabelecidas.
Detecta ataques como FrostyGoop que usam Modbus TCP para impacto operacional.
"""

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 import sniff, IP, TCP
except ImportError:
    print("Instale scapy: pip install scapy")
    sys.exit(1)


# Definições de códigos de função Modbus
MODBUS_READ_FUNCTIONS = {1, 2, 3, 4}
MODBUS_WRITE_FUNCTIONS = {5, 6, 15, 16}
MODBUS_DIAGNOSTIC_FUNCTIONS = {8, 17, 43}

MODBUS_FUNC_NAMES = {
    1: "Read Coils", 2: "Read Discrete Inputs",
    3: "Read Holding Registers", 4: "Read Input Registers",
    5: "Write Single Coil", 6: "Write Single Register",
    8: "Diagnostics", 15: "Write Multiple Coils",
    16: "Write Multiple Registers", 17: "Report Slave ID",
    22: "Mask Write Register", 23: "Read/Write Multiple Registers",
    43: "Encapsulated Interface Transport",
}


class ModbusAlert:
    """Representa uma anomalia Modbus detectada."""

    def __init__(self, severity: str, alert_type: str, src_ip: str,
                 dst_ip: str, unit_id: int, func_code: int,
                 description: str, mitre_technique: str = ""):
        self.timestamp = datetime.now().isoformat()
        self.severity = severity
        self.alert_type = alert_type
        self.src_ip = src_ip
        self.dst_ip = dst_ip
        self.unit_id = unit_id
        self.func_code = func_code
        self.func_name = MODBUS_FUNC_NAMES.get(func_code, f"FC Desconhecido {func_code}")
        self.description = description
        self.mitre_technique = mitre_technique

    def __str__(self):
        return (
            f"[{self.severity}] {self.alert_type} | {self.src_ip} -> {self.dst_ip} "
            f"| Unidade {self.unit_id} | {self.func_name} | {self.description}"
        )


class ModbusInjectionDetector:
    """Detecta ataques de injeção de comandos Modbus."""

    def __init__(self, baseline_file: Optional[str] = None):
        self.alerts: List[ModbusAlert] = []
        self.packet_count = 0
        self.modbus_count = 0

        # Dados da linha de base
        self.authorized_masters: Set[str] = set()
        self.authorized_pairs: Set[Tuple[str, str]] = set()
        self.allowed_write_sources: Set[str] = set()
        self.allowed_function_codes: Dict[str, Set[int]] = defaultdict(set)
        self.allowed_register_ranges: Dict[str, List[Tuple[int, int]]] = defaultdict(list)
        self.polling_intervals: Dict[str, float] = {}
        self.last_seen: Dict[str, float] = {}

        # Contadores para detecção de taxa
        self.write_counts: Dict[str, List[float]] = defaultdict(list)

        if baseline_file:
            self.load_baseline(baseline_file)

    def load_baseline(self, filepath: str):
        """Carrega a linha de base de comunicação Modbus estabelecida."""
        with open(filepath, "r") as f:

…