git clone https://github.com/icarax/ics-patch-tuesday-guide.git
cd ics-patch-tuesday-guide

# Create a virtual environment (recommended)
python -m venv .venv
source .venv/bin/activate   # on Windows: .venv\Scripts\activate

# Install dependencies
pip install --upgrade pip
pip install requests python-dotenv tqdm

# Initialize a new npm project (if you didn't clone the repo)
npm init -y

# Install core libraries
npm install axios dotenv

# Install TypeScript and type definitions (if you want TS)
npm install --save-dev typescript @types/node ts-node
npx tsc --init   # creates a basic tsconfig.json

ics-patch-tuesday-guide/
│
├─ python/
│   ├─ main.py
│   ├─ .env
│   └─ requirements.txt
│
├─ js/
│   ├─ src/
│   │   └─ index.ts
│   ├─ .env
│   ├─ package.json
│   └─ tsconfig.json
│
└─ README.md

#!/usr/bin/env python3
"""
ICS Patch Tuesday – Vendor feed aggregation & vulnerability check
=================================================================
* Reads Schneider Electric & Siemens patch feeds (mock endpoints)
* Compares against a local CSV inventory of OT assets
* Prints a JSON report of vulnerable assets and suggested remediation
"""

import os
import csv
import json
import logging
from typing import List, Dict, Any
from urllib.parse import urljoin

import requests
from dotenv import load_dotenv
from tqdm import tqdm

# ----------------------------------------------------------------------
# Configuration & helpers
# ----------------------------------------------------------------------
load_dotenv()  # loads .env into os.environ

SCHNEIDER_BASE = os.getenv("SCHNEIDER_API_BASE", "https://api.example.com/schneider")
SIEMENS_BASE   = os.getenv("SIEMENS_API_BASE",   "https://api.example.com/siemens")
API_KEY        = os.getenv("VENDOR_API_KEY")    # shared key for both mock services
INVENTORY_FILE = os.getenv("INVENTORY_FILE", "assets.csv")
REPORT_FILE    = os.getenv("REPORT_FILE", "vulnerability_report.json")

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s - %(message)s",
)
log = logging.getLogger(__name__)

def _auth_headers() -> Dict[str, str]:
    """Vendor‑specific auth – adjust per real API."""
    return {"Authorization": f"Bearer {API_KEY}", "Accept": "application/json"}

def fetch_patch_feed(base_url: str) -> List[Dict[str, Any]]:
    """
    Pull the patch feed from a vendor.
    Expected JSON shape (mock):
    [
      {
        "vendor": "Schneider Electric",
        "product": "Modicon M340",
        "version": "2.4.1",
        "cve": "CVE-2024-12345",
        "severity": "Critical",
        "fixedVersion": "2.4.2",
        "patchUrl": "https://support.schneider-electric.com/patch/..."
      },
      ...
    ]
    """
    url = urljoin(base_url, "/patches")
    log.info(f"Fetching patch feed from {url}")
    try:
        resp = requests.get(url, headers=_auth_headers(), timeout=15)
        resp.raise_for_status()
        data = resp.json()
        if not isinstance(data, list):
            raise ValueError("Expected a JSON list")
        log.info(f"Received {len(data)} patch entries from {base_url}")
        return data
    except requests.RequestException as exc:
        log.error(f"HTTP error while fetching {url}: {exc}")
        return []  # fail‑soft – you may want to raise instead
    except ValueError as exc:
        log.error(f"Invalid JSON from {url}: {exc}")
        return []

def load_inventory(csv_path: str) -> List[Dict[str, str]]:
    """
    Simple CSV inventory. Expected columns:
    asset_id, vendor, product, installedVersion, location, criticality
    """
    assets = []
    try:
        with open(csv_path, newline="", encoding="utf-8") as f:
            reader = csv.DictReader(f)
            for row in reader:
                assets.append({k.strip(): v.strip() for k, v in row.items()})
        log.info(f"Loaded {len(assets)} assets from {csv_path}")
    except FileNotFoundError:
        log.error(f"Inventory file not found: {csv_path}")
    except Exception as exc:
        log.error(f"Failed to read inventory: {exc}")
    return assets

def version_lt(v1: str, v2: str) -> bool:
    """
    Very naive semantic‑version comparison – sufficient for the demo.
    For production use `packaging.version` (pip install packaging).
    """
    def normalize(v):
        return [int(x) for x in v.split(".") if x.isdigit()]
    return normalize(v1) < normalize(v2)

def check_vulnerabilities(
    patches: List[Dict[str, Any]],
    assets: List[Dict[str, str]]
) -> List[Dict[str, Any]]:
    """
    Return a list of findings:
    {
      "asset_id": "...",
      "vendor": "...",
      "product": "...",
      "installedVersion": "...",
      "cve": "...",
      "severity": "...",
      "fixedVersion": "...",
      "patchUrl": "..."
    }
    """
    findings = []
    # Index patches by (vendor, product) for faster lookup
    patch_index: Dict[tuple, List[Dict[str, Any]]] = {}
    for p in patches:
        key = (p.get("vendor", "").strip().lower(), p.get("product", "").strip().lower())
        patch_index.setdefault(key, []).append(p)

    for asset in tqdm(assets, desc="Scanning assets"):
        key = (
            asset.get("vendor", "").strip().lower(),
            asset.get("product", "").strip().lower(),
        )
        installed = asset.get("installedVersion", "")
        if not installed:
            continue
        for patch in patch_index.get(key, []):
            fixed = patch.get("fixedVersion")
            if fixed and version_lt(installed, fixed):
                findings.append({
                    "asset_id": asset.get("asset_id"),
                    "vendor": asset.get("vendor"),
                    "product": asset.get("product"),
                    "installedVersion": installed,
                    "cve": patch.get("cve"),
                    "severity": patch.get("severity"),
                    "fixedVersion": fixed,
                    "patchUrl": patch.get("patchUrl"),
                })
                # Break after first matching patch – you may want to list all
                break
    return findings

def main() -> None:
    # 1️⃣ Pull feeds
    schneider_patches = fetch_patch_feed(SCHNEIDER_BASE)
    siemens_patches   = fetch_patch_feed(SIEMENS_BASE)
    all_patches = schneider_patches + siemens_patches

    # 2️⃣ Load inventory
    assets = load_inventory(INVENTORY_FILE)

    # 3️⃣ Correlate
    vulns = check_vulnerabilities(all_patches, assets)

    # 4️⃣ Output report
    report = {
        "generatedAt": __import__("datetime").datetime.utcnow().isoformat() + "Z",
        "totalAssetsScanned": len(assets),
        "totalVulnerabilitiesFound": len(vulns),
        "findings": vulns,
    }
    with open(REPORT_FILE, "w", encoding="utf-8") as f:
        json.dump(report, f, indent=2)
    log.info(f"Report written to {REPORT_FILE} ({len(vulns)} findings)")

if __name__ == "__main__":
    main()

#!/usr/bin/env node
/**
 * ICS Patch Tuesday – Node.js/TS version
 * Mirrors the Python logic but uses axios + dotenv.
 */

import * as dotenv from "dotenv";
import axios from "axios";
import * as fs from "fs";
import * as path from "path";
import { parse } from "csv-parse/sync"; // npm i csv-parse

dotenv.config();

interface PatchEntry {
  vendor: string;
  product: string;
  version: string; // vulnerable version advertised in feed
  cve: string;
  severity: string;
  fixedVersion: string;
  patchUrl: string;
}

interface Asset {
  asset_id: string;
  vendor: string;
  product: string;
  installedVersion: string;
  location?: string;
  criticality?: string;
}

interface Finding extends Asset {
  cve: string;
  severity: string;
  fixedVersion: string;
  patchUrl: string;
}

// ----------------------------------------------------------------------
// Config
// ----------------------------------------------------------------------
const SCHNEIDER_BASE = process.env.SCHNEIDER_API_BASE ?? "https://api.example.com/schneider";
const SIEMENS_BASE   = process.env.SIEMENS_API_BASE   ?? "https://api.example.com/siemens";
const API_KEY        = process.env.VENDOR_API_KEY ?? "";
const INVENTORY_FILE = process.env.INVENTORY_FILE ?? path.resolve(__dirname, "../../assets.csv");
const REPORT_FILE    = process.env.REPORT_FILE    ?? path.resolve(__dirname, "../../vulnerability_report.json");

if (!API_KEY) {
  console.error("❌ Vendor API key not set – check .env");
  process.exit(1);
}

const axiosInstance = axios.create({
  timeout: 15000,
  headers: {
    Authorization: `Bearer ${API_KEY}`,
    Accept: "application/json",
  },
});

// ----------------------------------------------------------------------
// Helpers
// ----------------------------------------------------------------------
async function fetchPatchFeed(baseUrl: string): Promise<PatchEntry[]> {
  const url = new URL("/patches", baseUrl).toString();
  console.log(`🔎 Fetching patch feed from ${url}`);
  try {
    const { data } = await axiosInstance.get<PatchEntry[]>(url);
    console.log(`✅ Received ${data.length} patches from ${baseUrl}`);
    return data;
  } catch (err: any) {
    if (axios.isAxiosError(err)) {
      console.error(`❌ HTTP error fetching ${url}: ${err.response?.status ?? err.message}`);
    } else {
      console.error(`❌ Unexpected error: ${err}`);
    }
    return []; // fail‑soft
  }
}

function loadInventory(csvPath: string): Asset[] {
  if (!fs.existsSync(csvPath)) {
    console.error(`❌ Inventory file not found: ${csvPath}`);
    return [];
  }
  const fileContent = fs.readFileSync(csvPath, { encoding: "utf8" });
  const records = parse(fileContent, {
    columns: true,
    skip_empty_lines: true,
    trim: true,
  }) as Asset[];
  console.log(`📦 Loaded ${records.length} assets from ${csvPath}`);
  return records;
}

/**
 * Very naive semver compare – replace with `semver` package in prod.
 */
function lt(v1: string, v2: string): boolean {
  const normalize = (v: string) => v.split(".").map(Number);
  return normalize(v1).some((n, i) => n < (normalize(v2)[i] ?? 0));
}

function checkVulnerabilities(
  patches: PatchEntry[],
  assets: Asset[]
): Finding[] {
  // Index patches by (vendor, product) → list
  const patchMap = new Map<
    string,
    PatchEntry[]
  >();
  for const p of patches {
    const key = `${p.vendor.toLowerCase()}|${p.product.toLowerCase()}`;
    patchMap.set(key, (patchMap.get(key) ?? []).concat(p));
  }

  const findings: Finding[] = [];
  for (const asset of assets) {
    const key = `${asset.vendor.toLowerCase()}|${asset.product.toLowerCase()}`;
    const installed = asset.installedVersion;
    if (!installed) continue;
    const candPatches = patchMap.get(key) ?? [];
    for (const p of candPatches) {
      if (p.fixedVersion && lt(installed, p.fixedVersion)) {
        findings.push({
          ...asset,
          cve: p.cve,
          severity: p.severity,
          fixedVersion: p.fixedVersion,
          patchUrl: p.patchUrl,
        });
        // break after first match – remove if you want all
        break;
      }
    }
  }
  return findings;
}

// ----------------------------------------------------------------------
// Main
// ----------------------------------------------------------------------
(async function main() {
  console.log("🚀 Starting ICS Patch Tuesday aggregation...");
  const [schneiderPatches, siemensPatches] = await Promise.all([
    fetchPatchFeed(SCHNEIDER_BASE),
    fetchPatchFeed(SIEMENS_BASE),
  ]);
  const allPatches = [...schneiderPatches, ...siemensPatches];

  const assets = loadInventory(INVENTORY_FILE);
  const vulns = checkVulnerabilities(allPatches, assets);

  const report = {
    generatedAt: new Date().toISOString(),
    totalAssetsScanned: assets.length,
    totalVulnerabilitiesFound: vulns.length,
    findings: vulns,
  };

  fs.writeFileSync(REPORT_FILE, JSON.stringify(report, null, 2), "utf8");
  console.log(
    `🗒️  Report written to ${REPORT_FILE} (${vulns.length} findings)`
  );
})();

# -------------------------------------------------
# Vendor API credentials (shared for both examples)
# -------------------------------------------------
VENDOR_API_KEY=your-super-secret-api-key-here

# Base URLs – replace with the real vendor endpoints when available
SCHNEIDER_API_BASE=https://api.example.com/schneider
SIEMENS_API_BASE=https://api.example.com/siemens

# Path to your OT asset inventory (CSV format)
INVENTORY_FILE=./assets.csv

# Where the final JSON report should be written
REPORT_FILE=./vulnerability_report.json

asset_id,vendor,product,installedVersion,location,criticality
OT-001,Schneider Electric,Modicon M340,2.4.1,Plant A,High
OT-002,Siemens,S7-1500 CPU,2.6.0,Plant B,Medium
OT-003,Schneider Electric,Altivar 71,3.2.0,Plant C,Low
OT-004,Siemens,SCALANCE X200,4.1.2,Plant D,High