performing-ot-vulnerability-assessment-with-claroty

使用 Claroty 执行 OT 漏洞评估

何时使用

• 当根据 IEC 62443 或 NERC CIP 要求执行定期 OT 漏洞评估时
• 当首次部署 Claroty xDome 并执行初始资产发现和风险评估时
• 当将新发布的 ICS-CERT 公告与您的 OT 资产清单进行关联时
• 当在有限的维护窗口内对 OT 漏洞修复进行优先级排序时
• 当为 CIP-010-4 漏洞评估要求生成合规证据时

请勿使用于对 PLC 和安全系统进行主动漏洞扫描（被动方法请参阅 performing-ot-network-security-assessment），仅限 IT 的漏洞管理（请参阅标准漏洞扫描器），或渗透测试（请参阅 performing-ics-penetration-testing）。

前提条件

• 已在 OT 网络上部署带有传感器的 Claroty xDome 或 CTD（持续威胁检测）
• 具有网络 SPAN/TAP 访问权限以进行被动资产发现
• 订阅 CISA ICS-CERT 公告以跟踪漏洞
• 包含所有 OT 设备固件版本的资产清单
• 在维护窗口期间部署补丁的变更管理流程

工作流

第 1 步：配置资产发现和漏洞关联

配置 Claroty 执行被动和主动安全发现，以构建包含固件版本的完整资产清单，用于漏洞关联。

#!/usr/bin/env python3
"""OT Vulnerability Assessment Manager.

Correlates OT asset inventory with ICS-CERT advisories and CVE data
to identify, prioritize, and track OT vulnerabilities. Designed to
integrate with Claroty xDome API or standalone operation.
"""

import json
import sys
from collections import defaultdict
from dataclasses import dataclass, field, asdict
from datetime import datetime

import requests


@dataclass
class OTAsset:
    asset_id: str
    name: str
    vendor: str
    model: str
    firmware_version: str
    asset_type: str  # PLC, HMI, RTU, historian, switch, etc.
    purdue_level: str
    ip_address: str
    protocol: str
    criticality: str  # critical, high, medium, low
    zone: str


@dataclass
class OTVulnerability:
    vuln_id: str
    cve_id: str
    title: str
    severity: str  # critical, high, medium, low
    cvss_score: float
    affected_vendor: str
    affected_product: str
    affected_versions: str
    description: str
    ics_cert_advisory: str = ""
    remediation: str = ""
    patch_available: bool = False
    compensating_controls: str = ""


@dataclass
class RiskAssessment:
    asset: OTAsset
    vulnerability: OTVulnerability
    risk_score: float = 0.0
    risk_rating: str = ""
    exploitability: str = ""
    operational_impact: str = ""
    compensating_controls: list = field(default_factory=list)
    remediation_priority: int = 0


class OTVulnerabilityAssessment:
    """OT vulnerability assessment and prioritization engine."""

    def __init__(self):
        self.assets = []
        self.vulnerabilities = []
        self.risk_assessments = []

    def load_assets(self, assets_data):
        """Load asset inventory from Claroty export or manual inventory."""
        for a in assets_data:
            self.assets.append(OTAsset(**a))
        print(f"[*] Loaded {len(self.assets)} OT assets")

    def fetch_ics_advisories(self):
        """Fetch latest ICS-CERT advisories from CISA."""
        print("[*] Fetching ICS-CERT advisories from CISA...")
        try:
            # CISA Known Exploited Vulnerabilities catalog
            url = "https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json"
            resp = requests.get(url, timeout=30)
            resp.raise_for_status()
            data = resp.json()

            ics_vulns = []
            for vuln in data.get("vulnerabilities", []):
                # Filter for ICS-relevant vendors
                ics_vendors = [
                    "siemens", "schneider", "rockwell", "honeywell",
                    "abb", "ge", "emerson", "yokogawa", "omron",
                    "mitsubishi", "phoenix", "moxa", "advantech",
                ]
                vendor = vuln.get("vendorProject", "").lower()
                if any(v in vendor for v in ics_vendors):
                    ics_vulns.append(vuln)

            print(f"  Found {len(ics_vulns)} ICS-relevant known exploited vulnerabilities")
            return ics_vulns

        except Exception as e:
            print(f"[WARN] Could not fetch advisories: {e}")
            return []

    def correlate_vulnerabilities(self):
        """Match vulnerabilities to assets based on vendor/model/firmware."""
        print("[*] Correlating vulnerabilities to assets...")

        for asset in self.assets:
            for vuln in self.vulnerabilities:
                if (vuln.affected_vendor.lower() in asset.vendor.lower() and
                    vuln.affected_product.lower() in asset.model.lower()):
                    # Check firmware version if specified
                    ra = RiskAssessment(asset=asset, vulnerability=vuln)
                    self._calculate_risk_score(ra)
                    self.risk_assessments.append(ra)

        print(f"  Correlated {len(self.risk_assessments)} asset-vulnerability pairs")

    def _calculate_risk_score(self, ra):
        """Calculate OT-specific risk score considering operational impact."""
        # Base score from CVSS
        base = ra.vulnerability.cvss_score

        # Criticality multiplier based on asset function
        criticality_weights = {
            "critical": 1.5,  # SIS, safety systems
            "high": 1.3,      # PLCs, primary control
            "medium": 1.0,    # HMIs, historians
            "low": 0.7,       # non-critical support systems
        }
        criticality = criticality_weights.get(ra.asset.criticality, 1.0)

        # Purdue level proximity factor (lower levels = higher risk)
        level_weights = {
            "Level 0-1": 1.5,
            "Level 2": 1.3,
            "Level 3": 1.0,
            "Level 3.5": 0.8,
            "Level 4": 0.6,
        }
        level_factor = level_weights.get(ra.asset.purdue_level, 1.0)

        # Network exposure reduction if compensating controls exist
        comp_reduction = 0.8 if ra.compensating_controls else 1.0

        ra.risk_score = round(base * criticality * level_factor * comp_reduction, 1)
        ra.risk_score = min(ra.risk_score, 10.0)

        if ra.risk_score >= 9.0:
            ra.risk_rating = "critical"
            ra.remediation_priority = 1
        elif ra.risk_score >= 7.0:
            ra.risk_rating = "high"
            ra.remediation_priority = 2
        elif ra.risk_score >= 4.0:
            ra.risk_rating = "medium"
            ra.remediation_priority = 3
        else:
            ra.risk_rating = "low"
            ra.remediation_priority = 4

    def generate_report(self):
        """Generate vulnerability assessment report."""
        # Sort by risk score descending
        sorted_ra = sorted(self.risk_assessments, key=lambda x: -x.risk_score)

        report = []
        report.append("=" * 70)
        report.append("OT VULNERABILITY ASSESSMENT REPORT")
        report.append(f"Date: {datetime.now().isoformat()}")
        report.append(f"Assets: {len(self.assets)} | Vulnerabilities: {len(self.vulnerabilities)}")
        report.append(f"Risk Assessments: {len(self.risk_assessments)}")
        report.append("=" * 70)

        for sev in ["critical", "high", "medium", "low"]:
            findings = [ra for ra in sorted_ra if ra.risk_rating == sev]
            if findings:
                report.append(f"\n--- {sev.upper()} RISK ({len(findings)}) ---")
                for ra in findings[:10]:
                    report.append(f"\n  Risk Score: {ra.risk_score}/10.0")
                    report.append(f"  Asset: {ra.asset.name} ({ra.asset.vendor} {ra.asset.model})")
                    report.append(f"  Zone: {ra.asset.zone} ({ra.asset.purdue_level})")
                    report.append(f"  CVE: {ra.vulnerability.cve_id} (CVSS: {ra.vulnerability.cvss_score})")
                    report.append(f"  Title: {ra.vulnerability.title}")
                    if ra.vulnerability.patch_available:
                        report.append(f"  Patch: Available - schedule for next maintenance window")
                    else:
                        report.append(f"  Patch: Not available - apply compensating controls")

        return "\n".join(report)

    def export_json(self, output_file):
        """Export assessment to JSON."""
        data = {
            "assessment_date": datetime.now().isoformat(),
            "asset_count": len(self.assets),
            "vulnerability_count": len(self.vulnerabilities),
            "risk_assessments": [
                {
                    "asset_name": ra.asset.name,
                    "asset_ip": ra.asset.ip_address,
                    "cve": ra.vulnerability.cve_id,
                    "risk_score": ra.risk_score,
                    "risk_rating": ra.risk_rating,
                    "priority": ra.remediation_priority,
                }
                for ra in sorted(self.risk_assessments, key=lambda x: -x.risk_score)
            ],
        }
        with open(output_file, "w") as f:
            json.dump(data, f, indent=2)


if __name__ == "__main__":
    assessment = OTVulnerabilityAssessment()
    advisories = assessment.fetch_ics_advisories()
    print(f"Fetched {len(advisories)} ICS advisories from CISA KEV catalog")

关键概念

术语	定义
Claroty xDome	网络物理系统保护平台，为 OT/IoT 环境提供资产发现、漏洞管理和威胁检测
被动发现	通过分析网络流量而不发送任何数据包来识别 OT 资产，对生产环境安全
安全主动查询	使用原生工业协议以安全速率查询 OT 设备，以收集详细资产信息而不中断运营
OT 风险评分	综合考虑 CVSS 基础评分、资产关键性、Purdue 层级和补偿控制的风险评级，用于适合 OT 的优先级排序
ICS-CERT 公告	CISA 发布的工业控制系统漏洞安全公告，包含特定于供应商的修复指导
虚拟补丁	在无法立即应用固件补丁时，部署 IPS/防火墙规则以阻止对已知漏洞的利用

工具与系统

• Claroty xDome：全面的 OT/IoT 资产发现、漏洞管理和持续威胁检测平台
• Claroty CTD：用于 OT 环境中被动网络监控的持续威胁检测传感器
• CISA ICS-CERT：发布 ICS 漏洞通知和缓解指导的美国政府公告服务
• Dragos Platform：具有资产可见性和漏洞管理能力的替代 OT 安全平台
• Nozomi Networks Guardian：具有漏洞关联和风险评分的 OT 监控平台

输出格式

OT Vulnerability Assessment Report
=====================================
Tool: Claroty xDome / Manual Assessment
Date: YYYY-MM-DD
Assets Scanned: [N]

RISK SUMMARY:
  Critical Risk: [N] vulnerabilities on [N] assets
  High Risk: [N] vulnerabilities on [N] assets
  Medium Risk: [N] vulnerabilities on [N] assets
  Low Risk: [N] vulnerabilities on [N] assets

TOP RISKS:
  [Risk Score] [CVE-ID] on [Asset Name] ([Zone])
    Remediation: [Patch/Compensating Control]
    Timeline: [Next maintenance window / Immediate]