Benchmark: Trivy 0.50 vs. Snyk 8 vs. Anchore 3 for Container Image Vulnerability Detection Accuracy

#!/usr/bin/env python3
"""
Automated Container Vulnerability Scanner Benchmark Runner
Compares Trivy 0.50, Snyk 8, and Anchore 3 against known CVE datasets
Version: 1.0.0
Dependencies: PyYAML, requests, pandas
"""

import subprocess
import json
import os
import time
from typing import Dict, List, Optional
import pandas as pd

# Benchmark configuration
BENCHMARK_CONFIG = {
    "image_list": "benchmark_images.txt",  # 1,200 images, one per line
    "known_cves": "nvd_2023_2024_cves.json",  # Pre-validated CVE dataset
    "output_dir": "./benchmark_results",
    "tools": {
        "trivy": {
            "binary": "/usr/local/bin/trivy",
            "version": "0.50.1",
            "cmd_template": "trivy image --format json --output {output} {image}"
        },
        "snyk": {
            "binary": "/usr/local/bin/snyk",
            "version": "8.25.0",
            "cmd_template": "snyk container test {image} --json > {output}"
        },
        "anchore": {
            "binary": "/usr/local/bin/grype",  # Anchore 3 uses Grype for scanning
            "version": "0.73.0",
            "cmd_template": "grype {image} -o json > {output}"
        }
    },
    "hardware": "AWS c7g.4xlarge (16 vCPU, 32GB RAM)"
}


def load_known_cves(cve_path: str) -> Dict[str, List[str]]:
    """Load pre-validated CVEs mapping image digest to known CVE IDs"""
    try:
        with open(cve_path, "r") as f:
            return json.load(f)
    except FileNotFoundError:
        raise RuntimeError(f"Known CVE file not found at {cve_path}")
    except json.JSONDecodeError:
        raise RuntimeError(f"Invalid JSON in CVE file {cve_path}")


def run_scan(tool: str, image: str, output_path: str) -> Optional[dict]:
    """Run vulnerability scan for a single tool and image, return parsed results"""
    config = BENCHMARK_CONFIG["tools"][tool]
    cmd = config["cmd_template"].format(output=output_path, image=image)

    try:
        start_time = time.time()
        result = subprocess.run(
            cmd,
            shell=True,
            capture_output=True,
            text=True,
            timeout=300  # 5 minute timeout per scan
        )
        scan_time = time.time() - start_time

        if result.returncode != 0:
            print(f"Scan failed for {tool} on {image}: {result.stderr}")
            return None

        # Parse output JSON
        with open(output_path, "r") as f:
            scan_data = json.load(f)

        return {
            "tool": tool,
            "image": image,
            "scan_time": scan_time,
            "raw_results": scan_data
        }
    except subprocess.TimeoutExpired:
        print(f"Scan timed out for {tool} on {image}")
        return None
    except Exception as e:
        print(f"Unexpected error scanning {image} with {tool}: {str(e)}")
        return None


def calculate_metrics(scan_results: List[dict], known_cves: Dict[str, List[str]]) -> pd.DataFrame:
    """Calculate TPR, FPR, scan time metrics for each tool"""
    metrics = []
    for tool in BENCHMARK_CONFIG["tools"].keys():
        tool_results = [r for r in scan_results if r["tool"] == tool and r is not None]
        total_cves = 0
        detected_cves = 0
        false_positives = 0

        for res in tool_results:
            image = res["image"]
            # Get known CVEs for this image (match by digest)
            image_digest = subprocess.run(
                f"docker inspect --format '{{{{.Id}}}}' {image}",
                shell=True,
                capture_output=True,
                text=True
            ).stdout.strip()
            known = known_cves.get(image_digest, [])
            total_cves += len(known)

            # Extract detected CVEs from scan results (tool-specific parsing)
            detected = []
            if tool == "trivy":
                detected = [v["VulnerabilityID"] for v in res["raw_results"]["Results"][0].get("Vulnerabilities", [])]
            elif tool == "snyk":
                detected = [v["id"] for v in res["raw_results"].get("vulnerabilities", [])]
            elif tool == "anchore":
                detected = [v["vulnerability"]["id"] for v in res["raw_results"]]

            detected_cves += len([c for c in detected if c in known])
            false_positives += len([c for c in detected if c not in known])

        tpr = (detected_cves / total_cves) * 100 if total_cves > 0 else 0
        fpr = (false_positives / (detected_cves + false_positives)) * 100 if (detected_cves + false_positives) > 0 else 0
        avg_scan_time = sum(r["scan_time"] for r in tool_results) / len(tool_results) if tool_results else 0

        metrics.append({
            "Tool": tool,
            "True Positive Rate (%)": round(tpr, 1),
            "False Positive Rate (%)": round(fpr, 1),
            "Avg Scan Time (s)": round(avg_scan_time, 1),
            "Total Scans": len(tool_results)
        })

    return pd.DataFrame(metrics)


if __name__ == "__main__":
    # Create output directory
    os.makedirs(BENCHMARK_CONFIG["output_dir"], exist_ok=True)

    # Load known CVEs
    known_cves = load_known_cves(BENCHMARK_CONFIG["known_cves"])

    # Load image list
    with open(BENCHMARK_CONFIG["image_list"], "r") as f:
        images = [line.strip() for line in f if line.strip()]

    print(f"Starting benchmark of {len(images)} images across {len(BENCHMARK_CONFIG['tools'])} tools")
    print(f"Hardware: {BENCHMARK_CONFIG['hardware']}")

    all_results = []
    for image in images:
        for tool in BENCHMARK_CONFIG["tools"].keys():
            output_path = os.path.join(BENCHMARK_CONFIG["output_dir"], f"{tool}_{image.replace('/', '_')}.json")
            result = run_scan(tool, image, output_path)
            if result:
                all_results.append(result)

    # Calculate and print metrics
    metrics_df = calculate_metrics(all_results, known_cves)
    print("\nBenchmark Results:")
    print(metrics_df.to_string(index=False))

    # Save results to CSV
    metrics_df.to_csv(os.path.join(BENCHMARK_CONFIG["output_dir"], "benchmark_metrics.csv"), index=False)

name: Container Vulnerability Scan Benchmark
on:
  push:
    branches: [ main ]
  pull_request:
    branches: [ main ]

jobs:
  benchmark-scans:
    runs-on: ubuntu-24.04
    strategy:
      matrix:
        tool: [trivy, snyk, anchore]
    env:
      IMAGE: "alpine:3.19"  # Test image with known CVEs (CVE-2024-2961)
    steps:
      - name: Checkout code
        uses: actions/checkout@v4

      - name: Set up Docker Buildx
        uses: docker/setup-buildx-action@v3

      - name: Pull test image
        run: docker pull ${{ env.IMAGE }}

      - name: Install Trivy 0.50
        if: matrix.tool == 'trivy'
        run: |
          curl -sfL https://raw.githubusercontent.com/aquasecurity/trivy/main/contrib/install.sh | sh -s -- -b /usr/local/bin v0.50.1
          trivy --version  # Verify version

      - name: Install Snyk 8
        if: matrix.tool == 'snyk'
        run: |
          npm install -g snyk@8.25.0
          snyk --version  # Verify version
          snyk auth ${{ secrets.SNYK_TOKEN }}  # Requires Snyk API token

      - name: Install Anchore Grype (Anchore 3)
        if: matrix.tool == 'anchore'
        run: |
          curl -sSfL https://raw.githubusercontent.com/anchore/grype/main/install.sh | sh -s -- -b /usr/local/bin v0.73.0
          grype --version  # Verify version

      - name: Run scan with Trivy
        if: matrix.tool == 'trivy'
        run: |
          trivy image --format json --output trivy_results.json ${{ env.IMAGE }}
          # Error handling: fail if no results
          if [ ! -f trivy_results.json ]; then
            echo "Trivy scan failed to produce output"
            exit 1
          fi
          # Count detected vulnerabilities
          TRIVY_VULNS=$(cat trivy_results.json | jq '[.Results[].Vulnerabilities[]] | length')
          echo "Trivy detected $TRIVY_VULNS vulnerabilities"

      - name: Run scan with Snyk
        if: matrix.tool == 'snyk'
        run: |
          snyk container test ${{ env.IMAGE }} --json > snyk_results.json
          # Error handling: check for Snyk errors
          if grep -q "error" snyk_results.json; then
            echo "Snyk scan failed: $(cat snyk_results.json)"
            exit 1
          fi
          # Count detected vulnerabilities
          SNYK_VULNS=$(cat snyk_results.json | jq '.vulnerabilities | length')
          echo "Snyk detected $SNYK_VULNS vulnerabilities"

      - name: Run scan with Anchore Grype
        if: matrix.tool == 'anchore'
        run: |
          grype ${{ env.IMAGE }} -o json > anchore_results.json
          # Error handling: check for empty output
          if [ ! -s anchore_results.json ]; then
            echo "Anchore scan failed to produce output"
            exit 1
          fi
          # Count detected vulnerabilities
          ANCHORE_VULNS=$(cat anchore_results.json | jq 'length')
          echo "Anchore detected $ANCHORE_VULNS vulnerabilities"

      - name: Upload scan results
        uses: actions/upload-artifact@v4
        with:
          name: ${{ matrix.tool }}-scan-results
          path: ${{ matrix.tool }}_results.json

      - name: Compare to known CVEs
        run: |
          # Known CVEs for alpine:3.19: CVE-2024-2961 (glibc), CVE-2024-2962 (zlib)
          KNOWN_CVES=("CVE-2024-2961" "CVE-2024-2962")
          if [ "${{ matrix.tool }}" == "trivy" ]; then
            DETECTED=$(cat trivy_results.json | jq -r '[.Results[].Vulnerabilities[].VulnerabilityID] | .[]')
          elif [ "${{ matrix.tool }}" == "snyk" ]; then
            DETECTED=$(cat snyk_results.json | jq -r '.vulnerabilities[].id')
          else
            DETECTED=$(cat anchore_results.json | jq -r '.[].vulnerability.id')
          fi
          # Check if known CVEs are detected
          for cve in "${KNOWN_CVES[@]}"; do
            if echo "$DETECTED" | grep -q "$cve"; then
              echo "✅ $cve detected by ${{ matrix.tool }}"
            else
              echo "❌ $cve missed by ${{ matrix.tool }}"
            fi
          done

#!/usr/bin/env python3
"""
Vulnerability Result Normalizer
Converts Trivy, Snyk, and Anchore scan results to a common schema for comparison
Version: 1.0.0
"""

import json
import os
from typing import Dict, List, Optional
from dataclasses import dataclass

@dataclass
class NormalizedVulnerability:
    """Common schema for vulnerability scan results"""
    cve_id: str
    severity: str
    package_name: str
    package_version: str
    fix_available: bool
    tool: str
    image: str


def normalize_trivy(trivy_json: dict, image: str) -> List[NormalizedVulnerability]:
    """Normalize Trivy 0.50 scan results to common schema"""
    normalized = []
    try:
        for result in trivy_json.get("Results", []):
            for vuln in result.get("Vulnerabilities", []):
                normalized.append(NormalizedVulnerability(
                    cve_id=vuln.get("VulnerabilityID"),
                    severity=vuln.get("Severity", "UNKNOWN").lower(),
                    package_name=vuln.get("PkgName"),
                    package_version=vuln.get("InstalledVersion"),
                    fix_available=vuln.get("FixedVersion") is not None,
                    tool="trivy",
                    image=image
                ))
    except Exception as e:
        print(f"Error normalizing Trivy results: {str(e)}")
    return normalized


def normalize_snyk(snyk_json: dict, image: str) -> List[NormalizedVulnerability]:
    """Normalize Snyk 8 scan results to common schema"""
    normalized = []
    try:
        for vuln in snyk_json.get("vulnerabilities", []):
            normalized.append(NormalizedVulnerability(
                cve_id=vuln.get("id"),
                severity=vuln.get("severity", "UNKNOWN").lower(),
                package_name=vuln.get("packageName"),
                package_version=vuln.get("version"),
                fix_available=vuln.get("isUpgradable", False) or vuln.get("isPatchable", False),
                tool="snyk",
                image=image
            ))
    except Exception as e:
        print(f"Error normalizing Snyk results: {str(e)}")
    return normalized


def normalize_anchore(anchore_json: List[dict], image: str) -> List[NormalizedVulnerability]:
    """Normalize Anchore 3 (Grype) scan results to common schema"""
    normalized = []
    try:
        for vuln in anchore_json:
            normalized.append(NormalizedVulnerability(
                cve_id=vuln.get("vulnerability", {}).get("id"),
                severity=vuln.get("vulnerability", {}).get("severity", "UNKNOWN").lower(),
                package_name=vuln.get("artifact", {}).get("name"),
                package_version=vuln.get("artifact", {}).get("version"),
                fix_available=vuln.get("vulnerability", {}).get("fixedInVersion") is not None,
                tool="anchore",
                image=image
            ))
    except Exception as e:
        print(f"Error normalizing Anchore results: {str(e)}")
    return normalized


def load_and_normalize(tool: str, file_path: str, image: str) -> List[NormalizedVulnerability]:
    """Load scan results from file and normalize to common schema"""
    if not os.path.exists(file_path):
        print(f"File not found: {file_path}")
        return []

    try:
        with open(file_path, "r") as f:
            raw_data = json.load(f)
    except json.JSONDecodeError:
        print(f"Invalid JSON in {file_path}")
        return []

    if tool == "trivy":
        return normalize_trivy(raw_data, image)
    elif tool == "snyk":
        return normalize_snyk(raw_data, image)
    elif tool == "anchore":
        return normalize_anchore(raw_data, image)
    else:
        print(f"Unknown tool: {tool}")
        return []


def compare_results(normalized_results: List[NormalizedVulnerability], known_cves: List[str]) -> Dict:
    """Compare normalized results to known CVEs and return metrics"""
    tool_results = {}
    for res in normalized_results:
        if res.tool not in tool_results:
            tool_results[res.tool] = {
                "detected_cves": set(),
                "false_positives": set(),
                "total": 0
            }
        tool_results[res.tool]["total"] += 1
        if res.cve_id in known_cves:
            tool_results[res.tool]["detected_cves"].add(res.cve_id)
        else:
            tool_results[res.tool]["false_positives"].add(res.cve_id)

    comparison = {}
    for tool, data in tool_results.items():
        tpr = (len(data["detected_cves"]) / len(known_cves)) * 100 if known_cves else 0
        fpr = (len(data["false_positives"]) / data["total"]) * 100 if data["total"] > 0 else 0
        comparison[tool] = {
            "true_positive_rate": round(tpr, 1),
            "false_positive_rate": round(fpr, 1),
            "total_detected": data["total"],
            "known_cves_found": len(data["detected_cves"])
        }
    return comparison


if __name__ == "__main__":
    # Example usage: Compare results for alpine:3.19
    KNOWN_CVES = ["CVE-2024-2961", "CVE-2024-2962"]
    IMAGE = "alpine:3.19"

    # Load and normalize results from benchmark
    trivy_results = load_and_normalize("trivy", "./benchmark_results/trivy_alpine_3.19.json", IMAGE)
    snyk_results = load_and_normalize("snyk", "./benchmark_results/snyk_alpine_3.19.json", IMAGE)
    anchore_results = load_and_normalize("anchore", "./benchmark_results/anchore_alpine_3.19.json", IMAGE)

    all_results = trivy_results + snyk_results + anchore_results
    comparison = compare_results(all_results, KNOWN_CVES)

    print("Normalized Result Comparison (alpine:3.19):")
    for tool, metrics in comparison.items():
        print(f"\n{tool}:")
        print(f"  True Positive Rate: {metrics['true_positive_rate']}%")
        print(f"  False Positive Rate: {metrics['false_positive_rate']}%")
        print(f"  Total Detected: {metrics['total_detected']}")
        print(f"  Known CVEs Found: {metrics['known_cves_found']}/{len(KNOWN_CVES)}")

# Ignore glibc CVEs for Alpine images
CVE-2024-2961
# Ignore low-severity zlib vulnerabilities with no fix
CVE-2024-2962:LOW

snyk container test $IMAGE --json --severity-threshold=high | jq '.vulnerabilities[] | select(.priorityScore >= 70)'

package anchore.policy.vulnerabilities

import future.keywords.if
import future.keywords.in

deny[msg] if {
    vuln := input.vulnerabilities[_]
    vuln.severity == "high" or vuln.severity == "critical"
    msg := sprintf("High/critical vulnerability %s found in %s", [vuln.id, vuln.package_name])
}