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

# Install required packages
pip install --upgrade pip
pip install requests python-dotenv tenacity

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

# Install dependencies
npm install axios dotenv
# For TypeScript add:
npm install --save-dev typescript @types/node ts-node
npx tsc --init   # creates tsconfig.json

#!/usr/bin/env python3
"""
fetch_cisco_critical.py
Retrieves Cisco Security Advisories marked as Critical from the last 30 days.
"""

import os
import sys
import time
from datetime import datetime, timedelta

import requests
from dotenv import load_dotenv
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

# ----------------------------------------------------------------------
# Load environment variables from .env (if present)
# ----------------------------------------------------------------------
load_dotenv()
CLIENT_ID = os.getenv("CISCO_CLIENT_ID")
CLIENT_SECRET = os.getenv("CISCO_CLIENT_SECRET")

if not CLIENT_ID or not CLIENT_SECRET:
    sys.exit("❌  Missing CISCO_CLIENT_ID or CISCO_CLIENT_SECRET in environment.")

# ----------------------------------------------------------------------
# Constants
# ----------------------------------------------------------------------
TOKEN_URL = "https://cloudsso.cisco.com/as/token.oauth2"
ADVISORY_URL = "https://api.cisco.com/security/advisories"
# Ask for advisories published within the last 30 days
START_DATE = (datetime.utcnow() - timedelta(days=30)).strftime("%Y-%m-%d")

# ----------------------------------------------------------------------
# Helper: fetch OAuth2 token (with retry)
# ----------------------------------------------------------------------
@retry(
    reraise=True,
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type((requests.ConnectionError, requests.Timeout)),
)
def get_access_token() -> str:
    """Exchange client credentials for a Bearer token."""
    payload = {
        "grant_type": "client_credentials",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
    }
    headers = {"Content-Type": "application/x-www-form-urlencoded"}
    resp = requests.post(TOKEN_URL, data=payload, headers=headers, timeout=10)
    resp.raise_for_status()
    data = resp.json()
    return data["access_token"]


# ----------------------------------------------------------------------
# Helper: fetch advisories (with retry & basic rate‑limit handling)
# ----------------------------------------------------------------------
@retry(
    reraise=True,
    stop=stop_after_attempt(4),
    wait=wait_exponential(multiplier=2, min=3, max=30),
    retry=retry_if_exception_type((requests.ConnectionError, requests.Timeout)),
)
def fetch_critical_advisories(token: str):
    """Return a list of Critical advisories published after START_DATE."""
    headers = {"Authorization": f"Bearer {token}", "Accept": "application/json"}
    params = {
        "publishedStartDate": START_DATE,
        "severity": "Critical",
        # The API supports pagination; we request a reasonable page size.
        "pageSize": 100,
    }

    advisories = []
    url = ADVISORY_URL
    while url:
        resp = requests.get(url, headers=headers, params=params, timeout=15)
        # Handle 429 (Too Many Requests) by respecting Retry-After header
        if resp.status_code == 429:
            retry_after = int(resp.headers.get("Retry-After", "5"))
            time.sleep(retry_after)
            continue  # retry the same request
        resp.raise_for_status()
        data = resp.json()
        advisories.extend(data.get("advisories", []))
        # Pagination: the API returns a `next` link if more pages exist
        url = data.get("metadata", {}).get("links", {}).get("next")
        # After first request we don't need to resend params (they're in the link)
        params = None
    return advisories


# ----------------------------------------------------------------------
# Main routine
# ----------------------------------------------------------------------
def main():
    try:
        token = get_access_token()
    except Exception as exc:
        sys.exit(f"❌  Failed to obtain OAuth token: {exc}")

    try:
        criticals = fetch_critical_advisories(token)
    except Exception as exc:
        sys.exit(f"❌  Error while fetching advisories: {exc}")

    if not criticals:
        print("✅  No Critical Cisco advisories in the last 30 days.")
        return

    print(f"🔎  Found {len(criticals)} Critical Cisco advisory(ies) (last 30 days):\n")
    for adv in criticals:
        adv_id = adv.get("advisoryId", "N/A")
        title = adv.get("title", "N/A")
        cvss = adv.get("cvss", {}).get("baseScore", "N/A")
        pub = adv.get("publicationDate", "N/A")
        print(f"- [{adv_id}] {title}")
        print(f"  CVSS: {cvss}   Published: {pub}")
        print()


if __name__ == "__main__":
    main()

// fetch_cisco_critical.ts
// Retrieves Cisco Security Advisories marked as Critical from the last 30 days.
import * as dotenv from "dotenv";
import axios, { AxiosInstance } from "axios";

dotenv.config();

const CLIENT_ID = process.env.CISCO_CLIENT_ID ?? "";
const CLIENT_SECRET = process.env.CISCO_CLIENT_SECRET ?? "";

if (!CLIENT_ID || !CLIENT_SECRET) {
  console.error("❌  Missing CISCO_CLIENT_ID or CISCO_CLIENT_SECRET in environment.");
  process.exit(1);
}

// ---------------------------------------------------------------------
// Configuration
// ---------------------------------------------------------------------
const TOKEN_URL = "https://cloudsso.cisco.com/as/token.oauth2";
const ADVISORY_URL = "https://api.cisco.com/security/advisories";
const START_DATE = new Date(
  Date.now() - 30 * 24 * 60 * 60 * 1000
).toISOString().split("T")[0]; // YYYY-MM-DD

// ---------------------------------------------------------------------
// Create an axios instance with default timeout
// ---------------------------------------------------------------------
const http: AxiosInstance = axios.create({
  timeout: 15000,
});

// ---------------------------------------------------------------------
// Helper: obtain OAuth2 token (with simple retry)
// ---------------------------------------------------------------------
async function getAccessToken(): Promise<string> {
  const payload = new URLSearchParams({
    grant_type: "client_credentials",
    client_id: CLIENT_ID,
    client_secret: CLIENT_SECRET,
  });

  try {
    const { data } = await http.post(TOKEN_URL, payload.toString(), {
      headers: { "Content-Type": "application/x-www-form-urlencoded" },
    });
    return data.access_token;
  } catch (err: any) {
    if (err.response?.status === 429) {
      const retryAfter = err.response.headers["retry-after"] ?? 5;
      await new Promise((r) => setTimeout(r, Number(retryAfter) * 1000));
      return getAccessToken(); // retry once after back‑off
    }
    throw new Error(`Token request failed: ${err.message}`);
  }
}

// ---------------------------------------------------------------------
// Helper: fetch advisories with pagination & rate‑limit handling
// ---------------------------------------------------------------------
async function fetchCriticalAdvisories(token: string) {
  const headers = { Authorization: `Bearer ${token}`, Accept: "application/json" };
  let url = ADVISORY_URL;
  let params = {
    publishedStartDate: START_DATE,
    severity: "Critical",
    pageSize: 100,
  };
  const advisories: any[] = [];

  while (url) {
    try {
      const { data } = await http.get(url, { headers, params });
      // First request uses `params`; subsequent pages use the `next` link.
      params = undefined;

      advisories.push(...(data.advisories ?? []));

      // Pagination: Cisco API returns a `next` link under metadata.links
      url = data.metadata?.links?.next ?? null;
    } catch (err: any) {
      if (err.response?.status === 429) {
        const retryAfter = err.response.headers["retry-after"] ?? 5;
        await new Promise((r) => setTimeout(r, Number(retryAfter) * 1000));
        continue; // retry same request
      }
      throw new Error(`Advisory request failed: ${err.message}`);
    }
  }
  return advisories;
}

// ---------------------------------------------------------------------
// Main
// ---------------------------------------------------------------------
(async () => {
  try {
    const token = await getAccessToken();
    const criticals = await fetchCriticalAdvisories(token);

    if (criticals.length === 0) {
      console.log("✅  No Critical Cisco advisories in the last 30 days.");
      return;
    }

    console.log(
      `🔎  Found ${criticals.length} Critical Cisco advisory(ies) (last 30 days):\n`
    );
    for (const adv of criticals) {
      const id = adv.advisoryId ?? "N/A";
      const title = adv.title ?? "N/A";
      const cvss = adv.cvss?.baseScore ?? "N/A";
      const pub = adv.publicationDate ?? "N/A";
      console.log(`- [${id}] ${title}`);
      console.log(`  CVSS: ${cvss}   Published: ${pub}`);
      console.log();
    }
  } catch (e: any) {
    console.error(`❌  Fatal error: ${e.message}`);
    process.exit(1);
  }
})();

# Compile & run (ts-node handles it)
npx ts-node fetch_cisco_critical.ts

npm run build   # if you added a "build": "tsc" script
node dist/fetch_cisco_critical.js

🔎  Found 12 Critical Cisco advisory(ies) (last 30 days):

- [cisco-sa-20240915-iosxe-rce] Multiple Vulnerabilities in Cisco IOS XE Software
  CVSS: 9.8   Published: 2024-09-15

- [cisco-sa-20240914-asa-auth-bypass] Authentication Bypass in Cisco ASA
  CVSS: 9.1   Published: 2024-09-14

...

# .env – never commit this file!
CISCO_CLIENT_ID=your-client-id-here
CISCO_CLIENT_SECRET=your-client-secret-here

export CISCO_CLIENT_ID=...
export CISCO_CLIENT_SECRET=...