performing-ics-asset-discovery-with-claroty

name	performing-ics-asset-discovery-with-claroty
description	使用Claroty xDome平台执行全面的ICS/OT资产发现，利用被动监控、Claroty Edge主动查询和集成生态系统，在Purdue模型各级别获得对工业控制系统资产（包括PLC、RTU、HMI和网络基础设施）的完整可见性。
domain	cybersecurity
subdomain	ot-ics-security
tags	["ot-security","ics","asset-discovery","claroty","xdome","scada","network-visibility","iec62443"]
version	1.0
author	mahipal
license	Apache-2.0

使用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网络流量而不影响运营。

#!/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(, [])

     ():
        
        params = {}
         site_id:
            params[] = site_id
         asset_type:
            params[] = asset_type

        resp = .session.get(, params=params)
        resp.raise_for_status()
         resp.json().get(, [])

     ():
        
        resp = .session.get()
        resp.raise_for_status()
         resp.json()

     ():
        
        resp = .session.get(
            
        )
        resp.raise_for_status()
         resp.json().get(, [])

     ():
        
        params = {: severity}
         site_id:
            params[] = site_id

        resp = .session.get(, params=params)
        resp.raise_for_status()
         resp.json().get(, [])

     ():
        
        assets = .get_assets(site_id=site_id)
          assets:
            ()
            

        fieldnames = [
            , , , , , ,
            , , , ,
            , , , , ,
            , ,
        ]

         (output_file, , newline=)  f:
            writer = csv.DictWriter(f, fieldnames=fieldnames)
            writer.writeheader()
             asset  assets:
                writer.writerow({
                    : asset.get(, ),
                    : asset.get(, ),
                    : asset.get(, ),
                    : asset.get(, ),
                    : asset.get(, ),
                    : asset.get(, ),
                    : asset.get(, ),
                    : asset.get(, ),
                    : asset.get(, ),
                    : asset.get(, ),
                    : asset.get(, ),
                    : .join(asset.get(, [])),
                    : asset.get(, ),
                    : asset.get(, ),
                    : asset.get(, ),
                    : asset.get(, ),
                    : asset.get(, ),
                })

        ()

     ():
        
        assets = .get_assets(site_id=site_id)
        levels = {: [], : [], : [], : [], : [], : [], : []}

         asset  assets:
            level = asset.get(, -)
             level  levels:
                levels[level].append(asset)

        ()
        ()
        ()
        ()
        ()
        ()
        ()

        level_names = {
            : ,
            : ,
            : ,
            : ,
            : ,
            : ,
            : ,
        }

         level, name  level_names.items():
            device_list = levels.get(level, [])
            ()
            ()
             device_list:
                vendors = (a.get(, )  a  device_list)
                types = (a.get(, )  a  device_list)
                ()
                ()
                high_risk = [a  a  device_list  a.get(, ) >= ]
                 high_risk:
                    ()
                     a  high_risk[:]:
                        ()


 __name__ == :
    discovery = ClarotyAssetDiscovery(
        base_url=,
        api_token=,
        verify_ssl=,
    )

    ()
    sites = discovery.get_sites()
     site  sites:
        ()

     sites:
        site_id = sites[][]
        ()
        discovery.generate_purdue_level_report(site_id)

        ()
        discovery.export_asset_inventory(
            ,
            site_id=site_id,
        )

        ()
        vulns = discovery.get_vulnerabilities(site_id=site_id, severity=)
        ()
         v  vulns[:]:
            ()

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

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

# 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
       []
       []
       
       
      
         
         
         

    
       
       []
       []
       
      
         
         
         
         

  
     []  
     
     
     
     
     

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

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

#!/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[]: row  row  csv.DictReader(f)}

        discovered_ips = {a[]  a  .discovered_assets  a[]}
        cmdb_ips = (cmdb_assets.keys())

        shadow_devices = discovered_ips - cmdb_ips
        missing_devices = cmdb_ips - discovered_ips

        ()
        ()
        ()
        ()
        ()
        ()
        ()

         shadow_devices:
            ()
             ip  (shadow_devices):
                asset = ((a  a  .discovered_assets  a[] == ip), {})
                ()
                .discrepancies.append({
                    : ,
                    : ,
                    : ip,
                    : ,
                })

         missing_devices:
            ()
             ip  (missing_devices):
                cmdb = cmdb_assets[ip]
                ()
                .discrepancies.append({
                    : ,
                    : ,
                    : ip,
                    : ,
                })

     ():
        
        vendor = asset.get(, ).lower()
        model = asset.get(, ).lower()
        firmware = asset.get(, )

          vendor   model:
             []

        search_term = 
        :
            resp = requests.get(
                ,
                params={: search_term, : },
                timeout=,
            )
             resp.status_code == :
                data = resp.json()
                 data.get(, [])
         requests.RequestException:
            
         []

     ():
        
        ()
        ()
        ()

        high_risk = [a  a  .discovered_assets  (a.get(, )) >= ]
        end_of_life = [a  a  .discovered_assets  a.get(, ).startswith()]
        no_encryption = [a  a  .discovered_assets    a.get(, ).lower()]

        ()
        ()
        ()
        ()


 __name__ == :
     (sys.argv) < :
        ()
        sys.exit()

    validator = AssetValidator(sys.argv[])
     (sys.argv) >= :
        validator.compare_with_cmdb(sys.argv[])
    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. [供应商] - [数量] 台设备

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