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
nist_ai_rmf	["MEASURE-2.7","MAP-5.1","MANAGE-2.4"]
atlas_techniques	["AML.T0070","AML.T0066","AML.T0082"]
nist_csf	["PR.IR-01","DE.CM-01","ID.AM-05","GV.OC-02"]

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.

Common Misconfigurations & Verification

• SPAN/TAP coverage gaps create silent blind spots. Guardian only sees mirrored traffic; oversubscribed SPAN ports drop packets under load, and serial/fieldbus segments below the switch are never seen. Verify coverage per segment against the asset inventory and check sensor packets/sec for drops, not just a green status.
• Passive-only misses quiet and one-way assets. Devices that rarely transmit may never be discovered; use Smart Polling sparingly and only with native protocols — never active scans against fragile PLCs.
• Zones not configured, so cross-zone alerts are meaningless. Behavioral anomaly detection and cross-zone link analysis depend on assets being assigned to zones; an unconfigured deployment reports flows it cannot judge.
• Stale threat intelligence or short learning window. Confirm the Threat Intelligence feed is current and the learning period spanned representative operating modes, or normal batch and shift changes generate false positives.
• Verify without disruption. Validate by reviewing discovered asset and protocol counts, replaying a known IOC into a test sensor, and confirming Splunk/Fortinet/ServiceNow forwarding with a synthetic alert — all passively, never by generating attack traffic on the live OT segment.

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


 json
 sys
 collections  defaultdict
 datetime  datetime
 typing  , , 

:
     requests
 ImportError:
    ()
    sys.exit()


 :
    

     ():
        .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()