# Detecting Stuxnet Style Attacks

> 本技能涵盖检测遵循Stuxnet攻击模式的复杂网络物理攻击——在修改PLC逻辑的同时欺骗传感器读数以向操作员隐藏操控行为。内容涉及PLC逻辑完整性监控、基于物理的过程异常检测、工程师工作站入侵指标、USB传播攻击向量，以及从IT到OT横向移动直至过程操控的多阶段攻击链检测。

- Skill: `killvxk/detecting-stuxnet-style-attacks` (Agent Skill, multi-file: 4 files)
- Install (CLI): `npx skillmds@latest add killvxk/detecting-stuxnet-style-attacks`
- Raw SKILL.md: https://api.skillmd.com/api/skills/killvxk/detecting-stuxnet-style-attacks/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- License: Apache-2.0
- Author: killvxk (https://skillmd.com/u/killvxk)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/killvxk/detecting-stuxnet-style-attacks

---


# 检测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式攻击链中映射检测机会。

```yaml
# 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:
      - "Step-7/TIA Portal目录的文件完整性监控"
      - "PLC编程软件中的DLL注入检测"
      - "监控s7otbxdx.dll的Stuxnet特定钩子"
      - "PLC项目文件的意外修改"
    indicators:
      - "西门子STEP 7安装目录中的修改DLL"
      - "工程师工作站上的Rootkit隐藏文件"
      - "PLC编程软件行为异常"

  stage_4_plc_logic_modification:
    technique: "向PLC程序注入恶意OB/FC块"
    mitre_ics: "T0839 - Module Firmware / T0833 - Modify Control Logic"
    detection:
      - "对比已知良好基线的PLC逻辑完整性比较"
      - "来自未授权来源的S7comm上传/下载流量"
      - "PLC程序中出现新的OB/FC/FB块"
      - "OB1（主扫描）或OB35（循环中断）的修改"
    indicators:
      - "PLC程序块数量变化"
      - "PLC程序大小变化"
      - "上传未知程序块"

  stage_5_process_manipulation:
    technique: "在欺骗传感器读数的同时操控物理过程"
    mitre_ics: "T0836 - Modify Parameter / T0856 - Spoof Reporting Message"
    detection:
      - "基于物理的异常检测（过程模型偏差）"
      - "独立传感器的交叉验证"
      - "振动分析和机械特征监控"
      - "PLC报告值与独立测量值的比较"
    indicators:
      - "电机/泵在正常参数范围之外运行"
      - "传感器读数偏离物理模型预测"
      - "过程效率指标出现意外偏差"
```

### 步骤 2：实现PLC逻辑完整性监控

通过将运行逻辑与已知良好基线比较，持续监控PLC程序完整性。

```python
#!/usr/bin/env python3
"""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  # OB, FC, FB, DB
    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 = {}  # plc_name -> PLCBlock列表
        self.alerts = []
        self.alert_counter = 1

    def load_baseline(self, plc_name, baseline_file):
        """Load known-good PLC program baseline."""
        with open(baseline_file) as f:
            data = json.load(f)
        blocks = [PLCBlock(**b) for b in data.get("blocks", [])]
        self.baselines[plc_name] = {
            "blocks": {f"{b.block_type}{b.block_number}": b for b in blocks},
            "total_blocks": len(blocks),
            "loaded_at": datetime.now().isoformat(),
        }
        print(f"[*] 已为 {plc_name} 加载基线: {len(blocks)} 个块")

    def check_integrity(self, plc_name, plc_ip, current_blocks):
        """Compare current PLC program against baseline."""
        baseline = self.baselines.get(plc_name)
        if not baseline:
            print(f"[WARN] {plc_name} 无基线")
            return

        baseline_blocks = baseline["blocks"]
        current_block_map = {f"{b.block_type}{b.block_number}": b for b in current_blocks}

        # 检查1：新增块（潜在逻辑注入）
        for key, block in current_block_map.items():
            if key not in baseline_blocks:
                self.alerts.append(IntegrityAlert(
                    alert_id=f"INT-{self.alert_counter:04d}",
                    timestamp=datetime.now().isoformat(),
                    severity="critical",
                    plc_name=plc_name,
                    plc_ip=plc_ip,
                    alert_type="NEW_BLOCK_DETECTED",
                    description=(
                        f"PLC中发现基线中不存在的新程序块 {key} ({block.name})。"
                        f"大小: {block.size_bytes} 字节。"
                    ),
                    baseline_value="基线中不存在该块",
                    current_value=f"{key}: {block.size_bytes} 字节, 校验和 {block.checksum}",
                    mitre_technique="T0839 - Module Firmware / T0833 - Modify Control Logic",
                ))
                self.alert_counter += 1

        # 检查2：删除的块
        for key in baseline_blocks:
            if key not in current_block_map:
                self.alerts.append(IntegrityAlert(
                    alert_id=f"INT-{self.alert_counter:04d}",
                    timestamp=datetime.now().isoformat(),
                    severity="high",
                    plc_name=plc_name,
                    plc_ip=plc_ip,
                    alert_type="BLOCK_REMOVED",
                    description=f"程序块 {key} 已从PLC中删除",
                    baseline_value=f"{key}: {baseline_blocks[key].size_bytes} 字节",
                    current_value="未找到该块",
                    mitre_technique="T0833 - Modify Control Logic",
                ))
                self.alert_counter += 1

        # 检查3：块内容被修改（校验和不匹配）
        for key in baseline_blocks:
            if key in current_block_map:
                baseline_block = baseline_blocks[key]
                current_block = current_block_map[key]

                if baseline_block.checksum != current_block.checksum:
                    self.alerts.append(IntegrityAlert(
                        alert_id=f"INT-{self.alert_counter:04d}",
                        timestamp=datetime.now().isoformat(),
                        severity="critical",
                        plc_name=plc_name,
                        plc_ip=plc_ip,
                        alert_type="BLOCK_MODIFIED",
                        description=(
                            f"程序块 {key} 校验和不匹配。"
                            f"自建立基线以来逻辑已被修改。"
                        ),
                        baseline_value=f"校验和: {baseline_block.checksum}, 大小: {baseline_block.size_bytes}",
                        current_value=f"校验和: {current_block.checksum}, 大小: {current_block.size_bytes}",
                        mitre_technique="T0833 - Modify Control Logic",
                    ))
                    self.alert_counter += 1

        # 检查4：块数量变化
        if len(current_blocks) != baseline["total_blocks"]:
            self.alerts.append(IntegrityAlert(
                alert_id=f"INT-{self.alert_counter:04d}",
                timestamp=datetime.now().isoformat(),
                severity="high",
                plc_name=plc_name,
                plc_ip=plc_ip,
                alert_type="BLOCK_COUNT_CHANGE",
                description=f"总块数从 {baseline['total_blocks']} 变为 {len(current_blocks)}",
                baseline_value=str(baseline["total_blocks"]),
                current_value=str(len(current_blocks)),
                mitre_technique="T0833 - Modify Control Logic",
            ))
            self.alert_counter += 1

    def generate_report(self):
        """Generate integrity monitoring report."""
        print(f"\n{'='*70}")
        print("PLC逻辑完整性监控报告")
        print(f"{'='*70}")
        print(f"已加载基线数: {len(self.baselines)}")
        print(f"告警数: {len(self.alerts)}")

        for a in self.alerts:
            print(f"\n  [{a.severity.upper()}] {a.alert_type}")
            print(f"    PLC: {a.plc_name} ({a.plc_ip})")
            print(f"    {a.description}")
            print(f"    基线: {a.baseline_value}")
            print(f"    当前: {a.current_value}")
            print(f"    MITRE: {a.mitre_technique}")


if __name__ == "__main__":
    monitor = PLCIntegrityMonitor()
    print("PLC逻辑完整性监控器")
    print("加载基线并定期调用 check_integrity()")
```

### 步骤 3：部署基于物理的过程异常检测

使用基于物理定律预测预期传感器值的模型监控物理过程行为。偏差表明设备故障或网络物理攻击。

```python
#!/usr/bin/env python3
"""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被欺骗，它将不匹配实际频率。
        """
        # 假设4极电机，典型3%转差率
        expected_rpm = (120 * frequency_hz / 4) * 0.97
        deviation = abs(reported_rpm - expected_rpm) / expected_rpm * 100

        if deviation > 5.0:
            self.alerts.append(PhysicsAlert(
                timestamp=datetime.now().isoformat(),
                severity="critical",
                alert_type="FREQUENCY_RPM_MISMATCH",
                sensor_tag="MOTOR.RPM vs VFD.FREQ",
                reported_value=reported_rpm,
                predicted_value=round(expected_rpm, 1),
                deviation_percent=round(deviation, 1),
                description=(
                    f"电机RPM（{reported_rpm}）与VFD频率（{frequency_hz} Hz）不匹配。"
                    f"预期约 {expected_rpm:.0f} RPM。"
                    f"可能在操控频率的同时欺骗RPM传感器。"
                ),
            ))

    def check_power_speed_correlation(self, rpm, power_kw):
        """验证功耗是否与转速匹配。

        对于离心负载，功率约与RPM的三次方成比例。
        """
        speed_ratio = rpm / self.rated_rpm
        expected_power = self.rated_power_kw * (speed_ratio ** 3)
        deviation = abs(power_kw - expected_power) / max(expected_power, 0.1) * 100

        if deviation > 15.0:
            self.alerts.append(PhysicsAlert(
                timestamp=datetime.now().isoformat(),
                severity="high",
                alert_type="POWER_SPEED_MISMATCH",
                sensor_tag="MOTOR.POWER vs MOTOR.RPM",
                reported_value=power_kw,
                predicted_value=round(expected_power, 1),
                deviation_percent=round(deviation, 1),
                description=(
                    f"功耗（{power_kw} kW）与RPM（{rpm}）不一致。"
                    f"预期约 {expected_power:.1f} kW。可能表明存在隐藏的转速变化。"
                ),
            ))

    def check_vibration_anomaly(self, rpm, vibration_mm_s):
        """检查振动特征是否与运行速度一致。

        在报告"正常"速度时的异常振动可能表明实际
        速度与传感器报告的不同。
        """
        # 平衡转子的正常振动随速度线性增加
        speed_ratio = rpm / self.rated_rpm
        expected_vibration = 2.0 * speed_ratio  # mm/s基线
        deviation = abs(vibration_mm_s - expected_vibration) / max(expected_vibration, 0.1) * 100

        if vibration_mm_s > 7.0:  # ISO 10816告警阈值
            self.alerts.append(PhysicsAlert(
                timestamp=datetime.now().isoformat(),
                severity="critical",
                alert_type="ABNORMAL_VIBRATION",
                sensor_tag="MOTOR.VIBRATION",
                reported_value=vibration_mm_s,
                predicted_value=round(expected_vibration, 1),
                deviation_percent=round(deviation, 1),
                description=(
                    f"RPM报告正常（{rpm}）时振动（{vibration_mm_s} mm/s）达到ISO告警级别。"
                    f"实际速度可能与报告不同。"
                ),
            ))

    def report(self):
        if self.alerts:
            print(f"\n{'='*60}")
            print("基于物理的异常检测告警")
            print(f"{'='*60}")
            for a in self.alerts:
                print(f"\n  [{a.severity.upper()}] {a.alert_type}")
                print(f"    {a.description}")
                print(f"    报告值: {a.reported_value} | 预测值: {a.predicted_value}")
                print(f"    偏差: {a.deviation_percent}%")


if __name__ == "__main__":
    model = CentrifugePhysicsModel(rated_rpm=1200, rated_frequency=50, rated_power_kw=75)

    # 正常运行 - 预期无告警
    model.check_frequency_rpm_correlation(50.0, 1164)
    model.check_power_speed_correlation(1164, 72.0)

    # Stuxnet式攻击：频率增加但RPM被欺骗为正常
    model.check_frequency_rpm_correlation(84.0, 1164)  # 频率增加，RPM被欺骗
    model.check_power_speed_correlation(1164, 180.0)    # 功耗揭示真实速度

    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]
```