Detecção de Ataques em Sistemas SCADA

Quando Usar

• Ao implantar recursos de detecção de intrusão em um ambiente SCADA pela primeira vez
• Ao investigar suspeitas de ciberataques contra sistemas de controle industrial
• Ao construir regras de detecção para padrões de ataque específicos de OT (Stuxnet, TRITON, Industroyer)
• Ao integrar o monitoramento de rede OT com um SOC corporativo para visibilidade unificada de ameaças
• Ao responder a alertas de ferramentas de monitoramento de segurança OT (Dragos, Nozomi, Claroty)

Não usar para detectar ataques em redes exclusivamente de TI sem componentes SCADA/ICS, para construir regras genéricas de IDS de rede (veja building-detection-rules-with-sigma) ou para procedimentos de resposta a incidentes após a confirmação de um ataque (veja performing-ot-incident-response).

Pré-requisitos

• Sensores de monitoramento passivo de rede implantados em portas SPAN/TAP nas fronteiras da rede OT
• Sistema de detecção de intrusão OT (Dragos Platform, Nozomi Guardian, Claroty xDome ou Suricata com conjuntos de regras OT)
• Compreensão dos protocolos industriais em uso (Modbus, DNP3, OPC UA, EtherNet/IP, S7comm)
• Linha de base de padrões de comunicação SCADA normais (intervalos de polling, códigos de função, faixas de registradores)
• Acesso a dados do histórico de processos para correlação de anomalias de processos físicos

Fluxo de Trabalho

Etapa 1: Estabelecer Linhas de Base de Comunicação SCADA

Antes de detectar anomalias, estabeleça como é o tráfego SCADA normal. Protocolos industriais são altamente determinísticos - o mesmo mestre consulta os mesmos escravos nos mesmos intervalos lendo os mesmos registradores.

#!/usr/bin/env python3
"""Construtor de Linha de Base de Comunicação SCADA.

Analisa o tráfego de rede OT para estabelecer linhas de base determinísticas para
comunicações Modbus/TCP, DNP3, EtherNet/IP e S7comm.
"""

import json
import sys
from collections import defaultdict
from datetime import datetime
from statistics import mean, stdev

try:
    from scapy.all import rdpcap, IP, TCP, UDP
except ImportError:
    print("Instale scapy: pip install scapy")
    sys.exit(1)

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 SCADABaselineBuilder:
    """Constrói linhas de base determinísticas a partir de capturas de tráfego SCADA."""

    def __init__(self):
        self.modbus_sessions = defaultdict(lambda: {
            "func_codes": defaultdict(int),
            "register_ranges": set(),
            "intervals": [],
            "last_seen": None,
            "request_count": 0,
        })
        self.communication_pairs = defaultdict(lambda: {
            "protocols": set(),
            "packet_count": 0,
            "first_seen": None,
            "last_seen": None,
        })

    def process_pcap(self, pcap_file):
        """Processa o arquivo pcap para construir linhas de base SCADA."""
        packets = rdpcap(pcap_file)
        print(f"[*] Processando {len(packets)} pacotes para linha de base...")

        for pkt in packets:
            if not pkt.haslayer(IP):
                continue

            src = pkt[IP].src
            dst = pkt[IP].dst
            ts = float(pkt.time)

            # Rastreia pares de comunicação
            pair_key = f"{src}->{dst}"
            pair = self.communication_pairs[pair_key]
            pair["packet_count"] += 1
            if pair["first_seen"] is None:
                pair["first_seen"] = ts
            pair["last_seen"] = ts

            # Analisa Modbus/TCP
            if pkt.haslayer(TCP) and pkt[TCP].dport == 502:
                self._analyze_modbus(pkt, src, dst, ts)

    def _analyze_modbus(self, pkt, src, dst, timestamp):
        """Extrai códigos de função Modbus e faixas de registradores."""
        payload = bytes(pkt[TCP].payload)
        if len(payload) < 8:
            return

        # Cabeçalho MBAP: transaction_id(2) + protocol_id(2) + length(2) + unit_id(1) + func_code(1)
        func_code = payload[7]
        session_key = f"{src}->{dst}"
        session = self.modbus_sessions[session_key]

        session["func_codes"][func_code] += 1
        session["request_count"] += 1
        session["protocols"] = {"Modbus/TCP"}

        # Rastreia intervalos de polling
        if session["last_seen"] is not None:

…