detecting-attacks-on-historian-servers

Files (4)

• LICENSE
• SKILL.md
• api-reference.md · references/
• agent.py · scripts/

SKILL.md (first lines)

# Detecting Attacks on Historian Servers

## When to Use

- When monitoring historian servers that bridge IT and OT networks for compromise indicators
- When detecting unauthorized queries or data manipulation in process historian databases
- When investigating lateral movement through historian servers between IT and OT zones
- When responding to alerts about exploitation of historian-specific vulnerabilities (CVE-2025-0921)
- When validating historian data integrity after a suspected OT security incident

**Do not use** for general database security monitoring (see database security skills), for historian deployment and configuration, or for IT-only data warehouse security.

## Prerequisites

- Historian server inventory (OSIsoft PI, Ignition, GE Proficy, Wonderware InSQL)
- Network monitoring on historian network segments (both IT-facing and OT-facing interfaces)
- Historian API access for data integrity validation
- Baseline of normal historian query patterns (which applications query which tags)
- Understanding of historian architecture (data sources, interfaces, client connections)

## Workflow

### Step 1: Monitor Historian for Attack Indicators

```python
#!/usr/bin/env python3
"""OT Historian Attack Detector.

Monitors historian servers for unauthorized access, data manipulation,
lateral movement indicators, and exploitation of historian-specific
vulnerabilities. Supports OSIsoft PI and Ignition platforms.
"""

import json
import sys
from collections import defaultdict
from datetime import datetime, timedelta
from typing import Dict, List, Optional

try:
    import requests
except ImportError:
    print("Install requests: pip install requests")
    sys.exit(1)


class HistorianAttackDetector:
    """Detects attacks targeting OT historian servers."""

    def __init__(self, historian_type: str, historian_url: str,
                 api_credentials: dict, verify_ssl: bool = False):
        self.historian_type = historian_type
        self.historian_url = historian_url.rstrip("/")
        self.credentials = api_credentials
        self.verify_ssl = verify_ssl
        self.alerts = []
        self.authorized_clients = set()
        self.authorized_queries = {}

    def set_baseline(self, authorized_clients: List[str],
                     authorized_query_patterns: Dict[str, List[str]]):
        """Set baseline of authorized historian clients and query patterns."""
        self.authorized_clients = set(authorized_clients)
        self.authorized_queries = authorized_query_patterns

    def check_active_connections(self) -> List[dict]:
        """Check for unauthorized connections to historian."""
        connections = []

        if self.historian_type == "osisoft_pi":
            try:
                resp = requests.get(
                    f"{self.historian_url}/piwebapi/system/status",
                    auth=(self.credentials.get("username"), self.credentials.get("password")),
                    verify=self.verify_ssl,
                    timeout=10,
                )
                if resp.status_code == 200:
                    data = resp.json()
                    connections = data.get("ConnectedClients", [])
            except requests.RequestException as e:
                print(f"[!] PI Web API error: {e}")

        elif self.historian_type == "ignition":
            try:
                resp = requests.get(
                    f"{self.historian_url}/data/status/connections",
                    headers={"Authorization": f"Bearer {self.credentials.get('token')}"},
                    verify=self.verify_ssl,
                    timeout=10,
                )
                if resp.status_code == 200:
                    connections = resp.json().get("connections", [])
            except requests.RequestException as e:
                print(f"[!] Ignition API error: {e}")

        # Check for unauthorized clients
        for conn in connections:
            client_ip = conn.get("client_ip", conn.get("address", ""))
            if self.authorized_clients and client_ip not in self.authorized_clients:
                self.alerts.append({
                    "severity": "HIGH",
                    "type": "UNAUTHORIZED_HISTORIAN_CLIENT",
                    "timestamp": datetime.now().isoformat(),
                    "source_ip": client_ip,
                    "details": f"Unauthorized client {client_ip} connected to {self.historian_type} historian",
                    "mitre": "T0802 - Automated Collection",
                })

        return connections

    def check_data_integrity(self, tags: List[str], hours_back: int = 24):
        """Check historian data for manipulation indicators."""
        print(f"[*] Checking data integrity for {len(tags)} tags over last {hours_back}h")

        integrity_issues = []
        for tag in tags:
            try:
                if self.historian_type == "osisoft_pi":

From mukul975/Anthropic-Cybersecurity-Skills, Apache-2.0.

More from mukul975/Anthropic-Cybersecurity-Skills