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网络分段设计工具。

分析流量基线数据，生成带VLAN分配、
防火墙规则和迁移计划的分段设计。
"""

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:
    """根据流量基线生成分段设计。"""

    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):
        """根据Purdue层级创建VLAN设计。"""
        self.vlans = [
            VLANDesign(10, "SIS-SAFETY", "1级（安全）",
                       "10.10.10.0/24", "10.10.10.1",
                       "安全仪表系统 - 气隙隔离或硬件隔离"),
            VLANDesign(20, "BPCS-FIELD", "0-1级（现场/控制）",
                       "10.10.20.0/24", "10.10.20.1",
                       "PLC、RTU、I/O模块、现场仪表"),
            VLANDesign(30, "BPCS-SUPERVISORY", "2级（监控）",
                       "10.10.30.0/24", "10.10.30.1",
                       "HMI、工程师工作站、本地历史服务器"),
            VLANDesign(40, "SITE-OPS", "3级（运营）",
                       "10.10.40.0/24", "10.10.40.1",
                       "站点历史服务器、OPC服务器、MES、报警管理"),
            VLANDesign(50, "OT-DMZ", "3.5级（DMZ）",
                       "172.16.50.0/24", "172.16.50.1",
                       "数据二极管、历史服务器镜像、跳板服务器、补丁服务器"),
            VLANDesign(60, "ENTERPRISE", "4级（企业）",
                       "10.0.60.0/24", "10.0.60.1",
                       "访问OT数据的企业IT系统"),
            VLANDesign(999, "QUARANTINE", "隔离区",
                       "10.10.99.0/24", "10.10.99.1",
                       "用于未授权或不受信任设备的隔离VLAN"),
        ]
        return self.vlans

    def generate_firewall_rules_from_baseline(self):
        """根据观察到的合法流量生成防火墙规则。"""
        self.rules = []

        # 各区域边界的默认拒绝规则
        zone_pairs = [
            ("2级", "0-1级"),
            ("3级", "2级"),
            ("3.5级", "3级"),
            ("4级", "3.5级"),
        ]

        # 从基线观察到的流量生成允许规则
        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"基线观察: {flow['src']} -> {flow['dst']}",
            ))
            self.rule_counter += 1

        # 在每个区域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"默认拒绝: {src_zone} -> {dst_zone}",
            ))
            self.rule_counter += 1

        return self.rules

    def _get_dpi_profile(self, port):
        """返回适合OT协议端口的DPI检测配置文件。"""
        dpi_profiles = {
            502: "modbus-inspect（仅允许来自L3的读取FC）",
            44818: "enip-inspect",
            4840: "opcua-inspect（需要SignAndEncrypt）",
            102: "s7comm-inspect",
            20000: "dnp3-inspect",
        }
        return dpi_profiles.get(port, "none")

    def generate_migration_plan(self):
        """生成网络分段的分阶段迁移计划。"""
        plan = {
            "phase_1": {
                "name": "DMZ实施（第1-2周）",
                "description": "在企业网络和OT网络之间部署DMZ",
                "steps": [
                    "部署DMZ防火墙对（内外两侧）",
                    "将历史服务器镜像迁移到DMZ",
                    "在DMZ中配置带MFA的跳板服务器",
                    "安装数据二极管用于单向历史服务器复制",
                    "通过DMZ路由企业到OT的流量",
                    "验证企业通过DMZ对历史数据的访问",
                ],
                "rollback": "移除DMZ防火墙规则，恢复直接路由",
            },
            "phase_2": {
                "name": "L3/L2分段（第3-4周）",
                "description": "将运营区（L3）与控制区（L2）分离",
                "steps": [
                    "在OT交换机上创建VLAN 30和VLAN 40",
                    "在L2和L3之间部署工业防火墙",
                    "以监控模式（仅记录，不阻断）配置防火墙",
                    "分析日志1周以验证规则完整性",
                    "在维护窗口期间切换到执行模式",
                    "验证所有HMI到PLC以及历史服务器到PLC的通信",
                ],
                "rollback": "还原VLAN分配，将防火墙设置为允许所有",
            },
            "phase_3": {
                "name": "现场设备隔离（第5-6周）",
                "description": "将0-1级现场设备与2级监控系统隔离",
                "steps": [
                    "为PLC和现场仪表创建VLAN 20",
                    "在PLC端口上配置MAC绑定的端口安全",
                    "应用Modbus功能码过滤（阻止来自L3的写入）",
                    "在维护窗口期间测试所有控制回路",
                    "验证从现场到HMI的报警传播",
                ],
                "rollback": "将VLAN 20合并回VLAN 30",
            },
            "phase_4": {
                "name": "SIS隔离（第7-8周）",
                "description": "完全隔离安全仪表系统",
                "steps": [
                    "验证SIS在专用VLAN 10上或气隙隔离",
                    "移除SIS与BPCS之间的所有网络路径",
                    "为SIS实施专用工程师工作站",
                    "在SIS EWS上应用USB和可移动介质控制",
                    "在隔离状态下测试SIS功能",
                ],
                "rollback": "不适用 - SIS隔离不应被撤销",
            },
        }
        return plan

    def export_design(self, output_file):
        """导出完整的分段设计。"""
        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"[*] 分段设计已导出到: {output_file}")
        print(f"    VLAN数量: {len(self.vlans)}")
        print(f"    防火墙规则数量: {len(self.rules)}")

        return design


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("用法: 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工业以太网4000/5000系列配置

# 创建与Purdue层级对齐的VLAN
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接入端口配置端口安全
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接入端口
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

# 到区域防火墙的中继
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

# 禁用并隔离所有未使用端口
interface range GigabitEthernet1/0/19-48
  description UNUSED - Shutdown
  switchport mode access
  switchport access vlan 999
  shutdown

步骤 3：验证分段有效性

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

#!/usr/bin/env python3
"""OT网络分段验证器。

运行自动化测试，在分段部署后验证区域隔离、
防火墙规则和协议执行情况。
"""

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"（被阻断）或"allowed"（被允许）
    actual_result: str = ""
    status: str = ""  # PASS或FAIL


class SegmentationValidator:
    """验证OT网络分段实施情况。"""

    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):
        """测试到目标的TCP连通性。"""
        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):
        """执行所有分段验证测试。"""
        print("=" * 60)
        print("OT分段验证")
        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"      目标: {test.target_ip}:{test.target_port}")
            print(f"      预期: {test.expected_result} | 实际: {actual} -> {test.status}")

        print(f"\n  结果: {passed} 通过, {failed} 失败, 共 {len(self.tests)} 个测试")
        return {"passed": passed, "failed": failed, "total": len(self.tests)}


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

    # 来自企业区（4级）的测试 - 应被阻止无法到达OT
    validator.add_test(ValidationTest(
        "SEG-001", "企业无法通过Modbus到达PLC",
        "4级", "10.10.20.10", 502, "blocked"))
    validator.add_test(ValidationTest(
        "SEG-002", "企业无法通过EtherNet/IP到达PLC",
        "4级", "10.10.20.10", 44818, "blocked"))
    validator.add_test(ValidationTest(
        "SEG-003", "企业可以访问DMZ跳板服务器",
        "4级", "172.16.50.10", 3389, "allowed"))
    validator.add_test(ValidationTest(
        "SEG-004", "企业可以访问DMZ历史服务器镜像",
        "4级", "172.16.50.20", 443, "allowed"))

    # 来自运营区（3级）的测试 - 对控制区的有限访问
    validator.add_test(ValidationTest(
        "SEG-005", "运营可以通过Modbus读取PLC",
        "3级", "10.10.20.10", 502, "allowed"))
    validator.add_test(ValidationTest(
        "SEG-006", "运营无法访问SIS控制器",
        "3级", "10.10.10.10", 1502, "blocked"))

    validator.run_all_tests()

核心概念

工具和系统

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

输出格式

OT网络分段报告
================================
实施日期: YYYY-MM-DD

VLAN架构:
  VLAN [ID] - [名称] ([Purdue层级])
    子网: [子网/掩码]
    设备数量: [数量]

防火墙规则:
  [区域A] -> [区域B]: [允许/拒绝数量]

验证结果:
  通过的测试: [N]/[总数]
  关键失败: [N]