1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

Scanner Scan Complete (webhook/schedule)
              │
              ▼
      Ingest findings via API
              │
              ▼
    Deduplicate & filter
    (suppress known exceptions)
              │
              ▼
    Enrich each CVE:
    - CVSS v3 base score
    - EPSS exploit probability
    - CISA KEV check (known exploited?)
    - Asset business criticality
              │
              ▼
    Calculate Priority Score
              │
       ┌──────┴───────┐
    Critical/       Medium/Low
    High            findings
       │                │
       ▼                ▼
  P1/P2 ticket     P3/P4 ticket
  assigned to      batched into
  patch team        weekly report
  SLA: 72hr        SLA: 30 days

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

Priority Score = (CVSS_Base × 0.4)
              + (EPSS_Probability × 100 × 0.3)
              + (KEV_Bonus × 0.2)
              + (Asset_Criticality × 0.1)

Where:
  CVSS_Base         = 0–10 (NVD score)
  EPSS_Probability  = 0.0–1.0 (FIRST.org daily model)
  KEV_Bonus         = 30 if in CISA KEV, else 0
  Asset_Criticality = 10 (critical), 7 (high), 4 (medium), 1 (low)

1

pip install requests tenable qualysapi python-docx jinja2

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

from tenable.sc import TenableSC

def fetch_tenable_findings(host, access_key, secret_key, days_back=7):
    sc = TenableSC(host, access_key=access_key, secret_key=secret_key)

    findings = []
    for vuln in sc.analysis.vulns(
        ('lastSeen', '=', f'0:{days_back * 86400}'),
        tool='vulndetails'
    ):
        findings.append({
            'cve':            vuln.get('cve', ''),
            'cvss_v3':        float(vuln.get('cvssV3BaseScore', 0) or 0),
            'plugin_id':      vuln['pluginID'],
            'plugin_name':    vuln['pluginName'],
            'asset_ip':       vuln['ip'],
            'asset_hostname': vuln.get('dnsName', vuln['ip']),
            'severity':       vuln['severity']['name'],
            'vuln_state':     vuln['vulnState'],
        })

    return findings

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

import requests

def enrich_epss(cve_list):
    """Fetch EPSS scores for a batch of CVEs from FIRST.org."""
    if not cve_list:
        return {}

    cve_param = ','.join(cve_list)
    url = f"https://api.first.org/data/1.0/epss?cve={cve_param}"
    response = requests.get(url)

    epss_scores = {}
    if response.status_code == 200:
        for item in response.json().get('data', []):
            epss_scores[item['cve']] = {
                'epss':       float(item['epss']),
                'percentile': float(item['percentile'])
            }
    return epss_scores

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

import requests

_KEV_CACHE = set()

def load_cisa_kev():
    """Load CISA KEV catalog — updated daily."""
    global _KEV_CACHE
    url = "https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json"
    response = requests.get(url)
    if response.status_code == 200:
        catalog = response.json()
        _KEV_CACHE = {v['cveID'] for v in catalog['vulnerabilities']}
    return _KEV_CACHE

def is_in_kev(cve_id):
    if not _KEV_CACHE:
        load_cisa_kev()
    return cve_id in _KEV_CACHE

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# Asset criticality is pulled from your CMDB or a local mapping file
ASSET_CRITICALITY = {
    'domain-controller-01': 10,
    'erp-server':           10,
    'web-server-prod':       7,
    'dev-workstation-01':    4,
}

def get_asset_criticality(hostname, ip):
    # Try hostname first, fall back to IP, default to medium
    return ASSET_CRITICALITY.get(hostname,
           ASSET_CRITICALITY.get(ip, 4))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

def calculate_priority_score(finding, epss_data, kev_set, cmdb):
    cve       = finding['cve']
    cvss      = finding['cvss_v3']
    epss      = epss_data.get(cve, {}).get('epss', 0)
    kev_bonus = 30 if cve in kev_set else 0
    asset_crit = get_asset_criticality(finding['asset_hostname'], finding['asset_ip'])

    score = (cvss * 0.4) + (epss * 100 * 0.3) + (kev_bonus * 0.2) + (asset_crit * 0.1)
    return round(score, 2)

def score_to_priority(score):
    if score >= 80: return 'P1', 'Critical',  '72 hours'
    if score >= 60: return 'P2', 'High',      '7 days'
    if score >= 40: return 'P3', 'Medium',    '30 days'
    return           'P4', 'Low',             '90 days'

def create_patch_ticket(finding, score, sn_client):
    priority, severity, sla = score_to_priority(score)
    cve = finding['cve']
    is_kev = is_in_kev(cve)

    description = f"""
**CVE:** {cve}
**CVSS v3:** {finding['cvss_v3']}
**EPSS:** {finding.get('epss', 0):.2%} probability of exploitation
**CISA KEV:** {'YES — actively exploited in the wild' if is_kev else 'No'}
**Asset:** {finding['asset_hostname']} ({finding['asset_ip']})
**Asset Criticality:** {get_asset_criticality(finding['asset_hostname'], finding['asset_ip'])}/10
**Priority Score:** {score}
**SLA:** Remediate within {sla}

**Remediation Steps:**
{get_remediation_guidance(cve)}
"""
    return sn_client.create_incident(
        priority=priority[1],
        short_desc=f"[{priority}] Patch required: {cve} on {finding['asset_hostname']}",
        description=description,
        assignment_group='Patch Management Team',
        due_date=calculate_due_date(sla)
    )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

def check_sla_compliance(sn_client):
    """
    Run daily — find overdue patch tickets and escalate.
    """
    overdue = sn_client.query_incidents(
        assignment_group='Patch Management Team',
        state='open',
        due_date_before=datetime.utcnow().isoformat()
    )

    for ticket in overdue:
        days_overdue = (datetime.utcnow() - ticket.due_date).days

        if days_overdue >= 7:
            # Escalate to manager
            sn_client.escalate(ticket.number, reason=f"Patch ticket overdue by {days_overdue} days")
            teams_notify(
                channel='#patch-management',
                message=f"⚠️ Ticket {ticket.number} is {days_overdue} days overdue. Escalated to manager."
            )
        elif days_overdue >= 1:
            teams_notify(
                channel='#patch-management',
                message=f"Reminder: Ticket {ticket.number} ({ticket.short_desc}) is overdue."
            )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

{
  "cve": "CVE-2024-3400",
  "cvss_v3": 10.0,
  "epss": 0.94,
  "in_cisa_kev": true,
  "asset_hostname": "domain-controller-01",
  "asset_criticality": 10,
  "priority_score": 99.4,
  "ticket_priority": "P1",
  "sla": "72 hours",
  "ticket_number": "INC0042891",
  "assigned_to": "Patch Management Team"
}