菜单
对西门子SIMATIC S7 PLC使用的S7comm和S7CommPlus协议进行安全分析，识别漏洞，包括重放攻击、完整性绕过、未授权的CPU停止命令以及针对S7-300、S7-400、S7-1200和S7-1500控制器弱点的程序下载操控。
默认使用会先检查来源的 Prompt；你也可以切换为直接命令，或下载本地副本。
决定是否安装前，请先阅读 SKILL.md，以及 SkillsMP 当前展示的配套文件。
用 Codex 或 Claude 帮你安装 复制这段 Prompt，粘贴到 Codex、Claude 或其他助手里，让它检查 Skill 页面并帮你完成安装。
直接命令不会经过审查 Prompt；运行前请先检查来源。
npx skills add https://github.com/killvxk/cybersecurity-skills-zh --skill performing-s7comm-protocol-security-analysis命令会保持在同一行。复制前请横向滚动并检查完整内容。
想先保存到本地？可下载 SkillsMP 当前能够提供的文件。
正在显示 SKILL.md
| 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 |
不适用于未经授权和测试计划扫描生产西门子PLC（可能导致控制器崩溃）、非西门子协议分析（Modbus请参见detecting-modbus-command-injection-attacks），或在生产环境中修改PLC程序。
#!/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功能，代表最高风险操作 |
背景：Dragos传感器对来自非授权TIA Portal工程师工作站IP地址的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. [监控增强]
基于 SOC 职业分类