implementing-dragos-platform-for-ot-monitoring

name	implementing-dragos-platform-for-ot-monitoring
description	为OT网络监控部署和配置Dragos平台，利用其600+工业协议解析器、情报驱动的威胁检测分析和资产可见性能力，保护ICS环境免受VOLTZITE、GRAPHITE和BAUXITE等威胁组织的攻击。
domain	cybersecurity
subdomain	ot-ics-security
tags	["ot-security","ics","dragos","threat-detection","ot-monitoring","scada","threat-intelligence","ndr"]
version	1.0
author	mahipal
license	Apache-2.0

为OT监控实现Dragos平台

适用场景

• 为工业环境部署OT专用网络检测与响应（NDR）解决方案
• 需要基于威胁情报检测已知ICS威胁组织（VOLTZITE、CHERNOVITE、KAMACITE）
• 构建使用专用工业安全工具的OT SOC能力
• 在单一平台上同时需要资产发现、漏洞管理和威胁检测
• 将OT安全监控与企业SIEM（Splunk、Sentinel、QRadar）集成

不适用于无ICS组件的纯IT网络监控、OT工作站的端点检测与响应（EDR），或已标准化使用Claroty或Nozomi的环境（参见相应技能）。

前置条件

• Dragos平台许可证和部署包
• OT网络边界的网络TAP或SPAN端口（每个监控网段一个传感器）
• Dragos传感器硬件（物理设备）或满足最低规格的虚拟设备
• 允许传感器到Dragos SiteStore通信的防火墙规则（仅从OT出站的加密连接）
• 用于威胁情报更新的Dragos知识包订阅

工作流程

步骤 1：部署Dragos传感器并配置监控

#!/usr/bin/env python3
"""Dragos平台部署验证和集成工具。

验证Dragos传感器部署、检查连通性，并
配置与企业SIEM的集成以进行OT告警转发。
"""

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

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


class DragosPlatformManager:
    """与Dragos平台API交互进行OT监控管理。"""

    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]:
        """获取所有已部署的Dragos传感器及其状态。"""
        resp = self.session.get(f"{self.base_url}/api/v1/sensors")
        resp.raise_for_status()
         resp.json().get(, [])

     () -> []:
        
        params = {}
         asset_type:
            params[] = asset_type
        resp = .session.get(, params=params)
        resp.raise_for_status()
         resp.json().get(, [])

     () -> []:
        
        params = {: severity, : limit}
        resp = .session.get(, params=params)
        resp.raise_for_status()
         resp.json().get(, [])

     () -> []:
        
        params = {: severity}
        resp = .session.get(, params=params)
        resp.raise_for_status()
         resp.json().get(, [])

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

     ():
        
        sensors = .get_sensors()
        assets = .get_assets()

        ()
        ()
        ()
        ()

        ()
        healthy_sensors = 
         sensor  sensors:
            status = sensor.get(, )
            icon =   status ==   
            ()
            ()
            ()
            ()
             status == :
                healthy_sensors += 

        ()

        ()
        ()
        asset_types = {}
         asset  assets:
            atype = asset.get(, )
            asset_types[atype] = asset_types.get(atype, ) + 
         atype, count  (asset_types.items(), key= x: -x[]):
            ()

        protocols = ()
         asset  assets:
            protocols.update(asset.get(, []))
        ()

        ()
        groups = .get_threat_groups()
        ()
         group  groups:
            ()
            ()
            ()

     ():
        
        configs = {
            : {
                : ,
                : ,
                : {
                    : ,
                    : ,
                    : ,
                    : ,
                    : ,
                },
                : ,
                : ,
                : [
                    , , , ,
                    , , , ,
                    , ,
                ],
            },
            : {
                : ,
                : ,
                : ,
                : {
                    : ,
                    : ,
                    : ,
                    : ,
                    : ,
                },
            },
        }

        config = configs.get(siem_type, configs[])
        ()
        (json.dumps(config, indent=))
         config


 __name__ == :
    manager = DragosPlatformManager(
        base_url=,
        api_key=,
        api_secret=,
        verify_ssl=,
    )

    manager.validate_deployment()
    manager.generate_siem_integration_config()

    ()
    notifications = manager.get_notifications(severity=, limit=)
     n  notifications:
        ()
        ()
        ()
        ()

步骤 2：配置检测分析和知识包

# Dragos平台检测配置
# 针对制造/能源环境调优

detection_configuration:
  knowledge_pack:
    auto_update: true
    update_schedule: "weekly"
    include_threat_groups:
      - "VOLTZITE"    # 针对能源行业，窃取OT图表
      - "GRAPHITE"    # 2025年新威胁组织，针对ICS
      - "BAUXITE"     # 2025年新威胁组织，针对ICS
      - "CHERNOVITE"  # 开发了PIPEDREAM/INCONTROLLER框架
      - "ELECTRUM"    # 与Industroyer/CrashOverride相关
      - "KAMACITE"    # 针对能源行业初始访问

  detection_categories:
    network_baseline:
      enabled: true
      learning_period_days: 30
      alert_on:
        - "new_communication_pair"
        - "new_protocol_detected"
        - "new_device_on_network"
        - "protocol_anomaly"

    threat_detection:
      enabled: true
      alert_on:
        - "known_malware_ioc"
        - "threat_group_ttp"
        - 
         
         

    
       
      
         
         

  
    
       
       
       
    
       
       
    
       
       
    
       
       
    
       
       

  
    
       [, ]
       
       
    
       []
       
       
    
       
    
       

核心概念

术语	定义
Dragos平台	专为OT网络安全打造的平台，具备ICS环境的资产可见性、威胁检测和漏洞管理能力
知识包（Knowledge Pack）	Dragos威胁情报更新包，包含针对ICS特有新威胁、恶意软件和漏洞利用的检测分析
SiteStore	Dragos中央管理服务器，汇聚站点内所有已部署传感器的数据
VOLTZITE	Dragos追踪的威胁组织，针对能源行业OT环境，窃取GIS数据和ICS网络图
PIPEDREAM/INCONTROLLER	由CHERNOVITE开发的模块化ICS攻击框架，针对Schneider/OMRON PLC和OPC UA服务器
邻域守护者（Neighborhood Keeper）	Dragos社区防御计划，在参与的OT环境之间共享匿名化威胁数据

常见场景

场景：检测能源公用事业中的VOLTZITE侦察活动

背景：部署在电力公司的Dragos传感器检测到异常的OPC UA浏览活动以及从工程师工作站窃取设备配置数据。

处理方法：

1. 查看Dragos通知中的MITRE ATT&CK ICS技术映射
2. 识别执行OPC UA浏览的源主机（检查是否为授权的工程师工作站）
3. 检查Dragos威胁情报中与VOLTZITE TTPs的关联
4. 检查被访问数据的范围（GIS数据、网络图、ICS配置文件）
5. 将被攻陷的工作站从OT网络隔离
6. 检查其他OT系统的横向移动指标
7. 如确认威胁组织归因，联系Dragos专业服务
8. 作为关键基础设施网络事件向CISA报告

注意事项：不要将OPC UA浏览告警视为误报——VOLTZITE专门使用这一技术进行预置。确保Dragos知识包是最新的以检测最新的VOLTZITE指标。在收集取证证据之前不要重新镜像被攻陷的工作站。

输出格式

DRAGOS OT监控部署报告
==========================================
站点: [站点名称]
日期: YYYY-MM-DD

传感器部署:
  传感器总数: [数量]
  正常运行: [数量]
  覆盖率: [已监控OT网段的百分比]

资产可见性:
  OT资产总数: [数量]
  PLC: [数量] | HMI: [数量] | 网络设备: [数量]
  协议: [列表]

威胁检测:
  相关活跃威胁组织: [数量]
  已加载检测分析: [数量]
  告警（过去30天）: [按严重程度分类的数量]

SIEM集成:
  状态: [已连接/未连接]
  已转发事件（过去24小时）: [数量]