detecting-anomalies-in-industrial-control-systems

name	detecting-anomalies-in-industrial-control-systems
description	Deploys anomaly detection for OT/ICS environments using machine learning on OT network baselines, physics-based process models, and Modbus/DNP3/OPC UA traffic analysis to flag deviations, rogue devices, and mismatches against historian data. Use for continuous OT monitoring, baselining deterministic SCADA polling, or investigating alerts from Nozomi Guardian/Dragos needing deeper protocol analysis.
domain	cybersecurity
subdomain	ot-ics-security
tags	["ot-security","ics","scada","industrial-control","iec62443","anomaly-detection","machine-learning"]
version	1.0.0
author	mahipal
license	Apache-2.0
atlas_techniques	["AML.T0043","AML.T0018"]
nist_ai_rmf	["MEASURE-2.7","MEASURE-2.5","MAP-5.1"]
nist_csf	["PR.IR-01","DE.CM-01","ID.AM-05","GV.OC-02"]
mitre_attack	["T0836","T0831","T0832","T0814","T0801"]

Detecting Anomalies in Industrial Control Systems

When to Use

• When deploying continuous monitoring for OT environments that lack intrusion detection
• When building behavior-based detection to complement signature-based IDS in OT networks
• When establishing baselines for deterministic SCADA communications to detect deviations
• When integrating machine learning anomaly detection with OT security monitoring platforms
• When investigating alerts from Nozomi Guardian or Dragos Platform that require deeper analysis

Do not use for signature-based detection of known exploits (see detecting-attacks-on-scada-systems), for IT network anomaly detection without OT protocols, or as a replacement for process safety systems (SIS).

Prerequisites

• Passive network monitoring sensors on OT network SPAN/TAP ports
• Minimum 2-4 weeks of baseline traffic capture during normal operations
• Python 3.9+ with scikit-learn, numpy, pandas for ML model training
• Process historian access for physical process correlation data
• Understanding of normal operational patterns including shift changes, batch processes, and maintenance windows

Workflow

Step 1: Build Multi-Dimensional Baseline Model

Capture and model the deterministic behavior of ICS communications across multiple dimensions: timing, protocol behavior, and network topology.

#!/usr/bin/env python3
"""ICS Anomaly Detection System.

Builds multi-dimensional baselines from OT network traffic and
detects anomalies using statistical and machine learning methods.
Designed for deterministic SCADA communication patterns.
"""

import json
import sys
import time
import warnings
from collections import defaultdict
from datetime import datetime, timedelta
from dataclasses import dataclass, field

import numpy as np
import pandas as pd
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing  StandardScaler

warnings.filterwarnings()



 :
    
    src_ip: 
    dst_ip: 
    protocol: 
    port: 
    avg_interval_ms:  = 
    std_interval_ms:  = 
    avg_payload_size:  = 
    function_codes:  = field(default_factory=)
    packets_per_minute:  = 
    first_seen:  = 
    last_seen:  = 


 :
    

     ():
        .profiles = {}
        .topology_baseline = ()
        .timing_model = 
        .isolation_forest = 
        .scaler = StandardScaler()
        .anomalies = []
        .training_data = []

     ():
        
        ()

         flow  pcap_data:
            key = 

             key   .profiles:
                .profiles[key] = CommunicationProfile(
                    src_ip=flow[],
                    dst_ip=flow[],
                    protocol=flow.get(, ),
                    port=flow[],
                    first_seen=flow.get(, ),
                )

            profile = .profiles[key]
            profile.last_seen = flow.get(, )

            
            fc = flow.get()
             fc   :
                profile.function_codes[fc] = profile.function_codes.get(fc, ) + 

            
            .topology_baseline.add((flow[], flow[], flow[]))

        
        ._calculate_timing_stats(pcap_data)

        ()
        ()

     ():
        
        timestamps = defaultdict()
         flow  flows:
            key = 
            ts = flow.get()
             ts:
                timestamps[key].append(ts)

         key, ts_list  timestamps.items():
             key  .profiles  (ts_list) > :
                ts_sorted = (ts_list)
                intervals = [
                    (ts_sorted[i+] - ts_sorted[i]) * 
                     i  ((ts_sorted) - )
                ]
                .profiles[key].avg_interval_ms = np.mean(intervals)
                .profiles[key].std_interval_ms = np.std(intervals)
                duration_min = (ts_sorted[-] - ts_sorted[]) / 
                 duration_min > :
                    .profiles[key].packets_per_minute = (ts_list) / duration_min

     ():
        
        ()

        feature_cols = [
            , , ,
            , ,
        ]

        available_cols = [c  c  feature_cols  c  features_df.columns]
        X = features_df[available_cols].fillna().values

        X_scaled = .scaler.fit_transform(X)

        .isolation_forest = IsolationForest(
            n_estimators=,
            contamination=,  
            random_state=,
            n_jobs=-,
        )
        .isolation_forest.fit(X_scaled)

        scores = .isolation_forest.decision_function(X_scaled)
        ()
        ()
        ()

     ():
        
         (src_ip, dst_ip, port)   .topology_baseline:
             {
                : ,
                : ,
                : ,
                : ,
            }
         

     ():
        
        key = 
        profile = .profiles.get(key)

         profile  profile.std_interval_ms > :
            z_score = (interval_ms - profile.avg_interval_ms) / profile.std_interval_ms
             z_score > :
                 {
                    : ,
                    : ,
                    : (
                        
                        
                    ),
                    : ,
                }
         

     ():
        
        key = 
        profile = .profiles.get(key)

         profile  func_code   profile.function_codes:
            severity =   func_code  {, , , , }  
             {
                : ,
                : severity,
                : (
                    
                    
                ),
                : ,
            }
         

     ():
        
        results = []

        
        topo = .detect_topology_anomaly(flow[], flow[], flow[])
         topo:
            results.append(topo)

        
           flow:
            timing = .detect_timing_anomaly(
                flow[], flow[], flow[], flow[])
             timing:
                results.append(timing)

        
           flow:
            fc = .detect_function_code_anomaly(
                flow[], flow[], flow[], flow[])
             fc:
                results.append(fc)

        .anomalies.extend(results)
         results

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

        severity_counts = defaultdict()
         a  .anomalies:
            severity_counts[a[]] += 

         sev  [, , , ]:
             severity_counts[sev]:
                ()

         a  .anomalies[:]:
            ()
            ()


 __name__ == :
    ()
    ()