Benchmark the supply chain of Vault and Sigstore: What Fails

import os
import time
import hvac
import random
import string
from concurrent.futures import ThreadPoolExecutor, as_completed
from prometheus_client import start_http_server, Summary, Counter

# Prometheus metrics for benchmarking
VAULT_CERT_LATENCY = Summary('vault_pki_cert_latency_ms', 'Vault PKI certificate issuance latency in ms')
VAULT_CERT_FAILURES = Counter('vault_pki_cert_failures_total', 'Total Vault PKI cert issuance failures')
VAULT_CERT_SUCCESS = Counter('vault_pki_cert_success_total', 'Total Vault PKI cert issuance successes')

def generate_random_cn():
    """Generate random common name for test certificates"""
    return f'benchmark-{''.join(random.choices(string.ascii_lowercase + string.digits, k=8))}.example.com'

def issue_vault_cert(vault_client, pki_path, role_name, cn):
    """
    Issue a PKI certificate from Vault's PKI secrets engine.
    Args:
        vault_client: Authenticated hvac Client instance
        pki_path: Path to mounted PKI engine (e.g., 'pki')
        role_name: Vault PKI role to use for issuance
        cn: Common Name for the certificate
    Returns:
        Tuple of (latency_ms, success_bool, error_msg)
    """
    start_time = time.time()
    try:
        # Request certificate with 1 hour TTL for benchmark consistency
        response = vault_client.secrets.pki.generate_certificate(
            role_name=role_name,
            common_name=cn,
            ttl='1h',
            mount_point=pki_path
        )
        latency_ms = (time.time() - start_time) * 1000
        VAULT_CERT_LATENCY.observe(latency_ms)
        VAULT_CERT_SUCCESS.inc()
        # Validate response has required fields
        if 'data' not in response or 'certificate' not in response['data']:
            raise ValueError('Invalid Vault PKI response: missing certificate data')
        return (latency_ms, True, '')
    except Exception as e:
        latency_ms = (time.time() - start_time) * 1000
        VAULT_CERT_FAILURES.inc()
        return (latency_ms, False, str(e))

def run_vault_benchmark(vault_addr, vault_token, pki_path, role_name, concurrency, iterations):
    """
    Run Vault PKI benchmark with specified concurrency.
    Args:
        vault_addr: Vault server address (e.g., https://vault.example.com:8200)
        vault_token: Vault authentication token
        pki_path: Mounted PKI engine path
        role_name: PKI role name
        concurrency: Number of concurrent threads
        iterations: Total number of certificate requests
    """
    # Initialize Vault client with TLS verification disabled for benchmark (not for prod!)
    client = hvac.Client(url=vault_addr, token=vault_token, verify=False)
    if not client.is_authenticated():
        raise RuntimeError('Failed to authenticate to Vault')

    # Start Prometheus metrics server on port 8000
    start_http_server(8000)
    print(f'Prometheus metrics available at http://localhost:8000/metrics')

    results = []
    # Use ThreadPoolExecutor for concurrent requests
    with ThreadPoolExecutor(max_workers=concurrency) as executor:
        futures = []
        for _ in range(iterations):
            cn = generate_random_cn()
            futures.append(executor.submit(issue_vault_cert, client, pki_path, role_name, cn))
        for future in as_completed(futures):
            results.append(future.result())

    # Calculate summary statistics
    latencies = [r[0] for r in results if r[1]]
    failures = [r for r in results if not r[1]]
    print(f'\nBenchmark Results ({iterations} iterations, {concurrency} concurrency):')
    print(f'Success Rate: {len(latencies)/iterations*100:.2f}%')
    if latencies:
        print(f'Mean Latency: {sum(latencies)/len(latencies):.2f}ms')
        print(f'P99 Latency: {sorted(latencies)[int(len(latencies)*0.99)] if latencies else 0:.2f}ms')
    if failures:
        print(f'Failures: {len(failures)}')
        for fail in failures[:3]:  # Print first 3 failures for debugging
            print(f'  Error: {fail[2]}')

if __name__ == '__main__':
    # Configuration from environment variables for reproducibility
    VAULT_ADDR = os.getenv('VAULT_ADDR', 'https://127.0.0.1:8200')
    VAULT_TOKEN = os.getenv('VAULT_TOKEN')
    PKI_PATH = os.getenv('VAULT_PKI_PATH', 'pki')
    ROLE_NAME = os.getenv('VAULT_PKI_ROLE', 'benchmark-role')
    CONCURRENCY = int(os.getenv('BENCH_CONCURRENCY', '100'))
    ITERATIONS = int(os.getenv('BENCH_ITERATIONS', '1000'))

    if not VAULT_TOKEN:
        raise ValueError('VAULT_TOKEN environment variable must be set')

    print(f'Starting Vault PKI benchmark: {ITERATIONS} iterations, {CONCURRENCY} concurrency')
    run_vault_benchmark(VAULT_ADDR, VAULT_TOKEN, PKI_PATH, ROLE_NAME, CONCURRENCY, ITERATIONS)

package main

import (
    "context"
    "crypto/rand"
    "flag"
    "fmt"
    "log"
    "os"
    "sync"
    "sync/atomic"
    "time"

    "github.com/google/go-containerregistry/pkg/name"
    "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/sigstore/pkg/oauthflow"
)

// Benchmark metrics
var (
    totalRequests   uint64
    successRequests uint64
    failedRequests  uint64
    totalLatencyMs  uint64
    p99Latencies    []time.Duration
    mu              sync.Mutex
)

func init() {
    // Initialize p99 latency slice
    p99Latencies = make([]time.Duration, 0)
}

func generateRandomImageRef() string {
    // Generate random image reference for benchmarking
    b := make([]byte, 4)
    rand.Read(b)
    return fmt.Sprintf("example.com/benchmark-%x:latest", b)
}

func signImage(ctx context.Context, imageRef string, opts cosign.SignOpts) (time.Duration, error) {
    start := time.Now()

    // Parse image reference
    ref, err := name.ParseReference(imageRef)
    if err != nil {
        return time.Since(start), fmt.Errorf("failed to parse image ref: %w", err)
    }

    // Sign the image using Cosign with ephemeral Fulcio certificate
    _, err = cosign.SignImage(ctx, ref, opts)
    if err != nil {
        return time.Since(start), fmt.Errorf("failed to sign image: %w", err)
    }

    return time.Since(start), nil
}

func runWorker(ctx context.Context, wg *sync.WaitGroup, iterations int, opts cosign.SignOpts) {
    defer wg.Done()

    for i := 0; i < iterations; i++ {
        atomic.AddUint64(&totalRequests, 1)
        imageRef := generateRandomImageRef()

        latency, err := signImage(ctx, imageRef, opts)
        latencyMs := latency.Milliseconds()

        mu.Lock()
        p99Latencies = append(p99Latencies, latency)
        mu.Unlock()

        atomic.AddUint64(&totalLatencyMs, uint64(latencyMs))

        if err != nil {
            atomic.AddUint64(&failedRequests, 1)
            log.Printf("Signing failed for %s: %v", imageRef, err)
        } else {
            atomic.AddUint64(&successRequests, 1)
        }

        // Small jitter to avoid thundering herd
        time.Sleep(time.Millisecond * time.Duration(rand.Intn(10)))
    }
}

func calculateP99(latencies []time.Duration) time.Duration {
    if len(latencies) == 0 {
        return 0
    }
    // Sort latencies to find p99
    mu.Lock()
    defer mu.Unlock()
    n := len(latencies)
    // Simple selection sort for small slices, use sort.Slice for production
    for i := 0; i < n; i++ {
        for j := i + 1; j < n; j++ {
            if latencies[i] > latencies[j] {
                latencies[i], latencies[j] = latencies[j], latencies[i]
            }
        }
    }
    p99Idx := int(float64(n) * 0.99)
    if p99Idx >= n {
        p99Idx = n - 1
    }
    return latencies[p99Idx]
}

func main() {
    // Command line flags
    concurrency := flag.Int("concurrency", 100, "Number of concurrent workers")
    iterations := flag.Int("iterations", 1000, "Total number of signing iterations per worker")
    oidcIssuer := flag.String("oidc-issuer", "https://accounts.google.com", "OIDC issuer for Fulcio auth")
    flag.Parse()

    ctx := context.Background()

    // Initialize Cosign options with Fulcio and Rekor
    opts := cosign.SignOpts{
        Registry: name.Registry{},
        Fulcio: &fulcio.Identity{
            Issuer: *oidcIssuer,
            ClientID: "sigstore",
            // Use device flow for OIDC authentication (interactive for benchmark setup)
            Flow: oauthflow.NewDeviceFlowTokenGetter(""),
        },
        Rekor: rekor.NewClient("https://rekor.sigstore.dev"),
    }

    var wg sync.WaitGroup
    workerIterations := *iterations / *concurrency
    if workerIterations < 1 {
        workerIterations = 1
    }

    fmt.Printf("Starting Sigstore Cosign benchmark: %d workers, %d iterations per worker\n", *concurrency, workerIterations)

    startTime := time.Now()
    for i := 0; i < *concurrency; i++ {
        wg.Add(1)
        go runWorker(ctx, &wg, workerIterations, opts)
    }
    wg.Wait()
    totalDuration := time.Since(startTime)

    // Calculate results
    total := atomic.LoadUint64(&totalRequests)
    success := atomic.LoadUint64(&successRequests)
    failed := atomic.LoadUint64(&failedRequests)
    meanLatency := float64(atomic.LoadUint64(&totalLatencyMs)) / float64(success)
    p99 := calculateP99(p99Latencies)

    fmt.Printf("\nBenchmark Results:\n")
    fmt.Printf("Total Requests: %d\n", total)
    fmt.Printf("Success Rate: %.2f%%\n", float64(success)/float64(total)*100)
    fmt.Printf("Mean Latency: %.2fms\n", meanLatency)
    fmt.Printf("P99 Latency: %v\n", p99)
    fmt.Printf("Total Duration: %v\n", totalDuration)
}

import subprocess
import time
import random
import json
from dataclasses import dataclass
from typing import List, Dict

@dataclass
class FailureScenario:
    name: str
    tool: str
    inject_cmd: str
    cleanup_cmd: str
    description: str

class SupplyChainFailureInjector:
    """Injects controlled failures into Vault and Sigstore components to measure resilience"""

    def __init__(self, vault_addr: str, sigstore_fulcio_addr: str, sigstore_rekor_addr: str):
        self.vault_addr = vault_addr
        self.fulcio_addr = sigstore_fulcio_addr
        self.rekor_addr = sigstore_rekor_addr
        self.scenarios: List[FailureScenario] = self._init_scenarios()

    def _init_scenarios(self) -> List[FailureScenario]:
        """Initialize predefined failure scenarios for benchmarking"""
        return [
            FailureScenario(
                name='vault-etcd-leader-election',
                tool='Vault',
                inject_cmd='docker exec etcd etcdctl --endpoints=https://127.0.0.1:2379 leader resign',
                cleanup_cmd='echo "Leader election triggered, etcd will auto-recover"',
                description='Force etcd leader election to simulate Vault storage backend instability'
            ),
            FailureScenario(
                name='sigstore-fulcio-timeout',
                tool='Sigstore',
                inject_cmd='iptables -A INPUT -p tcp --dport 8080 -j DROP',
                cleanup_cmd='iptables -D INPUT -p tcp --dport 8080 -j DROP',
                description='Drop all traffic to Fulcio (port 8080) to simulate CA outage'
            ),
            FailureScenario(
                name='sigstore-oidc-rate-limit',
                tool='Sigstore',
                inject_cmd='for i in {1..100}; do curl -s https://accounts.google.com/.well-known/openid-configuration > /dev/null; done',
                cleanup_cmd='echo "OIDC rate limit cooldown: wait 1 minute"',
                description='Burst OIDC discovery requests to trigger provider rate limits'
            ),
            FailureScenario(
                name='vault-pki-role-delete',
                tool='Vault',
                inject_cmd='vault delete pki/roles/benchmark-role',
                cleanup_cmd='vault write pki/roles/benchmark-role allowed_domains=example.com allow_subdomains=true max_ttl=1h',
                description='Delete Vault PKI role mid-benchmark to simulate misconfiguration'
            )
        ]

    def run_failure_scenario(self, scenario: FailureScenario, bench_cmd: str, duration_sec: int) -> Dict:
        """
        Run a failure scenario while executing a benchmark workload.
        Args:
            scenario: FailureScenario to inject
            bench_cmd: Command to run the benchmark (e.g., Vault or Sigstore bench script)
            duration_sec: How long to run the benchmark + failure
        Returns:
            Dict with scenario results: failure rate, latency impact, etc.
        """
        print(f'\nRunning scenario: {scenario.name} ({scenario.description})')
        results = {
            'scenario': scenario.name,
            'tool': scenario.tool,
            'success_pre': 0,
            'success_during': 0,
            'success_post': 0,
            'errors': []
        }

        # 1. Run baseline benchmark (no failure)
        print('Running baseline benchmark...')
        try:
            pre_output = subprocess.run(
                bench_cmd,
                shell=True,
                capture_output=True,
                text=True,
                timeout=duration_sec
            )
            results['success_pre'] = self._parse_success_rate(pre_output.stdout)
            results['baseline_latency'] = self._parse_mean_latency(pre_output.stdout)
        except Exception as e:
            results['errors'].append(f'Baseline failed: {str(e)}')

        # 2. Inject failure and run benchmark
        print(f'Injecting failure: {scenario.inject_cmd}')
        try:
            # Inject failure
            subprocess.run(scenario.inject_cmd, shell=True, check=True)
            time.sleep(2)  # Let failure take effect

            # Run benchmark during failure
            during_output = subprocess.run(
                bench_cmd,
                shell=True,
                capture_output=True,
                text=True,
                timeout=duration_sec
            )
            results['success_during'] = self._parse_success_rate(during_output.stdout)
            results['failure_latency'] = self._parse_mean_latency(during_output.stdout)
        except Exception as e:
            results['errors'].append(f'Failure injection failed: {str(e)}')
        finally:
            # Cleanup failure
            print(f'Cleaning up: {scenario.cleanup_cmd}')
            subprocess.run(scenario.cleanup_cmd, shell=True, check=False)

        # 3. Run post-failure benchmark
        print('Running post-failure benchmark...')
        try:
            post_output = subprocess.run(
                bench_cmd,
                shell=True,
                capture_output=True,
                text=True,
                timeout=duration_sec
            )
            results['success_post'] = self._parse_success_rate(post_output.stdout)
            results['recovery_latency'] = self._parse_mean_latency(post_output.stdout)
        except Exception as e:
            results['errors'].append(f'Post-failure failed: {str(e)}')

        return results

    def _parse_success_rate(self, output: str) -> float:
        """Parse success rate from benchmark output (expects 'Success Rate: XX.XX%')"""
        for line in output.split('\n'):
            if 'Success Rate:' in line:
                rate_str = line.split(':')[1].strip().replace('%', '')
                return float(rate_str)
        return 0.0

    def _parse_mean_latency(self, output: str) -> float:
        """Parse mean latency from benchmark output (expects 'Mean Latency: XX.XXms')"""
        for line in output.split('\n'):
            if 'Mean Latency:' in line:
                latency_str = line.split(':')[1].strip().replace('ms', '')
                return float(latency_str)
        return 0.0

    def run_all_scenarios(self, bench_cmd: str, duration_sec: int = 60):
        """Run all failure scenarios and print results"""
        all_results = []
        for scenario in self.scenarios:
            res = self.run_failure_scenario(scenario, bench_cmd, duration_sec)
            all_results.append(res)

        # Print summary table
        print('\n\n=== Failure Injection Summary ===')
        print(f'{"Scenario":<30} {"Tool":<10} {"Baseline Success":<18} {"During Failure":<18} {"Post-Recovery":<15}')
        for res in all_results:
            print(f'{res["scenario"]:<30} {res["tool"]:<10} {res["success_pre"]:>6.2f}% {res["success_during"]:>6.2f}% {res["success_post"]:>6.2f}%')

        # Save results to JSON
        with open('failure_injection_results.json', 'w') as f:
            json.dump(all_results, f, indent=2)
        print('\nResults saved to failure_injection_results.json')

if __name__ == '__main__':
    import argparse
    parser = argparse.ArgumentParser(description='Supply chain failure injection benchmark')
    parser.add_argument('--vault-addr', default='https://127.0.0.1:8200', help='Vault address')
    parser.add_argument('--fulcio-addr', default='http://127.0.0.1:8080', help='Fulcio address')
    parser.add_argument('--rekor-addr', default='http://127.0.0.1:3000', help='Rekor address')
    parser.add_argument('--bench-cmd', required=True, help='Benchmark command to run (e.g., python vault_bench.py)')
    parser.add_argument('--duration', type=int, default=60, help='Benchmark duration per scenario')
    args = parser.parse_args()

    injector = SupplyChainFailureInjector(args.vault_addr, args.fulcio_addr, args.rekor_addr)
    injector.run_all_scenarios(args.bench_cmd, args.duration)

# Cosign command to enable OIDC token caching with Vault as storage
export COSIGN_OIDC_TOKEN_CACHE=/tmp/cosign-oidc-cache.json
export COSIGN_OIDC_TOKEN_CACHE_ENCRYPT_KEY=$(vault kv get -field=enc-key secret/cosign/cache-key)

# Sign image with cached OIDC token
cosign sign --oidc-cache-token --oidc-cache-encrypt-key $COSIGN_OIDC_TOKEN_CACHE_ENCRYPT_KEY example.com/myimage:latest

# Vault config.hcl snippet to tune etcd connection pool
storage "etcd" {
  address = "https://etcd-1.example.com:2379"
  etcd_api = "v3"
  ha_enabled = "true"
  max_idle_connections = 10
  max_open_connections = 20
  tls_ca_file = "/etc/vault/etcd-ca.pem"
  tls_cert_file = "/etc/vault/etcd-client.pem"
  tls_key_file = "/etc/vault/etcd-client-key.pem"
}

# Rekor CLI command to verify an artifact's log entry
rekor verify --artifact myapp.jar --signature myapp.jar.sig --public-key myapp-pub-key.pem

# Kubernetes admission controller snippet to check Rekor
if ! rekor verify --artifact $IMAGE_DIGEST --type docker; then
  echo "Rejected: No valid Rekor entry for $IMAGE_DIGEST"
  exit 1
fi