vulnerability-data-processing

Vulnerability Data Processing and Extraction

Overview

This skill handles transforming raw Trivy JSON vulnerability scan results into structured data for reporting. It handles field extraction, CVSS score selection, and data validation.

Input Format

Trivy JSON output with structure:

{
  "Results": [
    {
      "Target": "package-lock.json",
      "Type": "npm",
      "Vulnerabilities": [
        {
          "VulnerabilityID": "CVE-2021-12345",
          "PkgName": "package-name",
          "InstalledVersion": "1.0.0",
          "FixedVersion": "1.0.1",
          "Severity": "HIGH",
          "Title": "Vulnerability description",
          "Description": "Detailed description",
          "CVSS": {
            "nvd": {
              "V3Vector": "CVSS:3.1/AV:N/AC:L/...",
              "V3Score": 7.5
            },
            "ghsa": {
              "V3Score": 7.4
            }
          },
          "References": ["https://..."]
        }
      ]
    }
  ]
}

Processing Steps

1. Parse Trivy JSON Output

Read the JSON file and validate structure:

• Check that Results array exists
• Filter to Results[].Vulnerabilities[] entries
• Skip if Vulnerabilities is empty or null

2. Filter by Severity

Keep only Severity == "HIGH" or Severity == "CRITICAL".

3. Extract CVSS Score

CVSS scores are nested under the CVSS object with multiple sources. Extract in priority:

def get_cvss_score(vuln):
    cvss = vuln.get("CVSS", {})

    # Priority: NVD > GHSA > RedHat
    if "nvd" in cvss and "V3Score" in cvss["nvd"]:
        return cvss["nvd"]["V3Score"]
    if "ghsa" in cvss and "V3Score" in cvss["ghsa"]:
        return cvss["ghsa"]["V3Score"]
    if "redhat" in cvss and "V3Score" in cvss["redhat"]:
        return cvss["redhat"]["V3Score"]

    return None  # Mark as N/A if missing

4. Extract Fixed Version

fixed = vuln.get("FixedVersion", "")
fixed_version = fixed if fixed else "N/A"

5. Extract Reference URL

Take the first valid HTTP/HTTPS URL from the References array:

references = vuln.get("References", [])
url = next((ref for ref in references if ref.startswith("http")), "N/A")

6. Build Output Record

Map each vulnerability to:

record = {
    "Package": vuln["PkgName"],
    "Version": vuln["InstalledVersion"],
    "CVE_ID": vuln["VulnerabilityID"],
    "Severity": vuln["Severity"],
    "CVSS_Score": cvss_score if cvss_score else "N/A",
    "Fixed_Version": fixed_version,
    "Title": vuln.get("Title", vuln.get("Description", "N/A")),
    "Url": url
}

7. Deduplication

If the same CVE appears multiple times (different packages), keep all entries. If the same package+CVE appears multiple times, keep first occurrence.

Validation Checks

• CVE_ID format: Should match CVE-\d{4}-\d{4,} pattern
• Severity: Must be "HIGH" or "CRITICAL"
• CVSS_Score: Should be numeric 0-10 or "N/A"
• Version fields: Non-empty strings
• Title length: Use first 200 chars if exceeds limit

Python Implementation Pattern

import json

def process_vulnerabilities(json_file):
    with open(json_file, 'r') as f:
        data = json.load(f)

    records = []
    for result in data.get("Results", []):
        for vuln in result.get("Vulnerabilities", []):
            if vuln["Severity"] not in ["HIGH", "CRITICAL"]:
                continue

            # Extract fields (use helper functions above)
            record = {
                "Package": vuln["PkgName"],
                "Version": vuln["InstalledVersion"],
                "CVE_ID": vuln["VulnerabilityID"],
                "Severity": vuln["Severity"],
                "CVSS_Score": get_cvss_score(vuln),
                "Fixed_Version": vuln.get("FixedVersion") or "N/A",
                "Title": vuln.get("Title", ""),
                "Url": get_first_url(vuln.get("References", []))
            }
            records.append(record)

    return records

Common Field Transformations

Source Field	Target Field	Transformation
VulnerabilityID	CVE_ID	Use as-is
PkgName	Package	Use as-is
InstalledVersion	Version	Use as-is
FixedVersion	Fixed_Version	Empty → "N/A"
Severity	Severity	Use as-is (pre-filtered)
CVSS.*.V3Score	CVSS_Score	Select first available, fallback "N/A"
Title/Description	Title	Prefer Title, fallback to Description
References[0]	Url	Extract first HTTP URL, fallback "N/A"

Error Handling

• Invalid JSON: Log error and exit — file must be valid JSON
• Missing fields: Use "N/A" as fallback
• Empty results: Return empty list (no vulnerabilities found)
• Malformed severity: Log warning and skip record

Output

Returns list of dictionaries ready for CSV export:

[
    {
        "Package": "express",
        "Version": "4.17.1",
        "CVE_ID": "CVE-2022-12345",
        "Severity": "HIGH",
        "CVSS_Score": 7.5,
        "Fixed_Version": "4.17.2",
        "Title": "Vulnerability in express",
        "Url": "https://nvd.nist.gov/vuln/detail/CVE-2022-12345"
    },
    ...
]

Next Steps

Pass the processed records to the CSV export skill for final report generation.