soc operations

Building Vulnerability Scanning Workflow

Builds a structured vulnerability scanning workflow using tools like Nessus, Qualys, and OpenVAS to discover, prioritize, and track remediation of security vulnerabilities across infrastructure. Use when SOC teams need to establish recurring vulnerability assessment processes, integrate scan results with SIEM alerting, and build remediation tracking dashboards.

cvssnessusopenvaspatch-managementqualysremediationsocvulnerability-scanning
Install this skill
npx skills add mukul975/Anthropic-Cybersecurity-Skills
Framework mappings

When to Use

Use this skill when:

  • SOC teams need to establish or improve recurring vulnerability scanning programs
  • Scan results require prioritization beyond raw CVSS scores using asset context and threat intelligence
  • Vulnerability data must be integrated into SIEM for correlation with exploitation attempts
  • Remediation tracking needs formalization with SLA-based dashboards and reporting

Do not use for penetration testing or active exploitation — vulnerability scanning identifies weaknesses, penetration testing validates exploitability.

Prerequisites

  • Vulnerability scanner (Tenable Nessus Professional, Qualys VMDR, or OpenVAS/Greenbone)
  • Asset inventory with criticality classifications (business-critical, standard, development)
  • Network access from scanner to all target segments (agent-based or network scan)
  • SIEM integration for scan result ingestion and correlation
  • Patch management system (WSUS, SCCM, Intune) for remediation tracking

Workflow

Step 1: Define Scan Scope and Scheduling

Create scan policies covering all asset types:

Nessus Scan Configuration (API):

import requests
 
nessus_url = "https://nessus.company.com:8834"
headers = {"X-ApiKeys": f"accessKey={access_key};secretKey={secret_key}"}
 
# Create scan policy
policy = {
    "uuid": "advanced",
    "settings": {
        "name": "SOC Weekly Infrastructure Scan",
        "description": "Weekly credentialed scan of all server and workstation segments",
        "scanner_id": 1,
        "policy_id": 0,
        "text_targets": "10.0.0.0/16, 172.16.0.0/12",
        "launch": "WEEKLY",
        "starttime": "20240315T020000",
        "rrules": "FREQ=WEEKLY;INTERVAL=1;BYDAY=SA",
        "enabled": True
    },
    "credentials": {
        "add": {
            "Host": {
                "Windows": [{
                    "domain": "company.local",
                    "username": "nessus_svc",
                    "password": "SCAN_SERVICE_PASSWORD",
                    "auth_method": "Password"
                }],
                "SSH": [{
                    "username": "nessus_svc",
                    "private_key": "/path/to/nessus_key",
                    "auth_method": "public key"
                }]
            }
        }
    }
}
 
response = requests.post(f"{nessus_url}/scans", headers=headers, json=policy,
                         verify=not os.environ.get("SKIP_TLS_VERIFY", "").lower() == "true")  # Set SKIP_TLS_VERIFY=true for self-signed certs in lab environments
scan_id = response.json()["scan"]["id"]
print(f"Scan created: ID {scan_id}")

Qualys VMDR Scan via API:

import qualysapi
 
conn = qualysapi.connect(
    hostname="qualysapi.qualys.com",
    username="api_user",
    password="API_PASSWORD"
)
 
# Launch vulnerability scan
params = {
    "action": "launch",
    "scan_title": "Weekly_Infrastructure_Scan",
    "ip": "10.0.0.0/16",
    "option_id": "123456",  # Scan profile ID
    "iscanner_name": "Internal_Scanner_01",
    "priority": "0"
}
 
response = conn.request("/api/2.0/fo/scan/", params)
print(f"Scan launched: {response}")

Step 2: Process and Prioritize Scan Results

Download results and apply risk-based prioritization:

import requests
import csv
 
# Export Nessus results
response = requests.get(
    f"{nessus_url}/scans/{scan_id}/export",
    headers=headers,
    params={"format": "csv"},
    verify=not os.environ.get("SKIP_TLS_VERIFY", "").lower() == "true",  # Set SKIP_TLS_VERIFY=true for self-signed certs in lab environments
)
 
# Parse and prioritize
vulns = []
reader = csv.DictReader(response.text.splitlines())
for row in reader:
    cvss = float(row.get("CVSS v3.0 Base Score", 0))
    asset_criticality = get_asset_criticality(row["Host"])  # From asset inventory
 
    # Risk-based priority calculation
    risk_score = cvss * asset_criticality_multiplier(asset_criticality)
 
    # Boost score if actively exploited (check CISA KEV)
    if row.get("CVE") in cisa_kev_list:
        risk_score *= 1.5
 
    vulns.append({
        "host": row["Host"],
        "plugin_name": row["Name"],
        "severity": row["Risk"],
        "cvss": cvss,
        "cve": row.get("CVE", "N/A"),
        "risk_score": round(risk_score, 1),
        "asset_criticality": asset_criticality,
        "kev": row.get("CVE") in cisa_kev_list
    })
 
# Sort by risk score
vulns.sort(key=lambda x: x["risk_score"], reverse=True)

CISA KEV (Known Exploited Vulnerabilities) Check:

import requests
 
kev_response = requests.get(
    "https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json"
)
kev_data = kev_response.json()
cisa_kev_list = {v["cveID"] for v in kev_data["vulnerabilities"]}
 
# Check if vulnerability is actively exploited
def is_actively_exploited(cve_id):
    return cve_id in cisa_kev_list

Step 3: Define Remediation SLAs

Apply SLA-based remediation timelines:

Priority CVSS Range Asset Type SLA Examples
P1 Critical 9.0-10.0 + KEV All assets 24 hours Log4Shell, EternalBlue on prod servers
P2 High 7.0-8.9 or 9.0+ non-KEV Business-critical 7 days RCE without known exploit
P3 Medium 4.0-6.9 Business-critical 30 days Authenticated privilege escalation
P4 Low 0.1-3.9 Standard 90 days Information disclosure, low-impact DoS
P5 Informational 0.0 Development Next cycle Best practice findings, config hardening

Step 4: Integrate with SIEM for Exploitation Detection

Correlate vulnerability scan data with SIEM alerts to detect active exploitation:

index=vulnerability sourcetype="nessus:scan"
| eval vuln_key = Host.":".CVE
| join vuln_key type=left [
    search index=ids_ips sourcetype="snort" OR sourcetype="suricata"
    | eval vuln_key = dest_ip.":".cve_id
    | stats count AS exploit_attempts, latest(_time) AS last_exploit_attempt by vuln_key
  ]
| where isnotnull(exploit_attempts)
| eval risk = "CRITICAL — Vulnerability being actively exploited"
| sort - exploit_attempts
| table Host, CVE, plugin_name, cvss_score, exploit_attempts, last_exploit_attempt, risk

Alert when KEV vulnerabilities are detected on critical assets:

index=vulnerability sourcetype="nessus:scan" severity="Critical"
| lookup cisa_kev_lookup.csv cve_id AS CVE OUTPUT kev_status, due_date
| where kev_status="active"
| lookup asset_criticality_lookup.csv ip AS Host OUTPUT criticality
| where criticality IN ("business-critical", "mission-critical")
| table Host, CVE, plugin_name, cvss_score, kev_status, due_date, criticality

Step 5: Build Remediation Tracking Dashboard

Splunk Dashboard for Vulnerability Metrics:

-- Open vulnerabilities by severity
index=vulnerability sourcetype="nessus:scan" status="open"
| stats count by severity
| eval order = case(severity="Critical", 1, severity="High", 2, severity="Medium", 3,
                    severity="Low", 4, 1=1, 5)
| sort order
 
-- SLA compliance tracking
index=vulnerability sourcetype="nessus:scan" status="open"
| eval sla_days = case(
    severity="Critical", 1,
    severity="High", 7,
    severity="Medium", 30,
    severity="Low", 90
  )
| eval days_open = round((now() - first_detected) / 86400)
| eval sla_status = if(days_open > sla_days, "OVERDUE", "Within SLA")
| stats count by severity, sla_status
 
-- Remediation trend over 90 days
index=vulnerability sourcetype="nessus:scan"
| eval is_open = if(status="open", 1, 0)
| eval is_closed = if(status="fixed", 1, 0)
| timechart span=1w sum(is_open) AS opened, sum(is_closed) AS remediated

Step 6: Automate Remediation Ticketing

Create tickets automatically for high-priority findings:

import requests
 
servicenow_url = "https://company.service-now.com/api/now/table/incident"
headers = {
    "Content-Type": "application/json",
    "Authorization": f"Bearer {snow_token}"
}
 
for vuln in vulns:
    if vuln["risk_score"] >= 8.0:
        ticket = {
            "short_description": f"[VULN] {vuln['cve']}{vuln['plugin_name']} on {vuln['host']}",
            "description": (
                f"Vulnerability: {vuln['plugin_name']}\n"
                f"CVE: {vuln['cve']}\n"
                f"CVSS: {vuln['cvss']}\n"
                f"Host: {vuln['host']}\n"
                f"Asset Criticality: {vuln['asset_criticality']}\n"
                f"CISA KEV: {'YES' if vuln['kev'] else 'NO'}\n"
                f"Risk Score: {vuln['risk_score']}\n"
                f"Remediation SLA: {'24 hours' if vuln['kev'] else '7 days'}"
            ),
            "urgency": "1" if vuln["kev"] else "2",
            "impact": "1" if vuln["asset_criticality"] == "business-critical" else "2",
            "assignment_group": "IT Infrastructure",
            "category": "Vulnerability"
        }
        response = requests.post(servicenow_url, headers=headers, json=ticket)
        print(f"Ticket created: {response.json()['result']['number']}")

Key Concepts

Term Definition
CVSS Common Vulnerability Scoring System — standardized severity rating (0-10) for vulnerabilities
CISA KEV Known Exploited Vulnerabilities catalog — CISA-maintained list of vulnerabilities with confirmed active exploitation
Credentialed Scan Vulnerability scan using authenticated access for deeper detection than network-only scanning
Asset Criticality Business impact classification determining remediation priority (mission-critical, business-critical, standard)
Remediation SLA Service Level Agreement defining maximum time allowed to patch vulnerabilities by severity
EPSS Exploit Prediction Scoring System — ML-based probability score predicting likelihood of exploitation

Tools & Systems

  • Tenable Nessus / Tenable.io: Enterprise vulnerability scanner with 200,000+ plugin checks and compliance auditing
  • Qualys VMDR: Cloud-based vulnerability management with asset discovery, prioritization, and patching integration
  • OpenVAS (Greenbone): Open-source vulnerability scanner with community-maintained vulnerability feed
  • CISA KEV Catalog: US government maintained list of actively exploited vulnerabilities requiring mandatory remediation
  • Rapid7 InsightVM: Vulnerability management platform with live dashboards and remediation project tracking

Common Scenarios

  • Zero-Day Response: New CVE published — run targeted scan for affected software, cross-reference with KEV and exploit databases
  • Compliance Audit Prep: Generate PCI DSS or HIPAA vulnerability report showing scan coverage and remediation status
  • Post-Patch Verification: Rescan patched systems to confirm vulnerability closure and update tracking dashboard
  • Network Expansion: New subnet added to infrastructure — onboard to scan scope with appropriate policy
  • Third-Party Risk: Scan externally-facing assets to validate vendor patch compliance before integration

Output Format

VULNERABILITY SCAN REPORT — Weekly Summary
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Scan Date:    2024-03-16 02:00 UTC
Scan Scope:   10.0.0.0/16 (1,247 hosts scanned)
Duration:     4h 23m
Coverage:     98.7% (16 hosts unreachable)
 
Findings:
  Severity     Count    New    CISA KEV
  Critical     23       5      3
  High         187      34     12
  Medium       892      78     0
  Low          1,456    112    0
  Info         3,891    201    0
 
Top Priority (P1 — 24hr SLA):
  CVE-2024-21762  FortiOS RCE           3 hosts   KEV: YES
  CVE-2024-1709   ConnectWise RCE       1 host    KEV: YES
  CVE-2024-3400   Palo Alto PAN-OS RCE  2 hosts   KEV: YES
 
SLA Compliance:
  Critical: 82% within SLA (4 overdue)
  High:     91% within SLA (17 overdue)
  Medium:   88% within SLA (107 overdue)
 
Tickets Created: 39 (ServiceNow)
Source materials

References and resources

Everything below is rendered for inspection. Script files are read-only and never run.

References 1

api-reference.md1.7 KB

API Reference: Vulnerability Scanning Workflow Agent

Overview

Orchestrates vulnerability scanning using Nmap and Nessus, enriches findings with CISA KEV data, applies risk-based prioritization, and generates remediation reports.

Dependencies

Package Version Purpose
python-nmap >=0.7.1 Nmap scan orchestration
requests >=2.28 REST API communication

CLI Usage

python agent.py --targets 192.168.1.0/24 --ports 1-1024 --output report.json
python agent.py --targets 10.0.0.0/16 --nessus-url https://nessus:8834 --nessus-keys "access;secret"

Key Functions

run_nmap_vuln_scan(targets, ports)

Runs Nmap with -sV --script=vulners,vulscan for service version detection and vulnerability matching.

fetch_cisa_kev()

Downloads the CISA Known Exploited Vulnerabilities JSON catalog and returns a set of CVE IDs.

launch_nessus_scan(nessus_url, api_keys, scan_name, targets)

Creates and launches a vulnerability scan via the Nessus REST API.

prioritize_vulnerabilities(vulns, kev_set, asset_criticality_map)

Applies risk scoring: risk_score = CVSS * asset_criticality * (1.5 if KEV). Assigns P1-P4 priority.

create_servicenow_ticket(snow_url, token, vuln)

Creates ServiceNow incident tickets for high-priority vulnerability findings.

generate_report(vulns)

Produces a formatted vulnerability scan summary with priority breakdown.

External APIs Used

API Endpoint Purpose
CISA KEV https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json Known exploited vulns
Nessus /scans, /scans/{id}/launch Scan management
ServiceNow /api/now/table/incident Ticket creation

Scripts 1

agent.py6.1 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""Vulnerability Scanning Workflow Agent - Automates scan orchestration and prioritization."""

import json
import logging
import os
import argparse
from datetime import datetime

import requests
import nmap

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)


def run_nmap_vuln_scan(targets, ports="1-1024"):
    """Run Nmap vulnerability scan against target hosts."""
    scanner = nmap.PortScanner()
    logger.info("Starting Nmap vuln scan on %s ports %s", targets, ports)
    scanner.scan(hosts=targets, ports=ports, arguments="-sV --script=vulners,vulscan")
    results = []
    for host in scanner.all_hosts():
        for proto in scanner[host].all_protocols():
            for port in scanner[host][proto]:
                service = scanner[host][proto][port]
                results.append({
                    "host": host,
                    "port": port,
                    "protocol": proto,
                    "state": service["state"],
                    "service": service.get("name", ""),
                    "version": service.get("version", ""),
                    "scripts": service.get("script", {}),
                })
    logger.info("Nmap scan complete: %d service entries", len(results))
    return results


def fetch_cisa_kev():
    """Fetch the CISA Known Exploited Vulnerabilities catalog."""
    url = "https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json"
    resp = requests.get(url, timeout=30)
    resp.raise_for_status()
    kev_data = resp.json()
    kev_set = {v["cveID"] for v in kev_data["vulnerabilities"]}
    logger.info("Loaded %d CISA KEV entries", len(kev_set))
    return kev_set


def launch_nessus_scan(nessus_url, api_keys, scan_name, targets):
    """Launch a vulnerability scan via Nessus REST API."""
    headers = {"X-ApiKeys": api_keys, "Content-Type": "application/json"}
    scan_config = {
        "uuid": "advanced",
        "settings": {
            "name": scan_name,
            "text_targets": targets,
            "launch": "ON_DEMAND",
            "enabled": True,
        },
    }
    resp = requests.post(
        f"{nessus_url}/scans", headers=headers, json=scan_config,
        verify=not os.environ.get("SKIP_TLS_VERIFY", "").lower() == "true", timeout=30  # Set SKIP_TLS_VERIFY=true for self-signed certs in lab environments
    )
    resp.raise_for_status()
    scan_id = resp.json()["scan"]["id"]
    logger.info("Nessus scan created: ID %d", scan_id)

    requests.post(
        f"{nessus_url}/scans/{scan_id}/launch", headers=headers,
        verify=not os.environ.get("SKIP_TLS_VERIFY", "").lower() == "true", timeout=30  # Set SKIP_TLS_VERIFY=true for self-signed certs in lab environments
    )
    logger.info("Nessus scan %d launched", scan_id)
    return scan_id


def prioritize_vulnerabilities(vulns, kev_set, asset_criticality_map):
    """Apply risk-based prioritization to vulnerability findings."""
    for vuln in vulns:
        cvss = vuln.get("cvss", 0.0)
        cve = vuln.get("cve", "")
        host = vuln.get("host", "")
        criticality = asset_criticality_map.get(host, 1.0)

        risk_score = cvss * criticality
        if cve in kev_set:
            risk_score *= 1.5
            vuln["kev"] = True
        else:
            vuln["kev"] = False

        vuln["risk_score"] = round(risk_score, 1)
        vuln["priority"] = (
            "P1" if risk_score >= 13.5 else
            "P2" if risk_score >= 7.0 else
            "P3" if risk_score >= 4.0 else
            "P4"
        )
    vulns.sort(key=lambda x: x["risk_score"], reverse=True)
    return vulns


def create_servicenow_ticket(snow_url, token, vuln):
    """Create a ServiceNow incident ticket for a high-priority vulnerability."""
    headers = {"Content-Type": "application/json", "Authorization": f"Bearer {token}"}
    ticket = {
        "short_description": f"[VULN] {vuln.get('cve', 'N/A')} on {vuln['host']}",
        "description": (
            f"CVE: {vuln.get('cve', 'N/A')}\n"
            f"CVSS: {vuln.get('cvss', 0)}\n"
            f"Host: {vuln['host']}\n"
            f"Risk Score: {vuln['risk_score']}\n"
            f"CISA KEV: {'YES' if vuln.get('kev') else 'NO'}"
        ),
        "urgency": "1" if vuln.get("kev") else "2",
        "category": "Vulnerability",
    }
    resp = requests.post(
        f"{snow_url}/api/now/table/incident", headers=headers, json=ticket, timeout=30
    )
    logger.info("ServiceNow ticket created: %s", resp.json().get("result", {}).get("number"))
    return resp.json()


def generate_report(vulns):
    """Generate vulnerability scan summary report."""
    p1 = [v for v in vulns if v.get("priority") == "P1"]
    p2 = [v for v in vulns if v.get("priority") == "P2"]
    lines = [
        "VULNERABILITY SCAN REPORT",
        "=" * 40,
        f"Date: {datetime.utcnow().strftime('%Y-%m-%d %H:%M UTC')}",
        f"Total Findings: {len(vulns)}",
        f"P1 Critical: {len(p1)}",
        f"P2 High: {len(p2)}",
        "",
        "TOP PRIORITY FINDINGS:",
    ]
    for v in p1[:10]:
        lines.append(f"  {v.get('cve', 'N/A'):20s} {v['host']:15s} Score: {v['risk_score']}")
    print("\n".join(lines))
    return lines


def main():
    parser = argparse.ArgumentParser(description="Vulnerability Scanning Workflow Agent")
    parser.add_argument("--targets", required=True, help="Target hosts/CIDR")
    parser.add_argument("--ports", default="1-1024", help="Port range to scan")
    parser.add_argument("--nessus-url", help="Nessus API URL")
    parser.add_argument("--nessus-keys", help="Nessus API keys (accessKey;secretKey)")
    parser.add_argument("--output", default="vuln_report.json")
    args = parser.parse_args()

    kev_set = fetch_cisa_kev()
    results = run_nmap_vuln_scan(args.targets, args.ports)

    if args.nessus_url and args.nessus_keys:
        launch_nessus_scan(args.nessus_url, args.nessus_keys, "Agent Scan", args.targets)

    prioritized = prioritize_vulnerabilities(results, kev_set, {})
    generate_report(prioritized)

    with open(args.output, "w") as f:
        json.dump(prioritized, f, indent=2)
    logger.info("Report saved to %s", args.output)


if __name__ == "__main__":
    main()
Keep exploring