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.
Installer avec Codex ou Claude Copiez ce prompt, collez-le dans Codex, Claude ou un autre assistant, puis laissez-le vérifier la page du skill et l'installer pour vous.
Une commande directe contourne le prompt de vérification. Examinez la source avant de l'exécuter.
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.
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.
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 workstationalertmodbusanyany->$OT_PLC_SUBNET502(msg:"OT-DETECTModbuswritefromunauthorizedsource";modbus_func:!read_coils;modbus_func:!read_discrete_inputs;modbus_func:!read_holding_registers;modbus_func:!read_input_registers;flow:to_server,established;threshold:typeboth,trackby_src,count1,seconds60;classtype:attempted-admin;sid:3000001;rev:1;)# Modbus diagnostic/restart command (FC 8) - potential PLC DoSalertmodbusanyany->$OT_PLC_SUBNET502(msg:"OT-DETECTModbusdiagnosticscommandtoPLC";modbus_func:diagnostics;flow:to_server,established;classtype:attempted-dos;sid:3000002;rev:1;)# Modbus broadcast write (unit ID 0) - affects all slavesalertmodbusanyany->$OT_PLC_SUBNET502(msg:"OT-CRITICALModbusbroadcastwritecommand";modbus_unit_id:0;flow:to_server,established;classtype:attempted-admin;sid:3000003;rev:1;priority:1;)# --- S7comm Attack Detection (Siemens) ---# S7comm CPU STOP command - shuts down PLC executionalerttcpanyany->$SIEMENS_PLC_SUBNET102(msg:"OT-CRITICALS7commCPUSTOPcommanddetected";content:"|0300|";offset:0;depth:2;content:"|29|";offset:17;depth:1;flow:to_server,established;classtype:attempted-dos;sid:3000010;rev:1;priority:1;)# S7comm PLC program upload (potential logic modification)alerttcpanyany->$SIEMENS_PLC_SUBNET102(msg:"OT-CRITICALS7commprogramdownloadtoPLC";content:"|0300|";offset:0;depth:2;content:"|1a|";offset:17;depth:1;flow:to_server,established;classtype:attempted-admin;sid:3000011;rev:1;priority:1;)# --- DNP3 Attack Detection ---# DNP3 cold restart commandalerttcpanyany->$OT_RTU_SUBNET20000(msg:"OT-CRITICALDNP3coldrestartcommand";content:"|0564|";offset:0;depth:2;content:"|0d|";offset:12;depth:1;flow:to_server,established;classtype:attempted-dos;sid:3000020;rev:1;priority:1;)# DNP3 firmware update command - potential PIPEDREAM indicatoralerttcpanyany->$OT_RTU_SUBNET20000(msg:"OT-CRITICALDNP3filetransfer/firmwareupdate";content:"|0564|";offset:0;depth:2;content:"|19|";offset:12;depth:1;flow:to_server,established;classtype:attempted-admin;sid:3000021;rev:1;priority:1;)# --- Network Anomaly Detection ---# New device communicating with PLCs (not in baseline)alertip!$AUTHORIZED_OT_HOSTSany->$OT_PLC_SUBNETany(msg:"OT-DETECTUnauthorizeddevicecommunicatingwithPLCsubnet";flow:to_server;threshold:typelimit,trackby_src,count1,seconds3600;classtype:network-scan;sid:3000030;rev:1;)# Port scan targeting OT protocolsalerttcpanyany->$OT_NETWORKany(msg:"OT-DETECTPortscantargetingindustrialprotocols";flags:S;threshold:typethreshold,trackby_src,count10,seconds60;classtype:network-scan;sid:3000031;rev:1;)
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 importOptionaltry:
import requests
except ImportError:
print("Install requests: pip install requests")
sys.exit(1)
@dataclassclassProcessVariable:
"""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@dataclassclassAnomaly:
"""Represents a detected process anomaly."""
timestamp: str
tag_name: str
anomaly_type: str
severity: str
current_value: float
expected_range: str
description: str
attack_pattern: str = ""classProcessAnomalyDetector:
"""Detects anomalies in SCADA process data from historian."""def__init__(self, historian_url, api_key=None):
self.historian_url = historian_url
self.api_key = api_key
self.variables = {}
self.history = defaultdict(lambda: deque(maxlen=1000))
self.anomalies = []
defadd_variable(self, var: ProcessVariable):
"""Register a process variable to monitor."""self.variables[var.tag_name] = var
deffetch_current_values(self):
"""Fetch current values from historian API."""
headers = {}
ifself.api_key:
headers["Authorization"] = f"Bearer {self.api_key}"
tag_list = list(self.variables.keys())
params = {"tags": ",".join(tag_list), "count": 1}
try:
resp = requests.get(
f"{self.historian_url}/api/v1/streams/values/current",
params=params,
headers=headers,
timeout=10,
verify=not os.environ.get("SKIP_TLS_VERIFY", "").lower() == "true", # Set SKIP_TLS_VERIFY=true for self-signed certs in lab environments
)
resp.raise_for_status()
return resp.json()
except requests.RequestException as e:
print(f"[ERROR] Historian API error: {e}")
return {}
defcheck_value(self, tag_name, value, timestamp):
"""Check a process variable value against all detection rules."""
var = self.variables.get(tag_name)
ifnot var:
returnself.history[tag_name].append((timestamp, value))
# Rule 1: Value out of engineering limitsif value < var.engineering_low or value > var.engineering_high:
self.anomalies.append(Anomaly(
timestamp=timestamp,
tag_name=tag_name,
anomaly_type="OUT_OF_RANGE",
severity="critical",
current_value=value,
expected_range=f"{var.engineering_low}-{var.engineering_high}{var.unit}",
description=f"{tag_name} ({var.description}) at {value}{var.unit} - outside engineering limits",
attack_pattern="Process manipulation - value driven outside safe operating range",
))
# Rule 2: Rate of change exceeds physical limits
history = list(self.history[tag_name])
iflen(history) >= 2:
prev_ts, prev_val = history[-2]
try:
dt = (datetime.fromisoformat(timestamp) - datetime.fromisoformat(prev_ts)).total_seconds()
if dt > 0:
rate = abs(value - prev_val) / dt
if rate > var.rate_of_change_limit:
self.anomalies.append(Anomaly(
timestamp=timestamp,
tag_name=tag_name,
anomaly_type="RATE_OF_CHANGE_VIOLATION",
severity="high",
current_value=value,
expected_range=f"Max rate: {var.rate_of_change_limit}{var.unit}/s",
description=(
f"{tag_name} changing at {rate:.2f}{var.unit}/s "f"(limit: {var.rate_of_change_limit}{var.unit}/s)"
),
attack_pattern="Possible sensor spoofing or actuator manipulation",
))
except (ValueError, TypeError):
pass# Rule 3: Flatline detection (sensor reading not changing when process is active)iflen(history) >= 20:
recent_values = [v for _, v inlist(history)[-20:]]
iflen(set(recent_values)) == 1:
self.anomalies.append(Anomaly(
timestamp=timestamp,
tag_name=tag_name,
anomaly_type="FLATLINE_DETECTED",
severity="high",
current_value=value,
expected_range="Expected variation during active process",
description=f"{tag_name} flatlined at {value} for 20+ consecutive readings",
attack_pattern="Stuxnet-style replay attack - frozen sensor value while process is manipulated",
))
# Rule 4: Statistical anomaly (z-score based)iflen(history) >= 50:
values = [v for _, v inlist(history)[-50:]]
avg = mean(values)
std = stdev(values) iflen(values) > 1else0if std > 0:
z_score = abs(value - avg) / std
if z_score > 3.5:
self.anomalies.append(Anomaly(
timestamp=timestamp,
tag_name=tag_name,
anomaly_type="STATISTICAL_ANOMALY",
severity="medium",
current_value=value,
expected_range=f"Mean: {avg:.2f}, StdDev: {std:.2f} (z={z_score:.1f})",
description=f"{tag_name} value {value} is {z_score:.1f} standard deviations from mean",
attack_pattern="Possible gradual process manipulation",
))
defreport_anomalies(self):
"""Print detected anomalies."""ifnotself.anomalies:
print("[*] No anomalies detected")
returnprint(f"\n{'='*70}")
print(f"PROCESS ANOMALY DETECTION REPORT - {len(self.anomalies)} anomalies")
print(f"{'='*70}")
for a inself.anomalies:
print(f"\n [{a.severity.upper()}] {a.anomaly_type}")
print(f" Time: {a.timestamp}")
print(f" Tag: {a.tag_name}")
print(f" Value: {a.current_value}")
print(f" Expected: {a.expected_range}")
print(f" Detail: {a.description}")
if a.attack_pattern:
print(f" Attack Pattern: {a.attack_pattern}")
if __name__ == "__main__":
from collections import defaultdict
detector = ProcessAnomalyDetector(
historian_url="https://10.30.1.50:5450",
)
# Define monitored process variables for a chemical reactor
detector.add_variable(ProcessVariable(
tag_name="REACTOR_01.TEMP",
description="Reactor 1 Temperature",
unit="C",
low_limit=150, high_limit=280,
rate_of_change_limit=5.0,
engineering_low=100, engineering_high=350,
))
detector.add_variable(ProcessVariable(
tag_name="REACTOR_01.PRESSURE",
description="Reactor 1 Pressure",
unit="bar",
low_limit=2.0, high_limit=8.0,
rate_of_change_limit=0.5,
engineering_low=0, engineering_high=12.0,
))
detector.add_variable(ProcessVariable(
tag_name="PUMP_03.FLOW",
description="Feed Pump 3 Flow Rate",
unit="m3/h",
low_limit=5.0, high_limit=25.0,
rate_of_change_limit=2.0,
engineering_low=0, engineering_high=30.0,
))
print("[*] Starting process anomaly monitoring...")
print("[*] Press Ctrl+C to stop and generate report")
try:
whileTrue:
data = detector.fetch_current_values()
for item in data.get("items", []):
detector.check_value(
item.get("tag"),
item.get("value"),
item.get("timestamp", datetime.now().isoformat()),
)
time.sleep(5)
except 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 advisoriesmalware_families:TRITON_TRISIS:description:"Targets Schneider Electric Triconex Safety Instrumented Systems"target:"Safety controllers (SIS)"network_indicators:-protocol:"TriStation"port:1502pattern:"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:2404pattern:"Rapid sequence of control commands outside normal polling"-protocol:"OPC DA"pattern:"Enumeration of OPC servers followed by write commands"host_indicators:-"haslo.exe (backdoor launcher)"-"61850.dll (IEC 61850 attack module)"-"OPC.dll (OPC DA attack module)"-"104.dll (IEC 104 attack module)"detection_rule:|
alert tcp any any -> $SUBSTATION_RTU 2404 (
msg:"OT-CRITICAL Rapid IEC 104 control commands - Industroyer pattern";
flow:to_server,established;
threshold:type threshold, track by_src, count 50, seconds 10;
sid:3000110; rev:1; priority:1;)
PIPEDREAM_INCONTROLLER:description:"Modular ICS attack framework targeting Schneider/OMRON PLCs and OPC UA"target:"Multiple PLC vendors (Schneider, OMRON) and OPC UA servers"network_indicators:-protocol:"CODESYS"port:1217pattern:"CODESYS runtime exploitation attempts"-protocol:"OPC UA"port:4840pattern:"OPC UA server enumeration and unauthorized method calls"-protocol:"Modbus"port:502pattern:"Rapid Modbus write commands to multiple unit IDs"host_indicators:-"TAGRUN tool for OPC UA scanning"-"CODECALL tool for CODESYS exploitation"-"OMSHELL tool for OMRON PLC interaction"detection_rule:|
alert tcp any any -> $OT_NETWORK 1217 (
msg:"OT-CRITICAL CODESYS runtime connection - PIPEDREAM indicator";
flow:to_server,established;
sid:3000120; rev:1; priority:1;)
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.
