Realizando Descoberta de Ativos ICS com Claroty

Quando Usar

• Ao obter visibilidade inicial em um ambiente de OT com ativos desconhecidos ou mal documentados
• Ao se preparar para uma avaliação de risco IEC 62443 que exige um inventário completo de ativos
• Ao integrar o Claroty xDome em um ambiente industrial brownfield
• Ao validar o inventário de ativos existente em relação às comunicações reais da rede
• Ao identificar dispositivos de OT "sombra" ou conexões não autorizadas na rede de controle

Não usar para descoberta de ativos apenas de TI (use ferramentas como Nessus ou Qualys), para varredura ativa de redes de PLC sensíveis sem aprovação do fornecedor, ou para ambientes onde o Claroty não é a plataforma implantada (veja implementando-análise-de-tráfego-de-rede-ot-com-nozomi).

Pré-requisitos

• Assinatura Claroty xDome SaaS ou implantação on-premises
• Porta TAP ou SPAN de rede configurada nas fronteiras da rede de OT (Níveis 1-3 do Modelo Purdue)
• Coletor Claroty Edge implantado para consulta ativa segura de segmentos de rede de difícil acesso
• Credenciais de integração para ferramentas CMDB (ServiceNow, BMC) se usadas
• Diagrama da arquitetura de rede mostrando VLANs, switches e zonas de firewall

Fluxo de Trabalho

Etapa 1: Configurar Monitoramento Passivo de Rede

Implante sensores Claroty em portas SPAN para observar passivamente todo o tráfego da rede de OT sem impactar as operações.

#!/usr/bin/env python3
"""Ferramenta de Configuração e Relatório de Descoberta de Ativos Claroty xDome.

Automatiza a configuração de sensores de monitoramento passivo e gera
relatórios de inventário de ativos a partir da API Claroty xDome.
"""

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

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


class ClarotyAssetDiscovery:
    """Interface com a API Claroty xDome para descoberta de ativos 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):
        """Recupera todos os sites monitorados."""
        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):
        """Recupera ativos descobertos com filtragem opcional.

        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):
        """Recupera informações detalhadas do ativo, incluindo firmware, módulos e CVEs."""
        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):
        """Recupera relacionamentos de comunicação entre ativos."""
        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"):
        """Recupera vulnerabilidades para ativos descobertos."""
        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):
        """Exporta o inventário completo de ativos para CSV para relatórios de conformidade."""
        assets = self.get_assets(site_id=site_id)
        if not assets:
            print("[!] Nenhum ativo encontrado")
            return

        fieldnames = [

…