performing-ot-network-security-assessment

What this skill does

# Performing OT Network Security Assessment

## When to Use

- When conducting an initial security baseline of an OT/ICS environment for a new client
- When evaluating the security posture of a facility after an IT/OT convergence initiative
- When preparing for IEC 62443 or NERC CIP compliance audits
- When assessing risk following a merger or acquisition involving industrial facilities
- When investigating whether an OT network has been compromised or has unmonitored pathways to corporate IT

**Do not use** for IT-only network assessments without OT components, for application-layer vulnerability scanning of IT web applications (see performing-web-app-penetration-test), or for active exploitation of live OT systems without explicit authorization and safety controls in place.

## Prerequisites

- Written authorization from the asset owner and operations management for all assessment activities
- Understanding of the Purdue Reference Model and IEC 62443 zone/conduit architecture
- Passive network monitoring tools (Nozomi Guardian, Dragos Platform, or Wireshark with industrial protocol dissectors)
- Access to network diagrams, firewall rule sets, and asset inventories (or the ability to perform passive discovery)
- Safety briefing on the physical processes controlled by the OT systems under assessment

## Workflow

### Step 1: Establish Assessment Scope and Safety Boundaries

Define the scope based on the Purdue Reference Model levels and identify safety-critical systems that must not be actively scanned. OT assessments differ fundamentally from IT assessments because active scanning can crash PLCs, disrupt safety instrumented systems (SIS), and cause physical harm.

```yaml
# OT Assessment Scope Definition
assessment:
  facility: "Chemical Processing Plant - Site Alpha"
  purdue_levels_in_scope:
    - level_0: "Physical process sensors and actuators (passive observation only)"
    - level_1: "PLCs, RTUs, safety controllers (passive only, no active scanning)"
    - level_2: "HMI stations, engineering workstations, historian (limited active with approval)"
    - level_3: "Site operations - OPC servers, application servers (active scanning permitted)"
    - level_3_5: "DMZ - data diodes, jump servers (active scanning permitted)"
    - level_4: "Enterprise IT connecting to OT (active scanning permitted)"

  safety_exclusions:
    - "Safety Instrumented Systems (SIS) - Triconex controllers"
    - "Emergency Shutdown (ESD) systems"
    - "Fire and Gas detection systems"
    - "Any Level 0/1 device during active production"

  authorized_activities:
    passive:
      - "Network traffic capture and analysis via SPAN ports"
      - "Industrial protocol deep packet inspection"
      - "Wireless spectrum analysis"
      - "Physical walkthrough and visual inspection"
    active_with_approval:
      - "Targeted Nmap scans of Level 2-4 systems during maintenance windows"
      - "Authentication testing on HMI and engineering workstations"
      - "Firewall rule verification between zones"
    prohibited:
      - "Active scanning of PLCs, RTUs, or SIS controllers"
      - "Fuzzing industrial protocols on live systems"
      - "Modifying PLC logic or firmware"
```

### Step 2: Perform Passive Network Discovery and Asset Inventory

Deploy passive monitoring to map all devices, communication flows, and protocols on the OT network without sending any traffic that could disrupt operations.

```python
#!/usr/bin/env python3
"""OT Network Passive Discovery and Asset Inventory Builder.

Uses pcap captures from SPAN ports to identify OT assets, protocols,
and communication patterns without active scanning.
"""

import json
import sys
from collections import defaultdict
from datetime import datetime

try:
    from scapy.all import rdpcap, IP, TCP, UDP
    from scapy.contrib.modbus import ModbusADURequest, ModbusADUResponse
except ImportError:
    print("Install scapy: pip install scapy")
    sys.exit(1)

# Industrial protocol port mappings
OT_PROTOCOL_PORTS = {
    502: "Modbus/TCP",
    102: "S7comm (Siemens)",
    44818: "EtherNet/IP (CIP)",
    2222: "EtherNet/IP (implicit)",
    4840: "OPC UA",
    20000: "DNP3",
    47808: "BACnet",
    1911: "Niagara Fox",
    789: "Crimson v3 (Red Lion)",
    2404: "IEC 60870-5-104",
    18245: "GE SRTP",
    5094: "HART-IP",
}

PURDUE_LEVEL_RANGES = {
    "Level 0-1 (Field Devices)": ["10.10.0.0/16", "192.168.10.0/24"],
    "Level 2 (Control Systems)": ["10.20.0.0/16", "192.168.20.0/24"],
    "Level 3 (Site Operations)": ["10.30.0.0/16", "192.168.30.0/24"],
    "Level 3.5 (DMZ)": ["172.16.0.0/16"],
    "Level 4 (Enterprise)": ["10.0.0.0/16"],
}


def classify_purdue_level(ip_addr):
    """Classify an IP address to its Purdue Reference Model level."""
    from ipaddress import ip_address, ip_network

    addr = ip_address(ip_addr)
    for level, subnets in PURDUE_LEVEL_RANGES.items():
        for subnet in subnets:
            if addr in ip_network(subnet):
                return level
    return "Unknown"


def analyze_ot_pcap(pcap_file):
    """Analyze pcap file to discover OT assets and communication patterns."""
    packets = rdpcap(pcap_file)

    assets = {}
    connections = defaultdict(lambda: {"count": 0, "protocols": set(), "ports": set()})
    protocol_stats = defaultdict(int)
    cross_zone_flows = []

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

        src_ip = pkt[IP].src
        dst_ip = pkt[IP].dst

        # Track assets
        for ip in (src_ip, dst_ip):
            if ip not in assets:
                assets[ip] = {
                    "ip": ip,
                    "purdue_level": classify_purdue_level(ip),
                    "protocols_observed": set(),
                    "roles": set(),
                    "first_seen": str(pkt.time),
                    "last_seen": str(pkt.time),
                    "mac": None,
                }
            assets[ip]["last_seen"] = str(pkt.time)

        # Identify OT protocols by port
        dst_port = None
        if pkt.haslayer(TCP):
            dst_port = pkt[TCP].dport
        elif pkt.haslayer(UDP):
            dst_port = pkt[UDP].dport

        if dst_port and dst_port in OT_PROTOCOL_PORTS:
            protocol_name = OT_PROTOCOL_PORTS[dst_port]
            protocol_stats[protocol_name] += 1
            assets[src_ip]["protocols_observed"].add(protocol_name)
            assets[dst_ip]["protocols_observed"].add(protocol_name)

            # Determine roles
            assets[src_ip]["roles"].add("client/master")
            assets[dst_ip]["roles"].add("server/slave")

        # Track connections
        conn_key = (src_ip, dst_ip)
        connections[conn_key]["count"] += 1
        if dst_port:
            connections[conn_key]["ports"].add(dst_port)
            if dst_port in OT_PROTOCOL_PORTS:
                connections[conn_key]["protocols"].add(OT_PROTOCOL_PORTS[dst_port])

        # Detect cross-zone communication
        src_level = classify_purdue_level(src_ip)
        dst_level = classify_purdue_level(dst_ip)
        if src_level != dst_level and src_level != "Unknown" and dst_level != "Unknown":
            cross_zone_flows.append({
                "src": src_ip,
                "src_level": src_level,
                "dst": dst_ip,
                "dst_level": dst_level,
                "protocol": OT_PROTOCOL_PORTS.get(dst_port, f"port/{dst_port}"),
            })

    return {
        "asset_count": len(assets),
        "assets": {ip: {k: list(v) if isinstance(v, set) else v for k, v in info.items()} for ip, info in assets.items()},
        "protocol_distribution": dict(protocol_stats),
        "total_connections": len(connections),
        "cross_zone_flows": cross_zone_flows[:50],
    }


def generate_assessment_report(results):
    """Generate the OT network assessment findings report."""
    report = []
    report.append("=" * 70)
    report.append("OT NETWORK PASSIVE DISCOVERY REPORT")
    report.append(f"Gener