Skip to main content
detecting-attacks-on-historian-servers Detect cyber attacks on OT historian servers (OSIsoft PI, Ignition, GE Proficy, Wonderware InSQL) using a Python detector that flags unauthorized queries, data manipulation, and lateral-movement indicators as historians pivot between IT and OT networks. Use when monitoring historians bridging IT/OT zones for compromise, investigating historian-specific CVE exploitation, or validating historian data integrity after a suspected OT incident.
Jump to install Skills Marketplace Discover and explore AI skills built by the community.
Install with Codex or Claude Copy this prompt, paste it into Codex, Claude, or another assistant, and let it review the skill page and install it for you.
Show prompt details A direct command skips the review prompt. Inspect the source before running it.
npx skills add https://github.com/mukul975/Anthropic-Cybersecurity-Skills --skill detecting-attacks-on-historian-serversThe command stays on one line. Scroll horizontally to inspect it before copying.
Prefer a local copy? Download the files currently available to SkillsMP.
Download Zip Downloading... Related occupations SOC
Based on SOC occupation classification
name detecting-attacks-on-historian-servers description Detect cyber attacks on OT historian servers (OSIsoft PI, Ignition, GE Proficy, Wonderware InSQL) using a Python detector that flags unauthorized queries, data manipulation, and lateral-movement indicators as historians pivot between IT and OT networks. Use when monitoring historians bridging IT/OT zones for compromise, investigating historian-specific CVE exploitation, or validating historian data integrity after a suspected OT incident.
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"] mitre_attack ["T0811","T0882","T0888","T0846","T0859"]
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.
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
"""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.
"""
import json
import sys
from collections import defaultdict
from datetime import datetime, timedelta
from typing import Dict , List , Optional
try :
import requests
except ImportError:
print ("Install requests: pip install requests" )
sys.exit(1 )
class HistorianAttackDetector :
"""Detects attacks targeting OT historian servers."""
def
__init__
self, historian_type: str , historian_url: str ,
api_credentials: dict , verify_ssl: bool = False
self
self
"/"
self
self
self
self
set
self
def
set_baseline
self, authorized_clients: List [str ],
authorized_query_patterns: Dict [str , List [str ]]
"""Set baseline of authorized historian clients and query patterns."""
self
set
self
def
check_active_connections
self
List
dict
"""Check for unauthorized connections to historian."""
if
self
"osisoft_pi"
try
f"{self.historian_url} /piwebapi/system/status"
self
"username"
self
"password"
self
10
if
200
"ConnectedClients"
except
as
print
f"[!] PI Web API error: {e} "
elif
self
"ignition"
try
f"{self.historian_url} /data/status/connections"
"Authorization"
f"Bearer {self.credentials.get('token' )} "
self
10
if
200
"connections"
except
as
print
f"[!] Ignition API error: {e} "
for
in
"client_ip"
"address"
""
if
self
and
not
in
self
self
"severity"
"HIGH"
"type"
"UNAUTHORIZED_HISTORIAN_CLIENT"
"timestamp"
"source_ip"
"details"
f"Unauthorized client {client_ip} connected to {self.historian_type} historian"
"mitre"
"T0802 - Automated Collection"
return
def
check_data_integrity
self, tags: List [str ], hours_back: int = 24
"""Check historian data for manipulation indicators."""
print
f"[*] Checking data integrity for {len (tags)} tags over last {hours_back} h"
for
in
try
if
self
"osisoft_pi"
f"{self.historian_url} /piwebapi/streams/{tag} /recorded"
"startTime"
f"*-{hours_back} h"
"endTime"
"*"
self
"username"
self
"password"
self
15
if
200
"Items"
if
len
0
"tag"
"issue"
"NO_DATA"
"detail"
"No data points in expected timeframe - possible deletion"
else
"Value"
0
for
in
if
isinstance
"Value"
int
float
if
and
len
set
1
and
len
100
"tag"
"issue"
"FLATLINE"
"detail"
f"Constant value {values[0 ]} for {len (values)} points - possible replay/spoofing"
except
pass
for
in
self
"severity"
"HIGH"
"type"
f"DATA_INTEGRITY_{issue['issue' ]} "
"timestamp"
"tag"
"tag"
"details"
"detail"
"mitre"
"T0809 - Data Destruction"
if
"issue"
"NO_DATA"
else
"T0832 - Manipulation of View"
return
def
check_lateral_movement_indicators
self
"""Check for indicators of historian being used as pivot point."""
"check"
"Outbound connections to PLC subnets"
"description"
"Historian initiating connections to Level 1 devices may indicate compromise"
"detection"
"Monitor firewall logs for historian IP connecting to PLC ports (502, 102, 44818)"
"check"
"Unauthorized processes on historian server"
"description"
"Attackers may install tools on historian for lateral movement"
"detection"
"Monitor process creation events (Sysmon EventID 1) on historian"
"check"
"Authentication from unexpected sources"
"description"
"Compromised IT systems authenticating to historian for pivoting"
"detection"
"Monitor Windows Security Event 4624 for logons from non-baseline sources"
return
def
generate_report
self
"""Generate historian attack detection report."""
print
f"\n{'=' *70 } "
print
"HISTORIAN ATTACK DETECTION REPORT"
print
f"{'=' *70 } "
print
f"Historian Type: {self.historian_type} "
print
f"Historian URL: {self.historian_url} "
print
f"Report Time: {datetime.now().isoformat()} "
print
f"Total Alerts: {len (self.alerts)} "
if
self
print
f"\n--- ALERTS ---"
for
in
self
print
f"\n [{alert['severity' ]} ] {alert['type' ]} "
print
f" Time: {alert['timestamp' ]} "
print
f" Detail: {alert['details' ]} "
print
f" MITRE ICS: {alert.get('mitre' , 'N/A' )} "
print
f"\n--- LATERAL MOVEMENT CHECKS ---"
for
in
self
print
f"\n Check: {indicator['check' ]} "
print
f" Risk: {indicator['description' ]} "
print
f" Detection: {indicator['detection' ]} "
if
"__main__"
"osisoft_pi"
"https://pi-server.plant.local"
"username"
"pi_reader"
"password"
"api_key_here"
"10.10.2.10"
"10.10.2.20"
"10.10.3.50"
"10.10.150.10"
"REACTOR_01.TEMP"
"PUMP_03.FLOW"
24
Key Concepts Term Definition OT Historian Database server (OSIsoft PI, Ignition, Wonderware) storing time-series process data from SCADA/DCS systems Pivot Point Historian's position between IT and OT networks makes it a prime target for attackers to move between zones Data Replay Attack Feeding historical data to an HMI to mask real-time process manipulation (Stuxnet technique) OSIsoft PI Most widely deployed OT historian, used by 65% of Global 500 process companies Ignition Inductive Automation SCADA platform with historian module, increasingly targeted due to Python scripting capabilities CVE-2025-0921 Ignition SCADA privileged file system vulnerability allowing escalation through malicious project files
Output Format HISTORIAN ATTACK DETECTION REPORT
====================================
Historian: [type and hostname]
Date: YYYY-MM-DD
CONNECTION ANALYSIS:
Authorized Clients: [count]
Unauthorized Clients Detected: [count with IPs]
DATA INTEGRITY:
Tags Checked: [count]
Integrity Issues: [count]
Flatline Detections: [count]
Data Gaps: [count]
LATERAL MOVEMENT INDICATORS:
Outbound PLC Connections: [found/not found]
Unauthorized Processes: [found/not found]
Anomalous Authentication: [found/not found]
More from this repository abusing-dpapi-for-credential-access Extract and decrypt Windows DPAPI-protected secrets (Credential Manager, browser logins/cookies, Wi-Fi credentials, KeePass keys) online or offline using SharpDPAPI, SharpChrome, Mimikatz, or Impacket's dpapi.py, including domain-wide decryption via the DPAPI backup key. Use during authorized red-team credential-access engagements after gaining a foothold or when triaging DPAPI blobs pulled from a host.
abusing-shadow-credentials-for-privesc Take over Active Directory accounts by writing attacker-controlled public keys to msDS-KeyCredentialLink (Shadow Credentials) with pyWhisker, Whisker, or Certipy, then authenticate via PKINIT to recover the target's NT hash without a password reset. Use when BloodHound shows GenericWrite/GenericAll/AddKeyCredentialLink over a target, as a stealthier alternative to ForceChangePassword, during authorized red-team engagements.
acquiring-disk-image-with-dd-and-dcfldd Create forensically sound bit-for-bit disk images with dd or dcfldd on a Linux forensic workstation, preserving evidence integrity through hash verification (MD5/SHA) during acquisition. Use when imaging a suspect drive, USB device, or memory card for investigation, preserving volatile disk evidence during incident response, or producing a verified copy for legal or law-enforcement proceedings before any destructive analysis.
