メニュー
通过监控未授权控制命令、固件更新尝试、协议违规以及与基线流量模式的偏差，使用深度包检测和机器学习方法检测SCADA系统中使用的DNP3（分布式网络协议3）通信中的异常。
デフォルトでは、最初にソースを確認する Prompt が選択されています。直接コマンドに切り替えるか、ローカルコピーをダウンロードすることもできます。
インストールを決める前に、SKILL.md と SkillsMP に表示されている付属ファイルをお読みください。
Codex または Claude でインストール この Prompt をコピーして Codex、Claude、または他のアシスタントに貼り付けると、Skill ページを確認してインストールできます。
直接コマンドでは確認用 Prompt が省略されます。実行前にソースを確認してください。
npx skills add https://github.com/killvxk/cybersecurity-skills-zh --skill detecting-dnp3-protocol-anomaliesコマンドは1行のまま表示されます。コピー前に横へスクロールして全体を確認してください。
ローカルで確認しますか？SkillsMP が現在取得できるファイルをダウンロードできます。
通过分析 Zeek dns.log 中的高熵子域名查询、超量查询量、超长查询长度以及异常 DNS 记录类型，检测 DNS 隧道和数据外泄中的隐蔽通道通信。适用于：当需要狩猎基于 DNS 的 C2 或数据外泄通道、调查异常 DNS 查询模式、或响应涉及 DNS 隧道工具（iodine、dnscat2、DNSExfiltrator）的威胁情报时使用。
实施 Google 的 BeyondCorp 零信任访问模型，通过 IAP、Access Context Manager 和 Chrome Enterprise Premium，消除网络边界的隐式信任，强制执行基于身份的访问控制，实现无 VPN 的安全应用访问。适用于将传统 VPN 替换为零信任架构、部署 Identity-Aware Proxy、配置设备信任策略、或为远程办公实施上下文感知访问控制时使用。
在授权的安全评估过程中，使用 Burp Suite 的扫描器、Intruder 和 Repeater 工具识别和验证跨站脚本（XSS）漏洞。适用于 Web 应用渗透测试中检测反射型、存储型和 DOM 型 XSS，验证自动化扫描器报告的 XSS 发现，以及评估 CSP 和 XSS 过滤器的有效性时使用。
SOC 職業分類に基づく
SKILL.md を表示中
| name | detecting-dnp3-protocol-anomalies |
| description | 通过监控未授权控制命令、固件更新尝试、协议违规以及与基线流量模式的偏差，使用深度包检测和机器学习方法检测SCADA系统中使用的DNP3（分布式网络协议3）通信中的异常。 |
| domain | cybersecurity |
| subdomain | ot-ics-security |
| tags | ["ot-security","ics","dnp3","scada","anomaly-detection","protocol-analysis","energy-sector","ids"] |
| version | 1.0 |
| author | mahipal |
| license | Apache-2.0 |
不适用于非DNP3协议监控（Modbus相关参见detecting-modbus-command-injection-attacks）、DNP3安全认证配置（独立实现），或与协议无关的网络异常检测。
#!/usr/bin/env python3
"""DNP3 Protocol Anomaly Detector.
Monitors DNP3 communications for unauthorized control commands,
protocol violations, and deviations from established baselines.
Supports both TCP and serial DNP3 deployments.
"""
import struct
import sys
import json
from collections import defaultdict
from datetime import datetime
from typing import Dict, List, Optional, Set
try:
from scapy.all import rdpcap, IP, TCP
except ImportError:
print("Install scapy: pip install scapy")
sys.exit(1)
# DNP3功能码
DNP3_FUNCTIONS = {
0x00: "Confirm", 0x01: "Read", 0x02: "Write",
0x03: "Select", 0x04: "Operate", 0x05: "Direct Operate",
0x06: "Direct Operate No Ack", 0x07: "Immediate Freeze",
0x08: "Immediate Freeze No Ack", 0x09: "Freeze and Clear",
0x0A: "Freeze and Clear No Ack", 0x0B: "Freeze at Time",
0x0C: , : ,
: , : ,
: , : ,
: , : ,
: , : ,
: , : ,
: , : ,
: , : ,
: , : ,
: , : , : ,
}
DNP3_CRITICAL_FUNCTIONS = {
,
, , , ,
,
,
,
,
,
, , ,
}
:
():
.alerts = []
.sessions = defaultdict(: {
: ,
: defaultdict(),
: ,
: ,
: ,
})
.packet_count =
.dnp3_count =
.authorized_masters: [] = ()
.authorized_pairs: [, []] = defaultdict()
.baseline_functions: [, []] = defaultdict()
baseline_file:
.load_baseline(baseline_file)
():
(filepath, ) f:
baseline = json.load(f)
entry baseline.get(, []):
master = entry[]
outstation = entry[]
.authorized_masters.add(master)
.authorized_pairs[master].add(outstation)
.baseline_functions[] = (
entry.get(, [, ])
)
() -> []:
(payload) < :
start_bytes = struct.unpack(, payload[:])[]
start_bytes != :
length = payload[]
control = payload[]
dest_addr = struct.unpack(, payload[:])[]
source_addr = struct.unpack(, payload[:])[]
direction = (control & )
result = {
: length,
: control,
: direction,
: dest_addr,
: source_addr,
: (control & ),
}
(payload) >= :
transport_header = payload[]
(payload) >= :
app_control = payload[]
func_code = payload[]
result[] = func_code
result[] = DNP3_FUNCTIONS.get(
func_code,
)
result
():
.packet_count +=
pkt.haslayer(IP) pkt.haslayer(TCP):
tcp = pkt[TCP]
tcp.dport != tcp.sport != :
payload = (tcp.payload)
payload:
dnp3 = .parse_dnp3_header(payload)
dnp3:
.dnp3_count +=
src_ip = pkt[IP].src
dst_ip = pkt[IP].dst
session_key =
session = .sessions[session_key]
session[] +=
func_code = dnp3.get()
func_code :
session[][func_code] +=
dnp3.get() .authorized_masters:
src_ip .authorized_masters:
.alerts.append({
: ,
: ,
: src_ip, : dst_ip,
: dnp3.get(),
: ,
: ,
})
func_code (, ):
session[] +=
restart_type = func_code ==
.alerts.append({
: ,
: ,
: src_ip, : dst_ip,
: ,
: ,
: ,
})
func_code (, , , , , ):
session[] +=
.alerts.append({
: ,
: ,
: src_ip, : dst_ip,
: dnp3.get(),
: ,
: ,
})
func_code (, , , ):
session[] +=
session_key .baseline_functions:
func_code .baseline_functions[session_key]:
.alerts.append({
: ,
: ,
: src_ip, : dst_ip,
: dnp3.get(),
: ,
: ,
})
session_key .baseline_functions:
func_code .baseline_functions[session_key]:
func_code (, , ):
.alerts.append({
: ,
: ,
: src_ip, : dst_ip,
: dnp3.get(),
: ,
: ,
})
():
()
()
()
()
()
()
()
()
key, session .sessions.items():
()
()
funcs = [DNP3_FUNCTIONS.get(f, ) f session[]]
()
()
()
()
.alerts:
()
alert .alerts:
()
()
()
()
()
__name__ == :
detector = DNP3AnomalyDetector(
baseline_file=sys.argv[] (sys.argv) >
)
(sys.argv) >= :
()
packets = rdpcap(sys.argv[])
pkt packets:
detector.analyze_packet(pkt)
detector.generate_report()
:
()
| 术语 | 定义 |
|---|---|
| DNP3 | 分布式网络协议版本3，能源行业SCADA系统中主站与外站通信的主要协议 |
| 外站（Outstation） | DNP3从站设备（通常为RTU或IED），响应主站的轮询和命令 |
| 先选后操（Select-Before-Operate） | DNP3安全机制，要求在操作前先发送选择命令，防止意外控制操作 |
| 冷重启（Cold Restart，FC 0x0D） | 完全重启外站的DNP3命令，重置所有配置——高风险的拒绝服务操作 |
| DNP3安全认证（DNP3 Secure Authentication） | 可选的DNP3扩展（SA v5），添加基于HMAC的认证以防止命令欺骗 |
| PIPEDREAM | 具有DNP3功能的ICS攻击框架，用于操控外站和执行固件更新 |
DNP3异常检测报告
================================
分析周期: [开始] 至 [结束]
监控点: [变电站/网段]
流量摘要:
DNP3数据包: [数量]
唯一主站-外站对: [数量]
控制命令: [数量]
文件操作: [数量]
告警:
[严重] 未授权DNP3主站 [IP]
[严重] 向外站 [地址] 发送冷重启命令
[高] 来自 [IP] 的意外控制命令
建议:
1. 部署DNP3安全认证（SA v5）
2. 在防火墙阻断未授权来源
3. 在工业防火墙启用DNP3深度包检测