# Detecting Attacks On Historian Servers > 检测针对OT历史数据服务器（OSIsoft PI、Ignition、Wonderware）的网络攻击，这些服务器位于IT/OT边界，是攻击者在企业网络和控制网络之间进行横向移动的跳板，包括数据篡改、未授权查询以及利用历史数据服务器特定漏洞的攻击。 - Skill: `killvxk/detecting-attacks-on-historian-servers` (Agent Skill, multi-file: 4 files) - Install (CLI): `npx skillmds@latest add killvxk/detecting-attacks-on-historian-servers` - Raw SKILL.md: https://api.skillmd.com/api/skills/killvxk/detecting-attacks-on-historian-servers/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools - License: Apache-2.0 - Author: killvxk (https://skillmd.com/u/killvxk) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/killvxk/detecting-attacks-on-historian-servers --- # 检测针对历史数据服务器的攻击 ## 适用场景 - 监控连接IT与OT网络的历史数据服务器，检测入侵指标 - 检测过程历史数据库中的未授权查询或数据篡改 - 调查通过历史数据服务器在IT和OT区域之间进行的横向移动 - 响应历史数据服务器特定漏洞利用告警（如CVE-2025-0921） - 在疑似OT安全事件后验证历史数据完整性 **不适用于**一般数据库安全监控（参见数据库安全技能）、历史数据服务器部署与配置，或仅限IT的数据仓库安全。 ## 前置条件 - 历史数据服务器清单（OSIsoft PI、Ignition、GE Proficy、Wonderware InSQL） - 历史数据服务器网段的网络监控（包括IT侧和OT侧接口） - 用于数据完整性验证的历史数据服务器API访问权限 - 正常历史查询模式基线（哪些应用程序查询哪些标签点） - 了解历史数据服务器架构（数据源、接口、客户端连接） ## 工作流程 ### 步骤 1：监控历史数据服务器的攻击指标 ```python #!/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. """ 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.historian_type = historian_type self.historian_url = historian_url.rstrip("/") self.credentials = api_credentials self.verify_ssl = verify_ssl self.alerts = [] self.authorized_clients = set() self.authorized_queries = {} 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.authorized_clients = set(authorized_clients) self.authorized_queries = authorized_query_patterns def check_active_connections(self) -> List[dict]: """Check for unauthorized connections to historian.""" connections = [] if self.historian_type == "osisoft_pi": try: resp = requests.get( f"{self.historian_url}/piwebapi/system/status", auth=(self.credentials.get("username"), self.credentials.get("password")), verify=self.verify_ssl, timeout=10, ) if resp.status_code == 200: data = resp.json() connections = data.get("ConnectedClients", []) except requests.RequestException as e: print(f"[!] PI Web API 错误: {e}") elif self.historian_type == "ignition": try: resp = requests.get( f"{self.historian_url}/data/status/connections", headers={"Authorization": f"Bearer {self.credentials.get('token')}"}, verify=self.verify_ssl, timeout=10, ) if resp.status_code == 200: connections = resp.json().get("connections", []) except requests.RequestException as e: print(f"[!] Ignition API 错误: {e}") # 检查未授权客户端 for conn in connections: client_ip = conn.get("client_ip", conn.get("address", "")) if self.authorized_clients and client_ip not in self.authorized_clients: self.alerts.append({ "severity": "HIGH", "type": "UNAUTHORIZED_HISTORIAN_CLIENT", "timestamp": datetime.now().isoformat(), "source_ip": client_ip, "details": f"未授权客户端 {client_ip} 连接到 {self.historian_type} 历史数据服务器", "mitre": "T0802 - Automated Collection", }) return connections def check_data_integrity(self, tags: List[str], hours_back: int = 24): """Check historian data for manipulation indicators.""" print(f"[*] 正在检查 {len(tags)} 个标签点过去 {hours_back} 小时的数据完整性") integrity_issues = [] for tag in tags: try: if self.historian_type == "osisoft_pi": resp = requests.get( f"{self.historian_url}/piwebapi/streams/{tag}/recorded", params={"startTime": f"*-{hours_back}h", "endTime": "*"}, auth=(self.credentials.get("username"), self.credentials.get("password")), verify=self.verify_ssl, timeout=15, ) if resp.status_code == 200: items = resp.json().get("Items", []) # 检查可疑模式 if len(items) == 0: integrity_issues.append({ "tag": tag, "issue": "NO_DATA", "detail": "预期时间段内无数据点 - 可能已被删除", }) else: values = [i.get("Value", 0) for i in items if isinstance(i.get("Value"), (int, float))] if values and len(set(values)) == 1 and len(values) > 100: integrity_issues.append({ "tag": tag, "issue": "FLATLINE", "detail": f"恒定值 {values[0]} 持续 {len(values)} 个数据点 - 可能是回放/欺骗攻击", }) except requests.RequestException: pass for issue in integrity_issues: self.alerts.append({ "severity": "HIGH", "type": f"DATA_INTEGRITY_{issue['issue']}", "timestamp": datetime.now().isoformat(), "tag": issue["tag"], "details": issue["detail"], "mitre": "T0809 - Data Destruction" if issue["issue"] == "NO_DATA" else "T0832 - Manipulation of View", }) return integrity_issues def check_lateral_movement_indicators(self): """Check for indicators of historian being used as pivot point.""" indicators = [] # 检查1：历史数据服务器向一级设备发起出站连接 # （历史数据服务器应接收数据，而不是主动发起到PLC的连接） indicators.append({ "check": "向PLC子网的出站连接", "description": "历史数据服务器向一级设备发起连接可能表明已被入侵", "detection": "监控防火墙日志，检查历史数据服务器IP是否连接PLC端口（502、102、44818）", }) # 检查2：历史数据服务器上的新进程或服务 indicators.append({ "check": "历史数据服务器上的未授权进程", "description": "攻击者可能在历史数据服务器上安装工具以进行横向移动", "detection": "监控历史数据服务器上的进程创建事件（Sysmon EventID 1）", }) # 检查3：历史数据服务器的异常身份验证 indicators.append({ "check": "来自意外来源的身份验证", "description": "已入侵的IT系统向历史数据服务器进行身份验证以进行跳板攻击", "detection": "监控Windows安全事件4624，检查非基线来源的登录", }) return indicators def generate_report(self): """Generate historian attack detection report.""" print(f"\n{'='*70}") print("历史数据服务器攻击检测报告") print(f"{'='*70}") print(f"历史数据服务器类型: {self.historian_type}") print(f"历史数据服务器URL: {self.historian_url}") print(f"报告时间: {datetime.now().isoformat()}") print(f"告警总数: {len(self.alerts)}") if self.alerts: print(f"\n--- 告警 ---") for alert in self.alerts: print(f"\n [{alert['severity']}] {alert['type']}") print(f" 时间: {alert['timestamp']}") print(f" 详情: {alert['details']}") print(f" MITRE ICS: {alert.get('mitre', 'N/A')}") print(f"\n--- 横向移动检查 ---") for indicator in self.check_lateral_movement_indicators(): print(f"\n 检查: {indicator['check']}") print(f" 风险: {indicator['description']}") print(f" 检测: {indicator['detection']}") if __name__ == "__main__": detector = HistorianAttackDetector( historian_type="osisoft_pi", historian_url="https://pi-server.plant.local", api_credentials={"username": "pi_reader", "password": "api_key_here"}, ) detector.set_baseline( authorized_clients=["10.10.2.10", "10.10.2.20", "10.10.3.50", "10.10.150.10"], authorized_query_patterns={}, ) detector.check_active_connections() detector.check_data_integrity(tags=["REACTOR_01.TEMP", "PUMP_03.FLOW"], hours_back=24) detector.generate_report() ``` ## 核心概念 | 术语 | 定义 | |------|------| | OT历史数据服务器（OT Historian） | 存储SCADA/DCS系统时间序列过程数据的数据库服务器（OSIsoft PI、Ignition、Wonderware） | | 跳板点（Pivot Point） | 历史数据服务器位于IT和OT网络之间，是攻击者在区域间移动的主要目标 | | 数据回放攻击（Data Replay Attack） | 向HMI馈送历史数据以掩盖实时过程操控（Stuxnet技术） | | OSIsoft PI | 最广泛部署的OT历史数据服务器，被全球500强过程工业企业中65%使用 | | Ignition | Inductive Automation的SCADA平台，带有历史数据模块，因Python脚本功能而日益成为攻击目标 | | CVE-2025-0921 | Ignition SCADA特权文件系统漏洞，允许通过恶意项目文件进行权限提升 | ## 输出格式 ``` 历史数据服务器攻击检测报告 ==================================== 历史数据服务器: [类型和主机名] 日期: YYYY-MM-DD 连接分析: 已授权客户端: [数量] 检测到未授权客户端: [数量及IP] 数据完整性: 检查的标签点: [数量] 完整性问题: [数量] 平线检测: [数量] 数据缺口: [数量] 横向移动指标: 出站PLC连接: [发现/未发现] 未授权进程: [发现/未发现] 异常身份验证: [发现/未发现] ```