War Story: We Implemented SBOMs with Syft 0.10 and Cut Compliance Audit Time 60% for 500 Services

#!/usr/bin/env python3
"""
Production SBOM generation pipeline for 500+ microservices.
Uses Syft 0.10.0 to scan container images, generate CycloneDX SBOMs,
and upload to S3 for compliance audit access.
"""

import subprocess
import json
import os
import boto3
from botocore.exceptions import ClientError, NoCredentialsError
import logging
from typing import List, Dict, Optional

# Configure logging for audit trails
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("sbom_pipeline.log"), logging.StreamHandler()],
)
logger = logging.getLogger(__name__)

# Configuration (loaded from env vars for 12-factor compliance)
SYFT_VERSION = "0.10.0"
S3_BUCKET = os.environ.get("SBOM_S3_BUCKET", "prod-sbom-registry")
S3_PREFIX = os.environ.get("SBOM_S3_PREFIX", "cyclonedx/v1")
SUPPORTED_IMAGE_REGISTRIES = ["docker.io", "quay.io", "our-internal-registry.io"]

def verify_syft_installation() -> bool:
    """Check if Syft 0.10.0 is installed and accessible."""
    try:
        result = subprocess.run(
            ["syft", "version"],
            capture_output=True,
            text=True,
            check=True
        )
        installed_version = result.stdout.strip().split(" ")[0]
        if installed_version != SYFT_VERSION:
            logger.error(f"Syft version mismatch: expected {SYFT_VERSION}, got {installed_version}")
            return False
        logger.info(f"Verified Syft installation: {installed_version}")
        return True
    except FileNotFoundError:
        logger.error("Syft not found in PATH. Install from https://github.com/anchore/syft")
        return False
    except subprocess.CalledProcessError as e:
        logger.error(f"Syft version check failed: {e.stderr}")
        return False

def scan_image(image_uri: str) -> Optional[Dict]:
    """
    Scan a container image with Syft 0.10.0, return CycloneDX SBOM as dict.
    Args:
        image_uri: Full OCI image URI (e.g., our-internal-registry.io/order-service:1.2.3)
    Returns:
        Parsed CycloneDX SBOM dict, or None if scan fails
    """
    if not any(reg in image_uri for reg in SUPPORTED_IMAGE_REGISTRIES):
        logger.warning(f"Skipping unsupported registry image: {image_uri}")
        return None

    try:
        # Syft 0.10.0 supports -o cyclonedx-json for deterministic output
        result = subprocess.run(
            ["syft", image_uri, "-o", "cyclonedx-json"],
            capture_output=True,
            text=True,
            check=True,
            timeout=300  # 5 minute timeout per large image
        )
        sbom = json.loads(result.stdout)
        logger.info(f"Scanned {image_uri}: found {len(sbom.get('components', []))} components")
        return sbom
    except subprocess.TimeoutExpired:
        logger.error(f"Scan timeout for {image_uri}")
        return None
    except subprocess.CalledProcessError as e:
        logger.error(f"Syft scan failed for {image_uri}: {e.stderr}")
        return None
    except json.JSONDecodeError:
        logger.error(f"Invalid JSON output from Syft for {image_uri}")
        return None

def upload_sbom_to_s3(sbom: Dict, service_name: str, image_tag: str) -> bool:
    """Upload SBOM to S3 with standardized naming."""
    s3_key = f"{S3_PREFIX}/{service_name}/{image_tag}/sbom.json"
    s3_client = boto3.client("s3")
    try:
        s3_client.put_object(
            Bucket=S3_BUCKET,
            Key=s3_key,
            Body=json.dumps(sbom, indent=2),
            ContentType="application/json",
            Metadata={
                "service-name": service_name,
                "image-tag": image_tag,
                "syft-version": SYFT_VERSION
            }
        )
        logger.info(f"Uploaded SBOM to s3://{S3_BUCKET}/{s3_key}")
        return True
    except NoCredentialsError:
        logger.error("AWS credentials not found for S3 upload")
        return False
    except ClientError as e:
        logger.error(f"S3 upload failed for {service_name}: {e.response['Error']['Message']}")
        return False

def main(service_manifest: str) -> None:
    """Main pipeline entrypoint: reads service manifest, scans all images."""
    if not verify_syft_installation():
        raise RuntimeError("Invalid Syft installation")

    # Load service manifest (list of image URIs)
    try:
        with open(service_manifest, "r") as f:
            services = json.load(f)
    except FileNotFoundError:
        logger.error(f"Service manifest not found: {service_manifest}")
        return
    except json.JSONDecodeError:
        logger.error(f"Invalid JSON in service manifest: {service_manifest}")
        return

    logger.info(f"Starting SBOM generation for {len(services)} services")
    success_count = 0
    fail_count = 0

    for service in services:
        image_uri = service.get("image_uri")
        service_name = service.get("name")
        if not image_uri or not service_name:
            logger.warning("Skipping service with missing image_uri or name")
            fail_count +=1
            continue

        sbom = scan_image(image_uri)
        if not sbom:
            fail_count +=1
            continue

        # Extract image tag from URI for S3 key
        image_tag = image_uri.split(":")[-1] if ":" in image_uri else "latest"
        if upload_sbom_to_s3(sbom, service_name, image_tag):
            success_count +=1
        else:
            fail_count +=1

    logger.info(f"Pipeline complete: {success_count} successes, {fail_count} failures")

if __name__ == "__main__":
    import sys
    if len(sys.argv) != 2:
        print(f"Usage: {sys.argv[0]} ")
        sys.exit(1)
    main(sys.argv[1])

#!/usr/bin/env python3
"""
SBOM compliance validation script for audit prep.
Checks CycloneDX SBOMs against internal compliance rules:
1. All components must have a declared license
2. No components with critical CVEs (via Grype 0.10.0 integration)
3. SBOM must be generated with Syft 0.10.0+
4. All top-level dependencies must be pinned to exact versions
"""

import json
import os
import subprocess
import logging
from typing import List, Dict, Optional, Tuple
from botocore.exceptions import ClientError
import boto3

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

# Compliance configuration
ALLOWED_LICENSES = ["Apache-2.0", "MIT", "BSD-3-Clause", "ISC"]
CRITICAL_CVE_THRESHOLD = 0  # Zero tolerance for critical CVEs
GRYPE_VERSION = "0.10.0"

def verify_grype_installation() -> bool:
    """Check Grype installation for CVE scanning."""
    try:
        result = subprocess.run(
            ["grype", "version"],
            capture_output=True,
            text=True,
            check=True
        )
        installed = result.stdout.strip().split(" ")[0]
        if installed != GRYPE_VERSION:
            logger.error(f"Grype version mismatch: expected {GRYPE_VERSION}, got {installed}")
            return False
        return True
    except FileNotFoundError:
        logger.error("Grype not found. Install from https://github.com/anchore/grype")
        return False
    except subprocess.CalledProcessError as e:
        logger.error(f"Grype check failed: {e.stderr}")
        return False

def download_sbom_from_s3(service_name: str, image_tag: str) -> Optional[Dict]:
    """Fetch SBOM from S3 registry."""
    s3_bucket = os.environ.get("SBOM_S3_BUCKET", "prod-sbom-registry")
    s3_key = f"cyclonedx/v1/{service_name}/{image_tag}/sbom.json"
    try:
        s3_client = boto3.client("s3")
        response = s3_client.get_object(Bucket=s3_bucket, Key=s3_key)
        sbom = json.loads(response["Body"].read().decode("utf-8"))
        logger.info(f"Downloaded SBOM for {service_name}:{image_tag}")
        return sbom
    except ClientError as e:
        if e.response["Error"]["Code"] == "NoSuchKey":
            logger.error(f"SBOM not found for {service_name}:{image_tag}")
        else:
            logger.error(f"S3 download failed: {e.response['Error']['Message']}")
        return None

def validate_licenses(sbom: Dict) -> Tuple[bool, List[str]]:
    """Check all components have allowed licenses."""
    violations = []
    for component in sbom.get("components", []):
        licenses = component.get("licenses", [])
        if not licenses:
            violations.append(f"Component {component.get('name')} has no license declared")
            continue
        # Extract license IDs (handle CycloneDX license objects)
        license_ids = []
        for lic in licenses:
            if "license" in lic:
                license_ids.append(lic["license"].get("id", ""))
            else:
                license_ids.append(lic.get("id", ""))
        if not any(lic_id in ALLOWED_LICENSES for lic_id in license_ids):
            violations.append(f"Component {component.get('name')} has unapproved license: {license_ids}")
    return (len(violations) == 0, violations)

def check_cves(image_uri: str) -> Tuple[bool, List[str]]:
    """Scan image for critical CVEs using Grype."""
    try:
        result = subprocess.run(
            ["grype", image_uri, "-o", "json"],
            capture_output=True,
            text=True,
            check=True,
            timeout=300
        )
        scan_result = json.loads(result.stdout)
        critical_cves = [
            vuln for vuln in scan_result.get("matches", [])
            if vuln.get("vulnerability", {}).get("severity", "").lower() == "critical"
        ]
        if len(critical_cves) > CRITICAL_CVE_THRESHOLD:
            return (False, [f"Critical CVE found: {cve['vulnerability']['id']}" for cve in critical_cves])
        return (True, [])
    except subprocess.CalledProcessError as e:
        logger.error(f"Grype scan failed for {image_uri}: {e.stderr}")
        return (False, [f"Grype scan failed: {e.stderr}"])
    except subprocess.TimeoutExpired:
        return (False, ["Grype scan timeout"])

def validate_sbom_metadata(sbom: Dict) -> Tuple[bool, List[str]]:
    """Check SBOM metadata (Syft version, format version)."""
    violations = []
    # Check CycloneDX version
    if sbom.get("bomFormat") != "CycloneDX":
        violations.append("SBOM is not CycloneDX format")
    if sbom.get("specVersion") != "1.4":
        violations.append(f"Unsupported CycloneDX version: {sbom.get('specVersion')}")
    # Check Syft version in metadata
    syft_version = sbom.get("metadata", {}).get("tools", {}).get("components", [{}])[0].get("version", "")
    if not syft_version.startswith("0.10."):
        violations.append(f"SBOM not generated with Syft 0.10.x: {syft_version}")
    return (len(violations) == 0, violations)

def main(service_manifest: str) -> None:
    """Run compliance validation on all services."""
    if not verify_grype_installation():
        raise RuntimeError("Invalid Grype installation")

    # Load service manifest
    try:
        with open(service_manifest, "r") as f:
            services = json.load(f)
    except Exception as e:
        logger.error(f"Failed to load service manifest: {e}")
        return

    logger.info(f"Validating SBOMs for {len(services)} services")
    compliant = 0
    non_compliant = 0
    all_violations = []

    for service in services:
        service_name = service.get("name")
        image_uri = service.get("image_uri")
        image_tag = image_uri.split(":")[-1] if ":" in image_uri else "latest"

        sbom = download_sbom_from_s3(service_name, image_tag)
        if not sbom:
            non_compliant +=1
            all_violations.append(f"{service_name}: SBOM not found")
            continue

        # Run all validation checks
        checks = [
            validate_sbom_metadata(sbom),
            validate_licenses(sbom),
            check_cves(image_uri)
        ]

        service_violations = []
        for passed, violations in checks:
            if not passed:
                service_violations.extend(violations)

        if service_violations:
            non_compliant +=1
            all_violations.extend([f"{service_name}: {v}" for v in service_violations])
            logger.warning(f"{service_name} has {len(service_violations)} violations")
        else:
            compliant +=1
            logger.info(f"{service_name} is compliant")

    # Generate audit report
    report = {
        "total_services": len(services),
        "compliant": compliant,
        "non_compliant": non_compliant,
        "violations": all_violations,
        "compliance_rate": f"{(compliant/len(services))*100:.1f}%"
    }
    with open("compliance_audit_report.json", "w") as f:
        json.dump(report, f, indent=2)
    logger.info(f"Audit report generated: {json.dumps(report, indent=2)}")

if __name__ == "__main__":
    import sys
    if len(sys.argv) !=2:
        print(f"Usage: {sys.argv[0]} ")
        sys.exit(1)
    main(sys.argv[1])

#!/bin/bash
#
# Orchestration script for nightly SBOM pipeline run.
# Triggers SBOM generation, validation, and audit report upload.
# Exits with non-zero code if any critical step fails.
#

set -euo pipefail  # Exit on error, undefined var, pipe failure

# Configuration
SERVICE_MANIFEST="service_manifest.json"
SBOM_GENERATOR="generate_sboms.py"
SBOM_VALIDATOR="validate_sboms.py"
AUDIT_REPORT_BUCKET="s3://prod-sbom-registry/audit-reports"
SLACK_WEBHOOK_URL="${SLACK_WEBHOOK_URL:-}"  # Optional Slack alerting
MAX_RETRIES=3
RETRY_DELAY=10

# Logging function
log() {
    echo "[$(date +'%Y-%m-%dT%H:%M:%S%z')] $1"
}

# Error handling function
handle_error() {
    local exit_code=$?
    log "ERROR: Pipeline failed at line $1 with exit code $exit_code"
    if [ -n "$SLACK_WEBHOOK_URL" ]; then
        curl -X POST -H 'Content-type: application/json' \
            --data "{"text":"SBOM Pipeline Failed: Exit code $exit_code"}" \
            "$SLACK_WEBHOOK_URL" || true  # Don't fail on Slack error
    fi
    exit $exit_code
}

trap 'handle_error $LINENO' ERR

# Verify dependencies
verify_dependencies() {
    log "Verifying dependencies..."
    local deps=("python3" "syft" "grype" "aws" "jq")
    for dep in "${deps[@]}"; do
        if ! command -v "$dep" &> /dev/null; then
            log "ERROR: Dependency $dep not found"
            exit 1
        fi
    done
    # Verify Syft version
    local syft_version
    syft_version=$(syft version | awk '{print $1}')
    if [[ "$syft_version" != "0.10.0" ]]; then
        log "ERROR: Syft version mismatch. Expected 0.10.0, got $syft_version"
        exit 1
    fi
    log "All dependencies verified"
}

# Retry function for flaky operations
retry() {
    local retries=$1
    shift
    local count=0
    until "$@"; do
        exit_code=$?
        count=$((count +1))
        if [ $count -lt $retries ]; then
            log "Retrying $count/$retries after $RETRY_DELAY seconds..."
            sleep $RETRY_DELAY
        else
            log "ERROR: Command failed after $retries retries"
            return $exit_code
        fi
    done
    return 0
}

# Generate service manifest from Kubernetes API
generate_service_manifest() {
    log "Generating service manifest from K8s API..."
    kubectl get pods -A -o json | jq -r '
        .items[] |
        select(.metadata.labels.app != null) |
        {
            name: .metadata.labels.app,
            image_uri: .spec.containers[0].image,
            namespace: .metadata.namespace
        }
    ' | jq -s '.' > "$SERVICE_MANIFEST"
    log "Generated service manifest with $(jq '. | length' "$SERVICE_MANIFEST") services"
}

# Run SBOM generation
run_sbom_generation() {
    log "Starting SBOM generation..."
    retry $MAX_RETRIES python3 "$SBOM_GENERATOR" "$SERVICE_MANIFEST"
    log "SBOM generation complete"
}

# Run compliance validation
run_validation() {
    log "Starting compliance validation..."
    retry $MAX_RETRIES python3 "$SBOM_VALIDATOR" "$SERVICE_MANIFEST"
    log "Compliance validation complete"
}

# Upload audit report to S3
upload_audit_report() {
    log "Uploading audit report to S3..."
    local report_file="compliance_audit_report.json"
    if [ ! -f "$report_file" ]; then
        log "ERROR: Audit report not found"
        return 1
    fi
    local timestamp
    timestamp=$(date +%Y%m%d_%H%M%S)
    retry $MAX_RETRIES aws s3 cp "$report_file" "${AUDIT_REPORT_BUCKET}/${timestamp}_audit_report.json"
    log "Audit report uploaded to ${AUDIT_REPORT_BUCKET}/${timestamp}_audit_report.json"
}

# Send success alert
send_success_alert() {
    if [ -n "$SLACK_WEBHOOK_URL" ]; then
        local compliant
        compliant=$(jq -r '.compliant' compliance_audit_report.json)
        local total
        total=$(jq -r '.total_services' compliance_audit_report.json)
        curl -X POST -H 'Content-type: application/json' \
            --data "{"text":"SBOM Pipeline Succeeded: $compliant/$total services compliant"}" \
            "$SLACK_WEBHOOK_URL" || true
    fi
}

# Main pipeline flow
main() {
    log "Starting nightly SBOM pipeline"
    verify_dependencies
    generate_service_manifest
    run_sbom_generation
    run_validation
    upload_audit_report
    send_success_alert
    log "Pipeline completed successfully"
}

main

# GitHub Actions step to pin Syft 0.10.0
- name: Install Syft 0.10.0
  uses: anchore/syft-action@v0.10.0
  with:
    version: 0.10.0  # Explicit patch pin
    debug: false

# Deterministic Syft scan command
syft our-internal-registry.io/order-service:1.2.3 \
  --source-name order-service \
  --source-version 1.2.3 \
  --exclude '**/*test*' \
  -o cyclonedx-json > order-service-1.2.3-sbom.json

# PR check step to validate SBOM Licenses
- name: Validate PR SBOM Licenses
  run: |
    syft ${{ steps.build-image.outputs.image-uri }} -o cyclonedx-json > pr-sbom.json
    jq -r '.components[].licenses[]?.license?.id' pr-sbom.json | \
    grep -v -E 'Apache-2.0|MIT|BSD-3-Clause' && exit 1 || echo "All licenses approved"