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设备发送任何数据包的情况下识别漏洞。

```python
#!/usr/bin/env python3
"""OT安全漏洞扫描协调器。

协调被动监控、原生协议查询和精心控制的主动扫描，
在不破坏工业运营的情况下进行OT漏洞评估。
"""

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

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


class OTVulnerabilityScanner:
    """安全OT漏洞扫描协调器。"""

    SCAN_SAFETY_LEVELS = {
        "passive": {
            "description": "仅观察网络流量，对设备零风险",
            "risk_level": "无",
            "methods": ["流量指纹", "协议分析", "版本检测"],
            "requires_window": False,
        },
        "native_query": {
            "description": "使用原生工业协议查询设备",
            "risk_level": "极低",
            "methods": ["modbus_device_id", "s7_szl_read", "cip_identity", "bacnet_whois"],
            "requires_window": True,
        },
        "controlled_active": {
            "description": "使用OT安全配置文件进行标准漏洞检查",
            "risk_level": "低-中",
            "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:
        """扫描前验证安全前提条件。"""
        checks = {
            "scan_level": scan_level,
            "target": target_subnet,
            "safety_level": self.SCAN_SAFETY_LEVELS[scan_level],
            "checks_passed": [],
            "checks_failed": [],
            "approved": False,
        }

        prerequisites = [
            {
                "name": "实验室验证完成",
                "description": "扫描配置文件已在实验室环境中针对每种设备类型测试",
                "required_for": ["native_query", "controlled_active"],
            },
            {
                "name": "供应商保修已验证",
                "description": "扫描不会使供应商支持协议失效",
                "required_for": ["native_query", "controlled_active"],
            },
            {
                "name": "变更管理已审批",
                "description": "扫描活动的变更工单已批准",
                "required_for": ["native_query", "controlled_active"],
            },
            {
                "name": "维护窗口已确认",
                "description": "运营团队确认可接受的扫描窗口",
                "required_for": ["controlled_active"],
            },
            {
                "name": "回滚计划已记录",
                "description": "停止扫描并在设备无响应时恢复的程序",
                "required_for": ["controlled_active"],
            },
            {
                "name": "SIS已排除在范围之外",
                "description": "安全仪表系统永远不进行主动扫描",
                "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):
        """使用流量分析运行被动漏洞评估。"""
        print(f"[*] 对站点 {site_id} 运行被动漏洞评估")
        print(f"[*] 安全级别: 无 - 不向OT设备发送数据包")

        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", "未知"),
                            "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"[+] 被动评估完成: 发现 {len(self.findings)} 个漏洞")
        except requests.RequestException as e:
            print(f"[!] API错误: {e}")

    def generate_prioritized_report(self, output_file: str):
        """生成OT环境的基于风险优先级的漏洞报告。"""
        self.findings.sort(key=lambda x: x.get("cvss", 0), reverse=True)

        print(f"\n{'='*70}")
        print("OT漏洞评估报告")
        print(f"{'='*70}")
        print(f"日期: {datetime.now().isoformat()}")
        print(f"发现总数: {len(self.findings)}")

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

        print(f"\n严重程度分布:")
        for sev in ["Critical", "High", "Medium", "Low"]:
            print(f"  {sev}: {severity_counts.get(sev, 0)}")

        # 考虑OT背景的基于风险的优先级排序
        print(f"\n--- 基于风险的优先级发现 ---")
        print(f"（按CVSS评分和OT影响排序）")
        for i, finding in enumerate(self.findings[:20], 1):
            print(f"\n  {i}. [{finding['severity']}] {finding['cve']}")
            print(f"     资产: {finding['asset']} ({finding['ip']})")
            print(f"     供应商: {finding['vendor']} | 类型: {finding['type']}")
            print(f"     CVSS: {finding['cvss']}")
            print(f"     检测方法: {finding['detection_method']}")
            print(f"     描述: {finding['description'][:100]}")
            if finding.get("remediation"):
                print(f"     修复: {finding['remediation'][:100]}")

        # 导出为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[+] 报告已导出到 {output_file}")


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

    # 始终从被动评估开始
    safety_check = scanner.check_safety_prerequisites("passive", "10.10.0.0/16")
    print(f"安全前提条件: {json.dumps(safety_check, indent=2)}")

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

## 核心概念

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

## 输出格式

```
OT漏洞评估报告
=====================================
日期: YYYY-MM-DD
范围: [网段]
方法: [被动/原生查询/受控主动]

漏洞摘要:
  严重: [数量]
  高: [数量]
  中: [数量]
  低: [数量]

主要风险发现:
  1. [CVE] - [CVSS] - [资产] - [描述]

无法打补丁需要补偿控制的资产:
  [资产] - [原因] - [推荐控制]

补丁优先级:
  立即: [列表]
  下次窗口: [列表]
  可接受风险: [带理由的列表]
```