git clone https://github.com/example/cisco-sdwan-watchdog.git
cd cisco-sdwan-watchdog

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

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

# Initialize a new Node project (if you don’t have one)
npm init -y

# Install core libs
npm install axios dotenv node-fetch slack-webhook tenacity

# (Optional) TypeScript support
npm install --save-dev typescript @types/node @types/node-fetch ts-node
npx tsc --init   # creates tsconfig.json

#!/usr/bin/env python3
"""
Cisco SD-WAN admin‑access watchdog.
Poll V‑Manage for recent login/admin events and alert on unexpected source IPs.
"""

import os
import time
import logging
from typing import Set, Dict, Any

import requests
from dotenv import load_dotenv
from slack_sdk.webhook import WebhookClient
from tenacity import retry, stop_after_attempt, wait_exponential

# ----------------------------------------------------------------------
# Load environment variables (.env file)
# ----------------------------------------------------------------------
load_dotenv()  # reads .env in cwd

VMANAGE_BASE = os.getenv("VMANAGE_BASE_URL")          # e.g. https://vmanage.example.com
VMANAGE_TOKEN = os.getenv("VMANAGE_TOKEN")            # API token (Bearer)
SLACK_WEBHOOK_URL = os.getenv("SLACK_WEBHOOK_URL")    # Optional: Slack alerting
POLL_INTERVAL = int(os.getenv("POLL_INTERVAL", "60")) # seconds
TRUSTED_ADMIN_IPS = set(
    ip.strip() for ip in os.getenv("TRUSTED_ADMIN_IPS", "").split(",") if ip.strip()
)
ADMIN_USERNAMES = set(
    name.strip() for name in os.getenv("ADMIN_USERNAMES", "admin,superuser").split(",") if name.strip()
)

# ----------------------------------------------------------------------
# Logging configuration
# ----------------------------------------------------------------------
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s - %(message)s",
)
log = logging.getLogger("sdwan-watchdog")

# ----------------------------------------------------------------------
# In‑memory dedup store (replace with Redis/DB for HA)
# ----------------------------------------------------------------------
_seen_events: Set[str] = set()


def _auth_headers() -> Dict[str, str]:
    """Return headers required for V‑Manage REST calls."""
    return {
        "Content-Type": "application/json",
        "Accept": "application/json",
        "Authorization": f"Bearer {VMANAGE_TOKEN}",
    }


@retry(reraise=True, stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def _fetch_recent_events() -> list[Dict[str, Any]]:
    """
    Pull the latest control/connections statistics.
    Adjust the endpoint if your version uses a different URI.
    """
    url = f"{VMANAGE_BASE}/dataservice/statistics/control/connections"
    log.debug("Fetching events from %s", url)
    resp = requests.get(url, headers=_auth_headers(), timeout=10)
    resp.raise_for_status()
    data = resp.json()
    # The API returns a list under "data"
    return data.get("data", [])


def _is_suspicious(event: Dict[str, Any]) -> bool:
    """
    Determine whether an event represents an admin login from an untrusted IP.
    Adjust field names according to your V‑Manage version.
    """
    username = event.get("username", "").lower()
    src_ip = event.get("sourceIP") or event.get("src_ip")
    # Some versions store the event type; we assume any entry here is a login attempt.
    if username not in ADMIN_USERNAMES:
        return False
    if not src_ip:
        log.warning("Event missing source IP: %s", event)
        return False
    return src_ip not in TRUSTED_ADMIN_IPS


def _event_key(event: Dict[str, Any]) -> str:
    """Create a stable, unique key for deduplication."""
    # Example: f"{username}:{src_ip}:{timestamp}"
    return f"{event.get('username')}:{event.get('sourceIP')}:{event.get('timestamp')}"


def _send_slack_alert(message: str) -> None:
    if not SLACK_WEBHOOK_URL:
        log.debug("Slack webhook not configured – skipping alert.")
        return
    try:
        webhook = WebhookClient(SLACK_WEBHOOK_URL)
        response = webhook.send(text=message)
        if response.status_code != 200:
            log.error("Failed to send Slack alert: %s %s", response.status_code, response.body)
        else:
            log.info("Slack alert sent.")
    except Exception as exc:  # pragma: no cover
        log.exception("Error while sending Slack alert: %s", exc)


def main() -> None:
    log.info("Starting Cisco SD‑WAN admin watchdog (poll every %ss)", POLL_INTERVAL)
    while True:
        try:
            events = _fetch_recent_events()
            new_suspicious = []
            for ev in events:
                if _is_suspicious(ev):
                    key = _event_key(ev)
                    if key not in _seen_events:
                        _seen_events.add(key)
                        new_suspicious.append(ev)

            if new_suspicious:
                for ev in new_suspicious:
                    msg = (
                        f":warning: *Suspicious admin access detected*\n"
                        f"> *User*: `{ev.get('username')}`\n"
                        f"> *Source IP*: `{ev.get('sourceIP')}`\n"
                        f"> *Time*: `{ev.get('timestamp')}`\n"
                        f"> *Device*: `{ev.get('deviceId', 'N/A')}`"
                    )
                    log.warning(msg)
                    _send_slack_alert(msg)
            else:
                log.debug("No new suspicious events in this poll.")
        except requests.HTTPError as http_err:
            log.error("HTTP error while contacting V‑Manage: %s", http_err)
        except Exception as exc:  # pragma: no cover
            log.exception("Unexpected error: %s", exc)

        time.sleep(POLL_INTERVAL)


if __name__ == "__main__":
    main()

/**
 * Cisco SD-WAN admin-access watchdog (Node.js)
 * Polls V‑Manage for admin login events from unexpected IPs and alerts via Slack.
 */

import dotenv from "dotenv";
import axios";
import { WebhookClient } from "@slack/webhook";
import { retry, exponentialBackoff } from "tenacity"; // tenacity works in JS too via npm package

dotenv.config();

// -------------------------- Configuration --------------------------
const VMANAGE_BASE = process.env.VMANAGE_BASE_URL; // e.g. https://vmanage.example.com
const VMANAGE_TOKEN = process.env.VMANAGE_TOKEN;   // Bearer token
const SLACK_WEBHOOK_URL = process.env.SLACK_WEBHOOK_URL; // optional
const POLL_INTERVAL = Number(process.env.POLL_INTERVAL) || 60; // seconds
const TRUSTED_ADMIN_IPS = new Set(
  (process.env.TRUSTED_ADMIN_IPS || "")
    .split(",")
    .map((ip) => ip.trim())
    .filter(Boolean)
);
const ADMIN_USERNAMES = new Set(
  (process.env.ADMIN_USERNAMES || "admin,superuser")
    .split(",")
    .map((u) => u.trim().toLowerCase())
    .filter(Boolean)
);

// -------------------------- Logging --------------------------
import { createLogger, format, transports } from "winston";
const log = createLogger({
  level: "info",
  format: format.combine(format.timestamp(), format.printf(({ timestamp, level, message }) => `${timestamp} [${level}] ${message}`)),
  transports: [new transports.Console()],
});

// -------------------------- In‑memory dedup --------------------------
const seenEvents = new Set();

// -------------------------- Helper functions --------------------------
function authHeaders() {
  return {
    "Content-Type": "application/json",
    Accept: "application/json",
    Authorization: `Bearer ${VMANAGE_TOKEN}`,
  };
}

/**
 * Fetch recent events from V‑Manage.
 * Adjust the endpoint as needed.
 */
async function fetchRecentEvents() {
  const url = `${VMANAGE_BASE}/dataservice/statistics/control/connections`;
  log.debug(`Fetching events from ${url}`);
  const response = await axios.get(url, { headers: authHeaders(), timeout: 10000 });
  return response.data.data || []; // API returns { data: [...] }
}

/**
 * Determine if an event is suspicious.
 */
function isSuspicious(event) {
  const username = (event.username || "").toLowerCase();
  const srcIp = event.sourceIP || event.src_ip;
  if (!ADMIN_USERNAMES.has(username)) return false;
  if (!srcIp) {
    log.warn("Event missing source IP:", event);
    return false;
  }
  return !TRUSTED_ADMIN_IPS.has(srcIp);
}

/**
 * Create a dedup key.
 */
function eventKey(event) {
  return `${event.username}:${event.sourceIP || event.src_ip}:${event.timestamp}`;
}

/**
 * Send a Slack alert (if webhook configured).
 */
async function sendSlackAlert(message) {
  if (!SLACK_WEBHOOK_URL) {
    log.debug("Slack webhook not configured – skipping alert.");
    return;
  }
  try {
    const webhook = new WebhookClient(SLACK_WEBHOOK_URL);
    const res = await webhook.send({ text: message });
    if (res.statusCode !== 200) {
      log.error(`Slack alert failed: ${res.statusCode} ${res.body}`);
    } else {
      log.info("Slack alert sent.");
    }
  } catch (err) {
    log.error("Error sending Slack alert:", err);
  }
}

/**
 * Main polling loop.
 */
async function main() {
  log.info(`Starting SD‑WAN watchdog (poll every ${POLL_INTERVAL}s)`);
  while (true) {
    try {
      const events = await fetchRecentEvents();
      const suspicious = [];

      for (const ev of events) {
        if (isSuspicious(ev)) {
          const key = eventKey(ev);
          if (!seenEvents.has(key)) {
            seenEvents.add(key);
            suspicious.push(ev);
          }
        }
      }

      if (suspicious.length) {
        for (const ev of suspicious) {
          const msg = [
            ":warning: *Suspicious admin access detected*",
            `> *User*: \`${ev.username}\``,
            `> *Source IP*: \`${ev.sourceIP || ev.src_ip}\``,
            `> *Time*: \`${ev.timestamp}\``,
            `> *Device*: \`${ev.deviceId || "N/A"}\``,
          ].join("\n");
          log.warn(msg);
          await sendSlackAlert(msg);
        }
      } else {
        log.debug("No new suspicious events this poll.");
      }
    } catch (err) {
      if (axios.isAxiosError(err)) {
        log.error(`HTTP error contacting V‑Manage: ${err.message}`);
      } else {
        log.error("Unexpected error:", err);
      }
    }

    // Wait before next poll – using a simple timeout; for production consider using node-cron or Agenda.
    await new Promise((resolve) => setTimeout(resolve, POLL_INTERVAL * 1000));
  }
}

// Run if invoked directly
if (require.main === module) {
  main().catch((err) => {
    log.fatal("Watchdog terminated due to uncaught error:", err);
    process.exit(1);
  });
}

# ----------------------- V‑Manage Connection -----------------------
VMANAGE_BASE_URL=https://vmanage.example.com   # Base URL without trailing slash
VMANAGE_TOKEN=your_long_lived_api_token_here   # Must have read‑only access to statistics/control

# ----------------------- Alerting -----------------------
# Slack Incoming Webhook (optional – leave blank to disable)
SLACK_WEBHOOK_URL=https://hooks.slack.com/services/T00000000/B00000000/XXXXXXXXXXXXXXXXXXXXXXXX

# ----------------------- Polling -----------------------
# How often (in seconds) to query V‑Manage
POLL_INTERVAL=60

# ----------------------- Trusted Admin IPs -----------------------
# Comma‑separated list of IPs/CIDRs you consider safe for admin logins.
# Example: 10.0.0.0/8,192.168.1.10,203.0.113.45
TRUSTED_ADMIN_IPS=10.0.0.0/8,192.168.1.10

# ----------------------- Admin Usernames -----------------------
# Comma‑separated list of usernames that grant admin/privileged rights.
ADMIN_USERNAMES=admin,superuser,netadmin