detecting-attacks-on-scada-systems

This skill covers detecting cyber attacks targeting Supervisory Control and Data Acquisition (SCADA) systems including man-in-the-middle attacks on industrial protocols, unauthorized command injection into PLCs, HMI compromise, historian data manipulation, and denial-of-service against control system communications. It leverages OT-specific intrusion detection systems, industrial protocol anomaly detection, and process data analytics to identify attacks that traditional IT security tools miss.

跳到安装

仓库: mukul975/Anthropic-Cybersecurity-Skills
最近来源活动: 2026年6月1日 10:13
检测到的 SKILL.md 语言: 英语
星标: 27,732
分支: 3,366

安装方式

默认使用会先检查来源的 Prompt；你也可以切换为直接命令，或下载本地副本。

用 Codex 或 Claude 帮你安装复制这段 Prompt，粘贴到 Codex、Claude 或其他助手里，让它检查 Skill 页面并帮你完成安装。

直接命令不会经过审查 Prompt；运行前请先检查来源。

npx skills add https://github.com/mukul975/Anthropic-Cybersecurity-Skills --skill detecting-attacks-on-scada-systems

命令会保持在同一行。复制前请横向滚动并检查完整内容。

想先保存到本地？可下载 SkillsMP 当前能够提供的文件。

检查来源文件

决定是否安装前，请先阅读 SKILL.md，以及 SkillsMP 当前展示的配套文件。

文件资源管理器

4 个文件

SKILL.md

来源说明 · 只读预览

返回搜索返回安装

abusing-dpapi-for-credential-access

Extract and decrypt Windows DPAPI-protected secrets (Credential Manager, browser logins/cookies, Wi-Fi credentials, KeePass keys) online or offline using SharpDPAPI, SharpChrome, Mimikatz, or Impacket's dpapi.py, including domain-wide decryption via the DPAPI backup key. Use during authorized red-team credential-access engagements after gaining a foothold or when triaging DPAPI blobs pulled from a host.

2026-08-02

abusing-shadow-credentials-for-privesc

Take over Active Directory accounts by writing attacker-controlled public keys to msDS-KeyCredentialLink (Shadow Credentials) with pyWhisker, Whisker, or Certipy, then authenticate via PKINIT to recover the target's NT hash without a password reset. Use when BloodHound shows GenericWrite/GenericAll/AddKeyCredentialLink over a target, as a stealthier alternative to ForceChangePassword, during authorized red-team engagements.

2026-08-02

acquiring-disk-image-with-dd-and-dcfldd

Create forensically sound bit-for-bit disk images with dd or dcfldd on a Linux forensic workstation, preserving evidence integrity through hash verification (MD5/SHA) during acquisition. Use when imaging a suspect drive, USB device, or memory card for investigation, preserving volatile disk evidence during incident response, or producing a verified copy for legal or law-enforcement proceedings before any destructive analysis.

2026-08-02

查看这个仓库的全部 Skills

Detecting Attacks on SCADA Systems

When to Use

When deploying intrusion detection capabilities in a SCADA environment for the first time
When investigating suspected cyber attacks against industrial control systems
When building detection rules for OT-specific attack patterns (Stuxnet, TRITON, Industroyer)
When integrating OT network monitoring with an enterprise SOC for unified threat visibility
When responding to alerts from OT security monitoring tools (Dragos, Nozomi, Claroty)

Do not use for detecting attacks on IT-only networks without SCADA/ICS components, for building generic network IDS rules (see building-detection-rules-with-sigma), or for incident response procedures after an attack is confirmed (see performing-ot-incident-response).

Prerequisites

Passive network monitoring sensors deployed on SPAN/TAP ports at OT network boundaries
OT intrusion detection system (Dragos Platform, Nozomi Guardian, Claroty xDome, or Suricata with OT rulesets)
Understanding of industrial protocols in use (Modbus, DNP3, OPC UA, EtherNet/IP, S7comm)
Baseline of normal SCADA communication patterns (polling intervals, function codes, register ranges)
Access to process historian data for physical process anomaly correlation

Workflow

Step 1: Establish SCADA Communication Baselines

Before detecting anomalies, establish what normal SCADA traffic looks like. Industrial protocols are highly deterministic - the same master polls the same slaves at the same intervals reading the same registers.

#!/usr/bin/env python3
"""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."""

    def __init__(self):
        self.modbus_sessions = defaultdict(lambda: {
            "func_codes": defaultdict(int),
            "register_ranges": set(),
            "intervals": [],
            "last_seen": None,
            "request_count": 0,
        })
        self.communication_pairs = defaultdict(lambda: {
            "protocols": set(),
            "packet_count": 0,
            "first_seen": None,
            "last_seen": None,
        })

    def process_pcap(self, pcap_file):
        """Process pcap file to build SCADA baselines."""
        packets = rdpcap(pcap_file)
        print(f"[*] Processing {len(packets)} packets for baseline...")

        for pkt in packets:
            if not pkt.haslayer(IP):
                continue

            src = pkt[IP].src
            dst = pkt[IP].dst
            ts = float(pkt.time)

            # Track communication pairs
            pair_key = f"{src}->{dst}"
            pair = self.communication_pairs[pair_key]
            pair["packet_count"] += 1
            if pair["first_seen"] is None:
                pair["first_seen"] = ts
            pair["last_seen"] = ts

            # Analyze Modbus/TCP
            if pkt.haslayer(TCP) and pkt[TCP].dport == 502:
                self._analyze_modbus(pkt, src, dst, ts)

    def _analyze_modbus(self, pkt, src, dst, timestamp):
        """Extract Modbus function codes and register ranges."""
        payload = bytes(pkt[TCP].payload)
        if len(payload) < 8:
            return

        # MBAP header: transaction_id(2) + protocol_id(2) + length(2) + unit_id(1) + func_code(1)
        func_code = payload[7]
        session_key = f"{src}->{dst}"
        session = self.modbus_sessions[session_key]

        session["func_codes"][func_code] += 1
        session["request_count"] += 1
        session["protocols"] = {"Modbus/TCP"}

        # Track polling intervals
        if session["last_seen"] is not None:
            interval = timestamp - session["last_seen"]
            if 0.01 < interval < 60:  # Reasonable polling interval
                session["intervals"].append(interval)
        session["last_seen"] = timestamp

        # Extract register range for read/write operations
        if len(payload) >= 12 and func_code in (1, 2, 3, 4, 5, 6, 15, 16):
            start_register = (payload[8] << 8) | payload[9]
            if func_code in (1, 2, 3, 4, 15, 16) and len(payload) >= 12:
                count = (payload[10] << 8) | payload[11]
                session["register_ranges"].add((func_code, start_register, start_register + count))

    def generate_baseline(self):
        """Generate the baseline profile from collected data."""
        baseline = {
            "generated": datetime.now().isoformat(),
            "modbus_baselines": {},
            "communication_pairs": {},
        }

        for session_key, session in self.modbus_sessions.items():
            avg_interval = mean(session["intervals"]) if session["intervals"] else 0
            interval_std = stdev(session["intervals"]) if len(session["intervals"]) > 1 else 0

            baseline["modbus_baselines"][session_key] = {
                "allowed_function_codes": list(session["func_codes"].keys()),
                "function_code_distribution": {
                    MODBUS_FUNC_NAMES.get(k, f"FC{k}"): v
                    for k, v in session["func_codes"].items()
                },
                "polling_interval_avg_sec": round(avg_interval, 3),
                "polling_interval_stddev": round(interval_std, 3),
                "register_ranges": [
                    {"func_code": r[0], "start": r[1], "end": r[2]}
                    for r in session["register_ranges"]
                ],
                "total_requests": session["request_count"],
            }

        return baseline

    def export_baseline(self, output_file):
        """Export baseline to JSON file."""
        baseline = self.generate_baseline()
        with open(output_file, "w") as f:
            json.dump(baseline, f, indent=2)
        print(f"[*] Baseline saved to: {output_file}")

        # Print summary
        print(f"\n{'='*60}")
        print("SCADA COMMUNICATION BASELINE SUMMARY")
        print(f"{'='*60}")
        for session, data in baseline["modbus_baselines"].items():
            print(f"\n  Session: {session}")
            print(f"    Function Codes: {data['allowed_function_codes']}")
            print(f"    Polling Interval: {data['polling_interval_avg_sec']}s (+/- {data['polling_interval_stddev']}s)")
            print(f"    Register Ranges: {len(data['register_ranges'])}")
            print(f"    Total Requests: {data['total_requests']}")


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python scada_baseline.py <pcap_file> [output.json]")
        sys.exit(1)

    builder = SCADABaselineBuilder()
    builder.process_pcap(sys.argv[1])
    output = sys.argv[2] if len(sys.argv) > 2 else "scada_baseline.json"
    builder.export_baseline(output)

Step 2: Deploy OT-Specific Detection Rules

Create detection rules for known SCADA attack patterns including those used by TRITON, Industroyer/CrashOverride, and PIPEDREAM/INCONTROLLER.

# Suricata Rules for SCADA Attack Detection
# Deploy on IDS sensor monitoring OT network SPAN port

# --- Modbus Attack Detection ---

# Unauthorized Modbus write to PLC from non-engineering workstation
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;
)

# Modbus diagnostic/restart command (FC 8) - potential PLC DoS
alert modbus any any -> $OT_PLC_SUBNET 502 (
  msg:"OT-DETECT Modbus diagnostics command to PLC";

Step 3: Implement Process Data Anomaly Detection

Monitor physical process data from the historian to detect attacks that manipulate the process while hiding their effects from operators (the Stuxnet attack pattern).

#!/usr/bin/env python3
"""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  # Maximum change per second
    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= os.environ.get(, ).lower() == ,  
            )
            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()

Step 4: Detect Known ICS Malware Indicators

Monitor for indicators of compromise (IOCs) associated with known ICS-targeting malware families.

# Known ICS Malware Detection Signatures
# Reference: MITRE ATT&CK for ICS, CISA ICS-CERT advisories

malware_families:
  TRITON_TRISIS:
    description: "Targets Schneider Electric Triconex Safety Instrumented Systems"
    target: "Safety controllers (SIS)"
    network_indicators:
      - protocol: "TriStation"
        port: 1502
        pattern: "Unusual TriStation commands from non-engineering workstation"
      - protocol: "TCP"
        pattern: "Connection to Triconex controller from unauthorized IP"
    host_indicators:
      - "trilog.exe present on engineering workstation"
      - "inject.bin in System32 directory"
      - "imain.bin payload targeting Triconex firmware"
    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: "Targets power grid SCADA via IEC 60870-5-101/104, IEC 61850, OPC DA"
    target: "Power grid substations and SCADA"
    network_indicators:
      - protocol: "IEC 60870-5-104"
        port: 2404
        pattern: "Rapid sequence of control commands outside normal polling"

Key Concepts

Term	Definition
SCADA	Supervisory Control and Data Acquisition - architecture for remote monitoring and control of industrial processes via RTUs and communication infrastructure
IDS/IPS for OT	Intrusion Detection/Prevention Systems designed for industrial protocols, using both signature-based and anomaly-based detection methods
Process Anomaly	Deviation in physical process behavior (temperature, pressure, flow) that may indicate cyber manipulation of control systems
Man-in-the-Middle (MITM)	Attack intercepting communication between SCADA master and field devices to modify commands or spoof sensor readings
Replay Attack	Capturing legitimate SCADA traffic and replaying it to mask malicious changes to the process (used by Stuxnet)
Protocol Anomaly	Deviation from expected industrial protocol behavior including unauthorized function codes, unusual polling patterns, or command sequences

Tools & Systems

Dragos Platform: OT cybersecurity platform with threat detection powered by Dragos threat intelligence on ICS-targeting activity groups
Nozomi Networks Guardian: OT/IoT visibility and threat detection using asset intelligence, anomaly detection, and vulnerability assessment
Claroty xDome: Cyber-physical systems protection with continuous threat monitoring and alert prioritization
Suricata with ET Open ICS rules: Open-source IDS/IPS with community-maintained rules for industrial protocol detection
Zeek (Bro) with OT scripts: Network security monitor with protocol analyzers for Modbus, DNP3, and BACnet

Common Scenarios

Scenario: Detecting TRITON-Style Attack on Safety Systems

Context: An OT security monitoring system alerts on unusual TriStation protocol traffic to a Triconex safety controller from an IP address that is not the authorized SIS engineering workstation.

Approach:

Immediately verify the source IP of the TriStation traffic - is it the authorized SIS engineering workstation or a compromised host?
Check if there is an authorized maintenance activity scheduled for the SIS controllers
Capture full packet payload of the TriStation communication for forensic analysis
Alert the process safety team - SIS compromise is a safety-critical event
If unauthorized, isolate the source host from the network immediately
Verify SIS controller logic integrity by comparing running logic against known-good backup
Check all engineering workstations in the facility for TRITON indicators (trilog.exe, inject.bin)

Pitfalls: Never assume SIS traffic anomalies are false positives - TRITON demonstrated that sophisticated attackers specifically target safety systems. Do not restart the SIS controller without first verifying firmware and logic integrity. Avoid alerting only the IT SOC; the process safety team must be immediately engaged for any SIS-related incident.

Output Format

SCADA Attack Detection Report
===============================
Detection Time: YYYY-MM-DD HH:MM:SS UTC
Detection Source: [IDS/Anomaly Detector/Process Monitor]

ALERT DETAILS:
  Alert ID: [unique identifier]
  Severity: Critical/High/Medium/Low
  Attack Category: [Protocol Anomaly/Process Manipulation/Unauthorized Access]
  MITRE ATT&CK for ICS: [Technique ID and name]

  Source: [IP/hostname]
  Target: [IP/hostname - device type]
  Protocol: [Modbus/DNP3/S7comm/etc]
  Detail: [Specific finding description]

BASELINE COMPARISON:
  Normal: [Expected behavior]
  Observed: [Actual behavior that triggered alert]
  Deviation: [How the observed differs from baseline]

RECOMMENDED RESPONSE:
  1. [Immediate containment action]
  2. [Verification step]
  3. [Escalation path]

name	detecting-attacks-on-scada-systems
description	This skill covers detecting cyber attacks targeting Supervisory Control and Data Acquisition (SCADA) systems including man-in-the-middle attacks on industrial protocols, unauthorized command injection into PLCs, HMI compromise, historian data manipulation, and denial-of-service against control system communications. It leverages OT-specific intrusion detection systems, industrial protocol anomaly detection, and process data analytics to identify attacks that traditional IT security tools miss.
domain	cybersecurity
subdomain	ot-ics-security
tags	["ot-security","ics","scada","industrial-control","iec62443","intrusion-detection","threat-detection"]
version	1.0.0
author	mahipal

detecting-attacks-on-scada-systems

来源信息

安装方式

检查来源文件

同仓库更多 Skills

同仓库更多 Skills

Detecting Attacks on SCADA Systems

When to Use

Prerequisites

Workflow

Step 1: Establish SCADA Communication Baselines

Step 2: Deploy OT-Specific Detection Rules

Step 3: Implement Process Data Anomaly Detection

Step 4: Detect Known ICS Malware Indicators

Key Concepts

Tools & Systems

Common Scenarios

Scenario: Detecting TRITON-Style Attack on Safety Systems

Output Format