Protegendo o Servidor Historian em Ambientes OT

Quando Usar

• Ao implantar um novo servidor historian em um ambiente OT e configurá-lo de forma segura desde o início
• Ao fortalecer um historian existente após uma avaliação de segurança identificar que ele é um alvo de alto risco
• Ao projetar a arquitetura de replicação de dados do historian através de uma DMZ para acesso de TI aos dados de processo
• Ao implementar controles de acesso para impedir a modificação não autorizada de dados históricos de processo
• Ao investigar suspeitas de comprometimento do historian ou problemas de integridade de dados

Não usar para segurança de banco de dados apenas de TI sem dados OT (veja o fortalecimento geral de bancos de dados), para segurança de transmissão de dados SCADA em tempo real (veja detecção-de-ataques-em-sistemas-scada), ou para decisões de seleção e dimensionamento do historian.

Pré-requisitos

• Plataforma Historian (OSIsoft PI, Honeywell PHD, GE Proficy, AVEVA Historian) instalada e operacional
• Segmentação de rede com o historian posicionado no Nível 3 (Operações do Site) de acordo com o Modelo Purdue
• Compreensão dos fluxos de dados: dispositivos de campo -> PLCs -> servidores OPC -> historian
• Acesso às credenciais de administração do historian
• Infraestrutura DMZ para replicação de dados voltada para TI

Fluxo de Trabalho

Etapa 1: Auditar a Configuração de Segurança Atual do Historian

Avalie a postura de segurança atual do servidor historian, incluindo exposição de rede, autenticação e controles de acesso.

#!/usr/bin/env python3
"""Ferramenta de Auditoria de Segurança do Historian.

Avalia a configuração de segurança de servidores historian de processo
incluindo exposição de rede, autenticação, controles de acesso
e proteções de integridade de dados.
"""

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:
    """Auditoria de segurança para servidores historian 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):
        """Verifica quais serviços de rede estão expostos pelo historian."""
        print(f"[*] Verificando exposição de rede: {self.ip}")

        # Portas comuns do historian
        ports_to_check = {
            5450: ("PI Data Archive", "Conexões PI SDK/API"),
            5457: ("PI AF Server", "PI Asset Framework"),
            5459: ("PI Notifications", "Serviço de Notificação PI"),
            443: ("HTTPS", "PI Vision / API Web"),
            80: ("HTTP", "Interface web não segura"),
            1433: ("MS SQL Server", "Acesso direto ao banco de dados"),
            5432: ("PostgreSQL", "Acesso direto ao banco de dados"),
            3389: ("RDP", "Área de Trabalho Remota"),
            135: ("RPC", "RPC do Windows"),
            445: ("SMB", "Compartilhamento de Arquivos do Windows"),
            8080: ("HTTP Alt", "Interface web alternativa"),
        }

        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

        # Sinaliza serviços desnecessariamente expostos
        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="Exposição de Rede",
                    title=f"Serviço desnecessário exposto: {svc['service']} (porta {svc['port']})",
                    detail=f"Porta {svc['port']} ({svc['description']}) está acessível no historian",
                    remediation=f"Desative {svc['service']} ou restrinja via firewall do host",
                ))
                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="Criptografia",
                title="Interface web do historian em HTTP não criptografado",

…