securing-historian-server-in-ot-environment

在 OT 环境中保护 Historian 服务器安全

何时使用

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

不要使用于没有 OT 数据的纯 IT 数据库安全（参见通用数据库加固），用于实时 SCADA 数据传输安全（参见检测针对 SCADA 系统的攻击），或用于 historian 选型和容量规划决策。

先决条件

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

工作流

步骤 1：审计当前 Historian 安全配置

评估 historian 服务器当前的安全态势，包括网络暴露、身份验证和访问控制。

#!/usr/bin/env python3
"""Historian Security Audit Tool.

Evaluates the security configuration of process historian servers
including network exposure, authentication, access controls,
and data integrity protections.
"""

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:
    """Security audit for OT historian servers."""

    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):
        """Check which network services are exposed by the historian."""
        print(f"[*] Checking network exposure: {self.ip}")

        # Common historian ports
        ports_to_check = {
            5450: ("PI Data Archive", "PI SDK/API connections"),
            5457: ("PI AF Server", "PI Asset Framework"),
            5459: ("PI Notifications", "PI Notification Service"),
            443: ("HTTPS", "PI Vision / Web API"),
            80: ("HTTP", "Unsecured web interface"),
            1433: ("MS SQL Server", "Direct database access"),
            5432: ("PostgreSQL", "Direct database access"),
            3389: ("RDP", "Remote Desktop"),
            135: ("RPC", "Windows RPC"),
            445: ("SMB", "Windows File Sharing"),
            8080: ("HTTP Alt", "Alternative web interface"),
        }

        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

        # Flag unnecessary exposed services
        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="Network Exposure",
                    title=f"Unnecessary service exposed: {svc['service']} (port {svc['port']})",
                    detail=f"Port {svc['port']} ({svc['description']}) is accessible on historian",
                    remediation=f"Disable {svc['service']} or restrict via host firewall",
                ))
                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="Encryption",
                title="Historian web interface on unencrypted HTTP",
                detail="Port 80 (HTTP) is open, exposing credentials and data in cleartext",
                remediation="Redirect HTTP to HTTPS; disable port 80",
            ))
            self.counter += 1

        return exposed

    def check_authentication(self):
        """Check historian authentication configuration."""
        print(f"[*] Checking authentication configuration")

        # Check if PI Trust authentication is still enabled (legacy, insecure)
        # PI Trust allows IP-based authentication without credentials
        checks = [
            {
                "check": "PI Trust Authentication",
                "risk": "PI Trust allows connections based on IP address alone without credentials",
                "severity": "critical",
                "remediation": "Migrate all PI Trust connections to Windows Integrated Security",
            },
            {
                "check": "Default piadmin account",
                "risk": "Default PI administrator account may have default or weak password",
                "severity": "critical",
                "remediation": "Disable piadmin; use named Windows accounts with PI mappings",
            },
            {
                "check": "PI SDK anonymous access",
                "risk": "Anonymous PI SDK connections may be permitted",
                "severity": "high",
                "remediation": "Require authentication for all PI SDK connections",
            },
        ]

        for check in checks:
            self.findings.append(AuditFinding(
                finding_id=f"HIST-{self.counter:03d}",
                severity=check["severity"],
                category="Authentication",
                title=f"Check: {check['check']}",
                detail=check["risk"],
                remediation=check["remediation"],
            ))
            self.counter += 1

    def check_data_integrity(self):
        """Check data integrity protections."""
        print(f"[*] Checking data integrity protections")

        integrity_checks = [
            AuditFinding(
                finding_id=f"HIST-{self.counter:03d}",
                severity="high",
                category="Data Integrity",
                title="Verify historical data modification audit trail",
                detail="Modifications to historical process data should be logged with before/after values",
                remediation="Enable PI audit trail for all data modifications; restrict edit permissions",
            ),
            AuditFinding(
                finding_id=f"HIST-{self.counter + 1:03d}",
                severity="medium",
                category="Data Integrity",
                title="Verify backup integrity and recovery testing",
                detail="Historian backups should be tested regularly for recovery capability",
                remediation="Implement automated backup verification with quarterly recovery testing",
            ),
        ]
        self.findings.extend(integrity_checks)
        self.counter += len(integrity_checks)

    def generate_report(self):
        """Generate historian security audit report."""
        report = []
        report.append("=" * 70)
        report.append("HISTORIAN SECURITY AUDIT REPORT")
        report.append(f"Target: {self.ip} ({self.type})")
        report.append(f"Date: {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"    Fix: {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：加固 Historian 服务器

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

# OSIsoft PI Server Hardening Script (Windows)
# Based on OSIsoft Security Best Practices Guide

# 1. Disable PI Trust authentication - migrate to Windows Integrated Security
# In PI SMT (System Management Tools):
# Security > Mappings & Trusts > Delete all Trust entries
# Create PI Mappings for Windows groups instead

# 2. Disable the default piadmin account
# In PI SMT: Security > Identities, Users & Groups
# Set piadmin account to disabled

# 3. Configure Windows Firewall for PI Server
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 "Allow PI SDK connections from OT zones only"

New-NetFirewallRule -DisplayName "PI AF Server" -Direction Inbound `
    -Protocol TCP -LocalPort 5457 -Action Allow `
    -RemoteAddress "10.30.0.0/16" `
    -Description "Allow PI AF connections from Operations zone"

New-NetFirewallRule -DisplayName "PI Vision HTTPS" -Direction Inbound `
    -Protocol TCP -LocalPort 443 -Action Allow `
    -RemoteAddress "172.16.0.0/16" `
    -Description "Allow PI Vision HTTPS from DMZ only"

# Block HTTP (force HTTPS)
New-NetFirewallRule -DisplayName "Block HTTP" -Direction Inbound `
    -Protocol TCP -LocalPort 80 -Action Block

# Block RDP from non-authorized sources
New-NetFirewallRule -DisplayName "RDP Restrict" -Direction Inbound `
    -Protocol TCP -LocalPort 3389 -Action Allow `
    -RemoteAddress "10.30.1.100" `
    -Description "Allow RDP from admin jump server only"

# 4. Enable Windows audit policies for CIP-007 compliance
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. Configure PI audit trail for data integrity
# In PI SMT: Audit > Enable auditing for security changes
# Enable auditing for: point creation/deletion, data edits, security changes

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

使用 PI-to-PI 接口或数据二极管配置通过 DMZ 的 historian 数据复制，以便在不暴露 OT historian 的情况下为 IT 提供过程数据访问。

# Historian DMZ Replication Architecture
#
# OT Historian (Level 3) --> Data Diode --> DMZ Historian (Level 3.5) <-- Enterprise (Level 4)
#
# Key principle: Enterprise users NEVER connect directly to the OT historian.
# Data flows unidirectionally from OT to DMZ.

architecture:
  ot_historian:
    location: "Level 3 - Site Operations"
    server: "PI-OT-01 (10.30.1.50)"
    role: "Primary data collection from OPC servers and PLCs"
    access: "OT operators and engineers only"

  data_diode:
    location: "Between Level 3 and Level 3.5"
    device: "Waterfall Security Unidirectional Gateway"
    direction: "OT -> DMZ (physically enforced one-way)"
    protocol: "PI-to-PI replication protocol"

  dmz_historian:
    location: "Level 3.5 - DMZ"
    server: "PI-DMZ-01 (172.16.1.50)"
    role: "Read-only mirror of OT historian for enterprise access"
    access: "Enterprise users via PI Vision (HTTPS)"
    data_delay: "Near real-time (typically 5-30 second delay)"

  enterprise_access:
    method: "PI Vision web application on DMZ historian"
    authentication: "Windows Integrated Security with MFA"
    protocol: "HTTPS (TLS 1.2+)"
    restrictions:
      - "Read-only access to process data"
      - "No write-back capability to OT historian"
      - "No direct database queries - PI Vision API only"
      - "Session timeout after 30 minutes of inactivity"

关键概念

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

工具与系统

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

输出格式

Historian Security Assessment Report
=====================================
Historian: [Type and Version]
Server: [Hostname/IP]
Network Zone: [Purdue Level]

AUTHENTICATION:
  PI Trust entries: [N] (should be 0)
  Default accounts: [enabled/disabled]
  Windows auth: [enabled/disabled]

NETWORK EXPOSURE:
  Open ports: [list]
  Unnecessary services: [list]

DATA INTEGRITY:
  Audit trail: [enabled/disabled]
  Backup tested: [date]

DMZ REPLICATION:
  Method: [PI-to-PI / Data Diode / VPN]
  Direction: [Unidirectional / Bidirectional]