Implementando a Plataforma Dragos para Monitoramento de OT

Quando Usar

• Ao implantar uma solução de detecção e resposta de rede (NDR) específica para OT em ambientes industriais
• Ao precisar de detecção baseada em inteligência de ameaças contra grupos de ameaças conhecidos de ICS (VOLTZITE, CHERNOVITE, KAMACITE)
• Ao construir uma capacidade de SOC de OT com ferramentas de segurança industrial desenvolvidas para esse fim
• Ao exigir descoberta de ativos e gerenciamento de vulnerabilidades juntamente com detecção de ameaças em uma única plataforma
• Ao integrar o monitoramento de segurança de OT com um SIEM corporativo (Splunk, Sentinel, QRadar)

Não usar para monitoramento de rede apenas de TI sem componentes de ICS, para detecção e resposta de endpoint (EDR) em estações de trabalho de OT, ou para ambientes padronizados em Claroty ou Nozomi (consulte as respectivas habilidades).

Pré-requisitos

• Licença da Plataforma Dragos e pacote de implantação
• Porta TAP ou SPAN de rede nos limites da rede de OT (um sensor por segmento monitorado)
• Hardware de sensor Dragos (dispositivo físico) ou dispositivo virtual que atenda às especificações mínimas
• Regras de firewall permitindo comunicação sensor-para-Dragos-SiteStore (criptografada, apenas de saída da OT)
• Assinatura do Dragos Knowledge Pack para atualizações de inteligência de ameaças

Fluxo de Trabalho

Etapa 1: Implantar Sensores Dragos e Configurar o Monitoramento

@@CODE<1>@@

#!/usr/bin/env python3
"""Dragos Platform Deployment Validator and Integration Tool.

Validates Dragos sensor deployment, checks connectivity, and
configures integration with enterprise SIEM for OT alert forwarding.
"""

import json
import sys
import csv
from datetime import datetime
from typing import Optional, List, Dict

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


class DragosPlatformManager:
    """Interface with Dragos Platform API for OT monitoring management."""

    def __init__(self, base_url: str, api_key: str, api_secret: str, verify_ssl: bool = True):
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        self.session.headers.update({
            "API-Key": api_key,
            "API-Secret": api_secret,
            "Content-Type": "application/json",
        })
        self.session.verify = verify_ssl

    def get_sensors(self) -> List[Dict]:
        """Retrieve all deployed Dragos sensors and their status."""
        resp = self.session.get(f"{self.base_url}/api/v1/sensors")
        resp.raise_for_status()
        return resp.json().get("sensors", [])

    def get_assets(self, asset_type: Optional[str] = None) -> List[Dict]:
        """Retrieve OT assets discovered by Dragos."""
        params = {}
        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_notifications(self, severity: str = "high", limit: int = 50) -> List[Dict]:
        """Retrieve threat detection notifications."""
        params = {"min_severity": severity, "limit": limit}
        resp = self.session.get(f"{self.base_url}/api/v1/notifications", params=params)
        resp.raise_for_status()
        return resp.json().get("notifications", [])

    def get_vulnerabilities(self, severity: str = "critical") -> List[Dict]:
        """Retrieve OT vulnerabilities with Dragos-specific context."""
        params = {"min_severity": severity}
        resp = self.session.get(f"{self.base_url}/api/v1/vulnerabilities", params=params)
        resp.raise_for_status()
        return resp.json().get("vulnerabilities", [])

    def get_threat_groups(self) -> List[Dict]:
        """Retrieve tracked ICS threat group activity relevant to the environment."""
        resp = self.session.get(f"{self.base_url}/api/v1/threat-groups")
        resp.raise_for_status()
        return resp.json().get("threat_groups", [])

    def validate_deployment(self):
        """Validate sensor deployment health and coverage."""
        sensors = self.get_sensors()
        assets = self.get_assets()

        print(f"\n{'='*65}")
        print("DRAGOS PLATFORM DEPLOYMENT VALIDATION")
        print(f"{'='*65}")
        print(f"Validation Time: {datetime.now().isoformat()}")

        print(f"\n--- SENSOR STATUS ---")
        healthy_sensors = 0
        for sensor in sensors:
            status = sensor.get("status", "unknown")
            icon = "[OK]" if status == "connected" else "[!!]"

…
@@CODE<1>@@