Detecção de Anomalias em Sistemas de Controle Industrial

Quando Usar

• Ao implantar monitoramento contínuo para ambientes de OT que não possuem detecção de intrusão
• Ao construir detecção baseada em comportamento para complementar IDS baseado em assinatura em redes de OT
• Ao estabelecer linhas de base para comunicações SCADA determinísticas para detectar desvios
• Ao integrar detecção de anomalias de aprendizado de máquina com plataformas de monitoramento de segurança de OT
• Ao investigar alertas do Nozomi Guardian ou Dragos Platform que exigem análise mais profunda

Não usar para detecção baseada em assinatura de exploits conhecidos (veja detecting-attacks-on-scada-systems), para detecção de anomalias em redes de TI sem protocolos de OT, ou como substituto para sistemas de segurança de processo (SIS).

Pré-requisitos

• Sensores de monitoramento de rede passivos em portas SPAN/TAP da rede de OT
• Mínimo de 2-4 semanas de captura de tráfego de linha de base durante operações normais
• Python 3.9+ com scikit-learn, numpy, pandas para treinamento de modelo de ML
• Acesso ao process historian para dados de correlação de processo físico
• Compreensão dos padrões operacionais normais, incluindo mudanças de turno, processos em lote e janelas de manutenção

Fluxo de Trabalho

Etapa 1: Construir Modelo de Linha de Base Multidimensional

Capturar e modelar o comportamento determinístico das comunicações de ICS em múltiplas dimensões: tempo, comportamento do protocolo e topologia de rede.

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

Constrói linhas de base multidimensionais a partir do tráfego de rede de OT e
detecta anomalias usando métodos estatísticos e de aprendizado de máquina.
Projetado para padrões de comunicação SCADA determinísticos.
"""

import json
import sys
import time
import warnings
from collections import defaultdict
from datetime import datetime, timedelta
from dataclasses import dataclass, field

import numpy as np
import pandas as pd
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler

warnings.filterwarnings("ignore")


@dataclass
class CommunicationProfile:
    """Perfil para um único par de comunicação mestre-escravo."""
    src_ip: str
    dst_ip: str
    protocol: str
    port: int
    avg_interval_ms: float = 0.0
    std_interval_ms: float = 0.0
    avg_payload_size: float = 0.0
    function_codes: dict = field(default_factory=dict)
    packets_per_minute: float = 0.0
    first_seen: str = ""
    last_seen: str = ""


class ICSAnomalyDetector:
    """Detecção de anomalias multidimensional para ambientes ICS."""

    def __init__(self):
        self.profiles = {}
        self.topology_baseline = set()
        self.timing_model = None
        self.isolation_forest = None
        self.scaler = StandardScaler()
        self.anomalies = []
        self.training_data = []

    def build_baseline_from_pcap(self, pcap_data):
        """Construir linhas de base a partir de dados de pcap analisados (lista de registros de fluxo)."""
        print("[*] Construindo linhas de base de comunicação ICS...")

        for flow in pcap_data:
            key = f"{flow['src']}->{flow['dst']}:{flow['port']}"

            if key not in self.profiles:
                self.profiles[key] = CommunicationProfile(
                    src_ip=flow["src"],
                    dst_ip=flow["dst"],
                    protocol=flow.get("protocol", "TCP"),
                    port=flow["port"],
                    first_seen=flow.get("timestamp", ""),
                )

            profile = self.profiles[key]
            profile.last_seen = flow.get("timestamp", "")

            # Rastrear códigos de função para protocolos industriais
            fc = flow.get("function_code")
            if fc is not None:
                profile.function_codes[fc] = profile.function_codes.get(fc, 0) + 1

            # Adicionar à linha de base da topologia
            self.topology_baseline.add((flow["src"], flow["dst"], flow["port"]))

        # Calcular estatísticas de intervalo
        self._calculate_timing_stats(pcap_data)

        print(f"  Pares de comunicação: {len(self.profiles)}")
        print(f"  Entradas de topologia: {len(self.topology_baseline)}")

    def _calculate_timing_stats(self, flows):
        """Calcular estatísticas de tempo de pacote por par de comunicação."""
        timestamps = defaultdict(list)
        for flow in flows:
            key = f"{flow['src']}->{flow['dst']}:{flow['port']}"
            ts = flow.get("timestamp_epoch")
            if ts:
                timestamps[key].append(ts)

        for key, ts_list in timestamps.items():
            if key in self.profiles and len(ts_list) > 1:

…