performing-ot-vulnerability-scanning-safely

安全地执行 OT 漏洞扫描

适用场景

• 在具有老旧控制器的 OT 环境中进行漏洞评估时
• 在不影响过程可用性的前提下实施持续漏洞监控时
• 为需要漏洞数据的 IEC 62443 或 NERC CIP 合规审计做准备时
• 评估 OT 资产的基于风险的补丁修复优先级时
• 验证补偿性控制措施能否保护未打补丁的 ICS 设备时

请勿使用于对生产 PLC 进行激进主动扫描（可能导致老旧控制器崩溃）、在 OT 网络上使用标准 Nessus 配置文件进行 IT 漏洞扫描，或对运行中的 OT 系统进行渗透测试（参见 performing-ics-penetration-testing）。

前提条件

• Tenable OT Security（原 Tenable.ot/Indegy）或同等的 OT 安全扫描平台
• 在 OT 网段的 SPAN/TAP 上部署的被动监控传感器
• 在生产使用前已针对每种设备类型验证过的实验室测试扫描配置文件
• 任何主动扫描的变更管理审批和维护窗口
• 供应商保修验证，确认扫描不会导致支持协议失效

工作流

步骤 1：部署被动漏洞检测

被动监控无需向 OT 设备发送任何数据包即可识别漏洞。

#!/usr/bin/env python3
"""OT Safe Vulnerability Scanner Orchestrator.

Coordinates passive monitoring, native protocol queries, and carefully
controlled active scanning for OT vulnerability assessment without
disrupting industrial operations.
"""

import json
import csv
import sys
from datetime import datetime
from typing import Dict, List, Optional

try:
    import requests
except ImportError:
    print("Install requests: pip install requests")
    sys.exit(1)


class OTVulnerabilityScanner:
    """Safe OT vulnerability scanning orchestrator."""

    SCAN_SAFETY_LEVELS = {
        "passive": {
            "description": "Observe network traffic only, zero risk to devices",
            "risk_level": "NONE",
            "methods": ["traffic_fingerprinting", "protocol_analysis", "version_detection"],
            "requires_window": False,
        },
        "native_query": {
            "description": "Query devices using native industrial protocols",
            "risk_level": "MINIMAL",
            "methods": ["modbus_device_id", "s7_szl_read", "cip_identity", "bacnet_whois"],
            "requires_window": True,
        },
        "controlled_active": {
            "description": "Standard vulnerability checks with OT-safe profiles",
            "risk_level": "LOW-MODERATE",
            "methods": ["credentialed_scan", "banner_grab", "service_detection"],
            "requires_window": True,
        },
    }

    def __init__(self, tenable_url: str, api_key: str, verify_ssl: bool = True):
        self.tenable_url = tenable_url.rstrip("/")
        self.session = requests.Session()
        self.session.headers.update({
            "X-ApiKeys": f"accessKey={api_key}",
            "Content-Type": "application/json",
        })
        self.session.verify = verify_ssl
        self.findings = []

    def check_safety_prerequisites(self, scan_level: str, target_subnet: str) -> dict:
        """Verify safety prerequisites before scanning."""
        checks = {
            "scan_level": scan_level,
            "target": target_subnet,
            "safety_level": self.SCAN_SAFETY_LEVELS[scan_level],
            "checks_passed": [],
            "checks_failed": [],
            "approved": False,
        }

        prerequisites = [
            {
                "name": "Lab validation complete",
                "description": "Scan profile tested against each device type in lab environment",
                "required_for": ["native_query", "controlled_active"],
            },
            {
                "name": "Vendor warranty verified",
                "description": "Scanning will not void vendor support agreements",
                "required_for": ["native_query", "controlled_active"],
            },
            {
                "name": "Change management approved",
                "description": "Change ticket approved for scanning activity",
                "required_for": ["native_query", "controlled_active"],
            },
            {
                "name": "Maintenance window confirmed",
                "description": "Operations team confirms acceptable scanning window",
                "required_for": ["controlled_active"],
            },
            {
                "name": "Rollback plan documented",
                "description": "Procedure to stop scan and recover if device becomes unresponsive",
                "required_for": ["controlled_active"],
            },
            {
                "name": "SIS excluded from scope",
                "description": "Safety Instrumented Systems are never actively scanned",
                "required_for": ["passive", "native_query", "controlled_active"],
            },
        ]

        for prereq in prerequisites:
            if scan_level in prereq["required_for"]:
                checks["checks_passed"].append(prereq["name"])

        return checks

    def run_passive_assessment(self, site_id: str):
        """Run passive vulnerability assessment using traffic analysis."""
        print(f"[*] Running passive vulnerability assessment for site {site_id}")
        print(f"[*] Safety Level: NONE - no packets sent to OT devices")

        try:
            resp = self.session.get(
                f"{self.tenable_url}/api/v1/assets",
                params={"site_id": site_id}
            )
            resp.raise_for_status()
            assets = resp.json().get("assets", [])

            for asset in assets:
                asset_id = asset.get("id")
                vuln_resp = self.session.get(
                    f"{self.tenable_url}/api/v1/assets/{asset_id}/vulnerabilities"
                )
                if vuln_resp.status_code == 200:
                    vulns = vuln_resp.json().get("vulnerabilities", [])
                    for vuln in vulns:
                        self.findings.append({
                            "asset": asset.get("name", "Unknown"),
                            "ip": asset.get("ip_address", ""),
                            "type": asset.get("type", ""),
                            "vendor": asset.get("vendor", ""),
                            "cve": vuln.get("cve_id", ""),
                            "severity": vuln.get("severity", ""),
                            "cvss": vuln.get("cvss_score", 0),
                            "description": vuln.get("description", ""),
                            "detection_method": "passive",
                            "remediation": vuln.get("remediation", ""),
                        })

            print(f"[+] Passive assessment complete: {len(self.findings)} vulnerabilities found")
        except requests.RequestException as e:
            print(f"[!] API error: {e}")

    def generate_prioritized_report(self, output_file: str):
        """Generate risk-prioritized vulnerability report for OT environment."""
        self.findings.sort(key=lambda x: x.get("cvss", 0), reverse=True)

        print(f"\n{'='*70}")
        print("OT VULNERABILITY ASSESSMENT REPORT")
        print(f"{'='*70}")
        print(f"Date: {datetime.now().isoformat()}")
        print(f"Total Findings: {len(self.findings)}")

        severity_counts = {}
        for f in self.findings:
            sev = f.get("severity", "Unknown")
            severity_counts[sev] = severity_counts.get(sev, 0) + 1

        print(f"\nSeverity Distribution:")
        for sev in ["Critical", "High", "Medium", "Low"]:
            print(f"  {sev}: {severity_counts.get(sev, 0)}")

        # Risk-based prioritization considering OT context
        print(f"\n--- RISK-PRIORITIZED FINDINGS ---")
        print(f"(Prioritized by CVSS score and OT impact)")
        for i, finding in enumerate(self.findings[:20], 1):
            print(f"\n  {i}. [{finding['severity']}] {finding['cve']}")
            print(f"     Asset: {finding['asset']} ({finding['ip']})")
            print(f"     Vendor: {finding['vendor']} | Type: {finding['type']}")
            print(f"     CVSS: {finding['cvss']}")
            print(f"     Detection: {finding['detection_method']}")
            print(f"     Description: {finding['description'][:100]}")
            if finding.get("remediation"):
                print(f"     Remediation: {finding['remediation'][:100]}")

        # Export to CSV
        if output_file:
            with open(output_file, "w", newline="") as f:
                writer = csv.DictWriter(f, fieldnames=self.findings[0].keys())
                writer.writeheader()
                writer.writerows(self.findings)
            print(f"\n[+] Report exported to {output_file}")


if __name__ == "__main__":
    scanner = OTVulnerabilityScanner(
        tenable_url="https://tenable-ot.plant.local",
        api_key="your-api-key-here",
        verify_ssl=True,
    )

    # Always start with passive assessment
    safety_check = scanner.check_safety_prerequisites("passive", "10.10.0.0/16")
    print(f"Safety prerequisites: {json.dumps(safety_check, indent=2)}")

    scanner.run_passive_assessment(site_id="plant-01")
    scanner.generate_prioritized_report("ot_vulnerabilities.csv")

关键概念

术语	定义
被动漏洞检测	通过分析镜像流量来识别漏洞，无需向 OT 设备发送任何数据包
原生协议查询	使用工业协议（Modbus FC43、S7 SZL Read、CIP Get Attribute）安全地提取设备信息
OT 安全扫描配置文件	经过设计和实验室测试、可避免导致工业控制器崩溃的漏洞扫描器配置
补偿性控制	保护未打补丁 OT 资产的替代安全措施（防火墙 DPI、网络隔离）
OT 场景下的 CVSS	考虑安全性、可用性和物理后果，针对 OT 影响调整后的标准 CVSS 评分
Tenable OT Security	采用被动和基于原生协议检测的专用 OT 漏洞管理平台

输出格式

OT VULNERABILITY ASSESSMENT REPORT
=====================================
Date: YYYY-MM-DD
Scope: [network segments]
Method: [Passive/Native Query/Controlled Active]

VULNERABILITY SUMMARY:
  Critical: [count]
  High: [count]
  Medium: [count]
  Low: [count]

TOP RISK FINDINGS:
  1. [CVE] - [CVSS] - [Asset] - [Description]

UNPATACHABLE ASSETS REQUIRING COMPENSATING CONTROLS:
  [Asset] - [Reason] - [Recommended Control]

PATCH PRIORITIZATION:
  Immediate: [list]
  Next Window: [list]
  Acceptable Risk: [list with justification]