securing-historian-server-in-ot-environment本技能涵盖在OT环境中加固和保护过程历史数据服务器（OSIsoft PI、Honeywell PHD、GE Proficy、AVEVA Historian）。涉及跨Purdue模型各层级的网络部署、历史数据服务器接口的访问控制、通过数据二极管或PI-to-PI连接器在DMZ中进行数据复制、历史数据查询中的SQL注入防护，以及用于安全分析、法规报告和过程优化的过程数据完整性保护。
Independent GitHub snapshot · Automated classification · Original documentation governs
不适用于没有OT数据的纯IT数据库安全（参见通用数据库加固）、实时SCADA数据传输安全（参见detecting-attacks-on-scada-systems），或历史数据服务器的选型和容量规划决策。
评估历史数据服务器当前的安全态势，包括网络暴露、认证和访问控制。
#!/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资产框架"),
5459: ("PI Notifications", "PI通知服务"),
443: ("HTTPS", "PI Vision / Web API"),
80: ("HTTP", "未加密Web界面"),
1433: ("MS SQL Server", "直接数据库访问"),
5432: ("PostgreSQL", "直接数据库访问"),
3389: ("RDP", "远程桌面"),
135: ("RPC", "Windows RPC"),
445: ("SMB", "Windows文件共享"),
8080: ("HTTP Alt", "备用Web界面"),
}
exposed = []
for port, (service, desc) in ports_to_check.items():
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(3)
result = sock.connect_ex((self.ip, port))
sock.close()
if result == 0:
exposed.append({"port": port, "service": service, "description": desc})
except Exception:
pass
# 标记不必要的暴露服务
for svc in exposed:
if svc["port"] in (80, 135, 445, 3389):
self.findings.append(AuditFinding(
finding_id=f"HIST-{self.counter:03d}",
severity="high",
category="网络暴露",
title=f"不必要的服务暴露: {svc['service']}（端口 {svc['port']}）",
detail=f"端口 {svc['port']}（{svc['description']}）在历史数据服务器上可访问",
remediation=f"禁用 {svc['service']} 或通过主机防火墙限制访问",
))
self.counter += 1
if any(s["port"] == 80 for s in exposed):
self.findings.append(AuditFinding(
finding_id=f"HIST-{self.counter:03d}",
severity="high",
category="加密",
title="历史数据服务器Web界面在未加密的HTTP上",
detail="端口80（HTTP）已开放，在明文中暴露凭据和数据",
remediation="将HTTP重定向到HTTPS；禁用端口80",
))
self.counter += 1
return exposed
def check_authentication(self):
"""检查历史数据服务器认证配置。"""
print(f"[*] 正在检查认证配置")
# 检查PI Trust认证是否仍然启用（旧版，不安全）
# PI Trust允许基于IP的认证，无需凭据
checks = [
{
"check": "PI Trust认证",
"risk": "PI Trust允许仅基于IP地址的连接，无需凭据",
"severity": "critical",
"remediation": "将所有PI Trust连接迁移到Windows集成安全",
},
{
"check": "默认piadmin账户",
"risk": "默认PI管理员账户可能使用默认或弱密码",
"severity": "critical",
"remediation": "禁用piadmin；使用带PI映射的命名Windows账户",
},
{
"check": "PI SDK匿名访问",
"risk": "可能允许匿名PI SDK连接",
"severity": "high",
"remediation": "要求所有PI SDK连接进行认证",
},
]
for check in checks:
self.findings.append(AuditFinding(
finding_id=f"HIST-{self.counter:03d}",
severity=check["severity"],
category="认证",
title=f"检查: {check['check']}",
detail=check["risk"],
remediation=check["remediation"],
))
self.counter += 1
def check_data_integrity(self):
"""检查数据完整性保护。"""
print(f"[*] 正在检查数据完整性保护")
integrity_checks = [
AuditFinding(
finding_id=f"HIST-{self.counter:03d}",
severity="high",
category="数据完整性",
title="验证历史数据修改审计跟踪",
detail="对历史过程数据的修改应记录修改前后的值",
remediation="为所有数据修改启用PI审计跟踪；限制编辑权限",
),
AuditFinding(
finding_id=f"HIST-{self.counter + 1:03d}",
severity="medium",
category="数据完整性",
title="验证备份完整性和恢复测试",
detail="历史数据服务器备份应定期测试恢复能力",
remediation="实施自动备份验证，每季度进行恢复测试",
),
]
self.findings.extend(integrity_checks)
self.counter += len(integrity_checks)
def generate_report(self):
"""生成历史数据服务器安全审计报告。"""
report = []
report.append("=" * 70)
report.append("历史数据服务器安全审计报告")
report.append(f"目标: {self.ip} ({self.type})")
report.append(f"日期: {datetime.now().isoformat()}")
report.append("=" * 70)
for sev in ["critical", "high", "medium", "low"]:
findings = [f for f in self.findings if f.severity == sev]
if findings:
report.append(f"\n--- {sev.upper()} ({len(findings)}) ---")
for f in findings:
report.append(f" [{f.finding_id}] {f.title}")
report.append(f" {f.detail}")
report.append(f" 修复: {f.remediation}")
return "\n".join(report)
if __name__ == "__main__":
target = sys.argv[1] if len(sys.argv) > 1 else "10.30.1.50"
audit = HistorianSecurityAudit(target, "PI")
audit.check_network_exposure()
audit.check_authentication()
audit.check_data_integrity()
print(audit.generate_report())
根据供应商安全指南和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中: 审计 > 启用安全变更审计
# 为以下项启用审计: 点创建/删除、数据编辑、安全变更
配置通过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"
- "不活跃30分钟后会话超时"
| 术语 | 定义 |
|---|---|
| 过程历史数据服务器（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中基于每个标签的访问控制，决定哪些用户/应用程序可以读取或写入特定过程数据点 |
历史数据服务器安全评估报告
=====================================
历史数据服务器: [类型和版本]
服务器: [主机名/IP]
网络区域: [Purdue层级]
认证:
PI Trust条目: [N]（应为0）
默认账户: [启用/禁用]
Windows认证: [启用/禁用]
网络暴露:
开放端口: [列表]
不必要的服务: [列表]
数据完整性:
审计跟踪: [启用/禁用]
备份已测试: [日期]
DMZ复制:
方法: [PI-to-PI / 数据二极管 / VPN]
方向: [单向 / 双向]