Image Scanning Automation

import subprocess
import json

def scan_image_trivy(image: str, severity: str = "CRITICAL,HIGH") -> dict:
    """Scan a Docker image with Trivy and return results."""
    cmd = [
        "trivy", "image",
        "--format", "json",
        "--severity", severity,
        "--quiet",
        image,
    ]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
    return json.loads(result.stdout)

def parse_vulnerabilities(scan_result: dict) -> list[dict]:
    """Extract vulnerabilities from Trivy scan results."""
    vulns = []
    for result in scan_result.get("Results", []):
        for vuln in result.get("Vulnerabilities", []):
            vulns.append({
                "id": vuln["VulnerabilityID"],
                "severity": vuln["Severity"],
                "package": vuln["PkgName"],
                "installed": vuln.get("InstalledVersion", "unknown"),
                "fixed": vuln.get("FixedVersion", "none"),
                "title": vuln.get("Title", ""),
            })
    return vulns

# Usage
scan = scan_image_trivy("myapp:latest")
vulns = parse_vulnerabilities(scan)
print(f"Found {len(vulns)} vulnerabilities")
for v in vulns:
    print(f"  [{v['severity']}] {v['id']}: {v['package']} {v['installed']}")

import subprocess
import json

def scan_image_grype(image: str) -> dict:
    """Scan a Docker image with Grype."""
    cmd = [
        "grype", image,
        "--output", "json",
        "--quiet",
    ]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
    return json.loads(result.stdout)

def summarize_grype_results(scan: dict) -> dict:
    """Summarize Grype scan results."""
    matches = scan.get("matches", [])
    summary = {"critical": 0, "high": 0, "medium": 0, "low": 0, "total": len(matches)}

    for match in matches:
        severity = match["vulnerability"]["severity"].lower()
        if severity in summary:
            summary[severity] += 1

    return summary

# Usage
scan = scan_image_grype("myapp:latest")
summary = summarize_grype_results(scan)
print(f"Critical: {summary['critical']}, High: {summary['high']}")
print(f"Total: {summary['total']} vulnerabilities")

import subprocess
import sys
import json

def ci_scan(image: str, fail_on: str = "CRITICAL") -> int:
    """Scan an image in CI and fail if vulnerabilities exceed threshold."""
    severity_map = {"CRITICAL": 4, "HIGH": 3, "MEDIUM": 2, "LOW": 1}
    threshold = severity_map.get(fail_on, 4)

    # Run Trivy
    scan = scan_image_trivy(image, severity="CRITICAL,HIGH,MEDIUM")
    vulns = parse_vulnerabilities(scan)

    # Filter by threshold
    failures = [
        v for v in vulns
        if severity_map.get(v["severity"].upper(), 0) >= threshold
    ]

    if failures:
        print(f"FAIL: Found {len(failures)} {fail_on}+ vulnerabilities")
        for v in failures:
            print(f"  [{v['severity']}] {v['id']}: {v['package']}")
        return 1

    print(f"PASS: No {fail_on}+ vulnerabilities found")
    return 0

# Exit with non-zero code if critical vulnerabilities are found
sys.exit(ci_scan("myapp:latest", fail_on="CRITICAL"))

from dataclasses import dataclass

@dataclass
class ScanPolicy:
    max_critical: int = 0
    max_high: int = 5
    max_medium: int = 20
    allowed_licenses: list[str] | None = None
    blocked_packages: list[str] | None = None

def enforce_policy(vulns: list[dict], policy: ScanPolicy) -> list[str]:
    """Check scan results against policy."""
    violations = []

    critical = [v for v in vulns if v["severity"] == "CRITICAL"]
    high = [v for v in vulns if v["severity"] == "HIGH"]
    medium = [v for v in vulns if v["severity"] == "MEDIUM"]

    if len(critical) > policy.max_critical:
        violations.append(f"Too many CRITICAL: {len(critical)} > {policy.max_critical}")

    if len(high) > policy.max_high:
        violations.append(f"Too many HIGH: {len(high)} > {policy.max_high}")

    if len(medium) > policy.max_medium:
        violations.append(f"Too many MEDIUM: {len(medium)} > {policy.max_medium}")

    if policy.blocked_packages:
        blocked = [v for v in vulns if v["package"] in policy.blocked_packages]
        if blocked:
            violations.append(f"Blocked packages found: {', '.join(set(v['package'] for v in blocked))}")

    return violations

# Usage
policy = ScanPolicy(max_critical=0, max_high=3)
violations = enforce_policy(vulns, policy)
if violations:
    print("Policy violations:")
    for v in violations:
        print(f"  - {v}")

from datetime import datetime
from pathlib import Path

def generate_scan_report(image: str, vulns: list[dict]) -> str:
    """Generate a Markdown scan report."""
    critical = [v for v in vulns if v["severity"] == "CRITICAL"]
    high = [v for v in vulns if v["severity"] == "HIGH"]
    medium = [v for v in vulns if v["severity"] == "MEDIUM"]

    report = [
        f"# Image Scan Report",
        f"",
        f"**Image:** {image}",
        f"**Scanned:** {datetime.utcnow().isoformat()}",
        f"**Total vulnerabilities:** {len(vulns)}",
        f"",
        f"## Summary",
        f"",
        f"| Severity | Count |",
        f"|----------|-------|",
        f"| Critical | {len(critical)} |",
        f"| High | {len(high)} |",
        f"| Medium | {len(medium)} |",
        f"",
    ]

    if critical:
        report.append("## Critical Vulnerabilities")
        report.append("")
        for v in critical:
            report.append(f"- **{v['id']}** ({v['package']}): {v['title']}")
            report.append(f"  - Installed: {v['installed']}, Fixed: {v['fixed']}")

    return "\n".join(report)

# Save report
report = generate_scan_report("myapp:latest", vulns)
Path("scan-report.md").write_text(report)