Implementing Container Signing with Cosign 2.0 and Kubernetes 1.32

package main

import (
    "context"
    "fmt"
    "log"
    "os"

    "github.com/sigstore/cosign/v2/cmd/cosign/cli/verify"
    "github.com/sigstore/cosign/v2/pkg/cosign"
    "github.com/sigstore/cosign/v2/pkg/cosign/fulcio"
    "github.com/sigstore/cosign/v2/pkg/cosign/rekor"
    "github.com/sigstore/cosign/v2/pkg/types"
    "github.com/sigstore/sigstore/pkg/signature/dsse"
)

// signImage signs an OCI image using Cosign 2.0 keyless OIDC flow
// imageRef: full OCI image reference (e.g., ghcr.io/owner/repo:tag)
// oidcIssuer: OIDC issuer URL (e.g., https://accounts.google.com for GCP)
// returns nil on success, error otherwise
func signImage(ctx context.Context, imageRef, oidcIssuer string) error {
    // Initialize OIDC credential provider
    credProvider, err := cosign.NewOIDCCredentialProvider(ctx, oidcIssuer)
    if err != nil {
        return fmt.Errorf("failed to initialize OIDC provider: %w", err)
    }

    // Get signing key from Fulcio (short-lived certificates)
    signer, err := fulcio.GetSigner(ctx, credProvider, fulcio.Options{})
    if err != nil {
        return fmt.Errorf("failed to get Fulcio signer: %w", err)
    }

    // Sign the image
    _, err = cosign.SignImage(ctx, signer, imageRef, cosign.SignOptions{
        RegistryOptions: cosign.RegistryOptions{
            AllowInsecure: false,
        },
        RekorURL: rekor.StagingURL, // Use rekor.sigstore.dev for production
    })
    if err != nil {
        return fmt.Errorf("failed to sign image %s: %w", imageRef, err)
    }

    log.Printf("Successfully signed image %s", imageRef)
    return nil
}

// verifyImage verifies an OCI image signature against a public key or Fulcio root
// imageRef: full OCI image reference
// publicKeyPath: path to Cosign public key (empty for keyless Fulcio verification)
func verifyImage(ctx context.Context, imageRef, publicKeyPath string) error {
    verifyOpts := verify.Options{
        RegistryOptions: cosign.RegistryOptions{
            AllowInsecure: false,
        },
        RekorURL: rekor.StagingURL,
    }

    // If public key is provided, use key-based verification
    if publicKeyPath != "" {
        pubKey, err := cosign.LoadPublicKey(ctx, publicKeyPath)
        if err != nil {
            return fmt.Errorf("failed to load public key: %w", err)
        }
        verifyOpts.PublicKey = pubKey
    } else {
        // Use Fulcio root CA for keyless verification
        fulcioRoots, err := fulcio.GetRoots()
        if err != nil {
            return fmt.Errorf("failed to get Fulcio roots: %w", err)
        }
        verifyOpts.RootCerts = fulcioRoots
    }

    // Execute verification
    _, err := verify.VerifyImage(ctx, imageRef, verifyOpts)
    if err != nil {
        return fmt.Errorf("image %s verification failed: %w", imageRef, err)
    }

    log.Printf("Successfully verified image %s", imageRef)
    return nil
}

func main() {
    ctx := context.Background()

    // Example usage: sign and verify a test image
    imageRef := "ghcr.io/example/myapp:v1.0.0"
    oidcIssuer := "https://accounts.google.com"
    publicKeyPath := "" // Empty for keyless verification

    // Sign the image
    if err := signImage(ctx, imageRef, oidcIssuer); err != nil {
        log.Fatalf("Signing failed: %v", err)
    }

    // Verify the image
    if err := verifyImage(ctx, imageRef, publicKeyPath); err != nil {
        log.Fatalf("Verification failed: %v", err)
    }

    // Example with static key pair (uncomment to use)
    // keyPairPath := "./cosign-key"
    // if err := generateKeyPair(keyPairPath); err != nil {
    //  log.Fatalf("Key generation failed: %v", err)
    // }
    // if err := signImageWithKey(ctx, imageRef, keyPairPath); err != nil {
    //  log.Fatalf("Key-based signing failed: %v", err)
    // }
}

// generateKeyPair generates a Cosign static key pair (private.key and public.key)
func generateKeyPair(outputPath string) error {
    _, privKey, err := cosign.GenerateKeyPair(nil) // nil uses default password prompt
    if err != nil {
        return fmt.Errorf("failed to generate key pair: %w", err)
    }

    // Write private key
    if err := os.WriteFile(outputPath+".key", privKey.PrivateBytes(), 0600); err != nil {
        return fmt.Errorf("failed to write private key: %w", err)
    }

    // Write public key
    if err := os.WriteFile(outputPath+".pub", privKey.PublicBytes(), 0644); err != nil {
        return fmt.Errorf("failed to write public key: %w", err)
    }

    log.Printf("Generated key pair at %s.key and %s.pub", outputPath, outputPath)
    return nil
}

import subprocess
import logging
import time
import sys
import json
from typing import List, Optional

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

# Cosign 2.0 CLI path (update if not in PATH)
COSIGN_CLI = "cosign"
# Max retries for transient Cosign errors
MAX_RETRIES = 3
# Retry delay in seconds
RETRY_DELAY = 5

def run_cosign_command(args: List[str], retries: int = MAX_RETRIES) -> subprocess.CompletedProcess:
    """Run a Cosign CLI command with retry logic for transient errors."""
    for attempt in range(retries):
        try:
            logger.info(f"Running cosign command: {' '.join([COSIGN_CLI] + args)}")
            result = subprocess.run(
                [COSIGN_CLI] + args,
                capture_output=True,
                text=True,
                check=True
            )
            logger.debug(f"Cosign output: {result.stdout}")
            return result
        except subprocess.CalledProcessError as e:
            logger.warning(f"Attempt {attempt + 1} failed: {e.stderr}")
            if attempt < retries - 1:
                time.sleep(RETRY_DELAY)
            else:
                raise RuntimeError(f"Cosign command failed after {retries} attempts: {e.stderr}") from e
        except FileNotFoundError:
            raise RuntimeError(f"Cosign CLI not found at {COSIGN_CLI}. Install Cosign 2.0 first.") from None

def sign_image(image_ref: str, key_path: Optional[str] = None, oidc_issuer: Optional[str] = None) -> bool:
    """
    Sign a single container image with Cosign 2.0.

    Args:
        image_ref: Full OCI image reference (e.g., ghcr.io/owner/repo:tag)
        key_path: Path to Cosign private key (None for keyless OIDC signing)
        oidc_issuer: OIDC issuer URL (required for keyless signing)

    Returns:
        True if signing succeeded, False otherwise
    """
    args = ["sign", "--yes"]

    if key_path:
        # Static key-based signing
        args += ["--key", key_path, image_ref]
    elif oidc_issuer:
        # Keyless OIDC signing
        args += ["--oidc-issuer", oidc_issuer, image_ref]
    else:
        logger.error("Either key_path or oidc_issuer must be provided for signing")
        return False

    try:
        run_cosign_command(args)
        logger.info(f"Successfully signed image: {image_ref}")
        return True
    except RuntimeError as e:
        logger.error(f"Failed to sign image {image_ref}: {e}")
        return False

def verify_image(image_ref: str, public_key_path: Optional[str] = None, oidc_issuer: Optional[str] = None) -> bool:
    """
    Verify a container image signature with Cosign 2.0.

    Args:
        image_ref: Full OCI image reference
        public_key_path: Path to Cosign public key (None for keyless verification)
        oidc_issuer: OIDC issuer URL (required for keyless verification)

    Returns:
        True if verification succeeded, False otherwise
    """
    args = ["verify"]

    if public_key_path:
        args += ["--key", public_key_path, image_ref]
    elif oidc_issuer:
        args += ["--oidc-issuer", oidc_issuer, image_ref]
    else:
        logger.error("Either public_key_path or oidc_issuer must be provided for verification")
        return False

    try:
        run_cosign_command(args)
        logger.info(f"Successfully verified image: {image_ref}")
        return True
    except RuntimeError as e:
        logger.error(f"Failed to verify image {image_ref}: {e}")
        return False

def batch_sign_registry(registry_prefix: str, tag_filter: str = "v*", key_path: Optional[str] = None) -> dict:
    """
    Batch sign all images in a registry matching a tag filter.

    Args:
        registry_prefix: Registry prefix (e.g., ghcr.io/example)
        tag_filter: Glob pattern for tags to sign
        key_path: Path to Cosign private key (None for keyless)

    Returns:
        Dict with success/failure counts and failed images
    """
    # List all images in the registry (using cosign images command)
    args = ["images", registry_prefix, "--filter-tag", tag_filter, "--output", "json"]
    try:
        result = run_cosign_command(args)
        images = json.loads(result.stdout)
    except RuntimeError as e:
        logger.error(f"Failed to list images in registry {registry_prefix}: {e}")
        return {"success": 0, "failure": 0, "failed": []}

    # Sign each image
    success = 0
    failure = 0
    failed = []
    for img in images:
        image_ref = img["image"]
        if sign_image(image_ref, key_path):
            success += 1
        else:
            failure += 1
            failed.append(image_ref)

    return {"success": success, "failure": failure, "failed": failed}

if __name__ == "__main__":
    # Example usage
    # Sign a single image with keyless OIDC (requires OIDC login via cosign login)
    # Uncomment to run:
    # sign_image(
    #     image_ref="ghcr.io/example/myapp:v1.0.0",
    #     oidc_issuer="https://accounts.google.com"
    # )

    # Batch sign all v1 tags in a registry
    # Uncomment to run:
    # results = batch_sign_registry(
    #     registry_prefix="ghcr.io/example",
    #     tag_filter="v1*"
    # )
    # logger.info(f"Batch sign results: {results}")

    # Verify an image
    # Uncomment to run:
    # verify_image(
    #     image_ref="ghcr.io/example/myapp:v1.0.0",
    #     oidc_issuer="https://accounts.google.com"
    # )

    logger.info("Example script completed. Uncomment sections to run real operations.")

#!/bin/bash
#
# Install and configure Cosign 2.0 with Kubernetes 1.32 admission controller
# Requires: kubectl 1.32+, cosign 2.0+, cluster admin access
set -euo pipefail  # Exit on error, undefined vars, pipe failures
trap 'echo "Script failed at line $LINENO"; exit 1' ERR

# Configuration variables
COSIGN_VERSION="2.0.3"  # Latest Cosign 2.0 stable as of 2024
K8S_VERSION="1.32.0"
NAMESPACE="cosign-system"
SIGNING_KEY_NAME="cosign-signing-key"
IMAGE_TO_SIGN="ghcr.io/example/myapp:unsigned"
SIGNED_IMAGE="ghcr.io/example/myapp:signed"

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

# Check prerequisites
check_prerequisites() {
    log "Checking prerequisites..."

    # Check kubectl version
    kubectl_version=$(kubectl version --client -o json | jq -r '.clientVersion.gitVersion' | sed 's/v//')
    if [[ "$kubectl_version" < "1.32.0" ]]; then
        log "ERROR: kubectl version must be 1.32.0 or higher. Found: $kubectl_version"
        exit 1
    fi

    # Check cosign version
    if ! command -v cosign &> /dev/null; then
        log "ERROR: cosign CLI not found. Install Cosign 2.0 first."
        exit 1
    fi
    cosign_version=$(cosign version | head -1 | awk '{print $2}' | sed 's/v//')
    if [[ "$cosign_version" < "2.0.0" ]]; then
        log "ERROR: cosign version must be 2.0.0 or higher. Found: $cosign_version"
        exit 1
    fi

    # Check cluster access
    if ! kubectl auth can-i create clusterroles; then
        log "ERROR: No cluster admin access. Please use a kubeconfig with admin privileges."
        exit 1
    fi

    log "Prerequisites satisfied."
}

# Install Cosign 2.0 admission controller in Kubernetes 1.32
install_cosign_admission() {
    log "Installing Cosign 2.0 admission controller..."

    # Create namespace
    kubectl create namespace "$NAMESPACE" --dry-run=client -o yaml | kubectl apply -f -

    # Enable Cosign admission controller alpha feature in K8s 1.32 (if not already enabled)
    # Note: Requires K8s API server flag --feature-gates=CosignAdmission=true
    log "Verifying Cosign admission feature gate is enabled..."
    if ! kubectl api-resources | grep -q "cosignadmissionpolicies"; then
        log "ERROR: CosignAdmission feature gate not enabled. Add --feature-gates=CosignAdmission=true to kube-apiserver flags."
        exit 1
    fi

    # Create Cosign signing key pair
    log "Generating Cosign signing key pair..."
    cosign generate-key-pair --k8s-namespace "$NAMESPACE" --k8s-secret "$SIGNING_KEY_NAME"

    # Create CosignAdmissionPolicy to enforce signing for all deployments
    log "Creating CosignAdmissionPolicy..."
    cat <