Postmortem: When Our Trivy 0.50 Scan and GitHub Actions 2.0 Pipeline Failed to Catch a Critical CVE

# trivy_scan_validator.py
# Validates Trivy scan results against NVD CVE feeds to detect false negatives
# Reproduces the CVE-2024-3094 false negative in Trivy 0.50

import subprocess
import json
import os
import sys
import logging
from datetime import datetime, timedelta
import requests

# Configure logging for audit trails
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)]
)

# Constants matching our production pipeline config
TRIVY_VERSION = "0.50.0"
NVD_API_URL = "https://services.nvd.nist.gov/rest/json/cves/2.0"
CVE_TARGET = "CVE-2024-3094"
LIBXML2_MIN_PATCHED = "2.11.6"
SCAN_TIMEOUT = 300  # 5 minutes per scan

def check_trivy_version():
    """Verify Trivy is installed and matches the target version"""
    try:
        result = subprocess.run(
            ["trivy", "--version"],
            capture_output=True,
            text=True,
            timeout=10
        )
        if result.returncode != 0:
            logging.error(f"Trivy not found: {result.stderr}")
            sys.exit(1)
        # Parse version from output (format: Trivy 0.50.0)
        version_line = [line for line in result.stdout.split("\n") if "Trivy" in line][0]
        installed_version = version_line.split()[1]
        if installed_version != TRIVY_VERSION:
            logging.warning(f"Installed Trivy version {installed_version} != target {TRIVY_VERSION}")
        else:
            logging.info(f"Trivy version {TRIVY_VERSION} confirmed")
    except subprocess.TimeoutExpired:
        logging.error("Trivy version check timed out")
        sys.exit(1)
    except Exception as e:
        logging.error(f"Version check failed: {str(e)}")
        sys.exit(1)

def fetch_nvd_cve_details(cve_id):
    """Retrieve official CVE details from NVD API"""
    try:
        response = requests.get(
            f"{NVD_API_URL}?cveId={cve_id}",
            headers={"User-Agent": "TrivyScanValidator/1.0"},
            timeout=30
        )
        response.raise_for_status()
        data = response.json()
        if not data.get("vulnerabilities"):
            logging.error(f"CVE {cve_id} not found in NVD")
            return None
        return data["vulnerabilities"][0]["cve"]
    except requests.exceptions.RequestException as e:
        logging.error(f"NVD API request failed: {str(e)}")
        return None
    except Exception as e:
        logging.error(f"Failed to parse NVD response: {str(e)}")
        return None

def run_trivy_scan(image_ref):
    """Run Trivy scan on target image and return parsed results"""
    try:
        result = subprocess.run(
            ["trivy", "image", "--format", "json", "--severity", "CRITICAL", image_ref],
            capture_output=True,
            text=True,
            timeout=SCAN_TIMEOUT
        )
        if result.returncode != 0:
            logging.error(f"Trivy scan failed for {image_ref}: {result.stderr}")
            return None
        scan_results = json.loads(result.stdout)
        return scan_results
    except subprocess.TimeoutExpired:
        logging.error(f"Trivy scan timed out for {image_ref}")
        return None
    except json.JSONDecodeError:
        logging.error(f"Failed to parse Trivy JSON output for {image_ref}")
        return None
    except Exception as e:
        logging.error(f"Scan failed: {str(e)}")
        return None

def check_for_cve(scan_results, cve_id):
    """Check if target CVE is present in scan results"""
    if not scan_results:
        return False
    for result in scan_results:
        for vuln in result.get("Vulnerabilities", []):
            if vuln.get("VulnerabilityID") == cve_id:
                return True
    return False

def main():
    # Pre-flight checks
    check_trivy_version()
    logging.info(f"Starting validation for {CVE_TARGET}")

    # Fetch official CVE details
    cve_details = fetch_nvd_cve_details(CVE_TARGET)
    if not cve_details:
        logging.error("Could not retrieve official CVE details, exiting")
        sys.exit(1)
    logging.info(f"Official CVE {CVE_TARGET} details retrieved: {cve_details['descriptions'][0]['value'][:50]}...")

    # Scan our production API image that contained the vulnerable libxml2
    target_image = "our-org/production-api:v1.2.3"
    logging.info(f"Scanning target image: {target_image}")
    scan_results = run_trivy_scan(target_image)

    # Check for false negative
    cve_found = check_for_cve(scan_results, CVE_TARGET)
    if not cve_found:
        logging.error(f"FALSE NEGATIVE: {CVE_TARGET} not found in Trivy {TRIVY_VERSION} scan of {target_image}")
        logging.error(f"Vulnerable libxml2 version {LIBXML2_MIN_PATCHED} or lower present in image")
        sys.exit(1)
    else:
        logging.info(f"TRUE POSITIVE: {CVE_TARGET} correctly identified in scan")
        sys.exit(0)

if __name__ == "__main__":
    main()

# fixed-security-scan.yml
# GitHub Actions 2.0 workflow fixing Trivy 0.50 false negatives
# Implements real-time CVE checks, cache busting for lockfiles, and NVD fallback

name: Fixed Security Scan Pipeline
run-name: Security Scan for ${{ github.sha }} by ${{ github.actor }}

on:
  push:
    branches: [main, release/*]
  pull_request:
    branches: [main]
  # Add webhook trigger for real-time CVE disclosures
  repository_dispatch:
    types: [cve-disclosure]

env:
  TRIVY_VERSION: "0.50.1"  # Patched version with 1h sync interval
  NVD_API_KEY: ${{ secrets.NVD_API_KEY }}
  CACHE_VERSION: "v2"  # Bump to bust existing caches

jobs:
  trivy-scan:
    runs-on: ubuntu-22.04
    permissions:
      contents: read
      security-events: write
      actions: read

    steps:
      - name: Checkout repository
        uses: actions/checkout@v4
        with:
          fetch-depth: 0  # Fetch full history for accurate diff checks

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

      - name: Log in to GitHub Container Registry
        uses: docker/login-action@v3
        with:
          registry: ghcr.io
          username: ${{ github.actor }}
          password: ${{ secrets.GITHUB_TOKEN }}

      - name: Extract metadata for Docker
        id: meta
        uses: docker/metadata-action@v5
        with:
          images: ghcr.io/${{ github.repository }}
          tags: |
            type=sha,format=long
            type=ref,event=branch
            type=ref,event=pr

      - name: Build Docker image for scanning
        uses: docker/build-push-action@v5
        with:
          context: .
          push: false
          load: true
          tags: ${{ steps.meta.outputs.tags }}
          labels: ${{ steps.meta.outputs.labels }}
          cache-from: type=gha,scope=${{ env.CACHE_VERSION }}-docker
          cache-to: type=gha,scope=${{ env.CACHE_VERSION }}-docker,mode=max

      - name: Install Trivy ${{ env.TRIVY_VERSION }}
        run: |
          # Download and install specific Trivy version with checksums
          TRIVY_CHECKSUM="a1b2c3d4e5f6789012345678901234567890abcdef1234567890abcdef123456"  # Replace with official checksum
          curl -sfL https://raw.githubusercontent.com/aquasecurity/trivy/main/contrib/install.sh | sh -s -- -b /usr/local/bin v${{ env.TRIVY_VERSION }}
          # Verify checksum to prevent supply chain attacks
          INSTALLED_CHECKSUM=$(sha256sum /usr/local/bin/trivy | awk '{print $1}')
          if [ "$INSTALLED_CHECKSUM" != "$TRIVY_CHECKSUM" ]; then
            echo "::error::Trivy checksum mismatch. Expected $TRIVY_CHECKSUM, got $INSTALLED_CHECKSUM"
            exit 1
          fi
          trivy --version

      - name: Force Trivy DB sync (fix 24h sync bug)
        run: |
          # Clear existing cache and force fresh DB download
          rm -rf ~/.cache/trivy
          trivy image --download-db-only --severity CRITICAL
          # Verify DB is up to date
          DB_AGE=$(find ~/.cache/trivy/db -mmin -60 2>/dev/null | wc -l)
          if [ "$DB_AGE" -eq 0 ]; then
            echo "::error::Trivy DB is older than 60 minutes"
            exit 1
          fi

      - name: Run Trivy scan with NVD fallback
        id: trivy-scan
        uses: aquasecurity/trivy-action@0.20.0
        with:
          image-ref: ${{ steps.meta.outputs.tags }}
          format: sarif
          output: trivy-results.sarif
          severity: CRITICAL,HIGH
          vuln-type: os,library
          # Bypass cache for security scans
          cache: false
          # Use NVD API as fallback for missing Trivy DB entries
          nvd-api-key: ${{ env.NVD_API_KEY }}

      - name: Check for critical CVEs
        run: |
          # Parse SARIF results for target CVEs
          if grep -q "CVE-2024-3094" trivy-results.sarif; then
            echo "::error::Critical CVE CVE-2024-3094 detected in image"
            exit 1
          fi
          # Fail on any critical vulnerabilities
          CRITICAL_COUNT=$(jq '[.runs[].results[] | select(.level == "error")] | length' trivy-results.sarif)
          if [ "$CRITICAL_COUNT" -gt 0 ]; then
            echo "::error::$CRITICAL_COUNT critical vulnerabilities found"
            exit 1
          fi

      - name: Upload SARIF results to GitHub Security
        uses: github/codeql-action/upload-sarif@v3
        if: always()
        with:
          sarif_file: trivy-results.sarif
          category: trivy

      - name: Notify on scan failure
        if: failure()
        uses: 8398a7/action-slack@v3
        with:
          status: ${{ job.status }}
          text: "Security scan failed for ${{ github.sha }}. Check Trivy results."
          webhook_url: ${{ secrets.SLACK_WEBHOOK }}

# cve_webhook_handler.py
# Flask webhook handler for real-time CVE disclosures from NVD/GitHub
# Triggers re-scans of all images containing affected dependencies

import os
import json
import logging
import subprocess
from datetime import datetime
from flask import Flask, request, jsonify
import requests
from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

app = Flask(__name__)

# Configuration from environment
NVD_API_KEY = os.getenv("NVD_API_KEY")
GITHUB_TOKEN = os.getenv("GITHUB_TOKEN")
TRIVY_VERSION = os.getenv("TRIVY_VERSION", "0.50.1")
SCAN_ORCHESTRATOR_URL = os.getenv("SCAN_ORCHESTRATOR_URL", "http://localhost:8080/trigger-scan")
SLACK_WEBHOOK = os.getenv("SLACK_WEBHOOK")

# In-memory cache of affected images (replace with Redis in prod)
affected_images_cache = {}

def verify_webhook_signature(request):
    """Verify webhook signature from NVD/GitHub (simplified for example)"""
    # In production, implement HMAC signature verification per provider docs
    # https://nvd.nist.gov/developers/webhooks
    # https://docs.github.com/en/developers/webhooks-and-events/webhooks/securing-your-webhooks
    signature = request.headers.get("X-Hub-Signature-256") or request.headers.get("X-NVD-Signature")
    if not signature:
        logger.warning("No webhook signature provided")
        return False
    # Simplified check: verify token exists (replace with actual HMAC verify)
    if os.getenv("WEBHOOK_SECRET") not in signature:
        logger.error("Invalid webhook signature")
        return False
    return True

def get_affected_dependency(cve_id):
    """Retrieve affected dependency from NVD API"""
    try:
        response = requests.get(
            f"https://services.nvd.nist.gov/rest/json/cves/2.0?cveId={cve_id}",
            headers={
                "User-Agent": "CVEWebhookHandler/1.0",
                "apiKey": NVD_API_KEY
            },
            timeout=30
        )
        response.raise_for_status()
        data = response.json()
        if not data.get("vulnerabilities"):
            return None
        cve = data["vulnerabilities"][0]["cve"]
        # Extract affected software from CVE configurations
        affected_software = []
        for config in cve.get("configurations", {}).get("nodes", []):
            for child in config.get("children", []):
                for match in child.get("cpeMatch", []):
                    if match.get("vulnerable"):
                        cpe = match.get("cpe23Uri", "")
                        # Parse CPE to get dependency name and version range
                        parts = cpe.split(":")
                        if len(parts) >= 5:
                            dep_name = parts[4]
                            affected_software.append(dep_name)
        return list(set(affected_software)) if affected_software else None
    except Exception as e:
        logger.error(f"Failed to get affected dependency for {cve_id}: {str(e)}")
        return None

def trigger_image_rescans(dependency_name, cve_id):
    """Trigger re-scans for all images using the affected dependency"""
    try:
        # Query container registry for images using the dependency (simplified)
        # In production, use registry API or SBOM index
        images = ["our-org/production-api:v1.2.3", "our-org/payment-svc:v2.1.0"]
        for image in images:
            logger.info(f"Triggering re-scan for {image} due to {cve_id}")
            # Call scan orchestrator or GitHub Actions API
            response = requests.post(
                SCAN_ORCHESTRATOR_URL,
                json={
                    "image": image,
                    "cve_id": cve_id,
                    "dependency": dependency_name,
                    "triggered_at": datetime.utcnow().isoformat()
                },
                headers={"Authorization": f"Bearer {GITHUB_TOKEN}"},
                timeout=10
            )
            response.raise_for_status()
            logger.info(f"Re-scan triggered for {image}: {response.status_code}")
        return True
    except Exception as e:
        logger.error(f"Failed to trigger re-scans: {str(e)}")
        return False

def send_slack_notification(cve_id, dependency_name, image_count):
    """Send Slack notification about new CVE and triggered scans"""
    if not SLACK_WEBHOOK:
        return
    try:
        message = {
            "text": f"🚨 New Critical CVE Disclosed: {cve_id}",
            "blocks": [
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": f"*CVE ID:* {cve_id}\n*Affected Dependency:* {dependency_name}\n*Triggered Re-scans:* {image_count} images\n*Action Required:* Verify scan results in GitHub Security tab"
                    }
                }
            ]
        }
        response = requests.post(SLACK_WEBHOOK, json=message, timeout=10)
        response.raise_for_status()
    except Exception as e:
        logger.error(f"Failed to send Slack notification: {str(e)}")

@app.route("/webhook/cve", methods=["POST"])
def handle_cve_webhook():
    """Handle incoming CVE disclosure webhooks"""
    # Verify webhook authenticity
    if not verify_webhook_signature(request):
        return jsonify({"error": "Invalid signature"}), 401

    # Parse webhook payload
    try:
        payload = request.json
        if not payload:
            return jsonify({"error": "Empty payload"}), 400
    except Exception as e:
        logger.error(f"Failed to parse payload: {str(e)}")
        return jsonify({"error": "Invalid JSON"}), 400

    # Extract CVE ID from payload (handle NVD and GitHub formats)
    cve_id = None
    if "cve" in payload and "CVE_data_meta" in payload["cve"]:
        cve_id = payload["cve"]["CVE_data_meta"]["ID"]
    elif "cve_id" in payload:
        cve_id = payload["cve_id"]
    else:
        logger.warning("No CVE ID found in payload")
        return jsonify({"error": "No CVE ID"}), 400

    logger.info(f"Received webhook for CVE: {cve_id}")

    # Get affected dependency
    affected_deps = get_affected_dependency(cve_id)
    if not affected_deps:
        logger.warning(f"No affected dependencies found for {cve_id}")
        return jsonify({"status": "no action"}), 200

    # Trigger re-scans for all affected images
    dependency_name = affected_deps[0]
    trigger_image_rescans(dependency_name, cve_id)
    send_slack_notification(cve_id, dependency_name, 2)

    return jsonify({"status": "success", "cve_id": cve_id, "affected_deps": affected_deps}), 200

@app.route("/health", methods=["GET"])
def health_check():
    """Health check endpoint"""
    return jsonify({"status": "healthy", "timestamp": datetime.utcnow().isoformat()}), 200

if __name__ == "__main__":
    logger.info("Starting CVE webhook handler on port 5000")
    app.run(host="0.0.0.0", port=5000, debug=os.getenv("DEBUG", "false").lower() == "true")

rm -rf ~/.cache/trivy
trivy image --download-db-only --severity CRITICAL
# Verify DB age is under 60 minutes
if [ $(find ~/.cache/trivy/db -mmin -60 | wc -l) -eq 0 ]; then
  echo "Trivy DB is stale, exiting"
  exit 1
fi

- name: Run Trivy Scan
  uses: aquasecurity/trivy-action@0.20.0
  with:
    image-ref: my-image:v1
    cache: false
    cache-scope: security-scan

syft dir:. -o spdx-json > sbom.spdx.json
# Cross-reference SBOM with NVD
curl -X POST https://our-sbom-checker/api/check \
  -H "Content-Type: application/json" \
  -d @sbom.spdx.json