vulnerability-data-processing

---
name: vulnerability-data-processing
description: Process vulnerability scan results from Trivy JSON output, extract HIGH/CRITICAL severity vulnerabilities with complete metadata (CVE, CVSS scores, fix versions, references). Use this skill whenever you need to transform raw vulnerability data into structured format with proper field mapping and CVSS score prioritization.
---

# 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:
```json
{
  "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:

```python
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

```python
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:

```python
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:

```python
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

```python
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:
```python
[
    {
        "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.