| 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流量并识别漏洞
"""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)
S7_ROSCTR = {
0x01: "Job（请求）",
0x02: "Ack",
0x03: "Ack_Data（响应）",
0x07: "Userdata",
}
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程序下载流量发出警报。
方法：
- 捕获完整的S7comm会话进行取证分析
- 识别源主机并确定其是否被攻陷或为恶意设备
- 将当前PLC程序与最后一个已知良好备份进行对比
- 检查PLC CPU模式是否发生变化（运行→停止→编程）
- 如果程序被修改，从已验证的备份中恢复
- 调查攻击链——攻击者如何到达S7comm网段
- 在所有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. [监控增强]
