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

name	implementing-ot-network-traffic-analysis-with-nozomi
description	Deploy Nozomi Networks Guardian sensors for passive OT network traffic analysis to achieve comprehensive asset visibility, real-time threat detection, and vulnerability assessment across industrial control systems without disrupting operations, leveraging behavioral anomaly detection and protocol-aware monitoring.
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

Implementing OT Network Traffic Analysis with Nozomi

When to Use

• When deploying passive OT network monitoring using Nozomi Networks Guardian sensors
• When requiring asset visibility without active scanning in sensitive ICS environments
• When building a Nozomi-based OT SOC with centralized management via Vantage or CMC
• When integrating OT network monitoring with Fortinet, Splunk, or ServiceNow ecosystems
• When monitoring compliance with IEC 62443 network segmentation policies

Do not use for active vulnerability scanning of OT devices (see performing-ot-vulnerability-scanning-safely), for environments standardized on Dragos (see implementing-dragos-platform-for-ot-monitoring), or for IT-only network monitoring.

Prerequisites

• Nozomi Networks Guardian sensor (hardware, VM, or container)
• Network TAP or SPAN port configured on monitored OT network segments
• Nozomi Vantage (cloud) or Central Management Console for multi-sensor management
• Nozomi Threat Intelligence subscription for updated detection signatures
• Network architecture documentation for sensor placement planning

Workflow

Step 1: Deploy Guardian Sensors for Passive Monitoring

#!/usr/bin/env python3
"""Nozomi Guardian Deployment Manager and Alert Analyzer.

Manages Nozomi Guardian sensor deployment validation, asset inventory
extraction, and threat alert analysis for OT environments.
"""

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

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


class NozomiGuardianManager:
    """Manages Nozomi Networks Guardian for OT monitoring."""

     ():
        .guardian_url = guardian_url.rstrip()
        .session = requests.Session()
        .session.headers.update({
            : ,
            : ,
        })
        .session.verify = verify_ssl

     () -> []:
        
        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()