Implementation Guide

ICARAX Tech Blog: Implementing Automated Threat Intelligence Monitoring for OT/ICS Environments

Topic: Dozens of Minnesota Water Utilities Targeted in Coordinated OT Attacks
Context: Following recent coordinated attacks on critical water infrastructure, security engineers must move from reactive to proactive monitoring. This guide provides a production-ready framework for integrating real-time threat intelligence feeds into your security operations center (SOC) to detect anomalies in Operational Technology (OT) networks.

Step 1: Prerequisites

Before implementing the monitoring engine, ensure you have the following:

API Access:
- Threat Intelligence Feed: Access to an aggregator (e.g., AlienVault OTX, VirusTotal, or a private MISP instance).
- Security Orchestration (SOAR) API: Access to your incident response platform.
Environment:
- Python 3.9+ or Node.js 18+ installed.
- A secure environment (Docker or VirtualEnv) to isolate the monitoring service.
Network Access:
- Egress access to threat intelligence endpoints.
- Note: In OT environments, this service should reside in a DMZ/IDMZ, never directly on the Control Network.

Step 2: Installation and Setup

Python Environment

# Create a virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install production-ready dependencies
pip install requests pydantic python-dotenv loguru

JavaScript/TypeScript Environment

# Initialize project
npm init -y

# Install dependencies
npm install axios dotenv zod

Step 3: Basic Implementation

We will implement a Threat Intelligence Integrator that fetches known malicious IP addresses/hashes and compares them against local system logs.

Python Implementation (Asynchronous)

This implementation uses pydantic for strict data validation—crucial when handling untrusted intelligence feeds.

import os
import asyncio
import httpx
from pydantic import BaseModel, ValidationError
from loguru import logger
from dotenv import load_dotenv

load_dotenv()

# --- Data Models ---
class ThreatIndicator(BaseModel):
    """Schema for validated threat intelligence."""
    indicator: str
    type: str  # e.g., 'ipv4', 'ha256'
    severity: str
    description: str

class OTThreatMonitor:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.threatintel.example.com/v1"
        self.headers = {"Authorization": f"Bearer {self.api_key}"}

    async def fetch_latest_threats(self) -> list[ThreatIndicator]:
        """Fetches and validates threat indicators from the provider."""
        async with httpx.AsyncClient() as client:
            try:
                # Simulating API call to a threat feed
                response = await client.get(f"{self.base_url}/indicators", headers=self.headers, timeout=10.0)
                response.raise_for_status()
                data = response.json()
                
                validated_indicators = []
                for item in data:
                    try:
                        validated_indicators.append(ThreatIndicator(**item))
                    except ValidationError as e:
                        logger.error(f"Data validation failed for item: {e}")
                
                return validated_indicators
            except httpx.HTTPError as e:
                logger.error(f"Network error fetching threats: {e}")
                return []

    async def scan_network_logs(self, logs: list[str]):
        """Cross-references logs with known indicators."""
        threats = await self.fetch_latest_threats()
        if not threats:
            logger.warning("No threat indicators to scan.")
            return

        logger.info(f"Starting scan against {len(logs)} log entries...")
        
        for log_entry in logs:
            for threat in threats:
                if threat.indicator in log_entry:
                    self.trigger_alert(threat, log_entry)

    def trigger_alert(self, threat: ThreatIndicator, log_entry: str):
        """High-priority alert triggering."""
        logger.critical(f"🚨 OT ATTACK DETECTED 🚨")
        logger.critical(f"Indicator: {threat.indicator} | Severity: {threat.severity}")
        logger.critical(f"Context: {log_entry}")
        # In production, this would call a Webhook or PagerDuty API

async def main():
    # Mock Data
    API_KEY = os.getenv("THREAT_INTEL_API_KEY", "default_key")
    mock_logs = [
        "2023-10-27 10:00:01 INFO Connection from 192.168.1.50",
        "2023-10-27 10:05:22 WARN Unauthorized access attempt from 45.33.22.11", # Malicious IP
        "2023-10-27 10:10:00 INFO PLC Heartbeat OK"
    ]

    monitor = OTThreatMonitor(API_KEY)
    await monitor.scan_network_logs(mock_logs)

if __name__ == "__main__":
    asyncio.run(main())

TypeScript Implementation (Modern/Strict)

Using Zod for schema validation and Axios for requests.

import axios from 'axios';
import { z } from 'zod';
import dotenv from 'dotenv';

dotenv.config();

// --- Schema Definitions ---
const ThreatIndicatorSchema = z.object({
  indicator: z.string(),
  type: z.enum(['ipv4', 'ha256', 'domain']),
  severity: z.enum(['low', 'edium', 'high', 'critical']),
  description: z.string(),
});

type ThreatIndicator = z.infer<typeof ThreatIndicatorSchema>;

class OTThreatService {
  private apiKey: string;
  private apiUrl: string;

  constructor() {
    this.apiKey = process.env.THREAT_INTEL_API_KEY || '';
    this.apiUrl = 'https://api.threatintel.example.com/v1';
  }

  /**
   * Fetches indicators with strict type safety
   */
  async getThreatIndicators(): Promise<ThreatIndicator[]> {
    try {
      const response = await axios.get(`${this.apiUrl}/indicators`, {
        headers: { Authorization: `Bearer ${this.apiKey}` },
        timeout: 5000
      });

      // Validate array of objects against schema
      return response.data.map((item: any) => ThreatIndicatorSchema.parse(item));
    } catch (error) {
      if (axios.isAxiosError(error)) {
        console.error(`API Error: ${error.message}`);
      } else {
        console.error(`Validation Error: ${error.message}`);
      }
      return [];
    }
  }

  /**
   * Scans incoming OT traffic logs
   */
  async monitorLogs(logs: string[]): Promise<void> {
    const threats = await this.getThreatIndicators();
    
    for (const log of logs) {
      const match = threats.find(t => log.includes(t.indicator));
      if (match) {
        this.raiseAlert(match, log);
      }
    }
  }

  private raiseAlert(threat: ThreatIndicator, context: string): void {
    console.error(`[ALERT] [${threat.severity.toUpperCase()}] Match found: ${threat.indicator}`);
    console.error(`[CONTEXT] ${context}`);
  }
}

// --- Execution ---
const service = new OTThreatService();
const sampleLogs = [
  "Login success: user=admin ip=10.0.0.5",
  "Malicious connection detected from 45.33.22.11",
  "Modbus Write Command: Register 40001"
];

service.monitorLogs(sampleLogs).catch(console.error);

Step 4: Configuration

Always use environment variables to prevent leaking sensitive credentials.

.env file template:

# Threat Intelligence API Credentials
THREAT_INTEL_API_KEY=your_secure_api_key_here

# Monitoring Settings
LOG_LEVEL=INFO
SCAN_INTERVAL_SECONDS=300

# Alerting Webhooks
PAGERDUTY_WEBHOOK_URL=https://events.pagerduty.com/v2/enqueue

Step 5: Common Patterns

The "Watchdog" Pattern

Instead of one-off scripts, use a scheduled task (Cron) or a long-running loop with an exponential backoff strategy for API calls to prevent getting rate-limited by your threat intelligence provider.

The "Validator" Pattern

Never trust API responses. Always use Pydantic (Python) or Zod (TS) to ensure the data structure hasn't changed, which could cause a crash during a critical security event.

Step 6: Troubleshooting

Error	Cause	Resolution
`ValidationError`	The API returned a field type that doesn't match our schema.	Update the `ThreatIndicator` model to accommodate new API fields.
`TimeoutError`	API provider is slow or network is congested.	Increase `timeout` in requests or check firewall egress rules.
`401 Unauthorized`	Invalid or expired API Key.	Check `.env` file and verify key status in provider dashboard.
`ConnectionRefused`	Service cannot reach the internet.	Verify the service is in a DMZ with appropriate proxy settings.

Step 7: Production Checklist

Isolate the Service: Ensure the monitoring service runs in a segmented network zone (DMZ) and cannot directly communicate with PLCs.
Fail-Safe Mechanism: If the threat intelligence API goes down, ensure the system logs a "Health Check Failed" alert but continues to monitor local logs.
Rate Limiting: Implement a throttle to avoid being blocked by your threat intel provider.
Secrets Management: In production, use AWS Secrets Manager, HashiCorp Vault, or Azure Key Vault instead of .env files.
Audit Logging: Every time a threat is detected, ensure the event is logged to a secure, immutable log server (e.g., Splunk or ELK stack).

Next Steps

Get API Access - Sign up at the official website
Try the Examples - Run the code snippets above
Read the Docs - Check official documentation
Join Communities - Discord, Reddit, GitHub discussions
Experiment - Build something cool!

Dozens of Minnesota Water Utilities Targeted in Coordinated OT Attacks

Implementation Guide

ICARAX Tech Blog: Implementing Automated Threat Intelligence Monitoring for OT/ICS Environments

Step 1: Prerequisites

Step 2: Installation and Setup

Python Environment

JavaScript/TypeScript Environment

Step 3: Basic Implementation

Python Implementation (Asynchronous)

TypeScript Implementation (Modern/Strict)

Step 4: Configuration

Step 5: Common Patterns

The "Watchdog" Pattern

The "Validator" Pattern

Step 6: Troubleshooting

Step 7: Production Checklist

Next Steps

Further Reading

Comments (0)

Similar Blogs

Ai • Technology • Machine-learning

Unpatched Fastjson Vulnerability Exploited in Attacks

Ai • Technology • Machine-learning

DentaQuest Data Breach Potentially Impacts Over 23 Million People

Ai • Technology • Machine-learning

Rockwell Patches Code Execution Flaws in Arena Simulation Software