Opinion: Compliance Is Dead – Use Automated Tools Like Kyverno 1.12 for DevSecOps

package main

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

    // Import Kyverno 1.12 policy engine packages
    // Canonical repo: https://github.com/kyverno/kyverno
    "github.com/kyverno/kyverno/pkg/policy"
    "github.com/kyverno/kyverno/pkg/policy/validation"
    "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
    "k8s.io/client-go/kubernetes/scheme"
    "sigs.k8s.io/yaml"
)

// validateKyvernoPolicy loads a Kyverno ClusterPolicy from a file, validates it against 1.12 schema
// Returns an error if the policy is invalid or fails schema checks
func validateKyvernoPolicy(policyPath string) error {
    // Read policy file
    data, err := os.ReadFile(policyPath)
    if err != nil {
        return fmt.Errorf("failed to read policy file %s: %w", policyPath, err)
    }

    // Decode YAML into unstructured object first to check type
    obj := &unstructured.Unstructured{}
    if err := yaml.Unmarshal(data, obj); err != nil {
        return fmt.Errorf("failed to unmarshal policy YAML: %w", err)
    }

    // Check if it's a ClusterPolicy
    if obj.GetAPIVersion() != "kyverno.io/v1" || obj.GetKind() != "ClusterPolicy" {
        return fmt.Errorf("invalid policy kind: expected ClusterPolicy, got %s/%s", obj.GetAPIVersion(), obj.GetKind())
    }

    // Convert to Kyverno Policy object
    policyObj, err := policy.NewFromUnstructured(obj)
    if err != nil {
        return fmt.Errorf("failed to convert to Kyverno policy: %w", err)
    }

    // Validate policy against Kyverno 1.12 schema and rules
    ctx := context.Background()
    if err := validation.ValidatePolicy(ctx, policyObj, nil); err != nil {
        return fmt.Errorf("policy validation failed: %w", err)
    }

    // Check for deprecated fields in 1.12
    if err := checkDeprecatedFields(policyObj); err != nil {
        return fmt.Errorf("deprecated field usage: %w", err)
    }

    fmt.Printf("Policy %s validated successfully against Kyverno 1.12 schema\n", policyObj.GetName())
    return nil
}

// checkDeprecatedFields returns an error if the policy uses fields deprecated in Kyverno 1.12
func checkDeprecatedFields(p policy.Policy) error {
    // Kyverno 1.12 deprecates `spec.validationFailureActionOverride` in favor of `spec.failureAction`
    for _, rule := range p.GetSpec().Rules {
        if rule.ValidationFailureActionOverride != "" {
            return fmt.Errorf("rule %s uses deprecated validationFailureActionOverride, use failureAction instead", rule.Name)
        }
    }
    return nil
}

func main() {
    if len(os.Args) < 2 {
        log.Fatal("Usage: policy-validator ")
    }

    policyPath := os.Args[1]
    if err := validateKyvernoPolicy(policyPath); err != nil {
        log.Fatalf("Policy validation failed: %v", err)
    }
}

#!/usr/bin/env python3
"""
Kyverno 1.12 Policy Test Runner
Requires: kyverno CLI v1.12+ installed (https://github.com/kyverno/kyverno/releases)
"""

import subprocess
import sys
import json
import os
from pathlib import Path

def run_kyverno_test(policy_dir: str, resource_dir: str) -> dict:
    """
    Runs `kyverno test` against policies in policy_dir and resources in resource_dir
    Returns parsed test results or raises an exception on failure
    """
    # Check if kyverno CLI is installed
    try:
        subprocess.run(["kyverno", "version"], capture_output=True, check=True)
    except FileNotFoundError:
        raise RuntimeError("Kyverno CLI not found. Install v1.12+ from https://github.com/kyverno/kyverno/releases")

    # Validate input directories exist
    policy_path = Path(policy_dir)
    resource_path = Path(resource_dir)
    if not policy_path.exists() or not policy_path.is_dir():
        raise FileNotFoundError(f"Policy directory {policy_dir} does not exist or is not a directory")
    if not resource_path.exists() or not resource_path.is_dir():
        raise FileNotFoundError(f"Resource directory {resource_dir} does not exist or is not a directory")

    # Run kyverno test with JSON output for parsing
    cmd = [
        "kyverno", "test",
        str(policy_path),
        str(resource_path),
        "--output", "json"
    ]

    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            check=True  # Raises CalledProcessError if return code != 0
        )
    except subprocess.CalledProcessError as e:
        # Kyverno test returns non-zero if tests fail, so parse output even on error
        if e.stdout:
            return json.loads(e.stdout)
        raise RuntimeError(f"Kyverno test failed: {e.stderr}") from e

    # Parse JSON output
    try:
        test_results = json.loads(result.stdout)
    except json.JSONDecodeError as e:
        raise RuntimeError(f"Failed to parse kyverno test output: {e}") from e

    # Print summary
    passed = test_results.get("passed", 0)
    failed = test_results.get("failed", 0)
    total = passed + failed
    print(f"Kyverno Test Results: {passed}/{total} passed, {failed} failed")

    # Raise error if any tests failed
    if failed > 0:
        raise AssertionError(f"{failed} policy tests failed")

    return test_results

def generate_test_resources(resource_type: str, count: int) -> list:
    """
    Generates sample Kubernetes resources for testing policies
    """
    resources = []
    for i in range(count):
        if resource_type == "Pod":
            resources.append({
                "apiVersion": "v1",
                "kind": "Pod",
                "metadata": {"name": f"test-pod-{i}", "namespace": "default"},
                "spec": {
                    "containers": [{
                        "name": "app",
                        "image": "nginx:1.25",
                        "ports": [{"containerPort": 80}]
                    }]
                }
            })
        elif resource_type == "Deployment":
            resources.append({
                "apiVersion": "apps/v1",
                "kind": "Deployment",
                "metadata": {"name": f"test-deploy-{i}", "namespace": "default"},
                "spec": {
                    "replicas": 1,
                    "selector": {"matchLabels": {"app": f"test-{i}"}},
                    "template": {
                        "metadata": {"labels": {"app": f"test-{i}"}},
                        "spec": {
                            "containers": [{"name": "app", "image": "nginx:1.25"}]
                        }
                    }
                }
            })
    return resources

if __name__ == "__main__":
    if len(sys.argv) != 3:
        print(f"Usage: {sys.argv[0]}  ", file=sys.stderr)
        sys.exit(1)

    policy_dir = sys.argv[1]
    resource_dir = sys.argv[2]

    try:
        results = run_kyverno_test(policy_dir, resource_dir)
        # Save results to file
        with open("kyverno-test-results.json", "w") as f:
            json.dump(results, f, indent=2)
        print("Test results saved to kyverno-test-results.json")
    except Exception as e:
        print(f"Test run failed: {e}", file=sys.stderr)
        sys.exit(1)

#!/bin/bash
#
# Kyverno 1.12 Compliance Automation Script
# Deploys policies, verifies image attestations, generates audit reports
# Requires: kubectl, kyverno CLI v1.12+, cosign v2+
#

set -euo pipefail  # Exit on error, undefined vars, pipe failures

# Configuration
KYVERNO_VERSION="1.12.0"
POLICY_REPO="https://github.com/kyverno/policies.git"
POLICY_DIR="kyverno-policies"
AUDIT_REPORT_DIR="./compliance-reports"
CLUSTER_NAME="prod-cluster-01"

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

# Error handling function
error_exit() {
    log "ERROR: $1" >&2
    exit 1
}

# Check prerequisites
check_prerequisites() {
    log "Checking prerequisites..."
    for cmd in kubectl kyverno cosign git; do
        if ! command -v "$cmd" &> /dev/null; then
            error_exit "Command $cmd not found. Please install it before running."
        fi
    done

    # Check kyverno version
    INSTALLED_KYVERNO=$(kyverno version | grep -oP 'v\K[0-9]+\.[0-9]+\.[0-9]+')
    if [[ "$INSTALLED_KYVERNO" < "$KYVERNO_VERSION" ]]; then
        error_exit "Kyverno version $INSTALLED_KYVERNO is installed, requires $KYVERNO_VERSION+"
    fi

    # Check kubectl connectivity
    if ! kubectl cluster-info &> /dev/null; then
        error_exit "Cannot connect to Kubernetes cluster. Check kubeconfig."
    fi
    log "Prerequisites satisfied."
}

# Clone Kyverno policy repository
clone_policies() {
    log "Cloning Kyverno policy repository..."
    if [[ -d "$POLICY_DIR" ]]; then
        log "Policy directory exists, pulling latest changes..."
        git -C "$POLICY_DIR" pull || error_exit "Failed to pull policy repo updates"
    else
        git clone "$POLICY_REPO" "$POLICY_DIR" || error_exit "Failed to clone policy repo"
    fi
    log "Policies cloned to $POLICY_DIR"
}

# Deploy Kyverno policies to cluster
deploy_policies() {
    log "Deploying Kyverno 1.12 policies to cluster $CLUSTER_NAME..."
    # Apply all production-ready policies from the repo
    find "$POLICY_DIR" -name "*.yaml" -path "*/production/*" | while read -r policy_file; do
        log "Applying policy: $policy_file"
        kubectl apply -f "$policy_file" || error_exit "Failed to apply policy $policy_file"
    done
    log "All policies deployed."
}

# Verify image attestations using Kyverno 1.12's native SLSA support
verify_attestations() {
    log "Verifying container image attestations..."
    # Get all running pods
    pods=$(kubectl get pods -A -o jsonpath='{range.items[*]}{.spec.containers[*].image}{"\n"}{end}' | sort -u)
    while IFS= read -r image; do
        if [[ -n "$image" ]]; then
            log "Verifying attestation for $image"
            # Use kyverno 1.12's attest command
            if ! kyverno attest "$image" --type slsa --level 3; then
                log "WARNING: No valid SLSA 3 attestation found for $image"
                # Add to non-compliant list
                echo "$image" >> "$AUDIT_REPORT_DIR/non-compliant-images.txt"
            fi
        fi
    done <<< "$pods"
    log "Attestation verification complete."
}

# Generate compliance audit report
generate_audit_report() {
    log "Generating compliance audit report..."
    mkdir -p "$AUDIT_REPORT_DIR"

    # Get all ClusterPolicies and their status
    kubectl get clusterpolicy -o json > "$AUDIT_REPORT_DIR/policies.json"

    # Get all policy violations from Kyverno
    kubectl get policyreport -A -o json > "$AUDIT_REPORT_DIR/policy-violations.json"

    # Generate summary
    total_policies=$(jq '.items | length' "$AUDIT_REPORT_DIR/policies.json")
    violated_policies=$(jq '[.items[] | select(.results.pass == false)] | length' "$AUDIT_REPORT_DIR/policy-violations.json")

    cat > "$AUDIT_REPORT_DIR/summary.txt" << EOF
Kyverno 1.12 Compliance Audit Report
Cluster: $CLUSTER_NAME
Date: $(date +'%Y-%m-%d')
Total Policies Deployed: $total_policies
Policy Violations: $violated_policies
Non-Compliant Images: $(wc -l < "$AUDIT_REPORT_DIR/non-compliant-images.txt" || echo 0)
EOF

    log "Audit report generated in $AUDIT_REPORT_DIR"
}

# Main execution
main() {
    log "Starting Kyverno 1.12 compliance automation for $CLUSTER_NAME"
    check_prerequisites
    clone_policies
    deploy_policies
    verify_attestations
    generate_audit_report
    log "Compliance automation complete. Report available at $AUDIT_REPORT_DIR/summary.txt"
}

main

# Kyverno 1.12 Ephemeral Container Policy Snippet
apiVersion: kyverno.io/v1
kind: ClusterPolicy
metadata:
  name: restrict-ephemeral-containers
spec:
  validationFailureAction: Enforce
  rules:
  - name: require-approved-debug-images
    match:
      any:
      - resources:
          kinds:
          - Pod
    validate:
      ephemeralContainers:
      - image: "gcr.io/distroless/debug:*"
        required: true
  - name: enforce-ephemeral-ttl
    match:
      any:
      - resources:
          kinds:
          - Pod
    validate:
      ephemeralContainers:
      - ttlSecondsAfterFinished: 3600

# Kyverno 1.12 SLSA 3 Attestation Policy Snippet
apiVersion: kyverno.io/v1
kind: ClusterPolicy
metadata:
  name: require-slsa-3-attestations
spec:
  validationFailureAction: Enforce
  rules:
  - name: verify-slsa-3
    match:
      any:
      - resources:
          kinds:
          - Pod
    verifyImages:
    - imageReferences:
      - "*"
      attestations:
      - type: slsa-provenance
        conditions:
        - all:
          - key: "level"
            operator: Equals
            value: "3"
      required: true

# Query Kyverno PolicyReports via kubectl
kubectl get policyreport -A \
  -o jsonpath='{range .items[*]}{.metadata.namespace}{"\t"}{.metadata.name}{"\t"}{.summary.pass}{"\t"}{.summary.fail}{"\n"}{end}' \
  > compliance-violations.csv