

In the world of cybersecurity, "Patch Tuesday" isn't just a routine update cycle; it is a high-stakes race against time. When Microsoft releases critical zero-day vulnerability details, the window between disclosure and active exploitation shrinks to minutes.
For security engineers and automated DevOps pipelines, the goal is Automated Vulnerability Triage. We need to programmatically ingest security advisories and cross-reference them against our current software inventory to prioritize patching.
This guide provides a production-ready implementation for an Automated Vulnerability Intelligence Engine.
Before implementing the triage engine, ensure you have the following:
.env files or AWS Secrets Manager).curl or Postman for testing API endpoints.# Create a virtual environment
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
# Install essential libraries
pip install requests python-dotenv pydantic
# Initialize project
npm init -y
# Install dependencies
npm install axios dotenv typescript ts-node @types/node
# Initialize TypeScript
npx tsc --init
We will build a Vulnerability Triage Engine that fetches recent critical advisories and compares them against a local "Inventory" to flag high-priority patches.
This implementation uses pydantic for strict data validation, ensuring that malformed API responses don't crash our triage pipeline.
import os
import requests
from typing import List, Optional
from pydantic import BaseModel, Field
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# --- Models ---
class Vulnerability(BaseModel):
"""Schema for a security advisory."""
cve_id: str = Field(..., alias="id")
description: str
severity: str
cvss_score: float
affected_component: str
class TriageEngine:
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.security-provider.io/v1/advisories" # Mock URL
def fetch_critical_advisories(self) -> List[Vulnerability]:
"""Fetches latest critical zero-day advisories."""
try:
# In a real scenario, this would be an authenticated GET request
# response = requests.get(self.base_url, headers={"Authorization": f"Bearer {self.api_key}"})
# response.raise_for_status()
# MOCK DATA for demonstration
mock_data = [
{"id": "CVE-2023-1234", "description": "Remote Code Execution in Windows Kernel", "severity": "CRITICAL", "cvss_score": 9.8, "affected_component": "Windows_Kernel"},
{"id": "CVE-2023-5678", "description": "Privilege Escalation in MS Office", "severity": "HIGH", "cvss_score": 7.5, "affected_component": "MS_Office"}
]
return [Vulnerability(**item) for item in mock_data]
except Exception as e:
print(f"[ERROR] Failed to fetch advisories: {e}")
return []
def triage(self, inventory: List[str], advisories: List[Vulnerability]):
"""Cross-references advisories against the current system inventory."""
alerts = []
for adv in advisories:
if adv.affected_component in inventory:
alerts.append(adv)
return alerts
# --- Execution Logic ---
if __name__ == "__main__":
# Mock system inventory (What we have installed)
SYSTEM_INVENTORY = ["Windows_Kernel", "Linux_Kernel", "Google_Chrome"]
engine = TriageEngine(api_key=os.getenv("SEC_API_KEY", "default_key"))
print("--- Starting Patch Tuesday Triage ---")
advisories = engine.fetch_critical_advisories()
critical_hits = engine.triage(SYSTEM_INVENTORY, advisories)
if not critical_hits:
print("✅ No critical matches found. Systems are secure.")
else:
print(f"🚨 ALERT: {len(critical_hits)} critical vulnerabilities detected!")
for hit in critical_hits:
print(f"[{hit.cve_id}] Severity: {hit.severity} | Component: {hit.affected_component}")
print(f"Description: {hit.description}\n")
Using axios for robust HTTP requests and strict typing for the triage logic.
import axios from 'axios';
import * as dotenv from 'dotenv';
dotenv.config();
// --- Interfaces ---
interface Vulnerability {
id: string;
description: string;
severity: 'LOW' | 'MEDIUM' | 'HIGH' | 'CRITICAL';
cvss_score: number;
affected_component: string;
}
// --- Engine ---
class TriageService {
private apiKey: string;
constructor() {
this.apiKey = process.env.SEC_API_KEY || '';
if (!this.apiKey) throw new Error("MISSING_API_KEY: SEC_API_KEY must be set.");
}
/**
* Simulates fetching data from a vulnerability feed
*/
async getAdvisories(): Promise<Vulnerability[]> {
try {
// In production: await axios.get(URL, { headers: { Authorization: `Bearer ${this.apiKey}` } });
return [
{ id: 'CVE-2023-9999', description: 'Zero-day in Windows Print Spooler', severity: 'CRITICAL', cvss_score: 10.0, affected_component: 'Windows_Print_Spooler' },
{ id: 'CVE-2023-8888', description: 'Buffer overflow in Chrome', severity: 'HIGH', cvss_score: 8.1, affected_component: 'Google_Chrome' }
];
} catch (error) {
console.error('Failed to fetch advisories:', error);
return [];
}
}
/**
* Identifies which advisories match our internal inventory
*/
async runTriage(inventory: string[]): Promise<Vulnerability[]> {
const advisories = await this.getAdvisories();
return advisories.filter(adv => inventory.includes(adv.affected_component));
}
}
// --- Execution ---
const run = async () => {
const service = new TriageService();
const myInventory = ['Google_Chrome', 'Windows_Print_Spooler'];
console.log('🔍 Scanning for Patch Tuesday vulnerabilities...');
const matches = await service.runTriage(myInventory);
if (matches.length > 0) {
console.log(`⚠️ FOUND ${matches.length} MATCHES:`);
matches.forEach(m => {
console.log(`- ${m.id}: ${m.description} (Score: ${m.cvss_score})`);
});
} else {
console.log('✅ No immediate threats detected.');
}
};
run().catch(console.error);
Never hardcode API keys. Use a .env file at the root of your project.
.env
# Security Intelligence API Key
SEC_API_KEY=sk_live_51NzX...your_secret_key...
# Environment Mode
APP_ENV=production
# Alerting Threshold
MIN_CVSS_SCORE=7.0
In production, you want to log every time a triage runs to maintain an audit trail for compliance (SOC2/ISO27001).
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("TriageEngine")
def log_triage_event(func):
def wrapper(*args, **kwargs):
logger.info(f"Executing triage: {func.__name__}")
result = func(*args, **kwargs)
logger.info(f"Triage completed. Found {len(result)} matches.")
return result
return wrapper
# Apply to the method
# @log_triage_event
# def triage(self,...):
| Error | Cause | Solution |
|---|---|---|
401 Unauthorized | Invalid or expired API Key | Check .env file and ensure the key has "Read" permissions. |
ValidationError (Python) | API returned unexpected format | Check if the provider changed their JSON schema. Update the Pydantic model. |
TimeoutError | Network latency or API downtime | Implement a Retry Strategy with exponential backoff. |
ModuleNotFoundError | Missing dependency | Run pip install -r requirements.txt or npm install. |
Before deploying this into a live CI/CD pipeline:
SYSTEM_INVENTORY is updated dynamically via your cloud provider (AWS/Azure) rather than being a static list.Source: Dark Reading
Follow ICARAX for more AI insights and tutorials.
