Skip to main content 首页 创作者 autohandai community-skills detecting-attacks-on-historian-servers
detecting-attacks-on-historian-servers 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.
跳到安装 Skills Marketplace 发现并探索由社区构建的 Agent Skills
用 Codex 或 Claude 帮你安装 复制这段 Prompt,粘贴到 Codex、Claude 或其他助手里,让它检查 Skill 页面并帮你完成安装。
直接命令不会经过审查 Prompt;运行前请先检查来源。
npx skills add https://github.com/autohandai/community-skills --skill detecting-attacks-on-historian-servers命令会保持在同一行。复制前请横向滚动并检查完整内容。
想先保存到本地?可下载 SkillsMP 当前能够提供的文件。
下载 Zip 下载中... 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
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."""
( ):
.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()
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]