implementing-purdue-model-network-segmentation

What this skill does

# Implementing Purdue Model Network Segmentation

## 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.

```python
#!/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",