# Performing S7comm Protocol Security Analysis > 对西门子SIMATIC S7 PLC使用的S7comm和S7CommPlus协议进行安全分析，识别漏洞，包括重放攻击、完整性绕过、未授权的CPU停止命令以及针对S7-300、S7-400、S7-1200和S7-1500控制器弱点的程序下载操控。 - Skill: `killvxk/performing-s7comm-protocol-security-analysis` (Agent Skill, multi-file: 4 files) - Install (CLI): `npx skillmds@latest add killvxk/performing-s7comm-protocol-security-analysis` - Raw SKILL.md: https://api.skillmd.com/api/skills/killvxk/performing-s7comm-protocol-security-analysis/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Security - License: Apache-2.0 - Author: killvxk (https://skillmd.com/u/killvxk) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/killvxk/performing-s7comm-protocol-security-analysis --- # 执行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流量并识别漏洞 ```python #!/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）", 0x29: "PLC停止", 0xF0: "建立通信", } # 安全相关的关键操作 CRITICAL_FUNCTIONS = {0x1A, 0x1B, 0x1C, 0x28, 0x29, 0x05} PROGRAM_FUNCTIONS = {0x1A, 0x1B, 0x1C, 0x1D, 0x1E, 0x1F} class S7commSecurityFinding: """表示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: """分析S7comm协议流量中的安全漏洞。""" 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]): """设置授权工程师工作站IP列表。""" self.authorized_engineering = set(ips) def parse_s7comm(self, payload: bytes) -> Optional[dict]: """从TCP有效载荷解析S7comm协议数据。""" # TPKT报头: 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报头紧跟TPKT之后 if len(payload) < 7: return None cotp_length = payload[4] cotp_type = payload[5] # S7comm在COTP之后开始 s7_offset = 4 + 1 + cotp_length if len(payload) < s7_offset + 10: return None # S7comm报头 protocol_id = payload[s7_offset] if protocol_id != 0x32: # S7comm魔术字节 return None rosctr = payload[s7_offset + 1] redundancy = struct.unpack(">H", payload[s7_offset + 2:s7_offset + 4])[0] pdu_ref = struct.unpack(">H", payload[s7_offset + 4:s7_offset + 6])[0] param_length = struct.unpack(">H", payload[s7_offset + 6:s7_offset + 8])[0] data_length = struct.unpack(">H", payload[s7_offset + 8:s7_offset + 10])[0] result = { "rosctr": rosctr, "rosctr_name": S7_ROSCTR.get(rosctr, f"未知 (0x{rosctr:02x})"), "pdu_ref": pdu_ref, "param_length": param_length, "data_length": data_length, } # 从参数中解析功能码 param_offset = s7_offset + 10 if rosctr in (0x01, 0x03) and param_length > 0 and len(payload) > param_offset: func_code = payload[param_offset] result["function_code"] = func_code result["function_name"] = S7_FUNCTIONS.get(func_code, f"未知 (0x{func_code:02x})") return result def analyze_packet(self, pkt): """分析数据包中的S7comm安全问题。""" self.packet_count += 1 if not pkt.haslayer(IP) or not pkt.haslayer(TCP): return tcp = pkt[TCP] if tcp.dport != 102 and tcp.sport != 102: return payload = bytes(tcp.payload) if not payload: return s7 = self.parse_s7comm(payload) if not s7: return src_ip = pkt[IP].src dst_ip = pkt[IP].dst session_key = f"{src_ip}->{dst_ip}" session = self.sessions[session_key] session["packets"] += 1 if session["first_seen"] is None: session["first_seen"] = float(pkt.time) session["last_seen"] = float(pkt.time) func_code = s7.get("function_code") if func_code is not None: session["functions_seen"].add(func_code) # 检查1：未授权工程师工作站 if tcp.dport == 102 and func_code in CRITICAL_FUNCTIONS: if self.authorized_engineering and src_ip not in self.authorized_engineering: self.findings.append(S7commSecurityFinding( severity="CRITICAL", finding_type="UNAUTHORIZED_ENGINEERING_ACCESS", src_ip=src_ip, dst_ip=dst_ip, function=s7.get("function_name", "未知"), description=( f"来自未授权源 {src_ip} 的关键S7comm操作。" f"功能: {s7.get('function_name')}。只有授权的TIA Portal " f"工作站才能发出这些命令。" ), recommendation="在工业防火墙处阻止未授权源。调查源主机是否被攻陷。", )) # 检查2：CPU停止命令 if func_code == 0x29: session["cpu_commands"] += 1 self.findings.append(S7commSecurityFinding( severity="CRITICAL", finding_type="CPU_STOP_COMMAND", src_ip=src_ip, dst_ip=dst_ip, function="PLC CPU停止 (0x29)", description=f"已向 {dst_ip} 处的PLC发送CPU停止命令。这会停止PLC程序执行。", cve="MITRE T0881 — 服务停止", recommendation="验证这是否是授权的维护操作。如果不是，立即隔离源。", )) # 检查3：程序下载 if func_code in (0x1A, 0x1B, 0x1C): session["program_downloads"] += 1 self.findings.append(S7commSecurityFinding( severity="CRITICAL", finding_type="PROGRAM_DOWNLOAD", src_ip=src_ip, dst_ip=dst_ip, function=s7.get("function_name", "下载"), description=( f"向 {dst_ip} 执行PLC程序下载操作。" f"这修改了PLC上运行的控制逻辑。" ), cve="MITRE T0843 — 程序下载", recommendation="对照变更管理记录验证。与已知良好的程序备份进行比对。", )) # 检查4：写入变量操作 if func_code == 0x05: session["writes"] += 1 # 检查5：程序上传（PLC代码泄露） if func_code in (0x1D, 0x1E, 0x1F): self.findings.append(S7commSecurityFinding( severity="HIGH", finding_type="PROGRAM_UPLOAD_EXFILTRATION", src_ip=src_ip, dst_ip=dst_ip, function=s7.get("function_name", "上传"), description=f"从 {dst_ip} 进行PLC程序上传（读取）。源 {src_ip} 正在提取PLC控制逻辑。", recommendation="验证这是否是授权的维护操作。未授权的上传表示侦察行为。", )) def check_known_vulnerabilities(self): """基于观察到的行为检查已知的西门子S7漏洞。""" vuln_checks = [ { "name": "S7-300/400重放攻击漏洞", "cve": "CVE-2019-13945", "description": "S7-300/400 PLC缺乏对S7comm会话的完整性检查，允许重放攻击", "affected": "S7-300、S7-400（所有固件版本）", "severity": "HIGH", }, { "name": "S7CommPlus完整性绕过", "cve": "研究发现（Biham等人）", "description": "S7CommPlusV3完整性机制可被能够观察到一个合法会话的攻击者绕过", "affected": "S7-1200（< V4.5）、S7-1500（< V2.9）", "severity": "HIGH", }, { "name": "不可修补的硬件信任根", "cve": "CVE-2022-38773", "description": "硬件漏洞允许绕过受保护的引导程序并持久修改固件", "affected": "S7-1500（特定硬件版本）", "severity": "CRITICAL", }, { "name": "通过端口102的远程DoS", "cve": "CVE-2019-10929", "description": "TCP端口102上的特制数据包可以远程崩溃S7 PLC", "affected": "S7-300、S7-400、S7-1200、S7-1500（特定固件）", "severity": "HIGH", }, ] return vuln_checks def generate_report(self): """生成综合S7comm安全分析报告。""" print(f"\n{'='*70}") print("S7COMM协议安全分析报告") print(f"{'='*70}") print(f"分析时间: {datetime.now().isoformat()}") print(f"已分析数据包: {self.packet_count}") print(f"S7comm会话: {len(self.sessions)}") print(f"安全发现: {len(self.findings)}") print(f"\n--- 会话摘要 ---") for key, session in self.sessions.items(): funcs = [S7_FUNCTIONS.get(f, f"0x{f:02x}") for f in session["functions_seen"]] print(f"\n {key}") print(f" 数据包: {session['packets']}") print(f" 功能: {', '.join(funcs)}") print(f" 写入次数: {session['writes']}") print(f" 程序下载次数: {session['program_downloads']}") print(f" CPU命令次数: {session['cpu_commands']}") if self.findings: print(f"\n--- 安全发现 ---") for f in self.findings: print(f"\n [{f.severity}] {f.finding_type}") print(f" 源: {f.src_ip} -> {f.dst_ip}") print(f" 功能: {f.function}") print(f" 详情: {f.description}") if f.cve: print(f" 参考: {f.cve}") if f.recommendation: print(f" 措施: {f.recommendation}") print(f"\n--- 已知漏洞评估 ---") for vuln in self.check_known_vulnerabilities(): print(f"\n [{vuln['severity']}] {vuln['name']}") print(f" CVE: {vuln['cve']}") print(f" 受影响: {vuln['affected']}") print(f" 详情: {vuln['description']}") if __name__ == "__main__": analyzer = S7commAnalyzer() analyzer.set_authorized_stations(["10.10.2.50", "10.10.2.51"]) if len(sys.argv) >= 2: print(f"[*] 正在分析捕获文件: {sys.argv[1]}") packets = rdpcap(sys.argv[1]) for pkt in packets: analyzer.analyze_packet(pkt) analyzer.generate_report() else: print("用法: python s7comm_analyzer.py ") print(" 分析S7comm流量中的安全漏洞") ``` ## 核心概念 | 术语 | 定义 | |------|------| | 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. [监控增强] ```