| name | detecting-stuxnet-style-attacks |
| description | 本技能涵盖检测遵循Stuxnet攻击模式的复杂网络物理攻击——在修改PLC逻辑的同时欺骗传感器读数以向操作员隐藏操控行为。内容涉及PLC逻辑完整性监控、基于物理的过程异常检测、工程师工作站入侵指标、USB传播攻击向量，以及从IT到OT横向移动直至过程操控的多阶段攻击链检测。
|
| domain | cybersecurity |
| subdomain | ot-ics-security |
| tags | ["ot-security","ics","scada","industrial-control","iec62443","stuxnet","plc-integrity","apt"] |
| version | 1.0.0 |
| author | mahipal |
| license | Apache-2.0 |
检测Stuxnet式攻击
适用场景
- 为高价值OT目标（核能、化工、关键基础设施）实施高级威胁检测
- 为针对PLC逻辑和过程操控的APT式攻击构建检测机制
- 建立PLC逻辑完整性监控以检测未授权修改
- 调查可能表明网络物理攻击的疑似过程异常
- 设计针对国家级OT威胁的纵深防御策略
不适用于基本OT入侵检测（参见detecting-attacks-on-scada-systems）、Stuxnet样本的恶意软件分析（参见恶意软件逆向工程技能），或PLC编程和逻辑开发。
前置条件
- 深入了解Stuxnet攻击链和MITRE ATT&CK for ICS框架
- 包含所有PLC程序已知良好基线副本的PLC逻辑备份仓库
- 具备OT感知的工程师工作站监控（EDR）
- 被控制物理过程的基于物理的过程模型
- 工业协议流量分析的网络监控
工作流程
步骤 1：了解Stuxnet攻击链
在多阶段Stuxnet式攻击链中映射检测机会。
attack_chain:
stage_1_initial_access:
technique: "针对气隙网络的USB传播恶意软件"
mitre_ics: "T0847 - Replication Through Removable Media"
detection:
- "工程师工作站的USB设备连接日志"
- "使用OT批准的AV扫描可移动媒体"
- "应用程序白名单阻止未授权可执行文件"
- "通过组策略禁用Windows自动运行"
indicators:
- "工程师工作站的新USB设备连接"
- "从可移动媒体执行未签名二进制文件"
- "LNK文件漏洞利用模式"
stage_2_lateral_movement:
technique: "利用Windows漏洞进行网络传播"
mitre_ics: "T0866 - Exploitation of Remote Services"
detection:
- "网络IDS检测漏洞利用流量（MS08-067、MS10-061）"
- "工程师工作站之间的异常SMB流量"
- "Windows事件日志显示权限提升"
- "新建计划任务或服务"
indicators:
- "3-4级Windows系统间的横向移动"
- "来自意外来源的WMI/PsExec执行"
- "哈希传递认证模式"
stage_3_ews_compromise:
technique: "入侵具有PLC编程软件的工程师工作站"
mitre_ics: "T0862 - Supply Chain Compromise (Step-7 hooking)"
detection:
步骤 2：实现PLC逻辑完整性监控
通过将运行逻辑与已知良好基线比较，持续监控PLC程序完整性。
"""PLC Logic Integrity Monitor.
Periodically retrieves PLC program block information and compares
against known-good baselines to detect unauthorized modifications
(Stuxnet-style logic injection).
"""
import hashlib
import json
import sys
import time
from dataclasses import dataclass, field, asdict
from datetime import datetime
@dataclass
class PLCBlock:
"""Represents a PLC program block."""
block_type: str
block_number: int
name: str
size_bytes: int
checksum: str
last_modified: str
author: str = ""
@dataclass
class IntegrityAlert:
alert_id: str
timestamp: str
severity: str
plc_name: str
plc_ip: str
alert_type: str
description: str
baseline_value: str
current_value: str
mitre_technique: str
class PLCIntegrityMonitor:
"""Monitors PLC program integrity against baselines."""
def __init__(self):
self.baselines = {}
self.alerts = []
.alert_counter =
():
(baseline_file) f:
data = json.load(f)
blocks = [PLCBlock(**b) b data.get(, [])]
.baselines[plc_name] = {
: {: b b blocks},
: (blocks),
: datetime.now().isoformat(),
}
()
():
baseline = .baselines.get(plc_name)
baseline:
()
baseline_blocks = baseline[]
current_block_map = {: b b current_blocks}
key, block current_block_map.items():
key baseline_blocks:
.alerts.append(IntegrityAlert(
alert_id=,
timestamp=datetime.now().isoformat(),
severity=,
plc_name=plc_name,
plc_ip=plc_ip,
alert_type=,
description=(
),
baseline_value=,
current_value=,
mitre_technique=,
))
.alert_counter +=
key baseline_blocks:
key current_block_map:
.alerts.append(IntegrityAlert(
alert_id=,
timestamp=datetime.now().isoformat(),
severity=,
plc_name=plc_name,
plc_ip=plc_ip,
alert_type=,
description=,
baseline_value=,
current_value=,
mitre_technique=,
))
.alert_counter +=
key baseline_blocks:
key current_block_map:
baseline_block = baseline_blocks[key]
current_block = current_block_map[key]
baseline_block.checksum != current_block.checksum:
.alerts.append(IntegrityAlert(
alert_id=,
timestamp=datetime.now().isoformat(),
severity=,
plc_name=plc_name,
plc_ip=plc_ip,
alert_type=,
description=(
),
baseline_value=,
current_value=,
mitre_technique=,
))
.alert_counter +=
(current_blocks) != baseline[]:
.alerts.append(IntegrityAlert(
alert_id=,
timestamp=datetime.now().isoformat(),
severity=,
plc_name=plc_name,
plc_ip=plc_ip,
alert_type=,
description=,
baseline_value=(baseline[]),
current_value=((current_blocks)),
mitre_technique=,
))
.alert_counter +=
():
()
()
()
()
()
a .alerts:
()
()
()
()
()
()
__name__ == :
monitor = PLCIntegrityMonitor()
()
()
步骤 3：部署基于物理的过程异常检测
使用基于物理定律预测预期传感器值的模型监控物理过程行为。偏差表明设备故障或网络物理攻击。
"""Physics-Based Cyber-Physical Attack Detector.
Uses simplified physics models to detect process manipulation
attacks where the attacker modifies the physical process while
spoofing sensor readings (the core Stuxnet attack pattern).
"""
import math
from dataclasses import dataclass
from datetime import datetime
@dataclass
class PhysicsAlert:
timestamp: str
severity: str
alert_type: str
sensor_tag: str
reported_value: float
predicted_value: float
deviation_percent: float
description: str
class CentrifugePhysicsModel:
"""离心机系统物理模型（Stuxnet目标类比）。
通过交叉关联检测操控:
- 电机频率（Hz）vs 报告RPM
- RPM vs 振动特征
- 功耗 vs 转速
"""
def __init__(self, rated_rpm=1200, rated_frequency=50, rated_power_kw=75):
self.rated_rpm = rated_rpm
self.rated_frequency = rated_frequency
self.rated_power_kw = rated_power_kw
self.alerts = []
def check_frequency_rpm_correlation(self, frequency_hz, reported_rpm):
"""验证电机频率与报告RPM是否匹配。
对于感应电机: RPM = 120 * 频率 / 极数
如果RPM被欺骗，它将不匹配实际频率。
"""
expected_rpm = (120 * frequency_hz / 4) * 0.97
deviation = abs(reported_rpm - expected_rpm) / expected_rpm *
deviation > :
.alerts.append(PhysicsAlert(
timestamp=datetime.now().isoformat(),
severity=,
alert_type=,
sensor_tag=,
reported_value=reported_rpm,
predicted_value=(expected_rpm, ),
deviation_percent=(deviation, ),
description=(
),
))
():
speed_ratio = rpm / .rated_rpm
expected_power = .rated_power_kw * (speed_ratio ** )
deviation = (power_kw - expected_power) / (expected_power, ) *
deviation > :
.alerts.append(PhysicsAlert(
timestamp=datetime.now().isoformat(),
severity=,
alert_type=,
sensor_tag=,
reported_value=power_kw,
predicted_value=(expected_power, ),
deviation_percent=(deviation, ),
description=(
),
))
():
speed_ratio = rpm / .rated_rpm
expected_vibration = * speed_ratio
deviation = (vibration_mm_s - expected_vibration) / (expected_vibration, ) *
vibration_mm_s > :
.alerts.append(PhysicsAlert(
timestamp=datetime.now().isoformat(),
severity=,
alert_type=,
sensor_tag=,
reported_value=vibration_mm_s,
predicted_value=(expected_vibration, ),
deviation_percent=(deviation, ),
description=(
),
))
():
.alerts:
()
()
()
a .alerts:
()
()
()
()
__name__ == :
model = CentrifugePhysicsModel(rated_rpm=, rated_frequency=, rated_power_kw=)
model.check_frequency_rpm_correlation(, )
model.check_power_speed_correlation(, )
model.check_frequency_rpm_correlation(, )
model.check_power_speed_correlation(, )
model.report()
核心概念
| 术语 | 定义 |
|---|
| 网络物理攻击（Cyber-Physical Attack） | 同时操控网络系统（PLC逻辑、传感器读数）和物理过程的攻击 |
| 逻辑注入（Logic Injection） | 将恶意代码块插入PLC程序以改变物理过程行为 |
| 传感器欺骗（Sensor Spoofing） | 回放或伪造传感器读数以向操作员隐藏过程操控 |
| 基于物理的检测（Physics-Based Detection） | 使用物理过程数学模型检测报告传感器值与实际物理的不一致 |
| PLC逻辑基线（PLC Logic Baseline） | 用于完整性比较的PLC程序块（OB、FC、FB、DB）的已知良好副本 |
| 气隙桥接（Air-Gap Bridging） | 通过USB驱动器跨越气隙网络的技术，如Stuxnet的初始访问方式 |
工具与系统
- Claroty xDome：具有基线比较和变更检测的持续PLC逻辑监控
- SIGA OT Solutions：电气层面的物理信号监控，用于检测过程操控
- Nozomi Guardian：具备PLC程序变更检测能力的OT监控平台
- Siemens SINEMA Remote Connect：带PLC项目版本跟踪的安全远程访问
输出格式
Stuxnet式攻击检测报告
========================================
监控的PLC数: [N]
监控周期: YYYY-MM-DD 至 YYYY-MM-DD
PLC完整性:
已验证基线: [N]/[N]
检测到逻辑修改: [N]
检测到新块: [N]
物理异常:
传感器关联违规: [N]
过程模型偏差: [N]
工程师工作站:
未授权修改: [N]
USB连接: [N]
