Skip to main content Home Creators killvxk cybersecurity-skills-zh detecting-anomalies-in-industrial-control-systems
detecting-anomalies-in-industrial-control-systems 本技能涵盖使用机器学习模型（基于OT网络基线训练）、基于物理过程模型以及工业协议通信行为分析，在工业控制环境中部署异常检测（anomaly detection）系统。内容涉及为SCADA轮询模式构建正常行为基线、检测Modbus/DNP3/OPC UA流量中的偏差、识别未授权设备，以及将网络异常与历史数据服务器的物理过程数据进行关联。
Jump to install Skills Marketplace Discover and explore AI skills built by the community.
Install with Codex or Claude Copy this prompt, paste it into Codex, Claude, or another assistant, and let it review the skill page and install it for you.
Show prompt details A direct command skips the review prompt. Inspect the source before running it.
npx skills add https://github.com/killvxk/cybersecurity-skills-zh --skill detecting-anomalies-in-industrial-control-systemsThe command stays on one line. Scroll horizontally to inspect it before copying.
Prefer a local copy? Download the files currently available to SkillsMP.
Download Zip Downloading... name detecting-anomalies-in-industrial-control-systems description 本技能涵盖使用机器学习模型（基于OT网络基线训练）、基于物理过程模型以及工业协议通信行为分析，在工业控制环境中部署异常检测（anomaly detection）系统。内容涉及为SCADA轮询模式构建正常行为基线、检测Modbus/DNP3/OPC UA流量中的偏差、识别未授权设备，以及将网络异常与历史数据服务器的物理过程数据进行关联。
domain cybersecurity subdomain ot-ics-security tags ["ot-security","ics","scada","industrial-control","iec62443","anomaly-detection","machine-learning"] version 1.0.0 author mahipal license Apache-2.0
检测工业控制系统中的异常行为
适用场景
为缺乏入侵检测的OT环境部署持续监控
构建基于行为的检测以补充OT网络中基于特征的入侵检测系统（IDS）
为确定性SCADA通信建立基线以检测偏差
将机器学习异常检测与OT安全监控平台集成
调查Nozomi Guardian或Dragos Platform告警时需要进行深入分析
不适用于 基于特征的已知漏洞利用检测（参见detecting-attacks-on-scada-systems）、不含OT协议的IT网络异常检测，或替代过程安全系统（SIS）。
前置条件
在OT网络SPAN/TAP端口上部署被动网络监控传感器
正常运营期间至少2-4周的基线流量采集
Python 3.9+，包含用于ML模型训练的scikit-learn、numpy、pandas
访问过程历史数据服务器以获取物理过程关联数据
了解正常操作模式，包括班次交接、批次处理和维护窗口
工作流程
步骤 1：构建多维基线模型
从多个维度捕获并建模ICS通信的确定性行为：时序、协议行为和网络拓扑。
"""ICS Anomaly Detection System.
Builds multi-dimensional baselines from OT network traffic and
detects anomalies using statistical and machine learning methods.
Designed for deterministic SCADA communication patterns.
"""
import json
import sys
import time
import warnings
from collections import defaultdict
from datetime import datetime, timedelta
from dataclasses import dataclass, field
import numpy as np
import pandas as pd
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
warnings.filterwarnings("ignore" )
@dataclass
class CommunicationProfile :
"""Profile for a single master-slave communication pair."""
src_ip: str
dst_ip: str
protocol: str
port: int
avg_interval_ms: float = 0.0
std_interval_ms: float = 0.0
avg_payload_size: float = 0.0
function_codes: dict = field(default_factory=dict )
packets_per_minute: float = 0.0
first_seen: str = ""
last_seen: str = ""
class ICSAnomalyDetector :
核心概念
术语 定义 确定性流量（Deterministic Traffic） ICS网络表现出高度可预测的通信模式，相同的主站以固定时间间隔、相同的功能码轮询相同的从站 孤立森林（Isolation Forest） 通过随机划分特征空间来隔离异常的无监督机器学习算法，适用于低异常率的OT流量 轮询间隔（Polling Interval） SCADA主站连续向从站设备发送请求之间的时间间隔，通常固定且可配置（100ms至10s） 功能码白名单（Function Code Allowlist） 每个通信对允许使用的工业协议操作集合，由异常检测规则强制执行 拓扑基线（Topology Baseline） OT网络中所有已授权设备间通信路径的完整映射 基于物理的检测（Physics-Based Detection） 使用物理过程模型（热力学、流体动力学）检测在欺骗传感器数据的同时操控过程的攻击
工具与系统
Nozomi Networks Guardian ：具备AI驱动基线学习和工业协议分析的OT异常检测平台Dragos Platform ：使用行为分析和ICS特定威胁情报的威胁检测平台Scikit-learn ：Python ML库，包含用于异常检测的孤立森林、单类SVM和局部离群因子算法Zeek with OT plugins ：具备Modbus、DNP3和BACnet协议分析器的网络安全监控工具，用于基线构建
输出格式
ICS 异常检测报告
==============================
检测周期: YYYY-MM-DD 至 YYYY-MM-DD
基线规模: [N] 个通信配置文件
检测到的异常数: [N]
严重: [N] 高: [N] 中: [N] 低: [N]
[严重级别] 异常类型
来源: [IP] -> 目标: [IP]:[端口]
详情: [偏离基线的描述]
基线: [预期行为]
观测: [实际行为]
"""Multi-dimensional anomaly detection for ICS environments."""
def
__init__
self
self
self
set
self
None
self
None
self
self
self
def
build_baseline_from_pcap
self, pcap_data
"""Build baselines from parsed pcap data (list of flow records)."""
print
"[*] 正在构建ICS通信基线..."
for
in
f"{flow['src' ]} ->{flow['dst' ]} :{flow['port' ]} "
if
not
in
self
self
"src"
"dst"
"protocol"
"TCP"
"port"
"timestamp"
""
self
"timestamp"
""
"function_code"
if
is
not
None
0
1
self
"src"
"dst"
"port"
self
print
f" 通信对数量: {len (self.profiles)} "
print
f" 拓扑条目数: {len (self.topology_baseline)} "
def
_calculate_timing_stats
self, flows
"""Calculate packet timing statistics per communication pair."""
list
for
in
f"{flow['src' ]} ->{flow['dst' ]} :{flow['port' ]} "
"timestamp_epoch"
if
for
in
if
in
self
and
len
1
sorted
1
1000
for
in
range
len
1
self
self
1
0
60
if
0
self
len
def
train_isolation_forest
self, features_df
"""Train Isolation Forest model on feature vectors from baseline traffic."""
print
"[*] 正在训练孤立森林（Isolation Forest）模型..."
"interval_ms"
"payload_size"
"packets_per_window"
"unique_func_codes"
"new_connection_flag"
for
in
if
in
0
self
self
200
0.01
42
1
self
self
print
f" 模型训练样本数: {len (X)} "
print
f" 异常分数范围: [{scores.min ():.4 f} , {scores.max ():.4 f} ]"
print
f" 阈值: {np.percentile(scores, 1 ):.4 f} "
def
detect_topology_anomaly
self, src_ip, dst_ip, port
"""Detect new/unauthorized communication pairs."""
if
not
in
self
return
"type"
"NEW_COMMUNICATION_PAIR"
"severity"
"high"
"detail"
f"新连接: {src_ip} -> {dst_ip} :{port} 不在基线中"
"recommendation"
"验证是否为已授权的新设备或配置变更"
return
None
def
detect_timing_anomaly
self, src_ip, dst_ip, port, interval_ms
"""Detect polling interval deviations."""
f"{src_ip} ->{dst_ip} :{port} "
self
if
and
0
abs
if
4.0
return
"type"
"TIMING_ANOMALY"
"severity"
"medium"
"detail"
f"间隔 {interval_ms:.1 f} ms 偏离基线 "
f"{profile.avg_interval_ms:.1 f} ms (z分数: {z_score:.1 f} )"
"recommendation"
"检查网络拥塞、设备故障或中间人攻击"
return
None
def
detect_function_code_anomaly
self, src_ip, dst_ip, port, func_code
"""Detect unauthorized Modbus/DNP3 function codes."""
f"{src_ip} ->{dst_ip} :{port} "
self
if
and
not
in
"critical"
if
in
5
6
15
16
8
else
"high"
return
"type"
"UNAUTHORIZED_FUNCTION_CODE"
"severity"
"detail"
f"来自 {src_ip} 到 {dst_ip} :{port} 的功能码 {func_code} "
f"不在基线中。允许的功能码: {list (profile.function_codes.keys())} "
"recommendation"
"调查来源 - 可能是命令注入攻击"
return
None
def
analyze_flow
self, flow
"""Analyze a single network flow against all detection models."""
self
"src"
"dst"
"port"
if
if
"interval_ms"
in
self
"src"
"dst"
"port"
"interval_ms"
if
if
"function_code"
in
self
"src"
"dst"
"port"
"function_code"
if
self
return
def
generate_report
self
"""Generate anomaly detection report."""
print
f"\n{'=' *60 } "
print
f"ICS 异常检测报告"
print
f"{'=' *60 } "
print
f"基线配置文件数: {len (self.profiles)} "
print
f"检测到的异常数: {len (self.anomalies)} "
int
for
in
self
"severity"
1
for
in
"critical"
"high"
"medium"
"low"
if
print
f" {sev.upper()} : {severity_counts[sev]} "
for
in
self
20
print
f"\n [{a['severity' ].upper()} ] {a['type' ]} "
print
f" {a['detail' ]} "
if
"__main__"
print
"ICS 异常检测系统"
print
"加载基线数据并调用 analyze_flow() 进行实时检测"
More from this repository
Related occupations SOC
Based on SOC occupation classification
