# Performing Ics Asset Discovery With Claroty

> 使用Claroty xDome平台执行全面的ICS/OT资产发现，利用被动监控、Claroty Edge主动查询和集成生态系统，在Purdue模型各级别获得对工业控制系统资产（包括PLC、RTU、HMI和网络基础设施）的完整可见性。

- Skill: `killvxk/performing-ics-asset-discovery-with-claroty` (Agent Skill, multi-file: 4 files)
- Install (CLI): `npx skillmds@latest add killvxk/performing-ics-asset-discovery-with-claroty`
- Raw SKILL.md: https://api.skillmd.com/api/skills/killvxk/performing-ics-asset-discovery-with-claroty/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Product & Planning
- License: Apache-2.0
- Author: killvxk (https://skillmd.com/u/killvxk)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/killvxk/performing-ics-asset-discovery-with-claroty

---


# 使用Claroty执行ICS资产发现

## 适用场景

- 对资产未知或记录不完整的OT环境进行初始可见性建立时
- 准备需要完整资产清单的IEC 62443风险评估时
- 将Claroty xDome引入棕地工业环境时
- 根据实际网络通信验证现有资产清单时
- 识别控制网络中的影子OT设备或未授权连接时

**不适用于**仅IT资产发现（使用Nessus或Qualys等工具）、未经供应商批准对敏感PLC网络进行主动扫描，或未部署Claroty的环境（参见implementing-ot-network-traffic-analysis-with-nozomi）。

## 前置条件

- Claroty xDome SaaS订阅或本地部署
- 在OT网络边界（Purdue模型1-3级）配置的网络TAP或SPAN端口
- 部署Claroty Edge采集器用于安全主动查询难以到达的网段
- CMDB工具（ServiceNow、BMC）的集成凭据（如使用）
- 显示VLAN、交换机和防火墙区域的网络架构图

## 工作流程

### 步骤 1：配置被动网络监控

在SPAN端口上部署Claroty传感器，被动观察所有OT网络流量而不影响运营。

```python
#!/usr/bin/env python3
"""Claroty xDome资产发现配置和报告工具。

自动化被动监控传感器的配置，并从
Claroty xDome API生成资产清单报告。
"""

import json
import sys
import csv
from datetime import datetime
from typing import Optional

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


class ClarotyAssetDiscovery:
    """通过Claroty xDome API进行ICS资产发现的接口。"""

    def __init__(self, base_url: str, api_token: str, verify_ssl: bool = True):
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_token}",
            "Content-Type": "application/json",
            "Accept": "application/json",
        })
        self.session.verify = verify_ssl

    def get_sites(self):
        """获取所有被监控的站点。"""
        resp = self.session.get(f"{self.base_url}/api/v1/sites")
        resp.raise_for_status()
        return resp.json().get("sites", [])

    def get_assets(self, site_id: Optional[str] = None, asset_type: Optional[str] = None):
        """获取已发现的资产（可选过滤）。

        asset_type: PLC, RTU, HMI, DCS, Engineering_Workstation,
                    Historian, Network_Device, IO_Module, Safety_Controller
        """
        params = {}
        if site_id:
            params["site_id"] = site_id
        if asset_type:
            params["type"] = asset_type

        resp = self.session.get(f"{self.base_url}/api/v1/assets", params=params)
        resp.raise_for_status()
        return resp.json().get("assets", [])

    def get_asset_detail(self, asset_id: str):
        """获取详细资产信息，包括固件、模块和CVE。"""
        resp = self.session.get(f"{self.base_url}/api/v1/assets/{asset_id}")
        resp.raise_for_status()
        return resp.json()

    def get_communication_map(self, site_id: str):
        """获取资产间的通信关系。"""
        resp = self.session.get(
            f"{self.base_url}/api/v1/sites/{site_id}/communications"
        )
        resp.raise_for_status()
        return resp.json().get("communications", [])

    def get_vulnerabilities(self, site_id: Optional[str] = None, severity: str = "critical"):
        """获取已发现资产的漏洞。"""
        params = {"min_severity": severity}
        if site_id:
            params["site_id"] = site_id

        resp = self.session.get(f"{self.base_url}/api/v1/vulnerabilities", params=params)
        resp.raise_for_status()
        return resp.json().get("vulnerabilities", [])

    def export_asset_inventory(self, output_file: str, site_id: Optional[str] = None):
        """将完整资产清单导出为CSV用于合规报告。"""
        assets = self.get_assets(site_id=site_id)
        if not assets:
            print("[!] 未找到资产")
            return

        fieldnames = [
            "asset_id", "name", "type", "vendor", "model", "firmware_version",
            "ip_address", "mac_address", "serial_number", "purdue_level",
            "zone", "protocol", "first_seen", "last_seen", "risk_score",
            "cve_count", "site_name",
        ]

        with open(output_file, "w", newline="") as f:
            writer = csv.DictWriter(f, fieldnames=fieldnames)
            writer.writeheader()
            for asset in assets:
                writer.writerow({
                    "asset_id": asset.get("id", ""),
                    "name": asset.get("name", "未知"),
                    "type": asset.get("type", ""),
                    "vendor": asset.get("vendor", ""),
                    "model": asset.get("model", ""),
                    "firmware_version": asset.get("firmware_version", ""),
                    "ip_address": asset.get("ip_address", ""),
                    "mac_address": asset.get("mac_address", ""),
                    "serial_number": asset.get("serial_number", ""),
                    "purdue_level": asset.get("purdue_level", ""),
                    "zone": asset.get("zone", ""),
                    "protocol": ", ".join(asset.get("protocols", [])),
                    "first_seen": asset.get("first_seen", ""),
                    "last_seen": asset.get("last_seen", ""),
                    "risk_score": asset.get("risk_score", 0),
                    "cve_count": asset.get("cve_count", 0),
                    "site_name": asset.get("site_name", ""),
                })

        print(f"[+] 已导出 {len(assets)} 个资产到 {output_file}")

    def generate_purdue_level_report(self, site_id: str):
        """按Purdue模型级别生成资产分布报告。"""
        assets = self.get_assets(site_id=site_id)
        levels = {0: [], 1: [], 2: [], 3: [], 3.5: [], 4: [], 5: []}

        for asset in assets:
            level = asset.get("purdue_level", -1)
            if level in levels:
                levels[level].append(asset)

        print(f"\n{'='*65}")
        print("PURDUE模型资产分布报告")
        print(f"{'='*65}")
        print(f"站点: {site_id}")
        print(f"已发现资产总数: {len(assets)}")
        print(f"报告生成时间: {datetime.now().isoformat()}")
        print(f"{'-'*65}")

        level_names = {
            0: "0级 - 物理过程（传感器/执行器）",
            1: "1级 - 基本控制（PLC/RTU）",
            2: "2级 - 监控控制（HMI/SCADA）",
            3: "3级 - 站点运营（历史服务器/MES）",
            3.5: "3.5级 - IT/OT DMZ",
            4: "4级 - 企业IT",
            5: "5级 - 企业网络/互联网",
        }

        for level, name in level_names.items():
            device_list = levels.get(level, [])
            print(f"\n  {name}")
            print(f"    数量: {len(device_list)}")
            if device_list:
                vendors = set(a.get("vendor", "未知") for a in device_list)
                types = set(a.get("type", "未知") for a in device_list)
                print(f"    供应商: {', '.join(vendors)}")
                print(f"    类型: {', '.join(types)}")
                high_risk = [a for a in device_list if a.get("risk_score", 0) >= 7]
                if high_risk:
                    print(f"    高风险资产: {len(high_risk)}")
                    for a in high_risk[:5]:
                        print(f"      - {a['name']} (风险: {a.get('risk_score')})")


if __name__ == "__main__":
    discovery = ClarotyAssetDiscovery(
        base_url="https://your-claroty-instance.claroty.cloud",
        api_token="your-api-token-here",
        verify_ssl=True,
    )

    print("[*] 获取站点...")
    sites = discovery.get_sites()
    for site in sites:
        print(f"  站点: {site['name']} (ID: {site['id']})")

    if sites:
        site_id = sites[0]["id"]
        print(f"\n[*] 为 {sites[0]['name']} 生成Purdue级别报告...")
        discovery.generate_purdue_level_report(site_id)

        print(f"\n[*] 导出资产清单...")
        discovery.export_asset_inventory(
            f"asset_inventory_{datetime.now().strftime('%Y%m%d')}.csv",
            site_id=site_id,
        )

        print(f"\n[*] 检查严重漏洞...")
        vulns = discovery.get_vulnerabilities(site_id=site_id, severity="critical")
        print(f"  严重漏洞: {len(vulns)}")
        for v in vulns[:10]:
            print(f"    - {v.get('cve_id')}: {v.get('description', '')[:80]}")
```

### 步骤 2：使用Claroty Edge配置主动发现

Claroty Edge使用原生工业协议（而非IT扫描）对OT设备执行安全、有针对性的查询，从被动监控单独无法完全识别的设备中提取详细资产信息。

```yaml
# Claroty Edge主动发现配置
# 使用原生工业协议进行安全主动查询

edge_configuration:
  deployment_mode: "on-premises"
  collection_schedule:
    frequency: "weekly"
    maintenance_window: "周日 02:00-06:00"
    max_concurrent_queries: 5

  protocol_queries:
    siemens_s7:
      enabled: true
      target_subnets: ["10.10.1.0/24", "10.10.2.0/24"]
      ports: [102]
      query_type: "SZL_read"
      information_collected:
        - "模块标识"
        - "固件版本"
        - "硬件配置"
        - "保护级别"

    rockwell_cip:
      enabled: true
      target_subnets: ["10.10.3.0/24"]
      ports: [44818]
      query_type: "CIP_identity"
      information_collected:
        - "产品名称和版本"
        - "序列号"
        - "设备类型"
        - "供应商ID"

    modbus:
      enabled: true
      target_subnets: ["10.10.4.0/24"]
      ports: [502]
      query_type: "read_device_identification"
      function_code: 43
      information_collected:
        - "供应商名称"
        - "产品代码"
        - "固件版本"

    bacnet:
      enabled: true
      target_subnets: ["10.10.5.0/24"]
      ports: [47808]
      query_type: "who_is"
      information_collected:
        - "设备名称"
        - "供应商标识符"
        - "型号名称"
        - "应用软件版本"

  safety_controls:
    excluded_subnets: ["10.10.100.0/24"]  # SIS网络 - 永不主动扫描
    rate_limiting: true
    max_packets_per_second: 10
    timeout_seconds: 5
    retry_count: 1
    abort_on_device_error: true
```

### 步骤 3：验证和丰富资产数据

将已发现的资产与已知清单进行交叉比对，并用漏洞数据进行丰富。

```python
#!/usr/bin/env python3
"""资产验证和丰富工具。

将Claroty发现结果与现有CMDB进行交叉比对，
并用NVD漏洞数据进行丰富。
"""

import json
import csv
import sys
from datetime import datetime

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


class AssetValidator:
    """验证和丰富OT资产清单。"""

    def __init__(self, inventory_file: str):
        self.discovered_assets = []
        self.load_inventory(inventory_file)
        self.discrepancies = []

    def load_inventory(self, filepath: str):
        """加载Claroty发现的资产清单。"""
        with open(filepath, "r") as f:
            reader = csv.DictReader(f)
            self.discovered_assets = list(reader)
        print(f"[*] 已加载 {len(self.discovered_assets)} 个已发现资产")

    def compare_with_cmdb(self, cmdb_file: str):
        """将已发现资产与CMDB记录进行比较。"""
        with open(cmdb_file, "r") as f:
            cmdb_assets = {row["ip_address"]: row for row in csv.DictReader(f)}

        discovered_ips = {a["ip_address"] for a in self.discovered_assets if a["ip_address"]}
        cmdb_ips = set(cmdb_assets.keys())

        shadow_devices = discovered_ips - cmdb_ips
        missing_devices = cmdb_ips - discovered_ips

        print(f"\n{'='*60}")
        print("资产清单验证报告")
        print(f"{'='*60}")
        print(f"已发现资产: {len(discovered_ips)}")
        print(f"CMDB记录: {len(cmdb_ips)}")
        print(f"影子OT设备（不在CMDB中）: {len(shadow_devices)}")
        print(f"缺失设备（在CMDB中但未发现）: {len(missing_devices)}")

        if shadow_devices:
            print(f"\n  影子设备（未授权/未记录）:")
            for ip in sorted(shadow_devices):
                asset = next((a for a in self.discovered_assets if a["ip_address"] == ip), {})
                print(f"    - {ip} | {asset.get('vendor', '未知')} {asset.get('model', '')} | 类型: {asset.get('type', '未知')}")
                self.discrepancies.append({
                    "type": "SHADOW_DEVICE",
                    "severity": "HIGH",
                    "ip": ip,
                    "detail": f"来自{asset.get('vendor', '未知供应商')}的未记录{asset.get('type', '设备')}",
                })

        if missing_devices:
            print(f"\n  缺失设备（预期存在但未发现）:")
            for ip in sorted(missing_devices):
                cmdb = cmdb_assets[ip]
                print(f"    - {ip} | {cmdb.get('name', '未知')} | CMDB最后更新: {cmdb.get('last_updated', 'N/A')}")
                self.discrepancies.append({
                    "type": "MISSING_DEVICE",
                    "severity": "MEDIUM",
                    "ip": ip,
                    "detail": f"CMDB资产 {cmdb.get('name', ip)} 在网络上未发现",
                })

    def check_firmware_vulnerabilities(self, asset):
        """检查NVD中匹配资产固件的已知漏洞。"""
        vendor = asset.get("vendor", "").lower()
        model = asset.get("model", "").lower()
        firmware = asset.get("firmware_version", "")

        if not vendor or not model:
            return []

        search_term = f"{vendor} {model}"
        try:
            resp = requests.get(
                "https://services.nvd.nist.gov/rest/json/cves/2.0",
                params={"keywordSearch": search_term, "resultsPerPage": 10},
                timeout=15,
            )
            if resp.status_code == 200:
                data = resp.json()
                return data.get("vulnerabilities", [])
        except requests.RequestException:
            pass
        return []

    def generate_risk_summary(self):
        """生成按风险优先级排序的发现摘要。"""
        print(f"\n{'='*60}")
        print("风险摘要")
        print(f"{'='*60}")

        high_risk = [a for a in self.discovered_assets if float(a.get("risk_score", 0)) >= 7]
        end_of_life = [a for a in self.discovered_assets if a.get("firmware_version", "").startswith("v1.")]
        no_encryption = [a for a in self.discovered_assets if "modbus" in a.get("protocol", "").lower()]

        print(f"  高风险资产（评分 >= 7）: {len(high_risk)}")
        print(f"  可能达到生命周期终止的固件: {len(end_of_life)}")
        print(f"  使用未加密协议的资产: {len(no_encryption)}")
        print(f"  清单差异数量: {len(self.discrepancies)}")


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("用法: python validate_assets.py <claroty_export.csv> [cmdb_export.csv]")
        sys.exit(1)

    validator = AssetValidator(sys.argv[1])
    if len(sys.argv) >= 3:
        validator.compare_with_cmdb(sys.argv[2])
    validator.generate_risk_summary()
```

## 核心概念

| 术语 | 定义 |
|------|------|
| 被动监控（Passive Monitoring） | 通过SPAN/TAP观察镜像网络流量而不注入数据包，对所有OT设备安全 |
| 主动查询（Active Querying） | 发送原生协议请求以提取详细设备信息；需要仔细规划 |
| Claroty Edge | Claroty的安全主动发现采集器，使用原生工业协议而非IT扫描 |
| Purdue级别（Purdue Level） | 工业网络资产的层次化分类，从0级（物理过程）到5级（企业） |
| 影子OT设备（Shadow OT Device） | 连接到OT网络但未在资产管理系统中记录的资产 |
| xDome | Claroty的基于SaaS的网络物理系统保护平台，提供可见性、风险管理和威胁检测 |

## 常见场景

### 场景：棕地工厂资产发现

**背景**：一家拥有20年设备添加历史的制造工厂需要为IEC 62443风险评估建立完整的OT资产清单。没有准确的资产记录。

**方法**：
1. 在每个主要网段（控制、监控、DMZ）的SPAN端口上部署Claroty传感器
2. 允许被动监控2-4周以捕获所有常规通信模式
3. 在计划的维护窗口期间安排Claroty Edge主动查询
4. 导出已发现的清单，并按Purdue级别、供应商和关键性对资产进行分类
5. 与任何现有文档（P&ID图、网络图纸）进行交叉比对
6. 识别影子设备并与工厂运营部门启动审查流程
7. 将验证后的清单纳入IEC 62443区域和通道风险评估

**注意事项**：不要在被动监控捕获基线流量模式之前急于进行主动发现。切勿直接对PLC或RTU使用IT漏洞扫描器（Nessus主动扫描）——这可能导致旧版控制器崩溃。始终将安全仪表系统（SIS）排除在主动查询之外。

## 输出格式

```
ICS资产发现报告
============================
日期: YYYY-MM-DD
平台: Claroty xDome
站点: [站点名称]

发现摘要:
  已发现资产总数: [数量]
  新资产（不在CMDB中）: [数量]
  高风险资产: [数量]

PURDUE级别分布:
  0级（过程）: [数量] 个资产
  1级（控制）: [数量] 个资产
  2级（监控）: [数量] 个资产
  3级（运营）: [数量] 个资产
  3.5级（DMZ）: [数量] 个资产
  4-5级（企业）: [数量] 个资产

主要供应商:
  1. [供应商] - [数量] 台设备
  2. [供应商] - [数量] 台设备

关键发现:
  - [影子设备描述]
  - [生命周期终止固件发现]
  - [未加密协议问题]
```