Análise de Segurança do Protocolo S7comm

Quando Usar

• Ao avaliar a postura de segurança de ambientes Siemens SIMATIC S7 PLC
• Ao construir regras de detecção para ataques baseados em S7comm contra controladores S7-300/400/1200/1500
• Ao realizar uma auditoria de segurança das comunicações Siemens Step 7/TIA Portal
• Ao investigar acesso não autorizado suspeito a programas Siemens PLC
• Ao avaliar mecanismos de integridade do S7CommPlus e seu potencial de bypass

Não use para escanear PLCs Siemens em produção sem autorização e um plano de testes (isso pode travar os controladores), para análise de protocolos não Siemens (veja detecting-modbus-command-injection-attacks para Modbus), ou para modificar programas PLC em um ambiente de produção.

Pré-requisitos

• Acesso à rede ao segmento de comunicação S7comm (porta TCP 102)
• Wireshark com dissector S7comm ou Zeek com analisador de protocolo S7comm
• Acesso autorizado para testes de segurança (nunca escaneie PLCs em produção sem autorização)
• Conhecimento dos modelos Siemens PLC e versões de firmware em escopo
• Compreensão da estrutura do protocolo S7comm (COTP, S7 PDU, códigos de função)

Fluxo de Trabalho

Etapa 1: Analisar Tráfego S7comm e Identificar Vulnerabilidades

#!/usr/bin/env python3
"""Analisador de Segurança do Protocolo S7comm.

Analisa o tráfego do protocolo Siemens S7comm para identificar
vulnerabilidades de segurança, padrões de acesso não autorizado e
indicadores potenciais de ataque contra SIMATIC S7 PLCs.
"""

import struct
import sys
import json
from collections import defaultdict
from datetime import datetime
from typing import Dict, List, Optional

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


# Definições S7comm ROSCTR (tipo PDU)
S7_ROSCTR = {
    0x01: "Job (Requisição)",
    0x02: "Ack",
    0x03: "Ack_Data (Resposta)",
    0x07: "Userdata",
}

# Códigos de função S7comm
S7_FUNCTIONS = {
    0x00: "Serviços da CPU",
    0x04: "Ler Variável",
    0x05: "Escrever Variável",
    0x1A: "Solicitar Download (Programa)",
    0x1B: "Baixar Bloco",
    0x1C: "Download Finalizado",
    0x1D: "Iniciar Upload (Ler Programa)",
    0x1E: "Carregar Bloco",
    0x1F: "Upload Finalizado",
    0x28: "Serviço PI (Iniciar/Parar CPU)",
    0x29: "Parar PLC",
    0xF0: "Configurar Comunicação",
}

# Operações críticas relevantes para segurança
CRITICAL_FUNCTIONS = {0x1A, 0x1B, 0x1C, 0x28, 0x29, 0x05}
PROGRAM_FUNCTIONS = {0x1A, 0x1B, 0x1C, 0x1D, 0x1E, 0x1F}


class S7commSecurityFinding:
    """Representa uma descoberta de segurança no tráfego S7comm."""

    def __init__(self, severity: str, finding_type: str, src_ip: str,
                 dst_ip: str, function: str, description: str,
                 cve: str = "", recommendation: str = ""):
        self.timestamp = datetime.now().isoformat()
        self.severity = severity
        self.finding_type = finding_type
        self.src_ip = src_ip
        self.dst_ip = dst_ip
        self.function = function
        self.description = description
        self.cve = cve
        self.recommendation = recommendation


class S7commAnalyzer:
    """Analisa o tráfego do protocolo S7comm em busca de vulnerabilidades de segurança."""

    def __init__(self):
        self.findings: List[S7commSecurityFinding] = []
        self.sessions: Dict[str, dict] = defaultdict(lambda: {
            "packets": 0,
            "functions_seen": set(),
            "writes": 0,
            "program_downloads": 0,
            "cpu_commands": 0,
            "first_seen": None,
            "last_seen": None,
        })
        self.authorized_engineering: set = set()
        self.packet_count = 0

    def set_authorized_stations(self, ips: List[str]):
        """Define a lista de IPs de estações de engenharia autorizadas."""
        self.authorized_engineering = set(ips)

    def parse_s7comm(self, payload: bytes) -> Optional[dict]:
        """Analisa dados do protocolo S7comm a partir do payload TCP."""
        # Cabeçalho TPKT: versão(1) + reservado(1) + comprimento(2)
        if len(payload) < 4:
            return None

        tpkt_version = payload[0]
        if tpkt_version != 3:
            return None

        tpkt_length = struct.unpack(">H", payload[2:4])[0]

        # Cabeçalho COTP segue TPKT
        if len(payload) < 7:
            return None

        cotp_length = payload[4]
        cotp_type = payload[5]

        # S7comm começa após COTP
        s7_offset = 4 + 1 + cotp_length
        if len(payload) < s7_offset + 10:
            return None

        # Cabeçalho S7comm

…