Deep Dive: Snyk 1.120's Vulnerability Detection Engine vs. Trivy 0.50's Static Analysis

// snyk-scan-ci.go
// Demonstrates integrating Snyk 1.120 CLI into a Go-based CI pipeline
// with structured error handling, result parsing, and GitHub Checks API integration
package main

import (
    "bytes"
    "context"
    "encoding/json"
    "fmt"
    "io"
    "net/http"
    "os"
    "os/exec"
    "time"
)

// SnykVulnerability represents a single Snyk detection result
type SnykVulnerability struct {
    ID          string `json:"id"`
    Severity    string `json:"severity"`
    PackageName string `json:"packageName"`
    Version     string `json:"version"`
    FixedIn     string `json:"fixedIn"`
}

// SnykScanResult represents the full Snyk CLI JSON output
type SnykScanResult struct {
    Vulnerabilities []SnykVulnerability `json:"vulnerabilities"`
    ProjectName    string               `json:"projectName"`
}

// GitHubCheckRequest represents a GitHub Checks API create request
type GitHubCheckRequest struct {
    Name        string `json:"name"`
    HeadSHA     string `json:"head_sha"`
    Status      string `json:"status"`
    Conclusion  string `json:"conclusion,omitempty"`
    Output      Output `json:"output,omitempty"`
}

// Output represents the markdown output for a GitHub Check
type Output struct {
    Title   string `json:"title"`
    Summary string `json:"summary"`
    Text    string `json:"text"`
}

func main() {
    // Validate required environment variables
    requiredEnvVars := []string{"GITHUB_TOKEN", "GITHUB_SHA", "SNYK_TOKEN"}
    for _, envVar := range requiredEnvVars {
        if os.Getenv(envVar) == "" {
            fmt.Fprintf(os.Stderr, "Missing required environment variable: %s\n", envVar)
            os.Exit(1)
        }
    }

    // Configure Snyk scan command for container image
    imageRef := os.Getenv("CONTAINER_IMAGE")
    if imageRef == "" {
        imageRef = "myapp:latest"
    }
    args := []string{"test", imageRef, "--json", "--all-projects"}
    cmd := exec.Command("snyk", args...)
    cmd.Env = append(os.Environ(), fmt.Sprintf("SNYK_TOKEN=%s", os.Getenv("SNYK_TOKEN")))

    // Capture stdout and stderr
    var stdout, stderr bytes.Buffer
    cmd.Stdout = &stdout
    cmd.Stderr = &stderr

    // Run Snyk scan with timeout
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
    defer cancel()
    cmd = exec.CommandContext(ctx, "snyk", args...)
    cmd.Env = append(os.Environ(), fmt.Sprintf("SNYK_TOKEN=%s", os.Getenv("SNYK_TOKEN")))
    cmd.Stdout = &stdout
    cmd.Stderr = &stderr

    err := cmd.Run()
    // Snyk returns exit code 1 if vulnerabilities are found, 2 for errors
    if err != nil && cmd.ProcessState.ExitCode() != 1 {
        fmt.Fprintf(os.Stderr, "Snyk scan failed: %v\nStderr: %s\n", err, stderr.String())
        os.Exit(1)
    }

    // Parse Snyk JSON output
    var scanResult SnykScanResult
    if err := json.Unmarshal(stdout.Bytes(), &scanResult); err != nil {
        fmt.Fprintf(os.Stderr, "Failed to parse Snyk output: %v\n", err)
        os.Exit(1)
    }

    // Generate GitHub Check output
    checkReq := GitHubCheckRequest{
        Name:    "Snyk 1.120 Vulnerability Scan",
        HeadSHA: os.Getenv("GITHUB_SHA"),
        Status:  "completed",
    }
    if len(scanResult.Vulnerabilities) > 0 {
        checkReq.Conclusion = "failure"
        summary := fmt.Sprintf("Found %d vulnerabilities in %s", len(scanResult.Vulnerabilities), imageRef)
        checkReq.Output = Output{
            Title:   "Vulnerabilities Detected",
            Summary: summary,
            Text:    generateVulnMarkdown(scanResult.Vulnerabilities),
        }
    } else {
        checkReq.Conclusion = "success"
        checkReq.Output = Output{
            Title:   "No Vulnerabilities Found",
            Summary: "Clean scan for " + imageRef,
            Text:    "All dependencies and container packages are up to date.",
        }
    }

    // Post check to GitHub API
    githubToken := os.Getenv("GITHUB_TOKEN")
    reqBody, _ := json.Marshal(checkReq)
    url := fmt.Sprintf("https://api.github.com/repos/%s/check-runs", os.Getenv("GITHUB_REPOSITORY"))
    req, _ := http.NewRequest("POST", url, bytes.NewBuffer(reqBody))
    req.Header.Set("Authorization", "Bearer "+githubToken)
    req.Header.Set("Accept", "application/vnd.github.v3+json")

    client := &http.Client{Timeout: 10 * time.Second}
    resp, err := client.Do(req)
    if err != nil {
        fmt.Fprintf(os.Stderr, "Failed to post GitHub check: %v\n", err)
        os.Exit(1)
    }
    defer resp.Body.Close()

    if resp.StatusCode != 201 {
        body, _ := io.ReadAll(resp.Body)
        fmt.Fprintf(os.Stderr, "GitHub API error: %d %s\n", resp.StatusCode, string(body))
        os.Exit(1)
    }

    fmt.Println("Snyk scan completed successfully. GitHub Check posted.")
}

// generateVulnMarkdown creates Markdown table for vulnerabilities
func generateVulnMarkdown(vulns []SnykVulnerability) string {
    md := "| CVE ID | Severity | Package | Version | Fixed In |\n"
    md += "|--------|----------|---------|---------|----------|\n"
    for _, v := range vulns {
        md += fmt.Sprintf("| %s | %s | %s | %s | %s |\n", v.ID, v.Severity, v.PackageName, v.Version, v.FixedIn)
    }
    return md
}

# trivy-k8s-operator.py
# Kubernetes operator sidecar for continuous Trivy 0.50 scanning of running pods
# Includes retry logic, metrics export to Prometheus, and Slack alerting
import os
import subprocess
import json
import time
import requests
from prometheus_client import start_http_server, Gauge
from kubernetes import client, config

# Prometheus metrics
VULN_COUNT = Gauge('trivy_vulnerability_count', 'Total vulnerabilities per pod', ['pod', 'namespace', 'severity'])
SCAN_DURATION = Gauge('trivy_scan_duration_seconds', 'Time taken to complete Trivy scan', ['pod', 'namespace'])

# Trivy output schema
class TrivyVulnerability:
    def __init__(self, vuln_data):
        self.vuln_id = vuln_data.get('VulnerabilityID')
        self.severity = vuln_data.get('Severity')
        self.pkg_name = vuln_data.get('PkgName')
        self.installed_version = vuln_data.get('InstalledVersion')
        self.fixed_version = vuln_data.get('FixedVersion', 'None')

class TrivyScanResult:
    def __init__(self, scan_data):
        self.results = scan_data.get('Results', [])
        self.vulnerabilities = []
        for res in self.results:
            for vuln in res.get('Vulnerabilities', []):
                self.vulnerabilities.append(TrivyVulnerability(vuln))

def run_trivy_scan(image_ref, retry_count=3):
    """Run Trivy CLI scan with retry logic for transient failures"""
    for attempt in range(retry_count):
        try:
            cmd = [
                'trivy', 'image',
                '--format', 'json',
                '--quiet',
                '--timeout', '10m',
                image_ref
            ]
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=600
            )
            if result.returncode != 0:
                raise RuntimeError(f"Trivy scan failed: {result.stderr}")
            scan_data = json.loads(result.stdout)
            return TrivyScanResult(scan_data)
        except (subprocess.TimeoutExpired, json.JSONDecodeError, RuntimeError) as e:
            print(f"Attempt {attempt + 1} failed: {e}")
            if attempt == retry_count - 1:
                raise
            time.sleep(2 ** attempt)  # Exponential backoff

def post_slack_alert(webhook_url, pod_name, namespace, vuln_count):
    """Send critical vulnerability alert to Slack"""
    if vuln_count == 0:
        return
    payload = {
        "text": f"🚨 Trivy 0.50 Scan Alert: {vuln_count} vulnerabilities found in {pod_name} ({namespace})",
        "attachments": [
            {
                "color": "#ff0000",
                "fields": [
                    {"title": "Pod", "value": pod_name, "short": True},
                    {"title": "Namespace", "value": namespace, "short": True},
                    {"title": "Vulnerability Count", "value": vuln_count, "short": True}
                ]
            }
        ]
    }
    try:
        resp = requests.post(webhook_url, json=payload, timeout=10)
        resp.raise_for_status()
    except Exception as e:
        print(f"Failed to post Slack alert: {e}")

def scan_pod(pod, namespace):
    """Scan all container images in a pod"""
    config.load_incluster_config()
    v1 = client.CoreV1Api()
    pod_spec = v1.read_namespaced_pod(pod, namespace).spec
    total_vulns = 0
    start_time = time.time()

    for container in pod_spec.containers:
        image_ref = container.image
        print(f"Scanning image {image_ref} for pod {pod}...")
        try:
            scan_result = run_trivy_scan(image_ref)
            # Update Prometheus metrics
            severity_counts = {"CRITICAL": 0, "HIGH": 0, "MEDIUM": 0, "LOW": 0}
            for vuln in scan_result.vulnerabilities:
                severity_counts[vuln.severity] = severity_counts.get(vuln.severity, 0) + 1
                VULN_COUNT.labels(pod=pod, namespace=namespace, severity=vuln.severity).set(severity_counts[vuln.severity])
            total_vulns += len(scan_result.vulnerabilities)
        except Exception as e:
            print(f"Failed to scan {image_ref}: {e}")
            total_vulns = -1  # Indicate scan failure

    # Record scan duration
    duration = time.time() - start_time
    SCAN_DURATION.labels(pod=pod, namespace=namespace).set(duration)

    # Post Slack alert if critical/high vulns
    slack_webhook = os.getenv('SLACK_WEBHOOK_URL')
    if slack_webhook and total_vulns > 0:
        post_slack_alert(slack_webhook, pod, namespace, total_vulns)

    return total_vulns

if __name__ == "__main__":
    # Start Prometheus metrics server
    start_http_server(8000)
    print("Trivy Kubernetes Operator started. Metrics on :8000")

    # Watch for pod events (simplified for example)
    config.load_incluster_config()
    v1 = client.CoreV1Api()
    pods = v1.list_pod_for_all_namespaces(watch=False)
    for pod in pods.items:
        scan_pod(pod.metadata.name, pod.metadata.namespace)

# benchmark-scanners.py
# Reproducible benchmark comparing Snyk 1.120 and Trivy 0.50 scan performance
# Tests speed, memory usage, and detection accuracy across 50 container images
import subprocess
import json
import time
import psutil
import pandas as pd
import matplotlib.pyplot as plt
from pathlib import Path

# Configuration
TEST_IMAGES = [
    "python:3.9-slim", "node:20-alpine", "golang:1.21", "openjdk:17-jdk-slim",
    "nginx:1.25-alpine", "postgres:16-alpine", "redis:7.2-alpine"
] * 7  # 49 images total
SNYK_CLI = "snyk"
TRIVY_CLI = "trivy"
RESULTS_DIR = Path("./benchmark-results")
RESULTS_DIR.mkdir(exist_ok=True)

class ScanResult:
    def __init__(self, tool, image, duration, memory_mb, vuln_count, error=None):
        self.tool = tool
        self.image = image
        self.duration = duration
        self.memory_mb = memory_mb
        self.vuln_count = vuln_count
        self.error = error

def get_memory_usage(process_pid):
    """Get peak memory usage of a process in MB"""
    try:
        proc = psutil.Process(process_pid)
        mem_info = proc.memory_info()
        return mem_info.rss / 1024 / 1024  # Convert to MB
    except Exception:
        return 0.0

def run_snyk_scan(image):
    """Run Snyk 1.120 scan and record metrics"""
    start_time = time.time()
    process = subprocess.Popen(
        [SNYK_CLI, "test", image, "--json", "--all-projects"],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        env={"SNYK_TOKEN": os.getenv("SNYK_TOKEN")}
    )
    # Monitor memory usage
    peak_mem = 0.0
    while process.poll() is None:
        mem = get_memory_usage(process.pid)
        if mem > peak_mem:
            peak_mem = mem
        time.sleep(0.1)
    end_time = time.time()
    duration = end_time - start_time
    stdout, stderr = process.communicate()
    # Snyk returns 1 for vulns found, 0 for clean, 2 for error
    if process.returncode == 2:
        return ScanResult("snyk", image, duration, peak_mem, 0, stderr.decode())
    try:
        result = json.loads(stdout)
        vuln_count = len(result.get("vulnerabilities", []))
        return ScanResult("snyk", image, duration, peak_mem, vuln_count)
    except json.JSONDecodeError:
        return ScanResult("snyk", image, duration, peak_mem, 0, "Failed to parse Snyk output")

def run_trivy_scan(image):
    """Run Trivy 0.50 scan and record metrics"""
    start_time = time.time()
    process = subprocess.Popen(
        [TRIVY_CLI, "image", "--format", "json", "--quiet", image],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE
    )
    # Monitor memory usage
    peak_mem = 0.0
    while process.poll() is None:
        mem = get_memory_usage(process.pid)
        if mem > peak_mem:
            peak_mem = mem
        time.sleep(0.1)
    end_time = time.time()
    duration = end_time - start_time
    stdout, stderr = process.communicate()
    if process.returncode != 0:
        return ScanResult("trivy", image, duration, peak_mem, 0, stderr.decode())
    try:
        result = json.loads(stdout)
        vuln_count = 0
        for res in result.get("Results", []):
            vuln_count += len(res.get("Vulnerabilities", []))
        return ScanResult("trivy", image, duration, peak_mem, vuln_count)
    except json.JSONDecodeError:
        return ScanResult("trivy", image, duration, peak_mem, 0, "Failed to parse Trivy output")

def run_benchmark():
    """Execute full benchmark suite"""
    results = []
    for idx, image in enumerate(TEST_IMAGES):
        print(f"Scanning {image} ({idx+1}/{len(TEST_IMAGES)})...")
        # Pull image first to avoid network skew
        subprocess.run(["docker", "pull", image], capture_output=True)
        # Run Snyk scan
        snyk_result = run_snyk_scan(image)
        results.append(snyk_result)
        # Run Trivy scan
        trivy_result = run_trivy_scan(image)
        results.append(trivy_result)
        # Save intermediate results
        with open(RESULTS_DIR / f"intermediate-{idx}.json", "w") as f:
            json.dump([vars(snyk_result), vars(trivy_result)], f)
    return results

def analyze_results(results):
    """Generate statistical summary and plots"""
    df = pd.DataFrame([vars(r) for r in results if r.error is None])
    # Summary stats
    summary = df.groupby("tool").agg({
        "duration": ["mean", "median", "std"],
        "memory_mb": ["mean", "median", "std"],
        "vuln_count": ["mean", "sum"]
    }).round(2)
    print("\n=== Benchmark Summary ===")
    print(summary)
    # Save to CSV
    df.to_csv(RESULTS_DIR / "benchmark-results.csv", index=False)
    # Generate speed comparison plot
    plt.figure(figsize=(10,6))
    df.boxplot(column="duration", by="tool", grid=False)
    plt.title("Scan Duration Comparison (Snyk 1.120 vs Trivy 0.50)")
    plt.ylabel("Duration (seconds)")
    plt.savefig(RESULTS_DIR / "speed-comparison.png")
    plt.close()
    # Generate memory comparison plot
    plt.figure(figsize=(10,6))
    df.boxplot(column="memory_mb", by="tool", grid=False)
    plt.title("Memory Usage Comparison (Snyk 1.120 vs Trivy 0.50)")
    plt.ylabel("Memory (MB)")
    plt.savefig(RESULTS_DIR / "memory-comparison.png")
    plt.close()

if __name__ == "__main__":
    import os
    if not os.getenv("SNYK_TOKEN"):
        print("Error: SNYK_TOKEN environment variable not set")
        exit(1)
    print("Starting benchmark: Snyk 1.120 vs Trivy 0.50")
    print(f"Test images: {len(TEST_IMAGES)}")
    print(f"Results directory: {RESULTS_DIR}")
    results = run_benchmark()
    analyze_results(results)
    print(f"Benchmark complete. Results saved to {RESULTS_DIR}")

    io.snyk
    snyk-maven-plugin
    2.0.0

        ${env.SNYK_TOKEN}
        high



            verify

                test

# .github/workflows/trivy-scan.yml
name: Trivy Container Scan
on: [push]
jobs:
  scan:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Run Trivy scan
        uses: aquasecurity/trivy-action@0.18.0
        with:
          image-ref: myapp:${{ github.sha }}
          format: sarif
          output: trivy-results.sarif
          severity: CRITICAL,HIGH
      - name: Upload SARIF to GitHub Security
        uses: github/codeql-action/upload-sarif@v3
        with:
          sarif_file: trivy-results.sarif

# merge-results.py
import json
def merge_vulns(snyk_json, trivy_json):
    merged = {}
    # Add Snyk vulns
    for vuln in snyk_json.get("vulnerabilities", []):
        merged[vuln["id"]] = {"source": "snyk", **vuln}
    # Add Trivy vulns, skip duplicates
    for res in trivy_json.get("Results", []):
        for vuln in res.get("Vulnerabilities", []):
            if vuln["VulnerabilityID"] not in merged:
                merged[vuln["VulnerabilityID"]] = {"source": "trivy", **vuln}
    return list(merged.values())