performing-ot-network-security-assessment

执行 OT 网络安全评估

何时使用

• 当为新客户进行 OT/ICS 环境的初始安全基线时
• 当在 IT/OT 融合计划之后评估设施的安全态势时
• 当为 IEC 62443 或 NERC CIP 合规审计做准备时
• 当在涉及工业设施的合并或收购后评估风险时
• 当调查 OT 网络是否已被入侵，或是否存在未受监控的通往企业 IT 的通道时

不要使用于没有 OT 组件的纯 IT 网络评估、IT Web 应用的应用层漏洞扫描（参见 performing-web-app-penetration-test），或在没有明确授权和安全控制措施的情况下对实时 OT 系统进行主动利用。

前置条件

• 资产所有者和运营管理层对所有评估活动的书面授权
• 理解 Purdue 参考模型和 IEC 62443 区域/通道架构
• 被动网络监控工具（Nozomi Guardian、Dragos Platform，或带工业协议解析器的 Wireshark）
• 访问网络拓扑图、防火墙规则集和资产清单（或执行被动发现的能力）
• 关于受评估 OT 系统所控制物理过程的安全简报

工作流

步骤 1：确定评估范围和安全边界

根据 Purdue 参考模型层级定义范围，并识别不得主动扫描的安全关键系统。OT 评估与 IT 评估有根本不同，因为主动扫描可能导致 PLC 崩溃、中断安全仪表系统（SIS），并造成人身伤害。

# OT Assessment Scope Definition
assessment:
  facility: "Chemical Processing Plant - Site Alpha"
  purdue_levels_in_scope:
    - level_0: "Physical process sensors and actuators (passive observation only)"
    - level_1: "PLCs, RTUs, safety controllers (passive only, no active scanning)"
    - level_2: "HMI stations, engineering workstations, historian (limited active with approval)"
    - level_3: "Site operations - OPC servers, application servers (active scanning permitted)"
    - level_3_5: "DMZ - data diodes, jump servers (active scanning permitted)"
    - level_4: "Enterprise IT connecting to OT (active scanning permitted)"

  safety_exclusions:
    - "Safety Instrumented Systems (SIS) - Triconex controllers"
    - "Emergency Shutdown (ESD) systems"
    - "Fire and Gas detection systems"
    - "Any Level 0/1 device during active production"

  authorized_activities:
    passive:
      - "Network traffic capture and analysis via SPAN ports"
      - "Industrial protocol deep packet inspection"
      - "Wireless spectrum analysis"
      - "Physical walkthrough and visual inspection"
    active_with_approval:
      - "Targeted Nmap scans of Level 2-4 systems during maintenance windows"
      - "Authentication testing on HMI and engineering workstations"
      - "Firewall rule verification between zones"
    prohibited:
      - "Active scanning of PLCs, RTUs, or SIS controllers"
      - "Fuzzing industrial protocols on live systems"
      - "Modifying PLC logic or firmware"

步骤 2：执行被动网络发现和资产盘点

部署被动监控，以映射 OT 网络上的所有设备、通信流和协议，而不发送任何可能中断运营的流量。

#!/usr/bin/env python3
"""OT Network Passive Discovery and Asset Inventory Builder.

Uses pcap captures from SPAN ports to identify OT assets, protocols,
and communication patterns without active scanning.
"""

import json
import sys
from collections import defaultdict
from datetime import datetime

try:
    from scapy.all import rdpcap, IP, TCP, UDP
    from scapy.contrib.modbus import ModbusADURequest, ModbusADUResponse
except ImportError:
    print("Install scapy: pip install scapy")
    sys.exit(1)

# Industrial protocol port mappings
OT_PROTOCOL_PORTS = {
    502: "Modbus/TCP",
    102: "S7comm (Siemens)",
    44818: "EtherNet/IP (CIP)",
    2222: "EtherNet/IP (implicit)",
    4840: "OPC UA",
    20000: "DNP3",
    47808: "BACnet",
    1911: "Niagara Fox",
    789: "Crimson v3 (Red Lion)",
    2404: "IEC 60870-5-104",
    18245: "GE SRTP",
    5094: "HART-IP",
}

PURDUE_LEVEL_RANGES = {
    "Level 0-1 (Field Devices)": ["10.10.0.0/16", "192.168.10.0/24"],
    "Level 2 (Control Systems)": ["10.20.0.0/16", "192.168.20.0/24"],
    "Level 3 (Site Operations)": ["10.30.0.0/16", "192.168.30.0/24"],
    "Level 3.5 (DMZ)": ["172.16.0.0/16"],
    "Level 4 (Enterprise)": ["10.0.0.0/16"],
}


def classify_purdue_level(ip_addr):
    """Classify an IP address to its Purdue Reference Model level."""
    from ipaddress import ip_address, ip_network

    addr = ip_address(ip_addr)
    for level, subnets in PURDUE_LEVEL_RANGES.items():
        for subnet in subnets:
            if addr in ip_network(subnet):
                return level
    return "Unknown"


def analyze_ot_pcap(pcap_file):
    """Analyze pcap file to discover OT assets and communication patterns."""
    packets = rdpcap(pcap_file)

    assets = {}
    connections = defaultdict(lambda: {"count": 0, "protocols": set(), "ports": set()})
    protocol_stats = defaultdict(int)
    cross_zone_flows = []

    for pkt in packets:
        if not pkt.haslayer(IP):
            continue

        src_ip = pkt[IP].src
        dst_ip = pkt[IP].dst

        # Track assets
        for ip in (src_ip, dst_ip):
            if ip not in assets:
                assets[ip] = {
                    "ip": ip,
                    "purdue_level": classify_purdue_level(ip),
                    "protocols_observed": set(),
                    "roles": set(),
                    "first_seen": str(pkt.time),
                    "last_seen": str(pkt.time),
                    "mac": None,
                }
            assets[ip]["last_seen"] = str(pkt.time)

        # Identify OT protocols by port
        dst_port = None
        if pkt.haslayer(TCP):
            dst_port = pkt[TCP].dport
        elif pkt.haslayer(UDP):
            dst_port = pkt[UDP].dport

        if dst_port and dst_port in OT_PROTOCOL_PORTS:
            protocol_name = OT_PROTOCOL_PORTS[dst_port]
            protocol_stats[protocol_name] += 1
            assets[src_ip]["protocols_observed"].add(protocol_name)
            assets[dst_ip]["protocols_observed"].add(protocol_name)

            # Determine roles
            assets[src_ip]["roles"].add("client/master")
            assets[dst_ip]["roles"].add("server/slave")

        # Track connections
        conn_key = (src_ip, dst_ip)
        connections[conn_key]["count"] += 1
        if dst_port:
            connections[conn_key]["ports"].add(dst_port)
            if dst_port in OT_PROTOCOL_PORTS:
                connections[conn_key]["protocols"].add(OT_PROTOCOL_PORTS[dst_port])

        # Detect cross-zone communication
        src_level = classify_purdue_level(src_ip)
        dst_level = classify_purdue_level(dst_ip)
        if src_level != dst_level and src_level != "Unknown" and dst_level != "Unknown":
            cross_zone_flows.append({
                "src": src_ip,
                "src_level": src_level,
                "dst": dst_ip,
                "dst_level": dst_level,
                "protocol": OT_PROTOCOL_PORTS.get(dst_port, f"port/{dst_port}"),
            })

    return {
        "asset_count": len(assets),
        "assets": {ip: {k: list(v) if isinstance(v, set) else v for k, v in info.items()} for ip, info in assets.items()},
        "protocol_distribution": dict(protocol_stats),
        "total_connections": len(connections),
        "cross_zone_flows": cross_zone_flows[:50],
    }


def generate_assessment_report(results):
    """Generate the OT network assessment findings report."""
    report = []
    report.append("=" * 70)
    report.append("OT NETWORK PASSIVE DISCOVERY REPORT")
    report.append(f"Generated: {datetime.now().isoformat()}")
    report.append("=" * 70)

    report.append(f"\nTotal Assets Discovered: {results['asset_count']}")
    report.append(f"Total Unique Connections: {results['total_connections']}")

    report.append("\n--- INDUSTRIAL PROTOCOL DISTRIBUTION ---")
    for proto, count in sorted(results["protocol_distribution"].items(), key=lambda x: -x[1]):
        report.append(f"  {proto}: {count} packets")

    report.append("\n--- CROSS-ZONE COMMUNICATION FLOWS ---")
    if results["cross_zone_flows"]:
        for flow in results["cross_zone_flows"][:20]:
            report.append(
                f"  {flow['src']} ({flow['src_level']}) -> "
                f"{flow['dst']} ({flow['dst_level']}) via {flow['protocol']}"
            )
    else:
        report.append("  No cross-zone flows detected (check subnet classifications)")

    report.append("\n--- FINDINGS ---")
    # Check for Level 4 to Level 0-1 direct connections (critical finding)
    for flow in results["cross_zone_flows"]:
        if "Level 4" in flow["src_level"] and "Level 0-1" in flow["dst_level"]:
            report.append(
                f"  [CRITICAL] Direct Enterprise-to-Field traffic: "
                f"{flow['src']} -> {flow['dst']} via {flow['protocol']}"
            )
        elif "Level 4" in flow["src_level"] and "Level 2" in flow["dst_level"]:
            report.append(
                f"  [HIGH] Enterprise-to-Control traffic bypassing DMZ: "
                f"{flow['src']} -> {flow['dst']} via {flow['protocol']}"
            )

    return "\n".join(report)


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python ot_network_discovery.py <pcap_file>")
        sys.exit(1)

    results = analyze_ot_pcap(sys.argv[1])
    print(generate_assessment_report(results))

    # Save detailed JSON
    output_file = sys.argv[1].replace(".pcap", "_inventory.json")
    with open(output_file, "w") as f:
        json.dump(results, f, indent=2, default=str)
    print(f"\nDetailed inventory saved to: {output_file}")

步骤 3：评估 Purdue 区域之间的防火墙规则

分析 OT 区域之间的防火墙配置，以识别过于宽松的规则、缺失的默认拒绝，以及违反 IEC 62443 区域模型的未授权通道。

#!/usr/bin/env python3
"""OT Zone Firewall Rule Analyzer.

Parses firewall rule exports (CSV format) and evaluates them against
IEC 62443 zone/conduit model requirements.
"""

import csv
import json
import sys
from dataclasses import dataclass, field
from typing import Optional


@dataclass
class FirewallRule:
    rule_id: str
    source_zone: str
    source_ip: str
    dest_zone: str
    dest_ip: str
    service: str
    port: str
    action: str
    enabled: bool
    comment: str = ""


# IEC 62443 zone communication policy
# Defines which zone pairs are allowed to communicate and through what conduit
ALLOWED_CONDUITS = {
    ("Level 4", "Level 3.5"): {
        "allowed_ports": [443, 3389, 22],
        "description": "Enterprise to DMZ - web services, jump hosts",
        "requires_inspection": True,
    },
    ("Level 3.5", "Level 3"): {
        "allowed_ports": [443, 1433, 5432, 8080],
        "description": "DMZ to Site Ops - historian mirror, OPC relay",
        "requires_inspection": True,
    },
    ("Level 3", "Level 2"): {
        "allowed_ports": [502, 44818, 4840, 102],
        "description": "Site Ops to Control - OPC UA, Modbus, S7comm",
        "requires_inspection": True,
    },
    ("Level 2", "Level 1"): {
        "allowed_ports": [502, 44818, 102, 2222],
        "description": "Control to Field - direct industrial protocols",
        "requires_inspection": False,
    },
}

# Prohibited direct connections
PROHIBITED_CONDUITS = [
    ("Level 4", "Level 3"),
    ("Level 4", "Level 2"),
    ("Level 4", "Level 1"),
    ("Level 4", "Level 0"),
    ("Level 3", "Level 1"),
    ("Level 3", "Level 0"),
    ("Internet", "Level 3.5"),
    ("Internet", "Level 3"),
    ("Internet", "Level 2"),
    ("Internet", "Level 1"),
]


def parse_firewall_rules(csv_file):
    """Parse firewall rules from CSV export."""
    rules = []
    with open(csv_file, "r") as f:
        reader = csv.DictReader(f)
        for row in reader:
            rules.append(FirewallRule(
                rule_id=row.get("rule_id", ""),
                source_zone=row.get("source_zone", ""),
                source_ip=row.get("source_ip", ""),
                dest_zone=row.get("dest_zone", ""),
                dest_ip=row.get("dest_ip", ""),
                service=row.get("service", ""),
                port=row.get("port", ""),
                action=row.get("action", ""),
                enabled=row.get("enabled", "true").lower() == "true",
                comment=row.get("comment", ""),
            ))
    return rules


def analyze_rules(rules):
    """Analyze firewall rules against IEC 62443 zone model."""
    findings = {"critical": [], "high": [], "medium": [], "low": [], "info": []}

    for rule in rules:
        if not rule.enabled or rule.action.lower() != "allow":
            continue

        zone_pair = (rule.source_zone, rule.dest_zone)
        port = int(rule.port) if rule.port.isdigit() else 0

        # Check for prohibited conduits
        if zone_pair in PROHIBITED_CONDUITS:
            findings["critical"].append({
                "rule_id": rule.rule_id,
                "finding": f"Prohibited direct connection: {rule.source_zone} -> {rule.dest_zone}",
                "detail": f"Rule allows {rule.source_ip} to reach {rule.dest_ip}:{rule.port} ({rule.service})",
                "remediation": "Remove rule. Route traffic through DMZ (Level 3.5) with application-layer inspection.",
            })

        # Check for overly broad rules (any/any)
        elif rule.source_ip in ("any", "0.0.0.0/0") or rule.dest_ip in ("any", "0.0.0.0/0"):
            findings["high"].append({
                "rule_id": rule.rule_id,
                "finding": f"Overly permissive rule with 'any' address",
                "detail": f"{rule.source_ip} -> {rule.dest_ip}:{rule.port} in {zone_pair}",
                "remediation": "Restrict to specific host IPs per IEC 62443 least-privilege conduit policy.",
            })

        # Check allowed conduits for port violations
        elif zone_pair in ALLOWED_CONDUITS:
            conduit = ALLOWED_CONDUITS[zone_pair]
            if port and port not in conduit["allowed_ports"]:
                findings["medium"].append({
                    "rule_id": rule.rule_id,
                    "finding": f"Unauthorized port in conduit {zone_pair}",
                    "detail": f"Port {port} ({rule.service}) not in allowed list {conduit['allowed_ports']}",
                    "remediation": f"Remove port {port} from conduit or justify in risk assessment.",
                })

    return findings


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python ot_firewall_analyzer.py <rules.csv>")
        sys.exit(1)

    rules = parse_firewall_rules(sys.argv[1])
    findings = analyze_rules(rules)

    print("=" * 70)
    print("OT ZONE FIREWALL RULE ANALYSIS")
    print("=" * 70)
    for severity in ["critical", "high", "medium", "low"]:
        if findings[severity]:
            print(f"\n--- {severity.upper()} FINDINGS ({len(findings[severity])}) ---")
            for f in findings[severity]:
                print(f"  [{f['rule_id']}] {f['finding']}")
                print(f"    Detail: {f['detail']}")
                print(f"    Fix: {f['remediation']}")

步骤 4：评估工业协议安全

评估正在使用的工业协议的安全配置，检查身份验证、加密和访问控制。

# Capture Modbus/TCP traffic for analysis
tcpdump -i eth0 -w ot_capture.pcap 'port 502 or port 44818 or port 4840 or port 102 or port 20000' -c 100000

# Use Wireshark with OT protocol dissectors for deep inspection
tshark -r ot_capture.pcap -Y "modbus" -T fields \
  -e ip.src -e ip.dst -e modbus.func_code -e modbus.reference_num \
  -e modbus.word_cnt > modbus_analysis.csv

# Check for unauthenticated Modbus write operations (function codes 5,6,15,16)
tshark -r ot_capture.pcap -Y "modbus.func_code >= 5 && modbus.func_code <= 16" \
  -T fields -e ip.src -e ip.dst -e modbus.func_code -e frame.time

# Scan for OPC UA servers and check security policies
# Only run against Level 3+ systems with explicit authorization
python3 -c "
from opcua import Client
server_url = 'opc.tcp://10.30.1.50:4840'
client = Client(server_url)
endpoints = client.connect_and_get_server_endpoints()
for ep in endpoints:
    print(f'Endpoint: {ep.EndpointUrl}')
    print(f'  Security Mode: {ep.SecurityMode}')
    print(f'  Security Policy: {ep.SecurityPolicyUri}')
    print(f'  Auth Tokens: {[t.TokenType for t in ep.UserIdentityTokens]}')
"

步骤 5：生成评估报告

将发现编译为符合 IEC 62443 和 NIST SP 800-82 Rev.3 的结构化报告。

OT Network Security Assessment Report
=======================================
Facility: Chemical Processing Plant - Site Alpha
Assessment Date: 2026-02-23
Standard: IEC 62443-3-3 / NIST SP 800-82r3
Assessor: [Assessor Name]

EXECUTIVE SUMMARY:
  The OT network assessment identified 47 assets across Purdue levels 0-4.
  12 critical and 23 high-severity findings were identified, primarily
  related to insufficient network segmentation, unauthenticated industrial
  protocols, and unauthorized cross-zone communication paths.

ASSET INVENTORY SUMMARY:
  Level 0-1 (Field):    18 devices (PLCs, RTUs, I/O modules)
  Level 2 (Control):     9 devices (HMIs, engineering workstations)
  Level 3 (Operations): 12 devices (historians, OPC servers, app servers)
  Level 3.5 (DMZ):       3 devices (data diode, jump server, patch server)
  Level 4 (Enterprise):  5 devices (domain controllers, file servers)

CRITICAL FINDINGS:
  [OT-001] Direct Enterprise-to-PLC communication detected
    Source: 10.0.5.22 (Level 4 - IT workstation)
    Dest: 10.10.1.15 (Level 1 - Allen-Bradley PLC)
    Protocol: EtherNet/IP (port 44818)
    Impact: An attacker on the corporate network could directly modify PLC logic
    Remediation: Block direct L4-L1 traffic; route through DMZ proxy

  [OT-002] Modbus/TCP write commands without authentication
    Affected: 8 PLCs accepting unauthenticated FC6 (Write Single Register)
    Impact: Any device on the OT network can modify process setpoints
    Remediation: Deploy Modbus-aware firewall; restrict write-capable sources

  [OT-003] Flat network - no segmentation between Purdue levels
    Detail: All OT devices share VLAN 100 (10.10.0.0/16)
    Impact: Compromised HMI has direct access to all PLCs and SIS
    Remediation: Implement zone-based segmentation per IEC 62443-3-2

RISK MATRIX:
  Critical: 12 findings (immediate remediation required)
  High:     23 findings (remediate within 30 days)
  Medium:   15 findings (remediate within 90 days)
  Low:       8 findings (remediate in next maintenance cycle)

关键概念

术语	定义
Purdue 参考模型	用于组织工业控制系统的分层架构模型（0-5 级），定义从物理过程到企业 IT 的安全区域
IEC 62443	工业自动化和控制系统（IACS）安全的国际标准系列，定义安全级别、区域、通道和安全要求
区域	由 IEC 62443-3-2 定义的、共享通用安全要求的逻辑或物理资产分组
通道	连接两个或多个区域的通信通道的逻辑分组，受通用安全策略约束
SCADA	监控与数据采集 - 用于工业过程高级过程监控管理的系统架构
DCS	分布式控制系统 - 控制元素分布在整个系统中的控制系统架构
空气隔离	OT 网络与 IT/互联网的物理隔离，正日益被带防火墙和数据二极管的受管理通道所取代
安全仪表系统（SIS）	设计用于在检测到危险状况时使过程进入安全状态的独立系统

工具与系统

• Nozomi Networks Guardian：被动式 OT 网络监控平台，为工业环境提供资产发现、漏洞评估和异常检测
• Dragos Platform：OT 网络安全平台，专为关键基础设施设计，具备资产可见性、威胁检测和漏洞管理功能
• Claroty xDome：网络物理系统保护平台，跨 OT、IoT 和 IIoT 提供全面的资产清单和风险评分
• Wireshark/tshark：网络协议分析器，配有适用于 Modbus、DNP3、S7comm、EtherNet/IP、OPC UA 和 BACnet 的工业协议解析器
• 带 OT 脚本的 Nmap：网络扫描器，配有用于 OT 协议枚举的 NSE 脚本（仅在获得授权的情况下于 Level 2+ 使用）
• Grassmarlin：NSA 开发的被动式 OT 网络测绘工具，用于识别 SCADA/ICS 网络拓扑

常见场景

场景：无分段的扁平 OT 网络

背景：某水务公司将所有 OT 设备置于单一 VLAN 上。被动网络监控发现 HMI、PLC、历史数据库和域控制器共享同一 Layer 2 广播域。企业网络与 OT 环境之间没有 DMZ。

方法：

1. 在 SPAN 端口部署被动监控，在 2-4 周内捕获完整的通信基线
2. 绘制所有设备间通信流，包括协议和数据量
3. 根据设备的功能和通信模式将其归类到 Purdue 层级
4. 依据 IEC 62443-3-2 设计包含 VLAN 和区域间防火墙的区域架构
5. 优先在 Level 3 和 Level 4 之间创建 DMZ，作为影响最大的分段措施
6. 提交分段方案，包含避免生产中断的迁移阶段

注意事项：在生产期间对 PLC 进行主动扫描可能导致通信超时和过程中断。在没有完整流量基线的情况下实施分段会破坏合法的控制系统通信。仅依赖网络层防火墙而不进行工业协议检查，会使 Modbus/TCP 和 EtherNet/IP 写命令不受检查。

输出格式

OT Network Security Assessment Report
=======================================
Facility: [Facility Name]
Assessment Date: YYYY-MM-DD
Standard: IEC 62443-3-3 / NIST SP 800-82r3

EXECUTIVE SUMMARY:
  [2-3 sentence overview of findings and risk level]

ASSET INVENTORY:
  Level 0-1: [count] field devices
  Level 2:   [count] control systems
  Level 3:   [count] operations systems
  Level 3.5: [count] DMZ systems
  Level 4:   [count] enterprise systems

FINDINGS BY SEVERITY:
  Critical: [count] (immediate action required)
  High:     [count] (30-day remediation)
  Medium:   [count] (90-day remediation)
  Low:      [count] (next maintenance window)

DETAILED FINDINGS:
  [OT-NNN] Finding Title
    Severity: Critical|High|Medium|Low
    Affected Assets: [list]
    IEC 62443 Reference: [section]
    NIST 800-82r3 Reference: [section]
    Description: [technical detail]
    Impact: [operational and safety impact]
    Remediation: [specific technical remediation steps]