L’interface est localisée ; les présentations et documents des Skills sont traduits automatiquement par IA dans votre langue. Le code, les commandes, chemins et termes techniques restent en version originale, et le document de l’auteur fait foi.
Explorer les Skills
Productivité · SKILL.md

performing-ot-vulnerability-assessment-with-claroty

performing-ot-vulnerability-assessment-with-claroty

本技能涵盖使用Claroty xDome平台在OT环境中执行漏洞评估,实现全面资产发现、风险评分、漏洞关联和修复优先级排序。内容涉及通过流量分析进行被动漏洞识别、OT设备安全主动查询、与CVE数据库和ICS-CERT公告集成,以及考虑运营影响和补偿控制措施的基于风险的优先级排序。

Independent GitHub snapshot · Automated classification · Original documentation governs

Documentation de la Skill

GitHub

name: performing-ot-vulnerability-assessment-with-claroty description: > 本技能涵盖使用Claroty xDome平台在OT环境中执行漏洞评估,实现全面资产发现、风险评分、漏洞关联和修复优先级排序。内容涉及通过流量分析进行被动漏洞识别、OT设备安全主动查询、与CVE数据库和ICS-CERT公告集成,以及考虑运营影响和补偿控制措施的基于风险的优先级排序。 domain: cybersecurity subdomain: ot-ics-security tags: [ot-security, ics, scada, industrial-control, iec62443, vulnerability-assessment, claroty] version: 1.0.0 author: mahipal license: Apache-2.0

使用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漏洞评估管理器。

将OT资产清单与ICS-CERT公告和CVE数据进行关联,
以识别、优先排序和跟踪OT漏洞。设计用于
与Claroty xDome API集成或独立运行。
"""

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等
    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漏洞评估和优先级排序引擎。"""

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

    def load_assets(self, assets_data):
        """从Claroty导出或手动清单加载资产。"""
        for a in assets_data:
            self.assets.append(OTAsset(**a))
        print(f"[*] 已加载 {len(self.assets)} 个OT资产")

    def fetch_ics_advisories(self):
        """从CISA获取最新ICS-CERT公告。"""
        print("[*] 从CISA获取ICS-CERT公告...")
        try:
            # CISA已知被利用漏洞目录
            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", []):
                # 过滤ICS相关供应商
                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"  发现 {len(ics_vulns)} 个ICS相关已知利用漏洞")
            return ics_vulns

        except Exception as e:
            print(f"[警告] 无法获取公告: {e}")
            return []

    def correlate_vulnerabilities(self):
        """基于供应商/型号/固件将漏洞与资产匹配。"""
        print("[*] 将漏洞与资产关联...")

        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()):
                    # 如有指定则检查固件版本
                    ra = RiskAssessment(asset=asset, vulnerability=vuln)
                    self._calculate_risk_score(ra)
                    self.risk_assessments.append(ra)

        print(f"  已关联 {len(self.risk_assessments)} 个资产-漏洞对")

    def _calculate_risk_score(self, ra):
        """计算考虑运营影响的OT特定风险评分。"""
        # 基于CVSS的基础评分
        base = ra.vulnerability.cvss_score

        # 基于资产功能的关键性乘数
        criticality_weights = {
            "critical": 1.5,  # SIS、安全系统
            "high": 1.3,      # PLC、主要控制
            "medium": 1.0,    # HMI、历史服务器
            "low": 0.7,       # 非关键支持系统
        }
        criticality = criticality_weights.get(ra.asset.criticality, 1.0)

        # Purdue级别接近度因子(级别越低 = 风险越高)
        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)

        # 如有补偿控制措施则降低网络暴露风险
        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):
        """生成漏洞评估报告。"""
        # 按风险评分降序排列
        sorted_ra = sorted(self.risk_assessments, key=lambda x: -x.risk_score)

        report = []
        report.append("=" * 70)
        report.append("OT漏洞评估报告")
        report.append(f"日期: {datetime.now().isoformat()}")
        report.append(f"资产: {len(self.assets)} | 漏洞: {len(self.vulnerabilities)}")
        report.append(f"风险评估: {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()} 风险 ({len(findings)}) ---")
                for ra in findings[:10]:
                    report.append(f"\n  风险评分: {ra.risk_score}/10.0")
                    report.append(f"  资产: {ra.asset.name} ({ra.asset.vendor} {ra.asset.model})")
                    report.append(f"  区域: {ra.asset.zone} ({ra.asset.purdue_level})")
                    report.append(f"  CVE: {ra.vulnerability.cve_id} (CVSS: {ra.vulnerability.cvss_score})")
                    report.append(f"  标题: {ra.vulnerability.title}")
                    if ra.vulnerability.patch_available:
                        report.append(f"  补丁: 可用 - 安排在下次维护窗口")
                    else:
                        report.append(f"  补丁: 不可用 - 应用补偿控制措施")

        return "\n".join(report)

    def export_json(self, output_file):
        """将评估导出为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"从CISA KEV目录获取了 {len(advisories)} 条ICS公告")

核心概念

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

工具和系统

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

输出格式

OT漏洞评估报告
=====================================
工具: Claroty xDome / 手动评估
日期: YYYY-MM-DD
已扫描资产: [N]

风险摘要:
  严重风险: [N]个漏洞影响[N]个资产
  高风险: [N]个漏洞影响[N]个资产
  中风险: [N]个漏洞影响[N]个资产
  低风险: [N]个漏洞影响[N]个资产

主要风险:
  [风险评分] [CVE-ID] 影响 [资产名称] ([区域])
    修复: [补丁/补偿控制措施]
    时限: [下次维护窗口/立即]
Skills enregistrées