# Implementing Ot Network Traffic Analysis With Nozomi > 部署Nozomi Networks Guardian传感器进行被动OT网络流量分析，在不中断运营的情况下实现对工业控制系统的全面资产可见性、实时威胁检测和漏洞评估，利用行为异常检测和协议感知监控能力。 - Skill: `killvxk/implementing-ot-network-traffic-analysis-with-nozomi` (Agent Skill, multi-file: 4 files) - Install (CLI): `npx skillmds@latest add killvxk/implementing-ot-network-traffic-analysis-with-nozomi` - Raw SKILL.md: https://api.skillmd.com/api/skills/killvxk/implementing-ot-network-traffic-analysis-with-nozomi/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools - License: Apache-2.0 - Author: killvxk (https://skillmd.com/u/killvxk) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/killvxk/implementing-ot-network-traffic-analysis-with-nozomi --- # 使用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传感器进行被动监控 ```python #!/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 = {} if node_type: params["type"] = node_type resp = self.session.get(f"{self.guardian_url}/api/v1/nodes", params=params) resp.raise_for_status() return resp.json().get("result", []) def get_alerts(self, severity: str = "high", limit: int = 100) -> List[Dict]: """获取安全告警。""" params = {"severity": severity, "limit": limit, "status": "open"} resp = self.session.get(f"{self.guardian_url}/api/v1/alerts", params=params) resp.raise_for_status() return resp.json().get("result", []) def get_links(self) -> List[Dict]: """获取节点之间的通信链接。""" resp = self.session.get(f"{self.guardian_url}/api/v1/links") resp.raise_for_status() return resp.json().get("result", []) def get_vulnerabilities(self) -> List[Dict]: """获取检测到的漏洞。""" resp = self.session.get(f"{self.guardian_url}/api/v1/vulnerabilities") resp.raise_for_status() return resp.json().get("result", []) def validate_deployment(self): """验证Guardian传感器部署和覆盖范围。""" print(f"\n{'='*65}") print("NOZOMI GUARDIAN部署验证") print(f"{'='*65}") print(f"Guardian URL: {self.guardian_url}") print(f"验证时间: {datetime.now().isoformat()}") # 检查系统状态 try: resp = self.session.get(f"{self.guardian_url}/api/v1/system/status") if resp.status_code == 200: status = resp.json() print(f"\n--- 系统状态 ---") print(f" 版本: {status.get('version', 'N/A')}") print(f" 运行时间: {status.get('uptime', 'N/A')}") print(f" 已处理数据包: {status.get('packets_processed', 'N/A')}") print(f" 威胁情报: {status.get('threat_intelligence_version', 'N/A')}") except requests.RequestException as e: print(f" [!] 系统状态不可用: {e}") # 资产发现摘要 nodes = self.get_nodes() print(f"\n--- 资产发现 ---") print(f" 发现的节点总数: {len(nodes)}") type_counts = defaultdict(int) vendor_counts = defaultdict(int) protocol_set = set() for node in nodes: type_counts[node.get("type", "unknown")] += 1 vendor_counts[node.get("vendor", "Unknown")] += 1 for proto in node.get("protocols", []): protocol_set.add(proto) print(f"\n 按类型:") for ntype, count in sorted(type_counts.items(), key=lambda x: -x[1]): print(f" {ntype}: {count}") print(f"\n 按供应商:") for vendor, count in sorted(vendor_counts.items(), key=lambda x: -x[1])[:10]: print(f" {vendor}: {count}") print(f"\n 已观察到的协议: {', '.join(sorted(protocol_set))}") # 告警摘要 alerts = self.get_alerts(severity="high") print(f"\n--- 告警摘要 ---") print(f" 高/严重告警: {len(alerts)}") alert_types = defaultdict(int) for alert in alerts: alert_types[alert.get("type_id", "unknown")] += 1 for atype, count in sorted(alert_types.items(), key=lambda x: -x[1])[:10]: print(f" {atype}: {count}") # 漏洞摘要 vulns = self.get_vulnerabilities() print(f"\n--- 漏洞摘要 ---") print(f" 漏洞总数: {len(vulns)}") sev_counts = defaultdict(int) for vuln in vulns: sev_counts[vuln.get("severity", "unknown")] += 1 for sev in ["critical", "high", "medium", "low"]: if sev in sev_counts: print(f" {sev.capitalize()}: {sev_counts[sev]}") def analyze_communication_patterns(self): """分析OT通信模式中的异常情况。""" links = self.get_links() nodes = {n.get("id"): n for n in self.get_nodes()} print(f"\n--- 通信分析 ---") print(f" 通信链接总数: {len(links)}") # 识别跨区域通信 cross_zone = [] for link in links: src_node = nodes.get(link.get("source_id"), {}) dst_node = nodes.get(link.get("destination_id"), {}) src_zone = src_node.get("zone", "unknown") dst_zone = dst_node.get("zone", "unknown") if src_zone != dst_zone and src_zone != "unknown" and dst_zone != "unknown": cross_zone.append({ "source": src_node.get("label", "Unknown"), "source_zone": src_zone, "destination": dst_node.get("label", "Unknown"), "dest_zone": dst_zone, "protocols": link.get("protocols", []), }) if cross_zone: print(f"\n 跨区域通信: {len(cross_zone)}") for comm in cross_zone[:10]: print(f" {comm['source']} ({comm['source_zone']}) -> " f"{comm['destination']} ({comm['dest_zone']}) " f"通过 {', '.join(comm['protocols'])}") if __name__ == "__main__": manager = NozomiGuardianManager( guardian_url="https://nozomi-guardian.plant.local", api_token="your-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名] 威胁检测: 严重告警: [数量] 高告警: [数量] 主要告警类别: [列表] 漏洞: 严重: [数量] 高: [数量] 网络分析: 通信链接: [数量] 跨区域流: [数量] ```