performing-ics-asset-discovery-with-claroty

Perform comprehensive ICS/OT asset discovery using Claroty xDome platform, leveraging passive monitoring, Claroty Edge active queries, and integration ecosystem to gain full visibility into industrial control system assets including PLCs, RTUs, HMIs, and network infrastructure across Purdue Model levels.

Zur Installation springen

Repository: autohandai/community-skills
Letzte Quellaktivität: 16. März 2026 um 00:59
Erkannte Sprache von SKILL.md: Englisch
Sterne: 10
Forks: 3

Installationsoptionen

Standardmäßig ist der Prompt ausgewählt, der zuerst die Quelle prüft. Sie können zu einem direkten Befehl wechseln oder eine lokale Kopie herunterladen.

Quelldateien prüfen

Lesen Sie SKILL.md und alle von SkillsMP angezeigten Begleitdateien, bevor Sie sich für eine Installation entscheiden.

Zurück zur Suche Zurück zur Installation

name	performing-ics-asset-discovery-with-claroty
description	Perform comprehensive ICS/OT asset discovery using Claroty xDome platform, leveraging passive monitoring, Claroty Edge active queries, and integration ecosystem to gain full visibility into industrial control system assets including PLCs, RTUs, HMIs, and network infrastructure across Purdue Model levels.
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

Performing ICS Asset Discovery with Claroty

When to Use

When gaining initial visibility into an OT environment with unknown or poorly documented assets
When preparing for an IEC 62443 risk assessment requiring a complete asset inventory
When onboarding Claroty xDome into a brownfield industrial environment
When validating existing asset inventory against actual network communications
When identifying shadow OT devices or unauthorized connections in the control network

Do not use for IT-only asset discovery (use tools like Nessus or Qualys), for active scanning of sensitive PLC networks without vendor approval, or for environments where Claroty is not the deployed platform (see implementing-ot-network-traffic-analysis-with-nozomi).

Prerequisites

Claroty xDome SaaS subscription or on-premises deployment
Network TAP or SPAN port configured at OT network boundaries (Levels 1-3 of Purdue Model)
Claroty Edge collector deployed for safe active querying of hard-to-reach network segments
Integration credentials for CMDB tools (ServiceNow, BMC) if used
Network architecture diagram showing VLANs, switches, and firewall zones

Workflow

Step 1: Configure Passive Network Monitoring

Deploy Claroty sensors on SPAN ports to passively observe all OT network traffic without impacting operations.

#!/usr/bin/env python3
"""Claroty xDome Asset Discovery Configuration and Reporting Tool.

Automates the configuration of passive monitoring sensors and generates
asset inventory reports from Claroty xDome API.
"""

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

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


class ClarotyAssetDiscovery:
    """Interface with Claroty xDome API for ICS asset discovery."""

     ():
        .base_url = base_url.rstrip()
        .session = requests.Session()
        .session.headers.update({
            : ,
            : ,
            : ,
        })
        .session.verify = verify_ssl

     ():
        
        resp = .session.get()
        resp.raise_for_status()
         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[:]:
            ()

# Claroty Edge Active Discovery Configuration
# Safe active queries using native industrial protocols

edge_configuration:
  deployment_mode: "on-premises"
  collection_schedule:
    frequency: "weekly"
    maintenance_window: "Sunday 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:
        - "Module identification"
        - "Firmware version"
        - "Hardware configuration"
        - "Protection level"

    rockwell_cip:
      enabled: true
      target_subnets: ["10.10.3.0/24"]
      ports: [44818]
      query_type: "CIP_identity"
      information_collected:
        - "Product name and revision"
        - "Serial number"
        - "Device type"
        - "Vendor ID"

    modbus:
      enabled: true
      target_subnets: ["10.10.4.0/24"]
      ports: [502]
      query_type: "read_device_identification"
      function_code: 43
      information_collected:
        - "Vendor name"
        - "Product code"
        - "Firmware revision"

    bacnet:
      enabled: true
      target_subnets: ["10.10.5.0/24"]
      ports: [47808]
      query_type: "who_is"
      information_collected:
        - "Device name"
        - "Vendor identifier"
        - "Model name"
        - "Application software version"

  safety_controls:
    excluded_subnets: ["10.10.100.0/24"]  # SIS network - never active scan
    rate_limiting: true
    max_packets_per_second: 10
    timeout_seconds: 5
    retry_count: 1
    abort_on_device_error: true

#!/usr/bin/env python3
"""Asset Validation and Enrichment Tool.

Cross-references Claroty discovery results against existing CMDB
and enriches with NVD vulnerability data.
"""

import json
import csv
import sys
from datetime import datetime

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


class AssetValidator:
    """Validates and enriches OT asset inventory."""

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

    def load_inventory(self, filepath: str):
        """Load Claroty-discovered asset inventory."""
        with open(filepath, "r") as f:
            reader = csv.DictReader(f)
            self.discovered_assets = list(reader)
        print(f"[*] Loaded {len(self.discovered_assets)} discovered assets")

    def compare_with_cmdb(self, cmdb_file: str):
        """Compare discovered assets against CMDB records."""
        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("ASSET INVENTORY VALIDATION REPORT")
        print(f"{'='*60}")
        print(f"Discovered assets: {len(discovered_ips)}")
        print(f"CMDB records: {len(cmdb_ips)}")
        print(f"Shadow OT devices (not in CMDB): {len(shadow_devices)}")
        print(f"Missing devices (in CMDB, not seen): {len(missing_devices)}")

        if shadow_devices:
            print(f"\n  SHADOW DEVICES (Unauthorized/Undocumented):")
            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', 'Unknown')} {asset.get('model', '')} | Type: {asset.get('type', 'Unknown')}")
                self.discrepancies.append({
                    "type": "SHADOW_DEVICE",
                    "severity": "HIGH",
                    "ip": ip,
                    "detail": f"Undocumented {asset.get('type', 'device')} from {asset.get('vendor', 'unknown vendor')}",
                })

        if missing_devices:
            print(f"\n  MISSING DEVICES (Expected but not seen):")
            for ip in sorted(missing_devices):
                cmdb = cmdb_assets[ip]
                print(f"    - {ip} | {cmdb.get('name', 'Unknown')} | Last CMDB update: {cmdb.get('last_updated', 'N/A')}")
                self.discrepancies.append({
                    "type": "MISSING_DEVICE",
                    "severity": "MEDIUM",
                    "ip": ip,
                    "detail": f"CMDB asset {cmdb.get('name', ip)} not seen on network",
                })

    def check_firmware_vulnerabilities(self, asset):
        """Check NVD for known vulnerabilities matching asset firmware."""
        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):
        """Generate risk-prioritized summary of findings."""
        print(f"\n{'='*60}")
        print("RISK SUMMARY")
        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"  High-risk assets (score >= 7): {len(high_risk)}")
        print(f"  Potentially end-of-life firmware: {len(end_of_life)}")
        print(f"  Assets using unencrypted protocols: {len(no_encryption)}")
        print(f"  Inventory discrepancies: {len(self.discrepancies)}")


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: 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()

Term	Definition
Passive Monitoring	Observing mirrored network traffic via SPAN/TAP without injecting packets, safe for all OT devices
Active Querying	Sending native protocol requests to extract detailed device information; requires careful scheduling
Claroty Edge	Claroty's safe active discovery collector that uses native industrial protocols rather than IT scanning
Purdue Level	Hierarchical classification of industrial network assets from Level 0 (physical process) to Level 5 (enterprise)
Shadow OT Device	Asset connected to the OT network that is not documented in the asset management system
xDome	Claroty's SaaS-based cyber-physical systems protection platform providing visibility, risk management, and threat detection

ICS ASSET DISCOVERY REPORT
============================
Date: YYYY-MM-DD
Platform: Claroty xDome
Site: [Site Name]

DISCOVERY SUMMARY:
  Total Assets Discovered: [count]
  New Assets (not in CMDB): [count]
  High-Risk Assets: [count]

PURDUE LEVEL DISTRIBUTION:
  Level 0 (Process): [count] assets
  Level 1 (Control): [count] assets
  Level 2 (Supervisory): [count] assets
  Level 3 (Operations): [count] assets
  Level 3.5 (DMZ): [count] assets
  Level 4-5 (Enterprise): [count] assets

TOP VENDORS:
  1. [Vendor] - [count] devices
  2. [Vendor] - [count] devices

CRITICAL FINDINGS:
  - [Shadow device description]
  - [End-of-life firmware finding]
  - [Unencrypted protocol concern]