Performing S7comm Protocol Security Analysis

When to Use

• When assessing the security posture of Siemens SIMATIC S7 PLC environments
• When building detection rules for S7comm-based attacks against S7-300/400/1200/1500 controllers
• When performing a security audit of Siemens Step 7/TIA Portal communications
• When investigating suspected unauthorized access to Siemens PLC programs
• When evaluating S7CommPlus integrity mechanisms and their bypass potential

Do not use for scanning production Siemens PLCs without authorization and a test plan (this can crash controllers), for non-Siemens protocol analysis (see detecting-modbus-command-injection-attacks for Modbus), or for modifying PLC programs in a production environment.

Prerequisites

• Network access to the S7comm communication segment (TCP port 102)
• Wireshark with S7comm dissector or Zeek with S7comm protocol analyzer
• Authorized access for security testing (never scan production PLCs without authorization)
• Knowledge of the Siemens PLC models and firmware versions in scope
• Understanding of S7comm protocol structure (COTP, S7 PDU, function codes)

Workflow

Step 1: Analyze S7comm Traffic and Identify Vulnerabilities

#!/usr/bin/env python3
"""S7comm Protocol Security Analyzer.

Analyzes Siemens S7comm protocol traffic to identify security
vulnerabilities, unauthorized access patterns, and potential
attack indicators against 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("Install scapy: pip install scapy")
    sys.exit(1)


# S7comm ROSCTR (PDU type) definitions
S7_ROSCTR = {
    0x01: "Job (Request)",
    0x02: "Ack",
    0x03: "Ack_Data (Response)",
    0x07: "Userdata",
}

# S7comm function codes
S7_FUNCTIONS = {
    0x00: "CPU services",
    0x04: "Read Variable",
    0x05: "Write Variable",
    0x1A: "Request Download (Program)",
    0x1B: "Download Block",
    0x1C: "Download Ended",
    0x1D: "Start Upload (Read Program)",
    0x1E: "Upload Block",
    0x1F: "Upload Ended",
    0x28: "PI Service (Start/Stop CPU)",
    0x29: "PLC Stop",
    0xF0: "Setup Communication",
}

# Critical security-relevant operations
CRITICAL_FUNCTIONS = {0x1A, 0x1B, 0x1C, 0x28, 0x29, 0x05}
PROGRAM_FUNCTIONS = {0x1A, 0x1B, 0x1C, 0x1D, 0x1E, 0x1F}


class S7commSecurityFinding:
    """Represents a security finding in S7comm traffic."""

    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:
    """Analyzes S7comm protocol traffic for security vulnerabilities."""

    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]):
        """Set list of authorized engineering workstation IPs."""
        self.authorized_engineering = set(ips)

    def parse_s7comm(self, payload: bytes) -> Optional[dict]:
        """Parse S7comm protocol data from TCP payload."""
        # TPKT header: version(1) + reserved(1) + length(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]

        # COTP header follows TPKT
        if len(payload) < 7:
            return None

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

        # S7comm starts after COTP
        s7_offset = 4 + 1 + cotp_length
        if len(payload) < s7_offset + 10:
            return None

        # S7comm header

…