القائمة
通过使用ICS感知IDS和协议深度包检测，监控未授权写入操作、异常功能码、格式错误的帧以及与已建立通信基线的偏差，检测ICS环境中针对Modbus TCP/RTU协议的命令注入攻击。
يُحدَّد Prompt الذي يراجع المصدر أولًا بشكل افتراضي. يمكنك التبديل إلى أمر مباشر أو تنزيل نسخة محلية.
اقرأ SKILL.md وأي ملفات مرافقة يعرضها SkillsMP قبل أن تقرر التثبيت.
التثبيت باستخدام Codex أو Claude انسخ هذا Prompt والصقه في Codex أو Claude أو مساعد آخر ليراجع صفحة Skill ويثبّتها لك.
يتجاوز الأمر المباشر Prompt المخصّص للمراجعة. افحص المصدر قبل تشغيله.
npx skills add https://github.com/killvxk/cybersecurity-skills-zh --skill detecting-modbus-command-injection-attacksيبقى الأمر في سطر واحد. مرّر أفقيًا لمراجعته كاملًا قبل النسخ.
تفضّل نسخة محلية؟ نزّل الملفات المتاحة حاليًا لدى 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-modbus-command-injection-attacks |
| description | 通过使用ICS感知IDS和协议深度包检测，监控未授权写入操作、异常功能码、格式错误的帧以及与已建立通信基线的偏差，检测ICS环境中针对Modbus TCP/RTU协议的命令注入攻击。 |
| domain | cybersecurity |
| subdomain | ot-ics-security |
| tags | ["ot-security","ics","modbus","command-injection","protocol-analysis","ids","scada","threat-detection"] |
| version | 1.0 |
| author | mahipal |
| license | Apache-2.0 |
不适用于检测非Modbus协议的攻击（DNP3相关参见detecting-dnp3-protocol-anomalies）、一般IT网络入侵检测，或Modbus设备配置（参见performing-ot-vulnerability-scanning-safely）。
捕获并分析正常Modbus流量，确定合法通信模式的构成。
#!/usr/bin/env python3
"""Modbus Command Injection Detector.
Monitors Modbus TCP traffic for unauthorized write operations, anomalous
function codes, and deviations from established communication baselines.
Detects attacks like FrostyGoop that use Modbus TCP for operational impact.
"""
import json
import struct
import sys
import time
from collections import defaultdict
from datetime import datetime
from typing import Dict, List, Optional, Set, Tuple
try:
from scapy.all import sniff, IP, TCP
except ImportError:
print("Install scapy: pip install scapy")
sys.exit(1)
# Modbus功能码定义
MODBUS_READ_FUNCTIONS = {1, 2, 3, 4}
MODBUS_WRITE_FUNCTIONS = {5, 6, 15, 16}
MODBUS_DIAGNOSTIC_FUNCTIONS = {8, 17, 43}
MODBUS_FUNC_NAMES = {
1: "Read Coils", 2: "Read Discrete Inputs",
3: "Read Holding Registers", 4: "Read Input Registers",
5: "Write Single Coil", 6: ,
: , : ,
: , : ,
: , : ,
: ,
}
:
():
.timestamp = datetime.now().isoformat()
.severity = severity
.alert_type = alert_type
.src_ip = src_ip
.dst_ip = dst_ip
.unit_id = unit_id
.func_code = func_code
.func_name = MODBUS_FUNC_NAMES.get(func_code, )
.description = description
.mitre_technique = mitre_technique
():
(
)
:
():
.alerts: [ModbusAlert] = []
.packet_count =
.modbus_count =
.authorized_masters: [] = ()
.authorized_pairs: [[, ]] = ()
.allowed_write_sources: [] = ()
.allowed_function_codes: [, []] = defaultdict()
.allowed_register_ranges: [, [[, ]]] = defaultdict()
.polling_intervals: [, ] = {}
.last_seen: [, ] = {}
.write_counts: [, []] = defaultdict()
baseline_file:
.load_baseline(baseline_file)
():
(filepath, ) f:
baseline = json.load(f)
session_key, data baseline.get(, {}).items():
src, dst = session_key.split()
.authorized_pairs.add((src.strip(), dst.strip()))
.authorized_masters.add(src.strip())
fc_set = (data.get(, []))
.allowed_function_codes[session_key] = fc_set
fc_set & MODBUS_WRITE_FUNCTIONS:
.allowed_write_sources.add(src.strip())
reg_range data.get(, []):
.allowed_register_ranges[session_key].append(
(reg_range[], reg_range[])
)
data.get():
.polling_intervals[session_key] = data[]
(
)
() -> []:
(payload) < :
transaction_id = struct.unpack(, payload[:])[]
protocol_id = struct.unpack(, payload[:])[]
length = struct.unpack(, payload[:])[]
unit_id = payload[]
func_code = payload[]
protocol_id != :
result = {
: transaction_id,
: protocol_id,
: length,
: unit_id,
: func_code,
}
(payload) >= func_code (, , , , , , , ):
result[] = struct.unpack(, payload[:])[]
result[] = struct.unpack(, payload[:])[]
result
():
.packet_count +=
pkt.haslayer(IP) pkt.haslayer(TCP):
tcp = pkt[TCP]
tcp.dport != tcp.sport != :
payload = (tcp.payload)
payload:
modbus = .parse_modbus_mbap(payload)
modbus:
.modbus_count +=
src_ip = pkt[IP].src
dst_ip = pkt[IP].dst
session_key =
now = time.time()
.authorized_masters src_ip .authorized_masters:
tcp.dport == :
.alerts.append(ModbusAlert(
severity=,
alert_type=,
src_ip=src_ip, dst_ip=dst_ip,
unit_id=modbus[],
func_code=modbus[],
description=,
mitre_technique=,
))
modbus[] MODBUS_WRITE_FUNCTIONS:
.allowed_write_sources src_ip .allowed_write_sources:
.alerts.append(ModbusAlert(
severity=,
alert_type=,
src_ip=src_ip, dst_ip=dst_ip,
unit_id=modbus[],
func_code=modbus[],
description=,
mitre_technique=,
))
.write_counts[src_ip].append(now)
recent_writes = [t t .write_counts[src_ip] now - t < ]
.write_counts[src_ip] = recent_writes
(recent_writes) > :
.alerts.append(ModbusAlert(
severity=,
alert_type=,
src_ip=src_ip, dst_ip=dst_ip,
unit_id=modbus[],
func_code=modbus[],
description=,
mitre_technique=,
))
session_key .allowed_function_codes:
modbus[] .allowed_function_codes[session_key]:
.alerts.append(ModbusAlert(
severity=,
alert_type=,
src_ip=src_ip, dst_ip=dst_ip,
unit_id=modbus[],
func_code=modbus[],
description=(
),
mitre_technique=,
))
modbus[] == modbus[] MODBUS_WRITE_FUNCTIONS:
.alerts.append(ModbusAlert(
severity=,
alert_type=,
src_ip=src_ip, dst_ip=dst_ip,
unit_id=,
func_code=modbus[],
description=,
mitre_technique=,
))
modbus session_key .allowed_register_ranges:
addr = modbus[]
qty = modbus.get(, )
in_range = (
start <= addr addr + qty <= end
start, end .allowed_register_ranges[session_key]
)
in_range:
.alerts.append(ModbusAlert(
severity=,
alert_type=,
src_ip=src_ip, dst_ip=dst_ip,
unit_id=modbus[],
func_code=modbus[],
description=,
mitre_technique=,
))
modbus[] MODBUS_DIAGNOSTIC_FUNCTIONS:
.alerts.append(ModbusAlert(
severity=,
alert_type=,
src_ip=src_ip, dst_ip=dst_ip,
unit_id=modbus[],
func_code=modbus[],
description=,
mitre_technique=,
))
():
()
()
()
()
()
()
()
.alerts:
severity_counts = defaultdict()
alert .alerts:
severity_counts[alert.severity] +=
()
sev [, , , ]:
sev severity_counts:
()
()
alert .alerts:
()
()
()
()
()
()
alert.mitre_technique:
()
():
()
()
:
sniff(
iface=interface,
=,
prn=.analyze_packet,
timeout=duration duration > ,
)
KeyboardInterrupt:
.print_report()
__name__ == :
detector = ModbusInjectionDetector(
baseline_file=sys.argv[] (sys.argv) >
)
(sys.argv) >= :
sys.argv[].endswith() sys.argv[].endswith():
scapy. rdpcap
()
packets = rdpcap(sys.argv[])
pkt packets:
detector.analyze_packet(pkt)
detector.print_report()
:
detector.start_live_monitoring(sys.argv[])
:
()
()
()