War Story: A Trivy 0.50 Missed Vulnerability Let a Hacker Access Our Cluster

# Original vulnerability scan orchestrator used in CI/CD (flawed Trivy 0.50 config)
# This script ran as a pre-deploy step, only scanned images tagged for production
# Missed CVE-2024-21626 due to default severity filter and stale DB

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

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

# Hardcoded config (part of the problem: no externalized config management)
TRIVY_VERSION = "0.50.1"
TRIVY_DB_REPO = "ghcr.io/aquasecurity/trivy-db"
SCAN_SEVERITY = "HIGH,CRITICAL"  # Excludes MEDIUM, which CVE-2024-21626 was initially rated
SCAN_VULN_TYPES = "os,library"  # Misses config and secret scans
PROD_IMAGE_REGEX = r"^prod-registry\.ourcorp\.com/.+:.+$"
SLACK_WEBHOOK = os.getenv("SLACK_WEBHOOK", "")

def run_trivy_scan(image_uri: str) -> dict:
    """Run Trivy scan on target image, return parsed results."""
    try:
        # Flawed: no --db-repo override, uses default stale DB
        # Flawed: --severity filter excludes medium CVEs
        cmd = [
            "trivy", "image",
            "--version", TRIVY_VERSION,
            "--severity", SCAN_SEVERITY,
            "--vuln-type", SCAN_VULN_TYPES,
            "--format", "json",
            "--quiet",
            image_uri
        ]
        logging.info(f"Scanning image {image_uri} with Trivy {TRIVY_VERSION}")
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            check=False  # Don't raise on non-zero exit, parse results instead
        )
        if result.returncode != 0:
            logging.error(f"Trivy scan failed for {image_uri}: {result.stderr}")
            return {"vulnerabilities": [], "error": result.stderr}
        scan_results = json.loads(result.stdout)
        return scan_results
    except Exception as e:
        logging.error(f"Unexpected error scanning {image_uri}: {str(e)}")
        return {"vulnerabilities": [], "error": str(e)}

def filter_prod_images(images: list) -> list:
    """Filter images matching production registry regex."""
    import re
    pattern = re.compile(PROD_IMAGE_REGEX)
    return [img for img in images if pattern.match(img)]

def send_slack_alert(vuln_count: int, image: str):
    """Send alert to Slack if vulnerabilities found."""
    if not SLACK_WEBHOOK:
        logging.warning("No Slack webhook configured, skipping alert")
        return
    import requests
    payload = {
        "text": f"⚠️ Trivy found {vuln_count} HIGH/CRITICAL vulnerabilities in {image}"
    }
    try:
        resp = requests.post(SLACK_WEBHOOK, json=payload, timeout=10)
        resp.raise_for_status()
        logging.info(f"Sent Slack alert for {image}")
    except Exception as e:
        logging.error(f"Failed to send Slack alert: {str(e)}")

def main():
    # Get list of images to scan from CI environment variable
    images_to_scan = os.getenv("CI_SCANNED_IMAGES", "").split(",")
    if not images_to_scan:
        logging.error("No images provided for scanning")
        sys.exit(1)

    # Filter to only production images (misses dev/staging images that later get promoted)
    prod_images = filter_prod_images(images_to_scan)
    logging.info(f"Found {len(prod_images)} production images to scan")

    total_vulns = 0
    for image in prod_images:
        scan_result = run_trivy_scan(image)
        if "error" in scan_result:
            logging.error(f"Scan failed for {image}, skipping")
            continue
        vulns = scan_result.get("vulnerabilities", [])
        total_vulns += len(vulns)
        if vulns:
            send_slack_alert(len(vulns), image)
            logging.warning(f"Found {len(vulns)} vulnerabilities in {image}")
        else:
            logging.info(f"No HIGH/CRITICAL vulnerabilities found in {image}")

    # Flawed: only fails pipeline if HIGH/CRITICAL vulns found
    # Misses medium vulns that could chain for compromise
    if total_vulns > 0:
        logging.error(f"Pipeline failed: {total_vulns} vulnerabilities found")
        sys.exit(1)
    else:
        logging.info("Pipeline passed: no HIGH/CRITICAL vulnerabilities found")

if __name__ == "__main__":
    main()

# Fixed vulnerability scan orchestrator with Trivy 0.52+ and Grype cross-check
# Implements defense-in-depth: scans all images (dev/staging/prod), all severities,
# updates DB before scan, cross-validates with Grype to catch scanner-specific misses

import subprocess
import sys
import json
import os
import hashlib
from datetime import datetime
import logging

# Externalized config (loaded from environment, not hardcoded)
TRIVY_VERSION = os.getenv("TRIVY_VERSION", "0.52.1")
GRYPE_VERSION = os.getenv("GRYPE_VERSION", "0.73.0")
TRIVY_DB_REPO = os.getenv("TRIVY_DB_REPO", "ghcr.io/aquasecurity/trivy-db")
SCAN_SEVERITY = os.getenv("SCAN_SEVERITY", "LOW,MEDIUM,HIGH,CRITICAL")  # Include all severities
SCAN_VULN_TYPES = os.getenv("SCAN_VULN_TYPES", "os,library,config,secret")  # Full scan scope
SLACK_WEBHOOK = os.getenv("SLACK_WEBHOOK", "")
VULN_THRESHOLDS = json.loads(os.getenv("VULN_THRESHOLDS", '{"critical": 0, "high": 2, "medium": 5}'))  # Fail on any critical, >2 high, >5 medium

# Configure structured logging
logging.basicConfig(
    level=logging.INFO,
    format=json.dumps({
        "timestamp": "%(asctime)s",
        "level": "%(levelname)s",
        "message": "%(message)s"
    }),
    handlers=[logging.StreamHandler(sys.stdout)]
)

def update_trivy_db():
    """Force update Trivy vulnerability database before scanning."""
    try:
        cmd = ["trivy", "db", "update", "--db-repo", TRIVY_DB_REPO]
        logging.info(f"Updating Trivy DB from {TRIVY_DB_REPO}")
        result = subprocess.run(cmd, capture_output=True, text=True, check=True, timeout=300)
        logging.info("Trivy DB updated successfully")
        return True
    except subprocess.TimeoutExpired:
        logging.error("Trivy DB update timed out after 5 minutes")
        return False
    except subprocess.CalledProcessError as e:
        logging.error(f"Trivy DB update failed: {e.stderr}")
        return False
    except Exception as e:
        logging.error(f"Unexpected error updating Trivy DB: {str(e)}")
        return False

def run_trivy_scan(image_uri: str) -> dict:
    """Run Trivy scan with full config, return parsed results."""
    try:
        cmd = [
            "trivy", "image",
            "--severity", SCAN_SEVERITY,
            "--vuln-type", SCAN_VULN_TYPES,
            "--format", "json",
            "--quiet",
            image_uri
        ]
        logging.info(f"Running Trivy scan for {image_uri}")
        result = subprocess.run(cmd, capture_output=True, text=True, check=False, timeout=600)
        if result.returncode != 0:
            logging.error(f"Trivy scan failed for {image_uri}: {result.stderr}")
            return {"vulnerabilities": [], "error": result.stderr}
        # Trivy returns list of scan results for multi-arch images
        scan_results = json.loads(result.stdout)
        if isinstance(scan_results, list):
            return scan_results[0] if scan_results else {}
        return scan_results
    except subprocess.TimeoutExpired:
        logging.error(f"Trivy scan timed out for {image_uri}")
        return {"vulnerabilities": [], "error": "scan timeout"}
    except Exception as e:
        logging.error(f"Unexpected error in Trivy scan: {str(e)}")
        return {"vulnerabilities": [], "error": str(e)}

def run_grype_scan(image_uri: str) -> dict:
    """Run Grype scan for cross-validation."""
    try:
        cmd = [
            "grype", image_uri,
            "-o", "json",
            "--quiet"
        ]
        logging.info(f"Running Grype scan for {image_uri}")
        result = subprocess.run(cmd, capture_output=True, text=True, check=False, timeout=600)
        if result.returncode != 0:
            logging.error(f"Grype scan failed for {image_uri}: {result.stderr}")
            return {"vulnerabilities": [], "error": result.stderr}
        return json.loads(result.stdout)
    except Exception as e:
        logging.error(f"Unexpected error in Grype scan: {str(e)}")
        return {"vulnerabilities": [], "error": str(e)}

def compare_scans(trivy_results: dict, grype_results: dict) -> list:
    """Find vulnerabilities missed by either scanner."""
    trivy_vulns = {v["vulnerabilityID"] for v in trivy_results.get("vulnerabilities", [])}
    grype_vulns = {v["id"] for v in grype_results.get("matches", [])}
    missed_by_trivy = grype_vulns - trivy_vulns
    missed_by_grype = trivy_vulns - grype_vulns
    return [
        {"scanner": "trivy", "missed": list(missed_by_trivy)},
        {"scanner": "grype", "missed": list(missed_by_grype)}
    ]

def main():
    # Update Trivy DB first to avoid stale data
    if not update_trivy_db():
        logging.error("Failed to update Trivy DB, exiting")
        sys.exit(1)

    images_to_scan = os.getenv("CI_SCANNED_IMAGES", "").split(",")
    if not images_to_scan:
        logging.error("No images provided for scanning")
        sys.exit(1)

    logging.info(f"Scanning {len(images_to_scan)} images with Trivy and Grype")
    pipeline_failed = False

    for image in images_to_scan:
        if not image.strip():
            continue
        # Run both scans in parallel (simplified here for readability)
        trivy_res = run_trivy_scan(image)
        grype_res = run_grype_scan(image)

        # Compare results
        scan_diffs = compare_scans(trivy_res, grype_res)
        for diff in scan_diffs:
            if diff["missed"]:
                logging.warning(f"{diff['scanner']} missed {len(diff['missed'])} vulnerabilities in {image}: {diff['missed']}")

        # Aggregate vulnerabilities
        all_vulns = []
        for v in trivy_res.get("vulnerabilities", []):
            all_vulns.append({
                "id": v["vulnerabilityID"],
                "severity": v["severity"],
                "scanner": "trivy"
            })
        for m in grype_res.get("matches", []):
            all_vulns.append({
                "id": m["vulnerability"]["id"],
                "severity": m["vulnerability"]["severity"],
                "scanner": "grype"
            })

        # Deduplicate by vulnerability ID
        deduped = {v["id"]: v for v in all_vulns}.values()
        severity_counts = {"critical": 0, "high": 0, "medium": 0, "low": 0}
        for v in deduped:
            sev = v["severity"].lower()
            if sev in severity_counts:
                severity_counts[sev] += 1

        # Check against thresholds
        if severity_counts["critical"] > VULN_THRESHOLDS["critical"]:
            logging.error(f"CRITICAL vulnerabilities found in {image}: {severity_counts['critical']}")
            pipeline_failed = True
        if severity_counts["high"] > VULN_THRESHOLDS["high"]:
            logging.error(f"Too many HIGH vulnerabilities in {image}: {severity_counts['high']}")
            pipeline_failed = True
        if severity_counts["medium"] > VULN_THRESHOLDS["medium"]:
            logging.error(f"Too many MEDIUM vulnerabilities in {image}: {severity_counts['medium']}")
            pipeline_failed = True

        logging.info(f"Image {image} scan summary: {json.dumps(severity_counts)}")

    if pipeline_failed:
        logging.error("Pipeline failed due to vulnerability threshold breaches")
        sys.exit(1)
    else:
        logging.info("Pipeline passed: all vulnerability thresholds met")

if __name__ == "__main__":
    main()

// Incident response audit tool to scan all K8s clusters for containers running
// vulnerable runc versions (CVE-2024-21626 affected runc < 1.1.12)
// Compiles with Go 1.22+, requires kubectl access and Trivy installed

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "os"
    "os/exec"
    "strings"
    "sync"
    "time"
)

// Config holds audit configuration
type Config struct {
    Clusters       []string `json:"clusters"`
    NamespaceRegex string   `json:"namespace_regex"`
    OutputFile     string   `json:"output_file"`
    TrivyVersion   string   `json:"trivy_version"`
}

// ScanResult holds per-container scan results
type ScanResult struct {
    Cluster     string    `json:"cluster"`
    Namespace   string    `json:"namespace"`
    Pod         string    `json:"pod"`
    Container   string    `json:"container"`
    Image       string    `json:"image"`
    Vulnerable  bool      `json:"vulnerable"`
    CVEs        []string  `json:"cves"`
    ScannedAt   time.Time `json:"scanned_at"`
    Error       string    `json:"error,omitempty"`
}

var (
    config     Config
    results    []ScanResult
    resultsMut sync.Mutex
    wg         sync.WaitGroup
)

func loadConfig() error {
    configPath := os.Getenv("AUDIT_CONFIG_PATH")
    if configPath == "" {
        configPath = "audit-config.json"
    }
    data, err := os.ReadFile(configPath)
    if err != nil {
        return fmt.Errorf("failed to read config: %w", err)
    }
    if err := json.Unmarshal(data, &config); err != nil {
        return fmt.Errorf("failed to parse config: %w", err)
    }
    if len(config.Clusters) == 0 {
        return fmt.Errorf("no clusters specified in config")
    }
    return nil
}

func getPods(cluster, namespaceRegex string) ([]string, error) {
    // Get all pods across all namespaces matching regex
    cmd := exec.Command("kubectl", "--context", cluster, "get", "pods", "--all-namespaces", "-o", "json")
    output, err := cmd.Output()
    if err != nil {
        return nil, fmt.Errorf("kubectl get pods failed: %w", err)
    }

    var podList struct {
        Items []struct {
            Metadata struct {
                Name      string `json:"name"`
                Namespace string `json:"namespace"`
            } `json:"metadata"`
            Spec struct {
                Containers []struct {
                    Image string `json:"image"`
                } `json:"containers"`
            } `json:"spec"`
        } `json:"items"`
    }
    if err := json.Unmarshal(output, &podList); err != nil {
        return nil, fmt.Errorf("failed to parse pod list: %w", err)
    }

    // Filter by namespace regex
    matchedPods := []string{}
    for _, pod := range podList.Items {
        // Simplified regex check (real impl uses regexp.MatchString)
        if strings.Contains(pod.Metadata.Namespace, namespaceRegex) || namespaceRegex == ".*" {
            for _, container := range pod.Spec.Containers {
                podInfo := fmt.Sprintf("%s|%s|%s|%s|%s", cluster, pod.Metadata.Namespace, pod.Metadata.Name, container.Image, pod.Metadata.Name)
                matchedPods = append(matchedPods, podInfo)
            }
        }
    }
    return matchedPods, nil
}

func scanImage(imageURI string) ([]string, error) {
    // Run Trivy scan for CVE-2024-21626 specifically
    cmd := exec.Command("trivy", "image", "--severity", "MEDIUM,HIGH,CRITICAL", "--format", "json", imageURI)
    output, err := cmd.Output()
    if err != nil {
        if exitErr, ok := err.(*exec.ExitError); ok {
            return nil, fmt.Errorf("trivy scan failed: %s", string(exitErr.Stderr))
        }
        return nil, fmt.Errorf("trivy scan error: %w", err)
    }

    var scanResults []struct {
        Vulnerabilities []struct {
            VulnerabilityID string `json:"vulnerabilityID"`
            Severity        string `json:"severity"`
        } `json:"vulnerabilities"`
    }
    if err := json.Unmarshal(output, &scanResults); err != nil {
        return nil, fmt.Errorf("failed to parse trivy output: %w", err)
    }

    cves := []string{}
    for _, res := range scanResults {
        for _, vuln := range res.Vulnerabilities {
            if vuln.VulnerabilityID == "CVE-2024-21626" {
                cves = append(cves, vuln.VulnerabilityID)
            }
        }
    }
    return cves, nil
}

func auditContainer(podInfo string) {
    defer wg.Done()
    parts := strings.Split(podInfo, "|")
    if len(parts) != 5 {
        log.Printf("Invalid pod info: %s", podInfo)
        return
    }
    cluster, namespace, pod, image, podName := parts[0], parts[1], parts[2], parts[3], parts[4]

    cves, err := scanImage(image)
    result := ScanResult{
        Cluster:   cluster,
        Namespace: namespace,
        Pod:       podName,
        Container: pod,
        Image:     image,
        ScannedAt: time.Now(),
    }

    if err != nil {
        result.Error = err.Error()
        log.Printf("Error scanning %s: %v", image, err)
    } else {
        result.CVEs = cves
        result.Vulnerable = len(cves) > 0
        if result.Vulnerable {
            log.Printf("⚠️ Vulnerable container found: %s/%s/%s - %s", cluster, namespace, podName, image)
        }
    }

    resultsMut.Lock()
    results = append(results, result)
    resultsMut.Unlock()
}

func main() {
    log.SetOutput(os.Stdout)
    log.SetFlags(log.LstdFlags | log.Lshortfile)

    if err := loadConfig(); err != nil {
        log.Fatalf("Failed to load config: %v", err)
    }

    log.Printf("Starting audit for %d clusters", len(config.Clusters))

    for _, cluster := range config.Clusters {
        log.Printf("Auditing cluster: %s", cluster)
        pods, err := getPods(cluster, config.NamespaceRegex)
        if err != nil {
            log.Printf("Failed to get pods for cluster %s: %v", cluster, err)
            continue
        }
        log.Printf("Found %d containers to scan in cluster %s", len(pods), cluster)

        for _, pod := range pods {
            wg.Add(1)
            go auditContainer(pod)
        }
    }

    wg.Wait()

    // Write results to file
    output, err := json.MarshalIndent(results, "", "  ")
    if err != nil {
        log.Fatalf("Failed to marshal results: %v", err)
    }
    if err := os.WriteFile(config.OutputFile, output, 0644); err != nil {
        log.Fatalf("Failed to write results file: %v", err)
    }

    // Summary
    vulnerableCount := 0
    for _, res := range results {
        if res.Vulnerable {
            vulnerableCount++
        }
    }
    log.Printf("Audit complete. Total containers scanned: %d, Vulnerable: %d", len(results), vulnerableCount)
}

# Run Trivy and Grype in parallel for a single image
trivy image --severity MEDIUM,HIGH,CRITICAL --format json alpine:3.19 > trivy-results.json &
grype alpine:3.19 -o json > grype-results.json &
wait
echo "Scans complete. Compare results with jq."

# Scan all severities, fail only on critical/high thresholds
trivy image \
  --severity LOW,MEDIUM,HIGH,CRITICAL \
  --format json \
  --vuln-type os,library,config,secret \
  my-prod-image:latest

# Update Trivy DB and verify it's less than 1 hour old
trivy db update --db-repo ghcr.io/aquasecurity/trivy-db
DB_AGE=$(trivy db info --format json | jq -r '.updatedAt | fromdateiso8601 | now - .')
if [ $(echo "$DB_AGE > 3600" | bc) -eq 1 ]; then
  echo "Error: Trivy DB is older than 1 hour"
  exit 1
fi