securing-historian-server-in-ot-environment

# 在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：审计当前历史数据服务器安全配置

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

```python
#!/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())
```

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

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

```powershell
# 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历史数据服务器。

```yaml
# 历史数据服务器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中基于每个标签的访问控制，决定哪些用户/应用程序可以读取或写入特定过程数据点 |

## 工具和系统

- **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]
  方向: [单向 / 双向]
```