implementing-network-segmentation-for-ot

为 OT 实施网络分段

何时使用

• 当 OT 安全评估发现一个扁平网络，Purdue 层级之间没有分段时
• 在完成风险评估（IEC 62443-3-2）后实施 IEC 62443 区域/管道架构时
• 作为 IT/OT 融合安全计划的一部分，分离 IT 和 OT 网络时
• 在企业 IT 和 OT 之间部署 DMZ，以保护工业系统免受源自 IT 的威胁时
• 将安全仪表系统（SIS）与基本过程控制系统（BPCS）分段时

请勿使用于没有 OT 组件的纯 IT 微分段（参见 implementing-zero-trust-in-cloud），也不要在未事先进行流量分析的情况下用于初始区域设计（请先参见 performing-ot-network-security-assessment）。

前提条件

• 通过被动监控获得的完整流量基线（至少 2-4 周的捕获数据）
• 包含所有 OT 设备 Purdue 层级分类的资产清单
• 支持 VLAN 和端口安全的工业级网络交换机
• 具备 OT 感知能力的防火墙（Cisco ISA-3000、Fortinet FortiGate Rugged、带 OT Security 的 Palo Alto）
• 网络变更的维护窗口安排
• 经运营管理层批准的回滚计划

工作流

步骤 1：基于流量基线设计分段架构

使用流量基线设计 VLAN 和防火墙架构，在隔离区域的同时保留所有合法通信路径。

#!/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",
                    "Configure port security with MAC binding on PLC ports",
                    "Apply Modbus function code filtering (block writes from L3)",
                    "Test all control loops during maintenance window",
                    "Verify alarm propagation from field to HMI",
                ],
                "rollback": "Merge VLAN 20 back into VLAN 30",
            },
            "phase_4": {
                "name": "SIS Isolation (Week 7-8)",
                "description": "Fully isolate Safety Instrumented Systems",
                "steps": [
                    "Verify SIS is on dedicated VLAN 10 or air-gapped",
                    "Remove any network path between SIS and BPCS",
                    "Implement dedicated engineering workstation for SIS",
                    "Apply USB and removable media controls on SIS EWS",
                    "Test SIS functionality in isolation",
                ],
                "rollback": "N/A - SIS isolation should not be reversed",
            },
        }
        return plan

    def export_design(self, output_file):
        """Export complete segmentation design."""
        design = {
            "vlans": [asdict(v) for v in self.vlans],
            "firewall_rules": [asdict(r) for r in self.rules],
            "migration_plan": self.generate_migration_plan(),
        }

        with open(output_file, "w") as f:
            json.dump(design, f, indent=2)

        print(f"[*] Segmentation design exported to: {output_file}")
        print(f"    VLANs: {len(self.vlans)}")
        print(f"    Firewall Rules: {len(self.rules)}")

        return design


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python segmentation_designer.py <baseline.json> [output.json]")
        sys.exit(1)

    designer = SegmentationDesigner(sys.argv[1])
    designer.design_vlans()
    designer.generate_firewall_rules_from_baseline()

    output = sys.argv[2] if len(sys.argv) > 2 else "segmentation_design.json"
    designer.export_design(output)

步骤 2：配置工业交换机 VLAN

将 VLAN 配置应用于工业以太网交换机，并启用端口安全和未使用端口加固。

# Cisco Industrial Ethernet 4000/5000 Series Configuration

# Create VLANs aligned with Purdue levels
vlan 10
  name SIS-SAFETY-L1
vlan 20
  name BPCS-FIELD-L01
vlan 30
  name BPCS-SUPERVISORY-L2
vlan 40
  name SITE-OPS-L3
vlan 50
  name OT-DMZ-L35
vlan 999
  name QUARANTINE

# PLC access ports with port security
interface range GigabitEthernet1/0/1-12
  description PLC Connections
  switchport mode access
  switchport access vlan 20
  switchport port-security
  switchport port-security maximum 1
  switchport port-security mac-address sticky
  switchport port-security violation shutdown
  storm-control broadcast level 10
  storm-control multicast level 10
  spanning-tree portfast
  spanning-tree bpduguard enable
  no cdp enable
  no lldp transmit
  no lldp receive

# HMI access ports
interface range GigabitEthernet1/0/13-18
  description HMI Stations
  switchport mode access
  switchport access vlan 30
  switchport port-security
  switchport port-security maximum 1
  switchport port-security mac-address sticky
  switchport port-security violation restrict
  spanning-tree portfast

# Trunk to zone firewall
interface TenGigabitEthernet1/0/1
  description Trunk to OT Zone Firewall
  switchport mode trunk
  switchport trunk allowed vlan 20,30,40,50
  switchport trunk native vlan 999
  switchport nonegotiate

# Disable and quarantine all unused ports
interface range GigabitEthernet1/0/19-48
  description UNUSED - Shutdown
  switchport mode access
  switchport access vlan 999
  shutdown

步骤 3：验证分段有效性

实施后，验证分段能够正确阻止未经授权的跨区域流量，同时允许所有合法操作。

#!/usr/bin/env python3
"""OT Network Segmentation Validator.

Runs automated tests to verify zone isolation, firewall rules,
and protocol enforcement after segmentation deployment.
"""

import json
import socket
import subprocess
import sys
import time
from dataclasses import dataclass, asdict


@dataclass
class ValidationTest:
    test_id: str
    description: str
    source_zone: str
    target_ip: str
    target_port: int
    expected_result: str  # "blocked" or "allowed"
    actual_result: str = ""
    status: str = ""  # PASS or FAIL


class SegmentationValidator:
    """Validates OT network segmentation implementation."""

    def __init__(self):
        self.tests = []
        self.results = []

    def add_test(self, test):
        self.tests.append(test)

    def run_connectivity_test(self, target_ip, target_port, timeout=3):
        """Test TCP connectivity to target."""
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(timeout)
            result = sock.connect_ex((target_ip, target_port))
            sock.close()
            return "reachable" if result == 0 else "blocked"
        except (socket.timeout, ConnectionRefusedError):
            return "blocked"
        except Exception:
            return "error"

    def run_all_tests(self):
        """Execute all segmentation validation tests."""
        print("=" * 60)
        print("OT SEGMENTATION VALIDATION")
        print("=" * 60)

        passed = 0
        failed = 0

        for test in self.tests:
            actual = self.run_connectivity_test(test.target_ip, test.target_port)
            test.actual_result = actual

            if actual == test.expected_result:
                test.status = "PASS"
                passed += 1
            else:
                test.status = "FAIL"
                failed += 1

            icon = "[+]" if test.status == "PASS" else "[-]"
            print(f"  {icon} {test.test_id}: {test.description}")
            print(f"      Target: {test.target_ip}:{test.target_port}")
            print(f"      Expected: {test.expected_result} | Actual: {actual} -> {test.status}")

        print(f"\n  Results: {passed} passed, {failed} failed out of {len(self.tests)} tests")
        return {"passed": passed, "failed": failed, "total": len(self.tests)}


if __name__ == "__main__":
    validator = SegmentationValidator()

    # Tests from Enterprise zone (Level 4) - should be blocked from OT
    validator.add_test(ValidationTest(
        "SEG-001", "Enterprise cannot reach PLCs via Modbus",
        "Level 4", "10.10.20.10", 502, "blocked"))
    validator.add_test(ValidationTest(
        "SEG-002", "Enterprise cannot reach PLCs via EtherNet/IP",
        "Level 4", "10.10.20.10", 44818, "blocked"))
    validator.add_test(ValidationTest(
        "SEG-003", "Enterprise can reach DMZ jump server",
        "Level 4", "172.16.50.10", 3389, "allowed"))
    validator.add_test(ValidationTest(
        "SEG-004", "Enterprise can reach DMZ historian mirror",
        "Level 4", "172.16.50.20", 443, "allowed"))

    # Tests from Operations zone (Level 3) - limited access to control
    validator.add_test(ValidationTest(
        "SEG-005", "Operations can read from PLCs via Modbus",
        "Level 3", "10.10.20.10", 502, "allowed"))
    validator.add_test(ValidationTest(
        "SEG-006", "Operations cannot reach SIS controllers",
        "Level 3", "10.10.10.10", 1502, "blocked"))

    validator.run_all_tests()

关键概念

术语	定义
VLAN	虚拟局域网（Virtual Local Area Network）—— 二层广播域隔离，用于在共享交换机基础设施上分隔 OT 区域
工业防火墙	具备工业协议（Modbus、DNP3、EtherNet/IP、OPC UA）深度包检测能力的防火墙
数据二极管	硬件强制单向网关，从物理上阻止反向数据流，用于 OT 运营与 DMZ 之间
端口安全	交换机功能，用于限制端口上的 MAC 地址数量并锁定分配，防止未经授权的设备连接
中继端口	使用 802.1Q 标记承载多个 VLAN 的交换机端口，用于跨区域边界连接交换机和防火墙
DMZ	企业 IT 与 OT 之间的隔离区 —— 所有跨域流量在此终止并接受检查的缓冲区

工具与系统

• Cisco ISA-3000：工业安全设备，具备 Modbus、DNP3 和 EtherNet/IP 深度包检测，用于 OT 区域防火墙
• Fortinet FortiGate Rugged 系列：加固型 NGFW，支持 OT 协议并具备工业环境认证
• Waterfall Security 单向网关：硬件数据二极管，用于强制 OT 到 IT 的单向数据流
• Cisco 工业以太网交换机：支持 VLAN、端口安全和工业协议的可管理交换机

输出格式

OT Network Segmentation Report
================================
Implementation Date: YYYY-MM-DD

VLAN ARCHITECTURE:
  VLAN [ID] - [Name] ([Purdue Level])
    Subnet: [subnet/mask]
    Devices: [count]

FIREWALL RULES:
  [Zone A] -> [Zone B]: [allow/deny count]

VALIDATION RESULTS:
  Tests Passed: [N]/[Total]
  Critical Failures: [N]