Install this skill

npx skills add mukul975/Anthropic-Cybersecurity-Skills

Framework mappings

MITRE ATT&CK

NIST CSF 2.0

When to Use

When designing or retrofitting network architecture for an ICS/SCADA environment
When implementing IEC 62443 zone and conduit requirements in a brownfield plant
When creating the IT/OT DMZ (Level 3.5) to control data flow between enterprise and control networks
When remediating audit findings about flat OT networks or direct IT-to-OT connectivity
When segmenting a converged IT/OT network after an acquisition or merger

Do not use for micro-segmentation within a single Purdue level (see implementing-zone-conduit-model-for-ics), for cloud-native environments without traditional ICS networks, or for network segmentation in purely IT environments.

Prerequisites

Complete OT asset inventory with Purdue level classification for each device
Network architecture diagram showing current topology, VLANs, and firewall placements
Industrial firewalls capable of deep packet inspection for OT protocols (Palo Alto, Fortinet, Cisco)
Understanding of required data flows between Purdue levels (historian replication, remote access, patch distribution)
Change management approval from plant operations for network modifications

Workflow

Step 1: Map Current Architecture to Purdue Levels

Classify all network assets and data flows according to the Purdue Model hierarchy.

#!/usr/bin/env python3
"""Purdue Model Network Segmentation Planner.
 
Maps existing OT/IT network assets to Purdue Model levels and generates
segmentation recommendations including firewall rules and VLAN assignments.
"""
 
import json
import csv
import sys
from collections import defaultdict
from datetime import datetime
from typing import Dict, List
 
 
PURDUE_LEVELS = {
    0: {
        "name": "Physical Process",
        "description": "Sensors, actuators, field instruments",
        "typical_devices": ["Sensors", "Actuators", "Drives", "Motor starters"],
        "vlan_range": "100-109",
        "allowed_protocols": ["HART", "Profibus", "Foundation Fieldbus", "IO-Link"],
    },
    1: {
        "name": "Basic Control",
        "description": "PLCs, RTUs, safety controllers",
        "typical_devices": ["PLC", "RTU", "Safety Controller", "DCS Controller"],
        "vlan_range": "110-119",
        "allowed_protocols": ["EtherNet/IP", "Profinet", "Modbus/TCP", "S7comm", "OPC UA"],
    },
    2: {
        "name": "Supervisory Control",
        "description": "HMI, SCADA servers, engineering workstations",
        "typical_devices": ["HMI", "SCADA Server", "Engineering Workstation", "Batch Server"],
        "vlan_range": "120-129",
        "allowed_protocols": ["OPC UA", "OPC DA", "Modbus/TCP", "DNP3", "HTTPS"],
    },
    3: {
        "name": "Site Operations",
        "description": "Historian, MES, asset management",
        "typical_devices": ["Historian", "MES Server", "Asset Management", "Alarm Server"],
        "vlan_range": "130-139",
        "allowed_protocols": ["OPC UA", "SQL", "HTTPS", "MQTT"],
    },
    3.5: {
        "name": "IT/OT DMZ",
        "description": "Demilitarized zone between IT and OT",
        "typical_devices": ["Jump Server", "Historian Mirror", "Patch Server", "AV Update Server", "Remote Access Gateway"],
        "vlan_range": "150-159",
        "allowed_protocols": ["HTTPS", "RDP (to jump server only)", "SSH", "SQL (read replica)"],
    },
    4: {
        "name": "Enterprise IT",
        "description": "Enterprise applications, email, ERP",
        "typical_devices": ["ERP Server", "Email Server", "Business Applications", "Active Directory"],
        "vlan_range": "200-249",
        "allowed_protocols": ["HTTPS", "LDAPS", "SMTP", "SQL"],
    },
    5: {
        "name": "Enterprise Network / Internet",
        "description": "External connections, cloud services, partner networks",
        "typical_devices": ["Internet Gateway", "VPN Concentrator", "Cloud Services"],
        "vlan_range": "250-254",
        "allowed_protocols": ["HTTPS", "IPsec VPN"],
    },
}
 
 
class PurdueSegmentationPlanner:
    """Plans Purdue Model network segmentation."""
 
    def __init__(self):
        self.assets = []
        self.data_flows = []
        self.firewall_rules = []
 
    def load_asset_inventory(self, filepath: str):
        """Load asset inventory from CSV."""
        with open(filepath, "r") as f:
            self.assets = list(csv.DictReader(f))
        print(f"[*] Loaded {len(self.assets)} assets")
 
    def classify_assets(self):
        """Classify assets into Purdue levels based on type and function."""
        classification = defaultdict(list)
        for asset in self.assets:
            level = asset.get("purdue_level", "")
            try:
                level = float(level)
            except (ValueError, TypeError):
                level = self._infer_purdue_level(asset)
 
            classification[level].append(asset)
            asset["assigned_level"] = level
 
        return classification
 
    def _infer_purdue_level(self, asset: dict) -> float:
        """Infer Purdue level from device type if not explicitly assigned."""
        device_type = asset.get("type", "").lower()
        mapping = {
            "sensor": 0, "actuator": 0, "drive": 0,
            "plc": 1, "rtu": 1, "safety": 1, "dcs": 1,
            "hmi": 2, "scada": 2, "engineering": 2,
            "historian": 3, "mes": 3, "alarm": 3,
            "jump": 3.5, "patch": 3.5, "remote_access": 3.5,
            "erp": 4, "email": 4, "directory": 4,
        }
        for keyword, level in mapping.items():
            if keyword in device_type:
                return level
        return -1
 
    def generate_vlan_plan(self, classification: dict) -> list:
        """Generate VLAN assignment plan based on Purdue levels."""
        vlan_plan = []
        for level, info in PURDUE_LEVELS.items():
            assets_at_level = classification.get(level, [])
            vlan_plan.append({
                "purdue_level": level,
                "level_name": info["name"],
                "vlan_range": info["vlan_range"],
                "asset_count": len(assets_at_level),
                "allowed_protocols": info["allowed_protocols"],
            })
        return vlan_plan
 
    def generate_firewall_rules(self) -> list:
        """Generate inter-level firewall rules enforcing Purdue Model boundaries."""
        rules = [
            {
                "rule_id": 1,
                "name": "Block direct IT-to-Level1",
                "action": "DENY",
                "source_zone": "Level_4_Enterprise",
                "dest_zone": "Level_1_Control",
                "service": "ANY",
                "log": True,
                "description": "No direct access from enterprise IT to basic control PLCs",
            },
            {
                "rule_id": 2,
                "name": "Block direct IT-to-Level2",
                "action": "DENY",
                "source_zone": "Level_4_Enterprise",
                "dest_zone": "Level_2_Supervisory",
                "service": "ANY",
                "log": True,
                "description": "No direct access from enterprise IT to HMI/SCADA",
            },
            {
                "rule_id": 3,
                "name": "Allow DMZ-to-Historian-Replica",
                "action": "ALLOW",
                "source_zone": "Level_3_Operations",
                "dest_zone": "Level_35_DMZ",
                "service": "SQL/1433 (read replica push)",
                "log": True,
                "description": "Historian pushes data to DMZ replica for IT consumption",
            },
            {
                "rule_id": 4,
                "name": "Allow IT-to-DMZ-JumpServer",
                "action": "ALLOW",
                "source_zone": "Level_4_Enterprise",
                "dest_zone": "Level_35_DMZ",
                "service": "RDP/3389, SSH/22",
                "log": True,
                "description": "IT users access OT via jump server in DMZ only",
            },
            {
                "rule_id": 5,
                "name": "Allow DMZ-JumpServer-to-Level2",
                "action": "ALLOW",
                "source_zone": "Level_35_DMZ",
                "dest_zone": "Level_2_Supervisory",
                "service": "RDP/3389 (from jump server IP only)",
                "log": True,
                "description": "Jump server provides controlled access to HMI/SCADA",
            },
            {
                "rule_id": 6,
                "name": "Allow Level2-to-Level1",
                "action": "ALLOW",
                "source_zone": "Level_2_Supervisory",
                "dest_zone": "Level_1_Control",
                "service": "Modbus/502, EtherNet-IP/44818, S7comm/102",
                "log": True,
                "description": "HMI/SCADA communicates with PLCs using industrial protocols",
            },
            {
                "rule_id": 7,
                "name": "Block Level1-outbound-internet",
                "action": "DENY",
                "source_zone": "Level_1_Control",
                "dest_zone": "Level_5_Internet",
                "service": "ANY",
                "log": True,
                "description": "PLCs must never reach the internet directly",
            },
            {
                "rule_id": 8,
                "name": "Allow patch distribution DMZ-to-Level2",
                "action": "ALLOW",
                "source_zone": "Level_35_DMZ",
                "dest_zone": "Level_2_Supervisory",
                "service": "WSUS/8530",
                "log": True,
                "description": "Patch server in DMZ distributes updates to supervisory systems",
            },
            {
                "rule_id": 9,
                "name": "Default deny all inter-zone",
                "action": "DENY",
                "source_zone": "ANY",
                "dest_zone": "ANY",
                "service": "ANY",
                "log": True,
                "description": "Default deny all traffic not explicitly permitted",
            },
        ]
        self.firewall_rules = rules
        return rules
 
    def print_segmentation_plan(self, classification: dict):
        """Print the complete segmentation plan."""
        print(f"\n{'='*70}")
        print("PURDUE MODEL NETWORK SEGMENTATION PLAN")
        print(f"{'='*70}")
        print(f"Generated: {datetime.now().isoformat()}")
 
        vlan_plan = self.generate_vlan_plan(classification)
        print(f"\n--- VLAN ASSIGNMENT ---")
        for v in vlan_plan:
            print(f"\n  {v['level_name']} (Purdue {v['purdue_level']})")
            print(f"    VLAN Range: {v['vlan_range']}")
            print(f"    Assets: {v['asset_count']}")
            print(f"    Allowed Protocols: {', '.join(v['allowed_protocols'])}")
 
        print(f"\n--- INTER-ZONE FIREWALL RULES ---")
        rules = self.generate_firewall_rules()
        for rule in rules:
            action_symbol = "+" if rule["action"] == "ALLOW" else "X"
            print(f"\n  [{action_symbol}] Rule {rule['rule_id']}: {rule['name']}")
            print(f"      {rule['source_zone']} -> {rule['dest_zone']}")
            print(f"      Service: {rule['service']}")
            print(f"      Reason: {rule['description']}")
 
 
if __name__ == "__main__":
    planner = PurdueSegmentationPlanner()
    if len(sys.argv) >= 2:
        planner.load_asset_inventory(sys.argv[1])
    classification = planner.classify_assets()
    planner.print_segmentation_plan(classification)

Step 2: Configure Industrial DMZ (Level 3.5)

The DMZ is the critical boundary between IT and OT. All data exchange must traverse it -- no direct connections are permitted.

# Level 3.5 DMZ Architecture Configuration
# All IT-OT data exchange flows through the DMZ
 
dmz_architecture:
  zone_name: "IT_OT_DMZ"
  purdue_level: 3.5
  vlan: 150
 
  components:
    historian_replica:
      purpose: "Read-only copy of OT historian data for IT/business access"
      direction: "OT pushes data TO DMZ (unidirectional)"
      ip: "10.10.150.10"
      services:
        - port: 1433
          protocol: "SQL"
          direction: "inbound from Level 3 historian only"
        - port: 443
          protocol: "HTTPS"
          direction: "outbound to Level 4 for IT consumers"
 
    jump_server:
      purpose: "Controlled remote access point for OT maintenance"
      ip: "10.10.150.20"
      services:
        - port: 3389
          protocol: "RDP"
          direction: "inbound from Level 4 with MFA"
        - port: 3389
          protocol: "RDP"
          direction: "outbound to Level 2 HMIs only"
      security_controls:
        - "Multi-factor authentication required"
        - "Session recording enabled"
        - "Maximum session duration: 4 hours"
        - "Approval-based access workflow"
 
    patch_server:
      purpose: "Staging area for tested patches before OT deployment"
      ip: "10.10.150.30"
      services:
        - port: 8530
          protocol: "WSUS"
          direction: "pulls from Level 4 WSUS, pushes to Level 2-3"
 
    antivirus_relay:
      purpose: "AV signature distribution to OT endpoints"
      ip: "10.10.150.40"
      services:
        - port: 443
          protocol: "HTTPS"
          direction: "pulls definitions from Level 4, distributes to Level 2-3"
 
  firewall_rules:
    north_firewall:  # Between DMZ and Level 4 Enterprise
      - allow: "Level 4 -> DMZ jump server:3389 (with MFA)"
      - allow: "Level 4 -> DMZ historian replica:443 (read-only)"
      - allow: "DMZ patch server -> Level 4 WSUS:8530 (pull only)"
      - deny: "ALL other traffic"
 
    south_firewall:  # Between DMZ and Level 3 Operations
      - allow: "Level 3 historian -> DMZ replica:1433 (push direction)"
      - allow: "DMZ jump server -> Level 2 HMI:3389 (session-limited)"
      - allow: "DMZ patch server -> Level 2/3:8530 (scheduled)"
      - deny: "ALL other traffic"
 
    critical_rule: "NO traffic passes through DMZ end-to-end. DMZ breaks all connections."

Key Concepts

Term	Definition
Purdue Model (PERA)	Hierarchical reference architecture organizing industrial networks into levels 0-5 based on function and trust
Level 3.5 DMZ	Demilitarized zone between IT (Level 4) and OT (Level 3), where all cross-boundary data exchange occurs
Defense in Depth	Layered security approach requiring attackers to breach multiple boundaries to reach critical control systems
Data Diode	Hardware-enforced unidirectional communication device ensuring data flows only from OT to IT, never reverse
Zone	Logical grouping of assets sharing common security requirements as defined by IEC 62443
Conduit	Controlled communication path between zones with defined security policies

Common Scenarios

Scenario: Flat OT Network Remediation

Context: An audit reveals that enterprise IT systems can directly communicate with PLCs on the control network. There is no DMZ and no firewall between IT and OT.

Approach:

Perform full traffic analysis to identify all legitimate data flows crossing IT/OT boundary
Design DMZ architecture with historian replica, jump server, and patch staging
Deploy industrial firewall between IT and DMZ (north firewall) and between DMZ and OT (south firewall)
Migrate data flows one at a time: start with historian replication through DMZ
Implement jump server for remote access, deprecating direct RDP to OT systems
Block direct IT-to-OT traffic on the north firewall after all flows migrate through DMZ
Validate with penetration test from IT network confirming no direct path to Level 1 controllers

Pitfalls: Do not cut over all traffic simultaneously -- migrate flow by flow with rollback plans. Legacy OT systems may use protocols that cannot traverse firewalls doing DPI; test thoroughly in a lab first. Never deploy the DMZ during active production without an agreed maintenance window.

Output Format

PURDUE MODEL SEGMENTATION REPORT
====================================
Assessment Date: YYYY-MM-DD
Facility: [Plant Name]
 
CURRENT STATE:
  Network Type: [Flat/Partially segmented/Fully segmented]
  IT-OT Boundary: [None/Firewall/DMZ with dual firewall]
  Direct IT-to-PLC paths: [count]
 
RECOMMENDED ARCHITECTURE:
  Level 0-1: VLAN 110 (Control Network)
  Level 2:   VLAN 120 (Supervisory Network)
  Level 3:   VLAN 130 (Operations Network)
  Level 3.5: VLAN 150 (IT/OT DMZ)
  Level 4-5: VLAN 200+ (Enterprise)
 
DMZ COMPONENTS:
  - Historian Replica Server
  - Jump Server (MFA-enabled)
  - Patch Staging Server
  - AV Relay Server
 
FIREWALL RULES: [count] rules generated
MIGRATION STEPS: [count] phases planned

Source materials

References and resources

Everything below is rendered for inspection. Script files are read-only and never run.

References 1

api-reference.md5.5 KB

API Reference: Purdue Model OT Network Segmentation Audit

Libraries Used

Library	Purpose
`scapy`	Network packet analysis and traffic flow validation
`requests`	Firewall API calls for rule review
`json`	Parse asset inventory and segmentation policy
`ipaddress`	Validate IP ranges and subnet assignments
`socket`	Port connectivity testing across Purdue levels

Installation

pip install scapy requests

Purdue Model Levels

Level	Name	Examples	Network Zone
L0	Process	Sensors, actuators, valves	Field Network
L1	Basic Control	PLCs, RTUs, safety controllers	Control Network
L2	Area Supervisory	HMIs, SCADA servers, historians	Supervisory Network
L3	Site Operations	Patch servers, AV, AD for OT	Operations Network
L3.5	DMZ	Data diodes, jump servers	Industrial DMZ
L4	Enterprise	ERP, email, business apps	Corporate Network
L5	Internet	Cloud, remote access, third parties	External

Core Audit Functions

Define Asset Zone Mapping

import ipaddress
 
PURDUE_ZONES = {
    "L0": [ipaddress.ip_network("10.10.0.0/24")],
    "L1": [ipaddress.ip_network("10.10.1.0/24")],
    "L2": [ipaddress.ip_network("10.10.2.0/24")],
    "L3": [ipaddress.ip_network("10.10.3.0/24")],
    "L3.5": [ipaddress.ip_network("10.10.35.0/24")],
    "L4": [ipaddress.ip_network("10.20.0.0/16")],
    "L5": [ipaddress.ip_network("0.0.0.0/0")],
}
 
def classify_ip(ip):
    addr = ipaddress.ip_address(ip)
    for level, subnets in PURDUE_ZONES.items():
        for subnet in subnets:
            if addr in subnet:
                return level
    return "UNKNOWN"

Validate Allowed Traffic Flows

# Purdue model: traffic should only flow between adjacent levels
ALLOWED_FLOWS = {
    ("L0", "L1"), ("L1", "L0"),
    ("L1", "L2"), ("L2", "L1"),
    ("L2", "L3"), ("L3", "L2"),
    ("L3", "L3.5"), ("L3.5", "L3"),
    ("L3.5", "L4"), ("L4", "L3.5"),
    ("L4", "L5"), ("L5", "L4"),
}
 
def validate_flow(src_ip, dst_ip):
    src_level = classify_ip(src_ip)
    dst_level = classify_ip(dst_ip)
    flow = (src_level, dst_level)
    return {
        "src_ip": src_ip,
        "dst_ip": dst_ip,
        "src_level": src_level,
        "dst_level": dst_level,
        "allowed": flow in ALLOWED_FLOWS or src_level == dst_level,
        "violation": flow not in ALLOWED_FLOWS and src_level != dst_level,
    }

Analyze Network Traffic for Segmentation Violations

from scapy.all import rdpcap, IP
 
def analyze_pcap_for_violations(pcap_path):
    packets = rdpcap(pcap_path)
    violations = []
    seen = set()
    for pkt in packets:
        if IP in pkt:
            flow_key = (pkt[IP].src, pkt[IP].dst)
            if flow_key in seen:
                continue
            seen.add(flow_key)
            result = validate_flow(pkt[IP].src, pkt[IP].dst)
            if result["violation"]:
                violations.append(result)
    return violations

Port Connectivity Test Across Levels

import socket
 
def test_segmentation(src_level_hosts, dst_level_hosts, ports):
    """Test that connections between non-adjacent levels are blocked."""
    results = []
    for src in src_level_hosts:
        for dst in dst_level_hosts:
            for port in ports:
                try:
                    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                    sock.settimeout(3)
                    result = sock.connect_ex((dst, port))
                    status = "open" if result == 0 else "closed"
                    sock.close()
                except socket.timeout:
                    status = "filtered"
                results.append({
                    "src": src, "dst": dst, "port": port,
                    "status": status,
                    "expected": "filtered",
                    "pass": status != "open",
                })
    return results

Audit Firewall Rules for DMZ Compliance

def audit_dmz_rules(firewall_rules):
    """Check that L3.5 DMZ properly isolates OT from IT."""
    findings = []
    for rule in firewall_rules:
        src_zone = classify_ip(rule["src_ip"])
        dst_zone = classify_ip(rule["dst_ip"])
 
        # Direct L4->L2 or L4->L1 bypasses DMZ
        if src_zone == "L4" and dst_zone in ("L0", "L1", "L2"):
            findings.append({
                "rule_id": rule["id"],
                "issue": f"Direct {src_zone}->{dst_zone} bypasses DMZ",
                "severity": "critical",
                "remediation": "Route through L3.5 DMZ",
            })
 
        # L5 direct to any OT level
        if src_zone == "L5" and dst_zone in ("L0", "L1", "L2", "L3"):
            findings.append({
                "rule_id": rule["id"],
                "issue": f"Internet ({src_zone}) directly reaches OT ({dst_zone})",
                "severity": "critical",
                "remediation": "Block all direct internet-to-OT traffic",
            })
    return findings

Output Format

{
  "audit_date": "2025-01-15",
  "total_flows_analyzed": 15420,
  "segmentation_violations": 12,
  "critical_violations": 3,
  "violations": [
    {
      "src_ip": "10.20.5.100",
      "dst_ip": "10.10.1.50",
      "src_level": "L4",
      "dst_level": "L1",
      "violation": true,
      "severity": "critical",
      "detail": "Enterprise host directly accessing PLC network"
    }
  ],
  "dmz_compliance": {
    "data_diode_present": true,
    "jump_server_hardened": true,
    "direct_ot_it_paths": 0
  }
}

Scripts 1

agent.py7.5 KB

Display-only source. This catalog never executes bundled scripts.

#!/usr/bin/env python3
"""Purdue model OT network segmentation audit agent.

Audits OT/ICS network segmentation against the Purdue Enterprise
Reference Architecture by testing connectivity between network zones,
verifying firewall rules, and mapping discovered hosts to Purdue levels.
"""
import argparse
import json
import os
import socket
import subprocess
import sys
from datetime import datetime, timezone


PURDUE_LEVELS = {
    0: {"name": "Process", "description": "Sensors, actuators, field devices"},
    1: {"name": "Basic Control", "description": "PLCs, RTUs, safety systems"},
    2: {"name": "Area Supervisory", "description": "HMIs, SCADA, historian"},
    3: {"name": "Site Operations", "description": "Patch mgmt, AV, file servers"},
    3.5: {"name": "DMZ", "description": "Industrial DMZ between IT and OT"},
    4: {"name": "Site Business", "description": "ERP, email, corporate apps"},
    5: {"name": "Enterprise", "description": "Internet, cloud, remote access"},
}

PROHIBITED_FLOWS = [
    {"from_level": 5, "to_level": 0, "description": "Internet to Process (critical violation)"},
    {"from_level": 5, "to_level": 1, "description": "Internet to Basic Control"},
    {"from_level": 5, "to_level": 2, "description": "Internet to SCADA"},
    {"from_level": 4, "to_level": 0, "description": "Corporate to Process"},
    {"from_level": 4, "to_level": 1, "description": "Corporate to PLC/RTU"},
]


def test_connectivity(source_ip, target_ip, ports, timeout=3):
    """Test TCP connectivity between two hosts on specified ports."""
    results = []
    for port in ports:
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(timeout)
            result = sock.connect_ex((target_ip, port))
            sock.close()
            reachable = result == 0
            results.append({
                "target": target_ip,
                "port": port,
                "reachable": reachable,
            })
        except (socket.error, OSError):
            results.append({"target": target_ip, "port": port, "reachable": False})
    return results


def audit_zone_separation(zone_map):
    """Audit network segmentation between Purdue zones."""
    findings = []
    print("[*] Auditing zone separation...")

    for flow in PROHIBITED_FLOWS:
        from_level = flow["from_level"]
        to_level = flow["to_level"]
        from_hosts = zone_map.get(str(from_level), [])
        to_hosts = zone_map.get(str(to_level), [])

        for src in from_hosts[:3]:
            for dst in to_hosts[:3]:
                common_ports = [22, 80, 443, 502, 102, 44818, 47808, 20000]
                results = test_connectivity(src, dst, common_ports)
                open_ports = [r for r in results if r["reachable"]]
                if open_ports:
                    findings.append({
                        "check": f"Zone {from_level} -> Zone {to_level}",
                        "severity": "CRITICAL",
                        "source": src,
                        "destination": dst,
                        "open_ports": [r["port"] for r in open_ports],
                        "detail": flow["description"],
                        "recommendation": "Block traffic between these zones via firewall",
                    })

    if not findings:
        findings.append({
            "check": "Prohibited zone flows",
            "severity": "INFO",
            "detail": "No prohibited cross-zone connectivity detected",
        })

    return findings


def audit_ot_protocols(target_ips):
    """Check for exposed OT/ICS protocols on target hosts."""
    findings = []
    ot_ports = {
        502: "Modbus TCP",
        102: "S7comm (Siemens)",
        44818: "EtherNet/IP",
        47808: "BACnet",
        20000: "DNP3",
        4840: "OPC UA",
        2222: "EtherCAT",
        1911: "Niagara Fox",
        9600: "OMRON FINS",
    }

    print(f"[*] Scanning {len(target_ips)} hosts for exposed OT protocols...")
    for ip in target_ips:
        for port, protocol in ot_ports.items():
            try:
                sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                sock.settimeout(2)
                result = sock.connect_ex((ip, port))
                sock.close()
                if result == 0:
                    findings.append({
                        "check": f"Exposed OT protocol: {protocol}",
                        "severity": "HIGH",
                        "host": ip,
                        "port": port,
                        "protocol": protocol,
                        "detail": f"{protocol} on {ip}:{port} is accessible",
                    })
            except (socket.error, OSError):
                pass

    print(f"[+] Found {len(findings)} exposed OT protocols")
    return findings


def load_zone_map(config_path):
    """Load zone-to-host mapping from config file."""
    with open(config_path, "r") as f:
        return json.load(f)


def format_summary(zone_findings, protocol_findings, zone_map):
    """Print audit summary."""
    all_findings = zone_findings + protocol_findings
    print(f"\n{'='*60}")
    print(f"  Purdue Model Network Segmentation Audit")
    print(f"{'='*60}")

    for level, info in sorted(PURDUE_LEVELS.items()):
        host_count = len(zone_map.get(str(level), []))
        print(f"  Level {level}: {info['name']:20s} ({host_count} hosts) - {info['description']}")

    print(f"\n  Zone Separation Findings : {len(zone_findings)}")
    print(f"  Protocol Exposure Findings: {len(protocol_findings)}")

    severity_counts = {}
    for f in all_findings:
        sev = f.get("severity", "INFO")
        severity_counts[sev] = severity_counts.get(sev, 0) + 1

    if all_findings:
        print(f"\n  Critical/High Issues:")
        for f in all_findings:
            if f["severity"] in ("CRITICAL", "HIGH"):
                print(f"    [{f['severity']:8s}] {f['check']}: {f.get('detail', '')[:50]}")

    return severity_counts


def main():
    parser = argparse.ArgumentParser(
        description="Purdue model OT network segmentation audit agent"
    )
    parser.add_argument("--zone-map", required=True,
                        help="JSON file mapping Purdue levels to host IPs")
    parser.add_argument("--scan-protocols", action="store_true",
                        help="Scan for exposed OT protocols")
    parser.add_argument("--output", "-o", help="Output JSON report")
    parser.add_argument("--verbose", "-v", action="store_true")
    args = parser.parse_args()

    zone_map = load_zone_map(args.zone_map)
    zone_findings = audit_zone_separation(zone_map)

    protocol_findings = []
    if args.scan_protocols:
        all_hosts = []
        for level_hosts in zone_map.values():
            all_hosts.extend(level_hosts)
        protocol_findings = audit_ot_protocols(list(set(all_hosts)))

    severity_counts = format_summary(zone_findings, protocol_findings, zone_map)

    report = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "tool": "Purdue Model Audit",
        "zone_map": zone_map,
        "zone_findings": zone_findings,
        "protocol_findings": protocol_findings,
        "severity_counts": severity_counts,
        "risk_level": (
            "CRITICAL" if severity_counts.get("CRITICAL", 0) > 0
            else "HIGH" if severity_counts.get("HIGH", 0) > 0
            else "LOW"
        ),
    }

    if args.output:
        with open(args.output, "w") as f:
            json.dump(report, f, indent=2)
        print(f"\n[+] Report saved to {args.output}")
    elif args.verbose:
        print(json.dumps(report, indent=2))


if __name__ == "__main__":
    main()

Keep exploring

MITRE ATT&CK

NIST CSF 2.0

When to Use

Prerequisites

Workflow

Step 1: Map Current Architecture to Purdue Levels

Step 2: Configure Industrial DMZ (Level 3.5)

Key Concepts

Common Scenarios

Scenario: Flat OT Network Remediation

Output Format

References and resources

References 1

API Reference: Purdue Model OT Network Segmentation Audit

Libraries Used

Installation

Purdue Model Levels

Core Audit Functions

Define Asset Zone Mapping

Validate Allowed Traffic Flows

Analyze Network Traffic for Segmentation Violations

Port Connectivity Test Across Levels

Audit Firewall Rules for DMZ Compliance

Output Format

Scripts 1

Related skills