performing-s7comm-protocol-security-analysis

name	performing-s7comm-protocol-security-analysis
description	Perform security analysis of Siemens S7comm and S7CommPlus protocols used by SIMATIC S7 PLCs to identify vulnerabilities including replay attacks, integrity bypass, unauthorized CPU stop commands, and program download manipulation exploiting weaknesses in S7-300, S7-400, S7-1200, and S7-1500 controllers.
domain	cybersecurity
subdomain	ot-ics-security
tags	["ot-security","ics","s7comm","siemens","plc-security","protocol-analysis","scada","vulnerability-assessment"]
version	1.0
author	mahipal
license	Apache-2.0
nist_csf	["PR.IR-01","DE.CM-01","ID.AM-05","GV.OC-02"]

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 = {
    : ,
    : ,
    : ,
    : ,
}


S7_FUNCTIONS = {
    : ,
    : ,
    : ,
    : ,
    : ,
    : ,
    : ,
    : ,
    : ,
    : ,
    : ,
    : ,
}


CRITICAL_FUNCTIONS = {, , , , , }
PROGRAM_FUNCTIONS = {, , , , , }


 :
    

     ():
        .timestamp = datetime.now().isoformat()
        .severity = severity
        .finding_type = finding_type
        .src_ip = src_ip
        .dst_ip = dst_ip
        .function = function
        .description = description
        .cve = cve
        .recommendation = recommendation


 :
    

     ():
        .findings: [S7commSecurityFinding] = []
        .sessions: [, ] = defaultdict(: {
            : ,
            : (),
            : ,
            : ,
            : ,
            : ,
            : ,
        })
        .authorized_engineering:  = ()
        .packet_count = 

     ():
        
        .authorized_engineering = (ips)

     () -> []:
        
        
         (payload) < :
             

        tpkt_version = payload[]
         tpkt_version != :
             

        tpkt_length = struct.unpack(, payload[:])[]

        
         (payload) < :
             

        cotp_length = payload[]
        cotp_type = payload[]

        
        s7_offset =  +  + cotp_length
         (payload) < s7_offset + :
             

        
        protocol_id = payload[s7_offset]
         protocol_id != :  
             

        rosctr = payload[s7_offset + ]
        redundancy = struct.unpack(, payload[s7_offset + :s7_offset + ])[]
        pdu_ref = struct.unpack(, payload[s7_offset + :s7_offset + ])[]
        param_length = struct.unpack(, payload[s7_offset + :s7_offset + ])[]
        data_length = struct.unpack(, payload[s7_offset + :s7_offset + ])[]

        result = {
            : rosctr,
            : S7_ROSCTR.get(rosctr, ),
            : pdu_ref,
            : param_length,
            : data_length,
        }

        
        param_offset = s7_offset + 
         rosctr  (, )  param_length >   (payload) > param_offset:
            func_code = payload[param_offset]
            result[] = func_code
            result[] = S7_FUNCTIONS.get(func_code, )

         result

     ():
        
        .packet_count += 

          pkt.haslayer(IP)   pkt.haslayer(TCP):
            

        tcp = pkt[TCP]
         tcp.dport !=   tcp.sport != :
            

        payload = (tcp.payload)
          payload:
            

        s7 = .parse_s7comm(payload)
          s7:
            

        src_ip = pkt[IP].src
        dst_ip = pkt[IP].dst
        session_key = 
        session = .sessions[session_key]
        session[] += 

         session[]  :
            session[] = (pkt.time)
        session[] = (pkt.time)

        func_code = s7.get()
         func_code   :
            session[].add(func_code)

            
             tcp.dport ==   func_code  CRITICAL_FUNCTIONS:
                 .authorized_engineering  src_ip   .authorized_engineering:
                    .findings.append(S7commSecurityFinding(
                        severity=,
                        finding_type=,
                        src_ip=src_ip, dst_ip=dst_ip,
                        function=s7.get(, ),
                        description=(
                            
                            
                            
                        ),
                        recommendation=,
                    ))

            
             func_code == :
                session[] += 
                .findings.append(S7commSecurityFinding(
                    severity=,
                    finding_type=,
                    src_ip=src_ip, dst_ip=dst_ip,
                    function=,
                    description=,
                    cve=,
                    recommendation=,
                ))

            
             func_code  (, , ):
                session[] += 
                .findings.append(S7commSecurityFinding(
                    severity=,
                    finding_type=,
                    src_ip=src_ip, dst_ip=dst_ip,
                    function=s7.get(, ),
                    description=(
                        
                        
                    ),
                    cve=,
                    recommendation=,
                ))

            
             func_code == :
                session[] += 

            
             func_code  (, , ):
                .findings.append(S7commSecurityFinding(
                    severity=,
                    finding_type=,
                    src_ip=src_ip, dst_ip=dst_ip,
                    function=s7.get(, ),
                    description=,
                    recommendation=,
                ))

     ():
        
        vuln_checks = [
            {
                : ,
                : ,
                : ,
                : ,
                : ,
            },
            {
                : ,
                : ,
                : ,
                : ,
                : ,
            },
            {
                : ,
                : ,
                : ,
                : ,
                : ,
            },
            {
                : ,
                : ,
                : ,
                : ,
                : ,
            },
        ]
         vuln_checks

     ():
        
        ()
        ()
        ()
        ()
        ()
        ()
        ()

        ()
         key, session  .sessions.items():
            funcs = [S7_FUNCTIONS.get(f, )  f  session[]]
            ()
            ()
            ()
            ()
            ()
            ()

         .findings:
            ()
             f  .findings:
                ()
                ()
                ()
                ()
                 f.cve:
                    ()
                 f.recommendation:
                    ()

        ()
         vuln  .check_known_vulnerabilities():
            ()
            ()
            ()
            ()


 __name__ == :
    analyzer = S7commAnalyzer()
    analyzer.set_authorized_stations([, ])

     (sys.argv) >= :
        ()
        packets = rdpcap(sys.argv[])
         pkt  packets:
            analyzer.analyze_packet(pkt)
        analyzer.generate_report()
    :
        ()
        ()