detecting-attacks-on-historian-servers

name	detecting-attacks-on-historian-servers
description	检测针对OT历史数据服务器（OSIsoft PI、Ignition、Wonderware）的网络攻击，这些服务器位于IT/OT边界，是攻击者在企业网络和控制网络之间进行横向移动的跳板，包括数据篡改、未授权查询以及利用历史数据服务器特定漏洞的攻击。
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

检测针对历史数据服务器的攻击

适用场景

• 监控连接IT与OT网络的历史数据服务器，检测入侵指标
• 检测过程历史数据库中的未授权查询或数据篡改
• 调查通过历史数据服务器在IT和OT区域之间进行的横向移动
• 响应历史数据服务器特定漏洞利用告警（如CVE-2025-0921）
• 在疑似OT安全事件后验证历史数据完整性

不适用于一般数据库安全监控（参见数据库安全技能）、历史数据服务器部署与配置，或仅限IT的数据仓库安全。

前置条件

• 历史数据服务器清单（OSIsoft PI、Ignition、GE Proficy、Wonderware InSQL）
• 历史数据服务器网段的网络监控（包括IT侧和OT侧接口）
• 用于数据完整性验证的历史数据服务器API访问权限
• 正常历史查询模式基线（哪些应用程序查询哪些标签点）
• 了解历史数据服务器架构（数据源、接口、客户端连接）

工作流程

步骤 1：监控历史数据服务器的攻击指标

#!/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: [],
                     authorized_query_patterns: [, []]):
        
        .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()

核心概念

术语	定义
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连接: [发现/未发现]
  未授权进程: [发现/未发现]
  异常身份验证: [发现/未发现]