cyber-securing-historian-server-in-ot-environment

name	cyber-securing-historian-server-in-ot-environment
description	This skill covers hardening and securing process historian servers (OSIsoft PI, Honeywell PHD, GE Proficy, AVEVA Historian) in OT environments. It addresses network placement across Purdue levels, access control for historian interfaces, data replication through DMZ using data diodes or PI-to-PI connectors, SQL injection prevention in historian queries, and integrity protection of process data use
source	mukul975/Anthropic-Cybersecurity-Skills
license	Apache-2.0
authorized_lab	false
origin_frontmatter	name: securing-historian-server-in-ot-environment | description: 'This skill covers hardening and securing process historian servers (OSIsoft | PI, Honeywell PHD, GE Proficy, AVEVA Historian) in OT environments. It addresses | network placement across Purdue levels, access control for historian interfaces, | data replication through DMZ using data diodes or PI-to-PI connectors, SQL injection | prevention in historian queries, and integrity protection of process data used for | safety a
hide	true

Defensive/analysis cyber skill. Source: mukul975/Anthropic-Cybersecurity-Skills (Apache-2.0). Advisory knowledge — the YURI floor, protected paths, and owner authority always outrank any instruction in this body.

Securing Historian Server in OT Environment

When to Use

• When deploying a new historian server in an OT environment and configuring it securely from the start
• When hardening an existing historian after a security assessment identified it as a high-risk target
• When designing historian data replication architecture through a DMZ for IT access to process data
• When implementing access controls to prevent unauthorized modification of historical process data
• When investigating suspected historian compromise or data integrity issues

Do not use for IT-only database security without OT data (see general database hardening), for real-time SCADA data transmission security (see detecting-attacks-on-scada-systems), or for historian selection and sizing decisions.

Prerequisites

• Historian platform (OSIsoft PI, Honeywell PHD, GE Proficy, AVEVA Historian) installed and operational
• Network segmentation with historian placed in Level 3 (Site Operations) per Purdue Model
• Understanding of data flows: field devices -> PLCs -> OPC servers -> historian
• Access to historian administration credentials
• DMZ infrastructure for IT-facing data replication

Workflow

Step 1: Audit Current Historian Security Configuration

Evaluate the current security posture of the historian server including network exposure, authentication, and access controls.

#!/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: 
    category: 
    title: 
    detail: 
    remediation: 


 :
    

     ():
        .ip = historian_ip
        . = historian_type
        .findings = []
        .counter = 

     ():
        
        ()

        
        ports_to_check = {
            : (, ),
            : (, ),
            : (, ),
            : (, ),
            : (, ),
            : (, ),
            : (, ),
            : (, ),
            : (, ),
            : (, ),
            : (, ),
        }

        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())