performing-s7comm-protocol-security-analysis

name	performing-s7comm-protocol-security-analysis
description	对西门子SIMATIC S7 PLC使用的S7comm和S7CommPlus协议进行安全分析，识别漏洞，包括重放攻击、完整性绕过、未授权的CPU停止命令以及针对S7-300、S7-400、S7-1200和S7-1500控制器弱点的程序下载操控。
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

执行S7comm协议安全分析

适用场景

• 评估西门子SIMATIC S7 PLC环境的安全态势时
• 为针对S7-300/400/1200/1500控制器的S7comm攻击构建检测规则时
• 对西门子Step 7/TIA Portal通信进行安全审计时
• 调查疑似对西门子PLC程序的未授权访问时
• 评估S7CommPlus完整性机制及其绕过可能性时

不适用于未经授权和测试计划扫描生产西门子PLC（可能导致控制器崩溃）、非西门子协议分析（Modbus请参见detecting-modbus-command-injection-attacks），或在生产环境中修改PLC程序。

前置条件

• 对S7comm通信网段（TCP端口102）的网络访问
• 带有S7comm解析器的Wireshark或带有S7comm协议分析器的Zeek
• 安全测试的授权访问（切勿在没有授权的情况下扫描生产PLC）
• 了解范围内的西门子PLC型号和固件版本
• 理解S7comm协议结构（COTP、S7 PDU、功能码）

工作流程

步骤 1：分析S7comm流量并识别漏洞

#!/usr/bin/env python3
"""S7comm协议安全分析器。

分析西门子S7comm协议流量，识别安全
漏洞、未授权访问模式以及针对SIMATIC S7 PLC的
潜在攻击指示符。
"""

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("请安装scapy: pip install scapy")
    sys.exit(1)


# S7comm ROSCTR（PDU类型）定义
S7_ROSCTR = {
    0x01: "Job（请求）",
    0x02: "Ack",
    0x03: "Ack_Data（响应）",
    0x07: "Userdata",
}

# S7comm功能码
S7_FUNCTIONS = {
    0x00: "CPU服务",
    0x04: "读取变量",
    0x05: "写入变量",
    0x1A: "请求下载（程序）",
    0x1B: "下载块",
    0x1C: "下载结束",
    0x1D: "开始上传（读取程序）",
    0x1E: "上传块",
    0x1F: "上传结束",
    0x28: "PI服务（启动/停止CPU）",
    : ,
    : ,
}


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()
    :
        ()
        ()

核心概念

术语	定义
S7comm	西门子专有协议，用于通过TCP端口102与SIMATIC S7 PLC通信，基于COTP/TPKT协议栈
S7CommPlus	S7comm的增强版本，用于S7-1200/1500，具有完整性保护机制
ROSCTR	S7comm报头中的远程操作服务控制字段，指示PDU类型（Job、Ack、Ack_Data、Userdata）
TIA Portal	全集成自动化门户（Totally Integrated Automation Portal）——西门子用于编程S7 PLC的工程软件
CPU停止（0x29）	停止PLC程序执行的S7comm功能，是一种关键的拒绝服务操作
程序下载（0x1A）	启动向PLC传输新控制逻辑的S7comm功能，代表最高风险操作

常见场景

场景：未授权PLC程序修改

背景：Dragos传感器对来自非授权TIA Portal工程师工作站IP地址的S7comm程序下载流量发出警报。

方法：

1. 捕获完整的S7comm会话进行取证分析
2. 识别源主机并确定其是否被攻陷或为恶意设备
3. 将当前PLC程序与最后一个已知良好备份进行对比
4. 检查PLC CPU模式是否发生变化（运行→停止→编程）
5. 如果程序被修改，从已验证的备份中恢复
6. 调查攻击链——攻击者如何到达S7comm网段
7. 在所有PLC上实施S7comm访问保护（知识保护、访问密码）

注意事项：S7-300/400 PLC没有密码学完整性保护——任何能够到达TCP端口102的设备都可以发送命令。不要仅依赖PLC密码，因为它们在S7comm中以明文传输（非S7CommPlus）。网络分段是主要防御手段。

输出格式

S7COMM安全分析报告
===================================
日期: YYYY-MM-DD
范围: [已分析的网段]

会话清单:
  工程师工作站: [数量和IP]
  通信中的PLC: [数量和IP]
  未授权源: [数量]

关键发现:
  CPU停止命令: [数量]
  程序下载: [来自未授权源的数量]
  重放攻击可能性: [评估]

漏洞评估:
  S7-300/400（无完整性保护）: [受影响PLC数量]
  S7-1200/1500（S7CommPlus）: [固件评估]
  适用的已知CVE: [列表]

建议:
  1. [最高优先级修复措施]
  2. [网络分段改进]
  3. [监控增强]