securing-historian-server-in-ot-environment

name	securing-historian-server-in-ot-environment
description	本技能涵盖在OT环境中加固和保护过程历史数据服务器（OSIsoft PI、Honeywell PHD、GE Proficy、AVEVA Historian）。涉及跨Purdue模型各层级的网络部署、历史数据服务器接口的访问控制、通过数据二极管或PI-to-PI连接器在DMZ中进行数据复制、历史数据查询中的SQL注入防护，以及用于安全分析、法规报告和过程优化的过程数据完整性保护。
domain	cybersecurity
subdomain	ot-ics-security
tags	["ot-security","ics","scada","industrial-control","iec62443","historian","osisoft-pi","data-integrity"]
version	1.0.0
author	mahipal
license	Apache-2.0

在OT环境中保护历史数据服务器

适用场景

• 在OT环境中部署新历史数据服务器并从一开始进行安全配置时
• 在安全评估将现有历史数据服务器识别为高风险目标后对其进行加固时
• 设计通过DMZ进行历史数据复制的架构以供IT访问过程数据时
• 实施访问控制以防止未授权修改历史过程数据时
• 调查疑似历史数据服务器入侵或数据完整性问题时

不适用于没有OT数据的纯IT数据库安全（参见通用数据库加固）、实时SCADA数据传输安全（参见detecting-attacks-on-scada-systems），或历史数据服务器的选型和容量规划决策。

前置条件

• 历史数据服务器平台（OSIsoft PI、Honeywell PHD、GE Proficy、AVEVA Historian）已安装并运行
• 按Purdue模型将历史数据服务器置于第3层（站点运营）的网络分段
• 了解数据流：现场设备 -> PLC -> OPC服务器 -> 历史数据服务器
• 历史数据服务器管理凭据访问权限
• 用于面向IT的数据复制的DMZ基础设施

工作流程

步骤 1：审计当前历史数据服务器安全配置

评估历史数据服务器当前的安全态势，包括网络暴露、认证和访问控制。

#!/usr/bin/env python3
"""历史数据服务器安全审计工具。

评估过程历史数据服务器的安全配置，
包括网络暴露、认证、访问控制
和数据完整性保护。
"""

import json
import socket
import ssl
import subprocess
import sys
from dataclasses import dataclass, field, asdict
from datetime import datetime


@dataclass
class AuditFinding:
    finding_id: str
    severity: str
    category: str
    title: str
    detail: str
    remediation: str


class HistorianSecurityAudit:
    """OT历史数据服务器安全审计。"""

    def __init__(self, historian_ip, historian_type="PI"):
        self.ip = historian_ip
        self.type = historian_type
        self.findings = []
        self.counter = 1

    def check_network_exposure(self):
        """检查历史数据服务器暴露的网络服务。"""
        print(f"[*] 正在检查网络暴露: {self.ip}")

        # 常见历史数据服务器端口
        ports_to_check = {
            5450: ("PI Data Archive", "PI SDK/API连接"),
            5457: ("PI AF Server", "PI资产框架"),
            : (, ),
            : (, ),
            : (, ),
            : (, ),
            : (, ),
            : (, ),
            : (, ),
            : (, ),
            : (, ),
        }

        exposed = []
         port, (service, desc)  ports_to_check.items():
            :
                sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                sock.settimeout()
                result = sock.connect_ex((.ip, port))
                sock.close()
                 result == :
                    exposed.append({: port, : service, : desc})
             Exception:
                

        
         svc  exposed:
             svc[]  (, , , ):
                .findings.append(AuditFinding(
                    finding_id=,
                    severity=,
                    category=,
                    title=,
                    detail=,
                    remediation=,
                ))
                .counter += 

         (s[] ==   s  exposed):
            .findings.append(AuditFinding(
                finding_id=,
                severity=,
                category=,
                title=,
                detail=,
                remediation=,
            ))
            .counter += 

         exposed

     ():
        
        ()

        
        
        checks = [
            {
                : ,
                : ,
                : ,
                : ,
            },
            {
                : ,
                : ,
                : ,
                : ,
            },
            {
                : ,
                : ,
                : ,
                : ,
            },
        ]

         check  checks:
            .findings.append(AuditFinding(
                finding_id=,
                severity=check[],
                category=,
                title=,
                detail=check[],
                remediation=check[],
            ))
            .counter += 

     ():
        
        ()

        integrity_checks = [
            AuditFinding(
                finding_id=,
                severity=,
                category=,
                title=,
                detail=,
                remediation=,
            ),
            AuditFinding(
                finding_id=,
                severity=,
                category=,
                title=,
                detail=,
                remediation=,
            ),
        ]
        .findings.extend(integrity_checks)
        .counter += (integrity_checks)

     ():
        
        report = []
        report.append( * )
        report.append()
        report.append()
        report.append()
        report.append( * )

         sev  [, , , ]:
            findings = [f  f  .findings  f.severity == sev]
             findings:
                report.append()
                 f  findings:
                    report.append()
                    report.append()
                    report.append()

         .join(report)


 __name__ == :
    target = sys.argv[]  (sys.argv) >   
    audit = HistorianSecurityAudit(target, )
    audit.check_network_exposure()
    audit.check_authentication()
    audit.check_data_integrity()
    (audit.generate_report())

步骤 2：加固历史数据服务器

根据供应商安全指南和IEC 62443要求应用安全加固。

# OSIsoft PI服务器加固脚本（Windows）
# 基于OSIsoft安全最佳实践指南

# 1. 禁用PI Trust认证 — 迁移到Windows集成安全
# 在PI SMT（系统管理工具）中:
# 安全 > 映射和信任 > 删除所有Trust条目
# 改为创建Windows组的PI映射

# 2. 禁用默认piadmin账户
# 在PI SMT中: 安全 > 身份、用户和组
# 将piadmin账户设置为禁用

# 3. 为PI服务器配置Windows防火墙
New-NetFirewallRule -DisplayName "PI Data Archive" -Direction Inbound `
    -Protocol TCP -LocalPort 5450 -Action Allow `
    -RemoteAddress "10.30.0.0/16","10.20.0.0/16" `
    -Description "仅允许来自OT区域的PI SDK连接"

New-NetFirewallRule -DisplayName "PI AF Server" -Direction Inbound `
    -Protocol TCP -LocalPort 5457 -Action Allow `
    -RemoteAddress "10.30.0.0/16" `
    -Description "允许来自运营区域的PI AF连接"

New-NetFirewallRule -DisplayName "PI Vision HTTPS" -Direction Inbound `
    -Protocol TCP -LocalPort 443 -Action Allow `
    -RemoteAddress "172.16.0.0/16" `
    -Description "仅允许来自DMZ的PI Vision HTTPS"

# 阻止HTTP（强制使用HTTPS）
New-NetFirewallRule -DisplayName "Block HTTP" -Direction Inbound `
    -Protocol TCP -LocalPort 80 -Action Block

# 仅允许从授权源进行RDP
New-NetFirewallRule -DisplayName "RDP Restrict" -Direction Inbound `
    -Protocol TCP -LocalPort 3389 -Action Allow `
    -RemoteAddress "10.30.1.100" `
    -Description "仅允许从管理跳板机进行RDP"

# 4. 启用Windows审计策略以符合CIP-007要求
auditpol /set /subcategory:"Logon" /success:enable /failure:enable
auditpol /set /subcategory:"Account Lockout" /success:enable /failure:enable
auditpol /set /subcategory:"File System" /success:enable /failure:enable
auditpol /set /subcategory:"Registry" /success:enable /failure:enable

# 5. 配置PI审计跟踪以确保数据完整性
# 在PI SMT中: 审计 > 启用安全变更审计
# 为以下项启用审计: 点创建/删除、数据编辑、安全变更

步骤 3：实施通过DMZ的安全数据复制

配置通过DMZ的历史数据复制，使用PI-to-PI接口或数据二极管为IT提供对过程数据的访问，而无需暴露OT历史数据服务器。

# 历史数据服务器DMZ复制架构
#
# OT历史数据服务器（第3层）--> 数据二极管 --> DMZ历史数据服务器（第3.5层）<-- 企业网络（第4层）
#
# 关键原则：企业用户永远不直接连接到OT历史数据服务器。
# 数据单向从OT流向DMZ。

architecture:
  ot_historian:
    location: "第3层 — 站点运营"
    server: "PI-OT-01 (10.30.1.50)"
    role: "从OPC服务器和PLC收集原始数据"
    access: "仅限OT操作员和工程师"

  data_diode:
    location: "第3层和第3.5层之间"
    device: "Waterfall Security单向网关"
    direction: "OT -> DMZ（物理强制单向）"
    protocol: "PI-to-PI复制协议"

  dmz_historian:
    location: "第3.5层 — DMZ"
    server: "PI-DMZ-01 (172.16.1.50)"
    role: "用于企业访问的OT历史数据服务器只读镜像"
    access: "企业用户通过PI Vision（HTTPS）"
    data_delay: "近实时（通常5-30秒延迟）"

  enterprise_access:
    method: "DMZ历史数据服务器上的PI Vision Web应用程序"
    authentication: "带MFA的Windows集成安全"
    protocol: "HTTPS（TLS 1.2+）"
    restrictions:
      - "对过程数据的只读访问"
      - "不具备向OT历史数据服务器写回的能力"
      - "不允许直接数据库查询 — 仅通过PI Vision API"
       

核心概念

术语	定义
过程历史数据服务器（Process Historian）	以高频率（亚秒到秒级）从工业控制系统收集、存储和提供时间序列过程数据的服务器
PI Trust	基于IP地址/主机名的旧版OSIsoft PI认证方法；不安全，应迁移到Windows集成安全
数据二极管（Data Diode）	硬件强制的单向网关，确保历史数据仅从OT流向DMZ，防止反向访问
PI-to-PI接口（PI-to-PI Interface）	OSIsoft复制机制，在服务器之间同步PI数据，用于DMZ数据镜像
审计跟踪（Audit Trail）	历史数据服务器功能，记录对历史数据的所有修改，包含修改前后的值、用户身份和时间戳
标签安全（Tag Security）	PI中基于每个标签的访问控制，决定哪些用户/应用程序可以读取或写入特定过程数据点

工具和系统

• OSIsoft PI Server: AVEVA（原OSIsoft）开发的行业领先过程历史数据服务器，在90%以上的大型工业设施中使用
• AVEVA Historian: 具有类SQL查询界面的过程数据时间序列数据库
• Waterfall Security: 用于单向历史数据复制的硬件数据二极管
• PI Vision: 基于Web的PI数据可视化工具，部署在DMZ中供企业访问

输出格式

历史数据服务器安全评估报告
=====================================
历史数据服务器: [类型和版本]
服务器: [主机名/IP]
网络区域: [Purdue层级]

认证:
  PI Trust条目: [N]（应为0）
  默认账户: [启用/禁用]
  Windows认证: [启用/禁用]

网络暴露:
  开放端口: [列表]
  不必要的服务: [列表]

数据完整性:
  审计跟踪: [启用/禁用]
  备份已测试: [日期]

DMZ复制:
  方法: [PI-to-PI / 数据二极管 / VPN]
  方向: [单向 / 双向]