| name | detecting-attacks-on-scada-systems |
| description | 本技能涵盖检测针对数据采集与监控（SCADA）系统的网络攻击，包括工业协议的中间人攻击、向PLC注入未授权命令、HMI入侵、历史数据操纵以及对控制系统通信的拒绝服务攻击。它利用OT专用入侵检测系统、工业协议异常检测和过程数据分析来识别传统IT安全工具无法发现的攻击。
|
| domain | cybersecurity |
| subdomain | ot-ics-security |
| tags | ["ot-security","ics","scada","industrial-control","iec62443","intrusion-detection","threat-detection"] |
| version | 1.0.0 |
| author | mahipal |
| license | Apache-2.0 |
检测针对SCADA系统的攻击
适用场景
- 首次在SCADA环境中部署入侵检测能力
- 调查针对工业控制系统的疑似网络攻击
- 为OT特定攻击模式（Stuxnet、TRITON、Industroyer）构建检测规则
- 将OT网络监控与企业安全运营中心（SOC）集成以实现统一威胁可视性
- 响应OT安全监控工具（Dragos、Nozomi、Claroty）的告警
不适用于检测不含SCADA/ICS组件的纯IT网络攻击、构建通用网络IDS规则（参见building-detection-rules-with-sigma），或确认攻击后的事件响应流程（参见performing-ot-incident-response）。
前置条件
- 在OT网络边界的SPAN/TAP端口部署被动网络监控传感器
- OT入侵检测系统（Dragos Platform、Nozomi Guardian、Claroty xDome，或带OT规则集的Suricata）
- 了解使用中的工业协议（Modbus、DNP3、OPC UA、EtherNet/IP、S7comm）
- 正常SCADA通信模式的基线（轮询间隔、功能码、寄存器范围）
- 访问过程历史数据以进行物理过程异常关联
工作流程
步骤 1：建立SCADA通信基线
在检测异常之前，先确定正常SCADA流量的形态。工业协议具有高度确定性——同一主站以相同的时间间隔轮询同一从站，读取相同的寄存器。
"""SCADA Communication Baseline Builder.
Analyzes OT network traffic to establish deterministic baselines for
Modbus/TCP, DNP3, EtherNet/IP, and S7comm communications.
"""
import json
import sys
from collections import defaultdict
from datetime import datetime
from statistics import mean, stdev
try:
from scapy.all import rdpcap, IP, TCP, UDP
except ImportError:
print("Install scapy: pip install scapy")
sys.exit(1)
MODBUS_FUNC_NAMES = {
1: "Read Coils", 2: "Read Discrete Inputs",
3: "Read Holding Registers", 4: "Read Input Registers",
5: "Write Single Coil", 6: "Write Single Register",
8: "Diagnostics", 15: "Write Multiple Coils",
16: "Write Multiple Registers", 17: "Report Slave ID",
22: "Mask Write Register", 23: "Read/Write Multiple Registers",
43: "Encapsulated Interface Transport",
}
class SCADABaselineBuilder:
"""Builds deterministic baselines from SCADA traffic captures."""
():
.modbus_sessions = defaultdict(: {
: defaultdict(),
: (),
: [],
: ,
: ,
})
.communication_pairs = defaultdict(: {
: (),
: ,
: ,
: ,
})
():
packets = rdpcap(pcap_file)
()
pkt packets:
pkt.haslayer(IP):
src = pkt[IP].src
dst = pkt[IP].dst
ts = (pkt.time)
pair_key =
pair = .communication_pairs[pair_key]
pair[] +=
pair[] :
pair[] = ts
pair[] = ts
pkt.haslayer(TCP) pkt[TCP].dport == :
._analyze_modbus(pkt, src, dst, ts)
():
payload = (pkt[TCP].payload)
(payload) < :
func_code = payload[]
session_key =
session = .modbus_sessions[session_key]
session[][func_code] +=
session[] +=
session[] = {}
session[] :
interval = timestamp - session[]
< interval < :
session[].append(interval)
session[] = timestamp
(payload) >= func_code (, , , , , , , ):
start_register = (payload[] << ) | payload[]
func_code (, , , , , ) (payload) >= :
count = (payload[] << ) | payload[]
session[].add((func_code, start_register, start_register + count))
():
baseline = {
: datetime.now().isoformat(),
: {},
: {},
}
session_key, session .modbus_sessions.items():
avg_interval = mean(session[]) session[]
interval_std = stdev(session[]) (session[]) >
baseline[][session_key] = {
: (session[].keys()),
: {
MODBUS_FUNC_NAMES.get(k, ): v
k, v session[].items()
},
: (avg_interval, ),
: (interval_std, ),
: [
{: r[], : r[], : r[]}
r session[]
],
: session[],
}
baseline
():
baseline = .generate_baseline()
(output_file, ) f:
json.dump(baseline, f, indent=)
()
()
()
()
session, data baseline[].items():
()
()
()
()
()
__name__ == :
(sys.argv) < :
()
sys.exit()
builder = SCADABaselineBuilder()
builder.process_pcap(sys.argv[])
output = sys.argv[] (sys.argv) >
builder.export_baseline(output)
步骤 2：部署OT专用检测规则
为已知SCADA攻击模式创建检测规则，包括TRITON、Industroyer/CrashOverride和PIPEDREAM/INCONTROLLER使用的攻击模式。
alert modbus any any -> $OT_PLC_SUBNET 502 (
msg:"OT-DETECT Modbus write from unauthorized source";
modbus_func:!read_coils; modbus_func:!read_discrete_inputs;
modbus_func:!read_holding_registers; modbus_func:!read_input_registers;
flow:to_server,established;
threshold:type both, track by_src, count 1, seconds 60;
classtype:attempted-admin;
sid:3000001; rev:1;
)
alert modbus any any -> $OT_PLC_SUBNET 502 (
msg:"OT-DETECT Modbus diagnostics command to PLC";
modbus_func:diagnostics;
flow:to_server,established;
步骤 3：实现过程数据异常检测
监控历史数据服务器的物理过程数据，检测在向操作员隐藏影响的同时操控过程的攻击（Stuxnet攻击模式）。
"""SCADA Process Data Anomaly Detector.
Monitors historian data to detect physical process anomalies
that may indicate cyber attacks manipulating control logic
while spoofing sensor readings (Stuxnet-style attacks).
"""
import json
import sys
import time
from collections import deque
from dataclasses import dataclass
from datetime import datetime
from statistics import mean, stdev
from typing import Optional
try:
import requests
except ImportError:
print("Install requests: pip install requests")
sys.exit(1)
@dataclass
class ProcessVariable:
"""Represents a monitored process variable."""
tag_name: str
description: str
unit: str
low_limit: float
high_limit: float
rate_of_change_limit: float
engineering_low: float
engineering_high: float
@dataclass
class Anomaly:
"""Represents a detected process anomaly."""
timestamp: str
tag_name: str
anomaly_type: str
severity: str
current_value: float
expected_range:
description:
attack_pattern: =
:
():
.historian_url = historian_url
.api_key = api_key
.variables = {}
.history = defaultdict(: deque(maxlen=))
.anomalies = []
():
.variables[var.tag_name] = var
():
headers = {}
.api_key:
headers[] =
tag_list = (.variables.keys())
params = {: .join(tag_list), : }
:
resp = requests.get(
,
params=params,
headers=headers,
timeout=,
verify=,
)
resp.raise_for_status()
resp.json()
requests.RequestException e:
()
{}
():
var = .variables.get(tag_name)
var:
.history[tag_name].append((timestamp, value))
value < var.engineering_low value > var.engineering_high:
.anomalies.append(Anomaly(
timestamp=timestamp,
tag_name=tag_name,
anomaly_type=,
severity=,
current_value=value,
expected_range=,
description=,
attack_pattern=,
))
history = (.history[tag_name])
(history) >= :
prev_ts, prev_val = history[-]
:
dt = (datetime.fromisoformat(timestamp) - datetime.fromisoformat(prev_ts)).total_seconds()
dt > :
rate = (value - prev_val) / dt
rate > var.rate_of_change_limit:
.anomalies.append(Anomaly(
timestamp=timestamp,
tag_name=tag_name,
anomaly_type=,
severity=,
current_value=value,
expected_range=,
description=(
),
attack_pattern=,
))
(ValueError, TypeError):
(history) >= :
recent_values = [v _, v (history)[-:]]
((recent_values)) == :
.anomalies.append(Anomaly(
timestamp=timestamp,
tag_name=tag_name,
anomaly_type=,
severity=,
current_value=value,
expected_range=,
description=,
attack_pattern=,
))
(history) >= :
values = [v _, v (history)[-:]]
avg = mean(values)
std = stdev(values) (values) >
std > :
z_score = (value - avg) / std
z_score > :
.anomalies.append(Anomaly(
timestamp=timestamp,
tag_name=tag_name,
anomaly_type=,
severity=,
current_value=value,
expected_range=,
description=,
attack_pattern=,
))
():
.anomalies:
()
()
()
()
a .anomalies:
()
()
()
()
()
()
a.attack_pattern:
()
__name__ == :
collections defaultdict
detector = ProcessAnomalyDetector(
historian_url=,
)
detector.add_variable(ProcessVariable(
tag_name=,
description=,
unit=,
low_limit=, high_limit=,
rate_of_change_limit=,
engineering_low=, engineering_high=,
))
detector.add_variable(ProcessVariable(
tag_name=,
description=,
unit=,
low_limit=, high_limit=,
rate_of_change_limit=,
engineering_low=, engineering_high=,
))
detector.add_variable(ProcessVariable(
tag_name=,
description=,
unit=,
low_limit=, high_limit=,
rate_of_change_limit=,
engineering_low=, engineering_high=,
))
()
()
:
:
data = detector.fetch_current_values()
item data.get(, []):
detector.check_value(
item.get(),
item.get(),
item.get(, datetime.now().isoformat()),
)
time.sleep()
KeyboardInterrupt:
detector.report_anomalies()
步骤 4：检测已知ICS恶意软件指标
监控与已知ICS定向恶意软件家族相关的失陷指标（IOC）。
malware_families:
TRITON_TRISIS:
description: "针对施耐德电气Triconex安全仪表系统（SIS）"
target: "安全控制器（SIS）"
network_indicators:
- protocol: "TriStation"
port: 1502
pattern: "来自非工程工作站的异常TriStation命令"
- protocol: "TCP"
pattern: "未授权IP连接到Triconex控制器"
host_indicators:
- "工程工作站上存在trilog.exe"
- "System32目录中有inject.bin"
- "针对Triconex固件的imain.bin载荷"
detection_rule: |
alert tcp !$SIS_ENGINEERING_WS any -> $SIS_CONTROLLERS 1502 (
msg:"OT-CRITICAL Unauthorized TriStation connection to SIS";
flow:to_server; sid:3000100; rev:1; priority:1;)
INDUSTROYER_CRASHOVERRIDE:
description: "通过IEC 60870-5-101/104、IEC 61850、OPC DA针对电网SCADA"
target: "电网变电站和SCADA"
network_indicators:
- protocol: "IEC 60870-5-104"
port: 2404
pattern: "超出正常轮询的快速控制命令序列"
- protocol: "OPC DA"
pattern: "OPC服务器枚举后跟写入命令"
核心概念
| 术语 | 定义 |
|---|
| SCADA | 数据采集与监控（Supervisory Control and Data Acquisition）——通过RTU和通信基础设施远程监控和控制工业过程的架构 |
| OT专用IDS/IPS | 针对工业协议设计的入侵检测/防御系统，同时使用基于特征和基于异常的检测方法 |
| 过程异常（Process Anomaly） | 物理过程行为（温度、压力、流量）的偏差，可能表明控制系统遭受网络操控 |
| 中间人攻击（Man-in-the-Middle） | 拦截SCADA主站与现场设备之间通信以修改命令或欺骗传感器读数的攻击 |
| 回放攻击（Replay Attack） | 捕获合法SCADA流量并重放以掩盖对过程恶意更改的攻击（Stuxnet使用的技术） |
| 协议异常（Protocol Anomaly） | 与预期工业协议行为的偏差，包括未授权功能码、异常轮询模式或命令序列 |
工具与系统
- Dragos Platform：OT网络安全平台，威胁检测由Dragos针对ICS定向活动组的威胁情报驱动
- Nozomi Networks Guardian：利用资产情报、异常检测和漏洞评估的OT/IoT可视性与威胁检测平台
- Claroty xDome：具有持续威胁监控和告警优先级排序的网络物理系统保护平台
- Suricata with ET Open ICS rules：带有社区维护的工业协议检测规则的开源IDS/IPS
- Zeek (Bro) with OT scripts：具备Modbus、DNP3和BACnet协议分析器的网络安全监控工具
常见场景
场景：检测针对安全系统的TRITON式攻击
场景背景：OT安全监控系统对一个不是已授权SIS工程工作站的IP地址发往Triconex安全控制器的异常TriStation协议流量发出告警。
方法：
- 立即验证TriStation流量的源IP——是已授权的SIS工程工作站还是被入侵的主机？
- 检查是否有针对SIS控制器的已授权维护活动在进行中
- 捕获TriStation通信的完整数据包载荷用于取证分析
- 告知过程安全团队——SIS入侵是安全关键事件
- 如果未授权，立即将源主机从网络隔离
- 通过与已知良好备份对比，验证SIS控制器逻辑完整性
- 检查设施内所有工程工作站是否存在TRITON指标（trilog.exe、inject.bin）
常见陷阱：永远不要假设SIS流量异常是误报——TRITON证明了老练的攻击者会专门针对安全系统。在验证固件和逻辑完整性之前不要重启SIS控制器。避免仅向IT SOC告警；任何SIS相关事件都必须立即让过程安全团队参与。
输出格式
SCADA攻击检测报告
===============================
检测时间: YYYY-MM-DD HH:MM:SS UTC
检测来源: [IDS/异常检测器/过程监控]
告警详情:
告警ID: [唯一标识符]
严重级别: 严重/高/中/低
攻击类别: [协议异常/过程操控/未授权访问]
MITRE ATT&CK for ICS: [技术ID和名称]
来源: [IP/主机名]
目标: [IP/主机名 - 设备类型]
协议: [Modbus/DNP3/S7comm/等]
详情: [具体发现描述]
基线对比:
正常: [预期行为]
观测: [触发告警的实际行为]
偏差: [观测值与基线的差异]
推荐响应:
1. [立即遏制行动]
2. [验证步骤]
3. [上报路径]
