detecting-attacks-on-historian-servers

name	detecting-attacks-on-historian-servers
description	Detect cyber attacks targeting OT historian servers (OSIsoft PI, Ignition, Wonderware) that sit at the IT/OT boundary and serve as pivot points for lateral movement between enterprise and control networks, including data manipulation, unauthorized queries, and exploitation of historian-specific vulnerabilities.
domain	cybersecurity
subdomain	ot-ics-security
tags	["ot-security","ics","historian","osisoft-pi","ignition","pivot-point","data-integrity","lateral-movement"]
version	1.0
author	mahipal
license	Apache-2.0
nist_csf	["PR.IR-01","DE.CM-01","ID.AM-05","GV.OC-02"]

Detecting Attacks on Historian Servers

When to Use

• When monitoring historian servers that bridge IT and OT networks for compromise indicators
• When detecting unauthorized queries or data manipulation in process historian databases
• When investigating lateral movement through historian servers between IT and OT zones
• When responding to alerts about exploitation of historian-specific vulnerabilities (CVE-2025-0921)
• When validating historian data integrity after a suspected OT security incident

Do not use for general database security monitoring (see database security skills), for historian deployment and configuration, or for IT-only data warehouse security.

Detection Gaps & Validation

• The SQL/ODBC path and the OPC/PI path are different attack surfaces. Queries arriving over the historian's SQL/ODBC interface (exfiltration, injection) look nothing like writes over OPC HDA/UA or the PI Web API. Monitoring only one interface misses the other; baseline both per client application and per tag set.
• Flatlined or replayed tags evade simple thresholds. A Stuxnet-style replay feeds prior good values to the HMI while the process is manipulated. The tell is a constant value across hundreds of points (PI compression effectively off), not an out-of-range alarm.
• The historian is a pivot, not just a database. Watch for the PI/OSIsoft or Ignition host initiating outbound connections to Level 1/2 (ports 502, 102, 44818, 20000) — it should receive data, never poll PLCs.
• How to confirm a hit safely. Validate detections against a non-production PI/Ignition instance: replay recorded archive data, simulate an unauthorized client IP, and test CVE indicators offline. Cross-check every alert against the change/maintenance log before escalating — routine interface restarts and archive backfills mimic the deletion/flatline signatures.

Prerequisites

• Historian server inventory (OSIsoft PI, Ignition, GE Proficy, Wonderware InSQL)
• Network monitoring on historian network segments (both IT-facing and OT-facing interfaces)
• Historian API access for data integrity validation
• Baseline of normal historian query patterns (which applications query which tags)
• Understanding of historian architecture (data sources, interfaces, client connections)

Workflow

Step 1: Monitor Historian for Attack Indicators

#!/usr/bin/env python3
"""OT Historian Attack Detector.

Monitors historian servers for unauthorized access, data manipulation,
lateral movement indicators, and exploitation of historian-specific
vulnerabilities. Supports OSIsoft PI and Ignition platforms.
"""

 json
 sys
 collections  defaultdict
 datetime  datetime, timedelta
 typing  , , 

:
     requests
 ImportError:
    ()
    sys.exit()


 :
    

     ():
        .historian_type = historian_type
        .historian_url = historian_url.rstrip()
        .credentials = api_credentials
        .verify_ssl = verify_ssl
        .alerts = []
        .authorized_clients = ()
        .authorized_queries = {}

     ():
        
        .authorized_clients = (authorized_clients)
        .authorized_queries = authorized_query_patterns

     () -> []:
        
        connections = []

         .historian_type == :
            :
                resp = requests.get(
                    ,
                    auth=(.credentials.get(), .credentials.get()),
                    verify=.verify_ssl,
                    timeout=,
                )
                 resp.status_code == :
                    data = resp.json()
                    connections = data.get(, [])
             requests.RequestException  e:
                ()

         .historian_type == :
            :
                resp = requests.get(
                    ,
                    headers={: },
                    verify=.verify_ssl,
                    timeout=,
                )
                 resp.status_code == :
                    connections = resp.json().get(, [])
             requests.RequestException  e:
                ()

        
         conn  connections:
            client_ip = conn.get(, conn.get(, ))
             .authorized_clients  client_ip   .authorized_clients:
                .alerts.append({
                    : ,
                    : ,
                    : datetime.now().isoformat(),
                    : client_ip,
                    : ,
                    : ,
                })

         connections

     ():
        
        ()

        integrity_issues = []
         tag  tags:
            :
                 .historian_type == :
                    resp = requests.get(
                        ,
                        params={: , : },
                        auth=(.credentials.get(), .credentials.get()),
                        verify=.verify_ssl,
                        timeout=,
                    )
                     resp.status_code == :
                        items = resp.json().get(, [])
                        
                         (items) == :
                            integrity_issues.append({
                                : tag, : ,
                                : ,
                            })
                        :
                            values = [i.get(, )  i  items  (i.get(), (, ))]
                             values  ((values)) ==   (values) > :
                                integrity_issues.append({
                                    : tag, : ,
                                    : ,
                                })
             requests.RequestException:
                

         issue  integrity_issues:
            .alerts.append({
                : ,
                : ,
                : datetime.now().isoformat(),
                : issue[],
                : issue[],
                :   issue[] ==   ,
            })

         integrity_issues

     ():
        
        indicators = []

        
        
        indicators.append({
            : ,
            : ,
            : ,
        })

        
        indicators.append({
            : ,
            : ,
            : ,
        })

        
        indicators.append({
            : ,
            : ,
            : ,
        })

         indicators

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

         .alerts:
            ()
             alert  .alerts:
                ()
                ()
                ()
                ()

        ()
         indicator  .check_lateral_movement_indicators():
            ()
            ()
            ()


 __name__ == :
    detector = HistorianAttackDetector(
        historian_type=,
        historian_url=,
        api_credentials={: , : },
    )

    detector.set_baseline(
        authorized_clients=[, , , ],
        authorized_query_patterns={},
    )

    detector.check_active_connections()
    detector.check_data_integrity(tags=[, ], hours_back=)
    detector.generate_report()