detecting-stuxnet-style-attacks

What this skill does

# Detecting Stuxnet-Style Attacks

## When to Use

- When implementing advanced threat detection for high-value OT targets (nuclear, chemical, critical infrastructure)
- When building detection for APT-style attacks targeting PLC logic and process manipulation
- When establishing PLC logic integrity monitoring to detect unauthorized modifications
- When investigating suspected process anomalies that may indicate cyber-physical attacks
- When designing defense-in-depth strategies against nation-state level OT threats

**Do not use** for basic OT intrusion detection (see detecting-attacks-on-scada-systems), for malware analysis of Stuxnet samples (see malware reverse engineering skills), or for PLC programming and logic development.

## Prerequisites

- Detailed understanding of the Stuxnet attack chain and MITRE ATT&CK for ICS framework
- PLC logic backup repository with known-good baseline copies of all PLC programs
- Engineering workstation monitoring (EDR with OT awareness)
- Physics-based process models for the controlled physical process
- Network monitoring for industrial protocol traffic analysis

## Workflow

### Step 1: Understand the Stuxnet Attack Chain

Map detection opportunities across the multi-stage Stuxnet-style attack chain.

```yaml
# Stuxnet-Style Attack Chain and Detection Points
attack_chain:
  stage_1_initial_access:
    technique: "USB-borne malware targeting air-gapped network"
    mitre_ics: "T0847 - Replication Through Removable Media"
    detection:
      - "USB device connection logging on engineering workstations"
      - "Removable media scanning with OT-approved AV"
      - "Application allowlisting blocking unauthorized executables"
      - "Windows autorun disabled via Group Policy"
    indicators:
      - "New USB device connections to engineering workstations"
      - "Execution of unsigned binaries from removable media"
      - "LNK file exploitation patterns"

  stage_2_lateral_movement:
    technique: "Exploitation of Windows vulnerabilities for network propagation"
    mitre_ics: "T0866 - Exploitation of Remote Services"
    detection:
      - "Network IDS detecting exploit traffic (MS08-067, MS10-061)"
      - "Unusual SMB traffic between engineering workstations"
      - "Windows event logs showing privilege escalation"
      - "New scheduled tasks or services created"
    indicators:
      - "Lateral movement between Level 3-4 Windows systems"
      - "WMI/PsExec execution from unexpected sources"
      - "Pass-the-hash authentication patterns"

  stage_3_ews_compromise:
    technique: "Compromise of engineering workstation with PLC programming software"
    mitre_ics: "T0862 - Supply Chain Compromise (Step-7 hooking)"
    detection:
      - "File integrity monitoring on Step-7/TIA Portal directories"
      - "DLL injection detection in PLC programming software"
      - "Monitoring s7otbxdx.dll for Stuxnet-specific hook"
      - "Unexpected modifications to PLC project files"
    indicators:
      - "Modified DLLs in Siemens STEP 7 installation directory"
      - "Rootkit hiding files on engineering workstation"
      - "PLC programming software behaving abnormally"

  stage_4_plc_logic_modification:
    technique: "Injecting malicious OB/FC blocks into PLC program"
    mitre_ics: "T0839 - Module Firmware / T0833 - Modify Control Logic"
    detection:
      - "PLC logic integrity comparison against known-good baseline"
      - "S7comm upload/download traffic from unauthorized sources"
      - "New OB/FC/FB blocks appearing in PLC program"
      - "Modification of OB1 (main scan) or OB35 (cyclic interrupt)"
    indicators:
      - "PLC program block count changes"
      - "PLC program size changes"
      - "Upload of unknown program blocks"

  stage_5_process_manipulation:
    technique: "Manipulating physical process while spoofing sensor readings"
    mitre_ics: "T0836 - Modify Parameter / T0856 - Spoof Reporting Message"
    detection:
      - "Physics-based anomaly detection (process model deviation)"
      - "Cross-validation of independent sensors"
      - "Vibration analysis and mechanical signature monitoring"
      - "Comparison of PLC-reported values vs independent measurements"
    indicators:
      - "Motor/pump operating outside normal parameters"
      - "Sensor readings diverging from physics model predictions"
      - "Process efficiency metrics deviating unexpectedly"
```

### Step 2: Implement PLC Logic Integrity Monitoring

Continuously monitor PLC program integrity by comparing running logic against known-good baselines.

```python
#!/usr/bin/env python3
"""PLC Logic Integrity Monitor.

Periodically retrieves PLC program block information and compares
against known-good baselines to detect unauthorized modifications
(Stuxnet-style logic injection).
"""

import hashlib
import json
import sys
import time
from dataclasses import dataclass, field, asdict
from datetime import datetime


@dataclass
class PLCBlock:
    """Represents a PLC program block."""
    block_type: str  # OB, FC, FB, DB
    block_number: int
    name: str
    size_bytes: int
    checksum: str
    last_modified: str
    author: str = ""


@dataclass
class IntegrityAlert:
    alert_id: str
    timestamp: str
    severity: str
    plc_name: str
    plc_ip: str
    alert_type: str
    description: str
    baseline_value: str
    current_value: str
    mitre_technique: str


class PLCIntegrityMonitor:
    """Monitors PLC program integrity against baselines."""

    def __init__(self):
        self.baselines = {}  # plc_name -> list of PLCBlock
        self.alerts = []
        self.alert_counter = 1

    def load_baseline(self, plc_name, baseline_file):
        """Load known-good PLC program baseline."""
        with open(baseline_file) as f:
            data = json.load(f)
        blocks = [PLCBlock(**b) for b in data.get("blocks", [])]
        self.baselines[plc_name] = {
            "blocks": {f"{b.block_type}{b.block_number}": b for b in blocks},
            "total_blocks": len(blocks),
            "loaded_at": datetime.now().isoformat(),
        }
        print(f"[*] Loaded baseline for {plc_name}: {len(blocks)} blocks")

    def check_integrity(self, plc_name, plc_ip, current_blocks):
        """Compare current PLC program against baseline."""
        baseline = self.baselines.get(plc_name)
        if not baseline:
            print(f"[WARN] No baseline for {plc_name}")
            return

        baseline_blocks = baseline["blocks"]
        current_block_map = {f"{b.block_type}{b.block_number}": b for b in current_blocks}

        # Check 1: New blocks added (potential logic injection)
        for key, block in current_block_map.items():
            if key not in baseline_blocks:
                self.alerts.append(IntegrityAlert(
                    alert_id=f"INT-{self.alert_counter:04d}",
                    timestamp=datetime.now().isoformat(),
                    severity="critical",
                    plc_name=plc_name,
                    plc_ip=plc_ip,
                    alert_type="NEW_BLOCK_DETECTED",
                    description=(
                        f"New program block {key} ({block.name}) found in PLC "
                        f"that does not exist in baseline. Size: {block.size_bytes} bytes."
                    ),
                    baseline_value="Block does not exist in baseline",
                    current_value=f"{key}: {block.size_bytes} bytes, checksum {block.checksum}",
                    mitre_technique="T0839 - Module Firmware / T0833 - Modify Control Logic",
                ))
                self.alert_counter += 1

        # Check 2: Blocks removed
        for key in baseline_blocks:
            if key not in current_block_map:
                self.alerts.append(IntegrityAlert(
                    alert_id=f"INT-{self.alert_counter:04d}",
                    timestamp=datetime.now().isoformat(),
                    severity="high",
                    plc_name=plc_name,