implementing-network-segmentation-for-ot

What this skill does

# Implementing Network Segmentation for OT

## When to Use

- When an OT security assessment reveals a flat network with no segmentation between Purdue levels
- When implementing IEC 62443 zone/conduit architecture after completing risk assessment (IEC 62443-3-2)
- When separating IT and OT networks as part of an IT/OT convergence security initiative
- When deploying a DMZ between corporate IT and OT to protect industrial systems from IT-originating threats
- When segmenting safety instrumented systems (SIS) from basic process control systems (BPCS)

**Do not use** for IT-only microsegmentation without OT components (see implementing-zero-trust-in-cloud), or for initial zone design without prior traffic analysis (see performing-ot-network-security-assessment first).

## Prerequisites

- Complete traffic baseline from passive monitoring (minimum 2-4 weeks of capture data)
- Asset inventory with Purdue level classifications for all OT devices
- Industrial-grade network switches with VLAN support and port security
- OT-aware firewalls (Cisco ISA-3000, Fortinet FortiGate Rugged, Palo Alto with OT Security)
- Maintenance window schedule for network changes
- Rollback plan approved by operations management

## Workflow

### Step 1: Design Segmentation Architecture Based on Traffic Baseline

Use the traffic baseline to design VLAN and firewall architecture that preserves all legitimate communication paths while isolating zones.

```python
#!/usr/bin/env python3
"""OT Network Segmentation Design Tool.

Analyzes traffic baseline data and generates a segmentation design
with VLAN assignments, firewall rules, and migration plan.
"""

import json
import sys
from collections import defaultdict
from dataclasses import dataclass, field, asdict
from ipaddress import ip_address, ip_network


@dataclass
class VLANDesign:
    vlan_id: int
    name: str
    purdue_level: str
    subnet: str
    gateway: str
    description: str
    devices: list = field(default_factory=list)


@dataclass
class FirewallRule:
    rule_id: int
    source_zone: str
    source_ip: str
    dest_zone: str
    dest_ip: str
    protocol: str
    port: int
    action: str
    dpi_profile: str = ""
    comment: str = ""


class SegmentationDesigner:
    """Generates segmentation design from traffic baseline."""

    def __init__(self, baseline_file):
        with open(baseline_file) as f:
            self.baseline = json.load(f)
        self.vlans = []
        self.rules = []
        self.rule_counter = 1

    def design_vlans(self):
        """Create VLAN design based on Purdue levels."""
        self.vlans = [
            VLANDesign(10, "SIS-SAFETY", "Level 1 (Safety)",
                       "10.10.10.0/24", "10.10.10.1",
                       "Safety Instrumented Systems - air-gapped or hardware-isolated"),
            VLANDesign(20, "BPCS-FIELD", "Level 0-1 (Field/Control)",
                       "10.10.20.0/24", "10.10.20.1",
                       "PLCs, RTUs, I/O modules, field instruments"),
            VLANDesign(30, "BPCS-SUPERVISORY", "Level 2 (Supervisory)",
                       "10.10.30.0/24", "10.10.30.1",
                       "HMIs, engineering workstations, local historian"),
            VLANDesign(40, "SITE-OPS", "Level 3 (Operations)",
                       "10.10.40.0/24", "10.10.40.1",
                       "Site historian, OPC server, MES, alarm management"),
            VLANDesign(50, "OT-DMZ", "Level 3.5 (DMZ)",
                       "172.16.50.0/24", "172.16.50.1",
                       "Data diode, historian mirror, jump server, patch server"),
            VLANDesign(60, "ENTERPRISE", "Level 4 (Enterprise)",
                       "10.0.60.0/24", "10.0.60.1",
                       "Enterprise IT systems accessing OT data"),
            VLANDesign(999, "QUARANTINE", "Quarantine",
                       "10.10.99.0/24", "10.10.99.1",
                       "Quarantine VLAN for unauthorized or untrusted devices"),
        ]
        return self.vlans

    def generate_firewall_rules_from_baseline(self):
        """Generate firewall rules based on observed legitimate traffic."""
        self.rules = []

        # Default deny rules for each zone boundary
        zone_pairs = [
            ("Level 2", "Level 0-1"),
            ("Level 3", "Level 2"),
            ("Level 3.5", "Level 3"),
            ("Level 4", "Level 3.5"),
        ]

        # Generate allow rules from baseline observed traffic
        for flow in self.baseline.get("cross_zone_flows", []):
            self.rules.append(FirewallRule(
                rule_id=self.rule_counter,
                source_zone=flow["src_level"],
                source_ip=flow["src"],
                dest_zone=flow["dst_level"],
                dest_ip=flow["dst"],
                protocol=flow.get("protocol", "TCP"),
                port=flow.get("port", 0),
                action="ALLOW",
                dpi_profile=self._get_dpi_profile(flow.get("port", 0)),
                comment=f"Baseline observed: {flow['src']} -> {flow['dst']}",
            ))
            self.rule_counter += 1

        # Add default deny rules at the end of each zone ACL
        for src_zone, dst_zone in zone_pairs:
            self.rules.append(FirewallRule(
                rule_id=self.rule_counter,
                source_zone=src_zone,
                source_ip="any",
                dest_zone=dst_zone,
                dest_ip="any",
                protocol="any",
                port=0,
                action="DENY",
                comment=f"Default deny: {src_zone} -> {dst_zone}",
            ))
            self.rule_counter += 1

        return self.rules

    def _get_dpi_profile(self, port):
        """Return the appropriate DPI inspection profile for an OT protocol port."""
        dpi_profiles = {
            502: "modbus-inspect (allow read FC only from L3)",
            44818: "enip-inspect",
            4840: "opcua-inspect (require SignAndEncrypt)",
            102: "s7comm-inspect",
            20000: "dnp3-inspect",
        }
        return dpi_profiles.get(port, "none")

    def generate_migration_plan(self):
        """Generate phased migration plan for network segmentation."""
        plan = {
            "phase_1": {
                "name": "DMZ Implementation (Week 1-2)",
                "description": "Deploy DMZ between enterprise and OT networks",
                "steps": [
                    "Deploy DMZ firewall pair (inside and outside)",
                    "Migrate historian mirror to DMZ",
                    "Configure jump server in DMZ with MFA",
                    "Install data diode for unidirectional historian replication",
                    "Route enterprise-to-OT traffic through DMZ",
                    "Verify enterprise access to historian data via DMZ",
                ],
                "rollback": "Remove DMZ firewall rules, restore direct routing",
            },
            "phase_2": {
                "name": "L3/L2 Segmentation (Week 3-4)",
                "description": "Separate operations (L3) from control (L2) zones",
                "steps": [
                    "Create VLAN 30 and VLAN 40 on OT switches",
                    "Deploy industrial firewall between L2 and L3",
                    "Configure firewall in monitor mode (log only, no blocking)",
                    "Analyze logs for 1 week to validate rule completeness",
                    "Switch to enforcement mode during maintenance window",
                    "Validate all HMI-to-PLC and historian-to-PLC communications",
                ],
                "rollback": "Revert VLAN assignments, set firewall to permit-any",
            },
            "phase_3": {
                "name": "Field Device Isolation (Week 5-6)",
                "description": "Isolate Level 0-1 field devices from Level 2 supervisory",
                "steps": [
                    "Create VLAN 20 for PLCs and field instruments",