git clone https://github.com/icarax-labs/sdwan-zero-day-monitor.git
cd sdwan-zero-day-monitor

# Create a virtual env (recommended)
python3 -m venv .venv
source .venv/bin/activate   # Windows: .venv\Scripts\activate

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

npm init -y
npm install axios dotenv typescript ts-node @types/node @types/axios --save-dev
# Create a basic tsconfig.json
npx tsc --init --rootDir src --outDir dist --esModuleInterop --resolveJsonModule --lib es2020,dom --module commonjs

#!/usr/bin/env python3
"""
SD‑WAN Zero‑Day Monitor – Python version
----------------------------------------
- Authenticates to Cisco vManage/DNA Center via basic auth (username/password)
- Retrieves device inventory
- Flags devices running known vulnerable software versions
- Sends a Slack alert (or logs to stdout if no webhook)
- Runs forever with a configurable interval and retry/backoff logic
"""

import os
import sys
import time
import logging
from typing import List, Dict

import requests
from requests.auth import HTTPBasicAuth
from dotenv import load_dotenv
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

# ----------------------------------------------------------------------
# Load configuration from .env (see Step 4)
# ----------------------------------------------------------------------
load_dotenv()  # pulls in VMANAGE_HOST, VMANAGE_USER, VMANAGE_PASS, SLACK_WEBHOOK, POLL_INTERVAL

VMANAGE_HOST = os.getenv("VMANAGE_HOST")
VMANAGE_USER = os.getenv("VMANAGE_USER")
VMANAGE_PASS = os.getenv("VMANAGE_PASS")
SLACK_WEBHOOK = os.getenv("SLACK_WEBHOOK")
POLL_INTERVAL = int(os.getenv("POLL_INTERVAL", "300"))  # seconds, default 5 min

# ----------------------------------------------------------------------
# Known vulnerable version ranges for the 7th SD‑WAN zero‑day (CVE‑2026‑XXXX)
# Format: (major, minor, patch_low, patch_high) – inclusive.
# Adjust if Cisco publishes a different list.
# ----------------------------------------------------------------------
VULNERABLE_RANGES = [
    ((20, 2, 0), (20, 4, 3)),   # 20.2.0 – 20.4.3
    ((21, 0, 0), (21, 1, 2)),   # 21.0.0 – 21.1.2
]

# ----------------------------------------------------------------------
# Logging setup
# ----------------------------------------------------------------------
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s - %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)],
)
log = logging.getLogger("sdwan-monitor")

# ----------------------------------------------------------------------
# Helper functions
# ----------------------------------------------------------------------
def is_version_vulnerable(version_str: str) -> bool:
    """
    Return True if version_str falls inside any of VULNERABLE_RANGES.
    Expects a dotted string like "20.3.1" or "21.0.5".
    """
    try:
        parts = tuple(map(int, version_str.split(".")))
        # Ensure we have at least three parts (major.minor.patch)
        if len(parts) < 3:
            parts = parts + (0,) * (3 - len(parts))
        for low, high in VULNERABLE_RANGES:
            if low <= parts <= high:
                return True
    except ValueError:
        log.warning(f"Unable to parse version string: {version_str}")
    return False


@retry(
    reraise=True,
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=2, max=30),
    retry=retry_if_exception_type((requests.ConnectionError, requests.Timeout)),
)
def get_auth_token() -> str:
    """
    Cisco vManage uses basic auth for the /j_security_check endpoint,
    then returns a JSESSIONID cookie. For simplicity we use HTTPBasicAuth
    on every request – the appliance accepts it for the REST API.
    """
    # No separate token needed; we will pass auth on each call.
    # This function exists to demonstrate retry logic.
    return f"{VMANAGE_USER}:{VMANAGE_PASS}"


@retry(
    reraise=True,
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=2, max=30),
    retry=retry_if_exception_type((requests.ConnectionError, requests.Timeout)),
)
def fetch_devices() -> List[Dict]:
    """
    Calls GET /dataservice/device/detail (vManage) to retrieve all devices.
    Returns a list of dicts; each dict contains at least 'deviceModel',
    'deviceIP', 'hostName', and 'version'.
    """
    url = f"https://{VMANAGE_HOST}/dataservice/device/detail"
    auth = HTTPBasicAuth(VMANAGE_USER, VMANAGE_PASS)
    resp = requests.get(url, auth=auth, verify=False, timeout=15)  # verify=False for self‑signed certs
    resp.raise_for_status()
    data = resp.json()
    # The API returns {"data": [ {...}, {...} ]}
    return data.get("data", [])


def send_slack_alert(message: str) -> None:
    """Posts a simple text message to the configured Slack webhook."""
    if not SLACK_WEBHOOK:
        log.info("Slack webhook not configured – printing alert instead:\n%s", message)
        return
    payload = {"text": message}
    try:
        r = requests.post(SLACK_WEBHOOK, json=payload, timeout=10)
        r.raise_for_status()
        log.info("Slack alert sent successfully.")
    except requests.RequestException as e:
        log.error(f"Failed to send Slack alert: {e}")


def build_alert(vuln_devices: List[Dict]) -> str:
    """Create a human‑readable alert message."""
    lines = [
        f"🚨 *Cisco SD‑WAN Zero‑Day Detected* ({len(vuln_devices)} device(s))",
        "",
    ]
    for d in vuln_devices:
        lines.append(
            f"- *Host*: {d.get('hostName','N/A')} "
            f"({d.get('deviceIP','N/A')}) "
            f"Model: {d.get('deviceModel','N/A')} "
            f"Version: {d.get('version','N/A')}"
        )
    lines.append("\n_Action:_ Immediately isolate affected devices and apply any available workaround from Cisco.")
    return "\n".join(lines)


def main() -> None:
    log.info("Starting SD‑WAN zero‑day monitor (poll every %s seconds)", POLL_INTERVAL)
    while True:
        try:
            devices = fetch_devices()
            log.info("Fetched %d devices from vManage", len(devices))

            vuln = [d for d in devices if is_version_vulnerable(d.get("version", ""))]
            if vuln:
                alert = build_alert(vuln)
                log.warning("Zero‑day vulnerable devices found!")
                send_slack_alert(alert)
            else:
                log.info("No vulnerable devices detected in this poll.")

        except Exception as exc:  # pragma: no cover – catch‑all for unexpected errors
            log.exception("Unexpected error during monitoring loop: %s", exc)

        time.sleep(Poll_INTERVAL)


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        log.info("Monitor stopped by user.")

/**
 * SD‑WAN Zero‑Day Monitor – Node.js / TypeScript version
 * ------------------------------------------------------
 * Mirrors the Python logic: authenticates, fetches device inventory,
 * checks for vulnerable versions, and posts a Slack alert.
 *
 * Requires Node.js ≥18 and the packages installed in Step 2.
 */

import "dotenv/config";
import axios, { AxiosInstance, AxiosError } from "axios";
import { exponentialBackoff, jitter } from "axios-retry";

// ---------------------------------------------------------------------
// Configuration (read from .env)
// ---------------------------------------------------------------------
const VMANAGE_HOST = process.env.VMANAGE_HOST!;
const VMANAGE_USER = process.env.VMANAGE_USER!;
const VMANAGE_PASS = process.env.VMANAGE_PASS!;
const SLACK_WEBHOOK = process.env.SLACK_WEBHOOK ?? "";
const POLL_INTERVAL = Number(process.env.POLL_INTERVAL) || 300; // seconds

// ---------------------------------------------------------------------
// Known vulnerable version ranges (same as Python)
// ---------------------------------------------------------------------
type VersionRange = { low: [number, number, number]; high: [number, number, number] };
const VULNERABLE_RANGES: VersionRange[] = [
  { low: [20, 2, 0], high: [20, 4, 3] },
  { low: [21, 0, 0], high: [21, 1, 2] },
];

// ---------------------------------------------------------------------
// Logging helper (simple console logger with timestamps)
// ---------------------------------------------------------------------
function log(level: string, msg: string): void {
  console.log(`[${new Date().toISOString()}] ${level.padEnd(5)} ${msg}`);
}

// ---------------------------------------------------------------------
// Axios instance with basic auth and retry logic
// ---------------------------------------------------------------------
const api: AxiosInstance = axios.create({
  baseURL: `https://${VMANAGE_HOST}`,
  auth: { username: VMANAGE_USER, password: VMANAGE_PASS },
  httpsAgent: new (require("https")).Agent({ rejectUnauthorized: false }), // self‑signed certs
  timeout: 15000,
});

// Retry on network errors (axios-retry)
exponentialBackoff(api, {
  retries: 5,
  retryDelay: (retryCount) => {
    const delay = Math.pow(2, retryCount) * 1000; // 1s,2s,4s,8s,16s
    return delay + jitter(); // add jitter to avoid thundering herd
  },
  retryCondition: (error: AxiosError) => {
    // Retry on connection/timeout errors or 5xx responses
    return !error.response || error.response!.status >= 500;
  },
});

// ---------------------------------------------------------------------
// Version utilities
// ---------------------------------------------------------------------
function parseVersion(vstr: string): [number, number, number] | null {
  const parts = vstr.split(".").map((p) => parseInt(p, 10));
  if (parts.some(isNaN)) return null;
  // Ensure we have at least three components (major.minor.patch)
  while (parts.length < 3) parts.push(0);
  return [parts[0], parts[1], parts[2]];
}

function isVersionVulnerable(vstr: string): boolean {
  const v = parseVersion(vstr);
  if (!v) {
    log("WARN", `Unable to parse version "${vstr}"`);
    return false;
  }
  return VULNERABLE_RANGES.some(
    (range) =>
      v[0] > range.low[0] ||
      (v[0] === range.low[0] &&
        (v[1] > range.low[1] ||
          (v[1] === range.low[1] && v[2] >= range.low[2]))) &&
      v[0] < range.high[0] ||
      (v[0] === range.high[0] &&
        (v[1] < range.high[1] ||
          (v[1] === range.high[1] && v[2] <= range.high[2])))
  );
}

// ---------------------------------------------------------------------
// Fetch device inventory from vManage
// ---------------------------------------------------------------------
async function fetchDevices(): Promise<any[]> {
  try {
    const { data } = await api.get("/dataservice/device/detail");
    return data.data ?? [];
  } catch (err: any) {
    if (axios.isAxiosError(err)) {
      log("ERROR", `Axios error: ${err.message}`);
    } else {
      log("ERROR", `Unexpected error: ${err}`);
    }
    throw err; // let the outer loop handle it
  }
}

// ---------------------------------------------------------------------
// Slack alerting
// ---------------------------------------------------------------------
async function sendSlackAlert(text: string): Promise<void> {
  if (!SLACK_WEBHOOK) {
    log("INFO", "Slack webhook not set – printing alert:\n" + text);
    return;
  }
  try {
    await axios.post(SLACK_WEBHOOK, { text });
    log("INFO", "Slack alert sent.");
  } catch (e: any) {
    log("ERROR", `Failed to send Slack alert: ${e?.message ?? e}`);
  }
}

function buildAlert(devices: any[]): string {
  const lines = [
    `🚨 *Cisco SD‑WAN Zero‑Day Detected* (${devices.length} device(s))`,
    "",
  ];
  for (const d of devices) {
    lines.push(
      `- *Host*: ${d.hostName ?? "N/A"} (${d.deviceIP ?? "N/A"}) ` +
        `Model: ${d.deviceModel ?? "N/A"} Version: ${d.version ?? "N/A"}`
    );
  }
  lines.push("\n_Action:_ Immediately isolate affected devices and apply any available workaround from Cisco.");
  return lines.join("\n");
}

// ---------------------------------------------------------------------
// Main monitoring loop
// ---------------------------------------------------------------------
async function monitorLoop(): Promise<void> {
  log("INFO", `Starting SD‑WAN zero‑day monitor (poll every ${POLL_INTERVAL}s)`);
  while (true) {
    try {
      const devices = await fetchDevices();
      log("INFO", `Fetched ${devices.length} devices from vManage`);

      const vulnerable = devices.filter((d) =>
        isVersionVulnerable(d.version ?? "")
      );

      if (vulnerable.length > 0) {
        const alert = buildAlert(vulnerable);
        log("WARN", `Zero‑day vulnerable devices found! Sending alert...`);
        await sendSlackAlert(alert);
      } else {
        log("INFO", "No vulnerable devices detected in this poll.");
      }
    } catch (err) {
      log("ERROR", `Error during monitoring iteration: ${err}`);
    }

    // Wait for the next interval
    await new Promise((resolve) => setTimeout(resolve, POLL_INTERVAL * 1000));
  }
}

// Graceful shutdown on SIGINT/SIGTERM
process.on("SIGINT", () => {
  log("INFO", "Received SIGINT – shutting down monitor.");
  process.exit(0);
});
process.on("SIGTERM", () => {
  log("INFO", "Received SIGTERM – shutting down monitor.");
  process.exit(0);
});

// Start the loop
monitorLoop().catch((e) => {
  log("FATAL", `Unhandled error in monitorLoop: ${e}`);
  process.exit(1);
});

# Cisco vManage / DNA Center connection
VMANAGE_HOST=your-vmanage.example.com   # hostname only, no scheme
VMANAGE_USER=apiuser
VMANAGE_PASS=SuperSecretPassword123!

# Optional: If you use token‑based auth (DNA Center), replace the above with:
# DNAC_HOST=your-dnac.example.com
# DNAC_CLIENT_ID=your-client-id
# DNAC_CLIENT_SECRET=your-client-secret

# Slack Incoming Webhook (https://api.slack.com/messaging/webhooks)
SLACK_WEBHOOK=https://hooks.slack.com/services/T00000000/B00000000/XXXXXXXXXXXXXXXXXXXXXXXX

# How often to poll (seconds). Default 300 = 5 minutes.
POLL_INTERVAL=300

python src/monitor.py

# Build (if using TS)
npx tsc
node dist/monitor.js   # or directly: npx ts-node src/monitor.ts