Detecção de Anomalias no Protocolo Modbus

Quando Usar

• Ao implantar detecção de intrusão específica para Modbus em um ambiente de OT (Tecnologia Operacional)
• Ao construir modelos de linha de base para padrões determinísticos de polling Modbus
• Ao investigar tráfego Modbus suspeito sinalizado por ferramentas de monitoramento de OT
• Ao implementar lista de permissões (allowlisting) de códigos de função em firewalls industriais
• Ao detectar comandos de escrita Modbus não autorizados que podem manipular setpoints de processo

Não usar para proteger comunicações Modbus de ponta a ponta (Modbus não possui segurança nativa; consulte implementing-network-segmentation-for-ot para controles baseados em firewall), para monitoramento de protocolos não Modbus (consulte detecting-anomalies-in-industrial-control-systems para multi-protocolo) ou para fuzzing ativo de implementações Modbus (consulte performing-plc-firmware-security-analysis).

Pré-requisitos

• Acesso SPAN/TAP de rede para monitorar tráfego Modbus/TCP (porta 502)
• Zeek (anteriormente Bro) com analisador de protocolo Modbus ou Suricata com conjuntos de regras de OT
• Python 3.9+ com scapy e pymodbus para análise personalizada
• Captura de linha de base de tráfego Modbus normal (mínimo de 1-2 semanas)
• Documentação de clientes Modbus autorizados, códigos de função e mapas de registradores

Fluxo de Trabalho

Etapa 1: Capturar e Analisar Tráfego Modbus

Implante monitoramento passivo para capturar todo o tráfego Modbus/TCP e analisá-lo em registros estruturados para análise.

#!/usr/bin/env python3
"""Sistema de Detecção de Anomalias no Protocolo Modbus.

Monitora o tráfego Modbus/TCP em busca de anomalias, incluindo
códigos de função não autorizados, acesso incomum a registradores,
desvios de tempo e dispositivos clientes maliciosos.
"""

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:
    print("Instale scapy: pip install scapy")
    sys.exit(1)


MODBUS_FUNCTION_CODES = {
    1: ("Read Coils", "read"),
    2: ("Read Discrete Inputs", "read"),
    3: ("Read Holding Registers", "read"),
    4: ("Read Input Registers", "read"),
    5: ("Write Single Coil", "write"),
    6: ("Write Single Register", "write"),
    7: ("Read Exception Status", "diagnostic"),
    8: ("Diagnostics", "diagnostic"),
    11: ("Get Comm Event Counter", "diagnostic"),
    12: ("Get Comm Event Log", "diagnostic"),
    15: ("Write Multiple Coils", "write"),
    16: ("Write Multiple Registers", "write"),
    17: ("Report Slave ID", "diagnostic"),
    22: ("Mask Write Register", "write"),
    23: ("Read/Write Multiple Registers", "read_write"),
    43: ("Encapsulated Interface Transport", "diagnostic"),
}


@dataclass
class ModbusAnomaly:
    timestamp: str
    anomaly_type: str
    severity: str
    src_ip: str
    dst_ip: str
    unit_id: int
    func_code: int
    detail: str
    mitre_technique: str = ""


@dataclass
class ModbusSession:
    """Rastreia o estado de uma sessão mestre-escravo Modbus."""
    src_ip: str
    dst_ip: str
    func_codes_seen: dict = field(default_factory=lambda: defaultdict(int))
    register_ranges: set = field(default_factory=set)
    intervals: list = field(default_factory=lambda: deque(maxlen=500))
    last_timestamp: float = 0
    request_count: int = 0
    write_count: int = 0


class ModbusAnomalyDetector:
    """Detecta anomalias no tráfego Modbus/TCP."""

    def __init__(self):
        self.sessions = {}
        self.baseline_sessions = {}
        self.anomalies = []
        self.authorized_clients = set()
        self.authorized_func_codes = {}  # FCs permitidos por sessão
        self.packet_count = 0

    def set_authorized_clients(self, clients):
        """Define a lista de IPs de clientes Modbus autorizados."""
        self.authorized_clients = set(clients)

    def set_authorized_func_codes(self, session_key, func_codes):
        """Define os códigos de função permitidos para uma sessão específica."""
        self.authorized_func_codes[session_key] = set(func_codes)

    def load_baseline(self, baseline_file):
        """Carrega perfis de linha de base de análises de captura anteriores."""
        with open(baseline_file) as f:
            baseline = json.load(f)
        for key, data in baseline.get("modbus_baselines", {}).items():
            self.baseline_sessions[key] = data
            self.authorized_func_codes[key] = set(data.get("allowed_function_codes", []))
        print(f"[*] Carregados {len(self.baseline_sessions)} baselines Modbus")

    def process_packet(self, pkt):

…