implementing-ot-network-traffic-analysis-with-nozomi

name	implementing-ot-network-traffic-analysis-with-nozomi
description	部署Nozomi Networks Guardian传感器进行被动OT网络流量分析，在不中断运营的情况下实现对工业控制系统的全面资产可见性、实时威胁检测和漏洞评估，利用行为异常检测和协议感知监控能力。
domain	cybersecurity
subdomain	ot-ics-security
tags	["ot-security","ics","nozomi","guardian","network-monitoring","asset-visibility","anomaly-detection","ndr"]
version	1.0
author	mahipal
license	Apache-2.0

使用Nozomi实施OT网络流量分析

适用场景

• 使用Nozomi Networks Guardian传感器部署被动OT网络监控
• 在敏感ICS环境中需要资产可见性但不进行主动扫描
• 通过Vantage或CMC集中管理构建基于Nozomi的OT SOC
• 将OT网络监控与Fortinet、Splunk或ServiceNow生态系统集成
• 监控IEC 62443网络分段策略的合规性

不适用于OT设备的主动漏洞扫描（参见performing-ot-vulnerability-scanning-safely）、已标准化使用Dragos的环境（参见implementing-dragos-platform-for-ot-monitoring），或仅IT网络监控。

前置条件

• Nozomi Networks Guardian传感器（硬件、VM或容器）
• 在监控的OT网段上配置的网络TAP或SPAN端口
• 用于多传感器管理的Nozomi Vantage（云端）或中央管理控制台
• 用于更新检测签名的Nozomi威胁情报订阅
• 用于传感器布置规划的网络架构文档

工作流程

步骤 1：部署Guardian传感器进行被动监控

#!/usr/bin/env python3
"""Nozomi Guardian部署管理器和告警分析器。

管理Nozomi Guardian传感器部署验证、资产清单
提取，以及OT环境的威胁告警分析。
"""

import json
import sys
from collections import defaultdict
from datetime import datetime
from typing import Dict, List, Optional

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


class NozomiGuardianManager:
    """管理Nozomi Networks Guardian进行OT监控。"""

    def __init__(self, guardian_url: str, api_token: str, verify_ssl: bool = False):
        self.guardian_url = guardian_url.rstrip("/")
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_token}",
            "Content-Type": "application/json",
        })
        self.session.verify = verify_ssl

    def get_nodes(self, node_type: Optional[str] = None) -> List[Dict]:
        """获取已发现的网络节点（资产）。"""
        params = {}
         node_type:
            params[] = node_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(, [])

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

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

     ():
        
        ()
        ()
        ()
        ()
        ()

        
        :
            resp = .session.get()
             resp.status_code == :
                status = resp.json()
                ()
                ()
                ()
                ()
                ()
         requests.RequestException  e:
            ()

        
        nodes = .get_nodes()
        ()
        ()

        type_counts = defaultdict()
        vendor_counts = defaultdict()
        protocol_set = ()
         node  nodes:
            type_counts[node.get(, )] += 
            vendor_counts[node.get(, )] += 
             proto  node.get(, []):
                protocol_set.add(proto)

        ()
         ntype, count  (type_counts.items(), key= x: -x[]):
            ()

        ()
         vendor, count  (vendor_counts.items(), key= x: -x[])[:]:
            ()

        ()

        
        alerts = .get_alerts(severity=)
        ()
        ()

        alert_types = defaultdict()
         alert  alerts:
            alert_types[alert.get(, )] += 

         atype, count  (alert_types.items(), key= x: -x[])[:]:
            ()

        
        vulns = .get_vulnerabilities()
        ()
        ()

        sev_counts = defaultdict()
         vuln  vulns:
            sev_counts[vuln.get(, )] += 

         sev  [, , , ]:
             sev  sev_counts:
                ()

     ():
        
        links = .get_links()
        nodes = {n.get(): n  n  .get_nodes()}

        ()
        ()

        
        cross_zone = []
         link  links:
            src_node = nodes.get(link.get(), {})
            dst_node = nodes.get(link.get(), {})
            src_zone = src_node.get(, )
            dst_zone = dst_node.get(, )

             src_zone != dst_zone  src_zone !=   dst_zone != :
                cross_zone.append({
                    : src_node.get(, ),
                    : src_zone,
                    : dst_node.get(, ),
                    : dst_zone,
                    : link.get(, []),
                })

         cross_zone:
            ()
             comm  cross_zone[:]:
                (
                      
                      )


 __name__ == :
    manager = NozomiGuardianManager(
        guardian_url=,
        api_token=,
    )

    manager.validate_deployment()
    manager.analyze_communication_patterns()

核心概念

术语	定义
Guardian	Nozomi Networks被动传感器，通过SPAN/TAP监控OT网络流量，不产生额外流量
Vantage	Nozomi基于云的中央管理平台，用于汇聚多个Guardian传感器的数据
行为异常检测（BAD）	Nozomi基于AI的方法，检测与学习到的正常OT网络行为的偏差
智能轮询（Smart Polling）	Nozomi使用原生协议安全提取额外设备详情的主动查询功能
资产情报（Asset Intelligence）	Nozomi从网络流量自动识别和分类OT/IoT资产
威胁情报订阅源	Nozomi Labs维护的OT特定威胁指标订阅源，基于全球蜜罐数据更新

输出格式

NOZOMI GUARDIAN OT监控报告
=======================================
站点: [站点名称]
日期: YYYY-MM-DD

资产可见性:
  总资产数: [数量]
  PLC: [数量] | HMI: [数量] | 交换机: [数量]
  协议: [列表]
  供应商: [前5名]

威胁检测:
  严重告警: [数量]
  高告警: [数量]
  主要告警类别: [列表]

漏洞:
  严重: [数量]
  高: [数量]

网络分析:
  通信链接: [数量]
  跨区域流: [数量]