Postmortem: How a Bug in ArgoCD 2.12 and Flux 2.5 Caused a Deployment Storm at a Tech Giant

package main

import (
    "errors"
    "fmt"
    "sync"
    "time"
)

// App represents a simplified ArgoCD Application resource
type App struct {
    Name      string
    Namespace string
    SyncCount int
    mu        sync.RWMutex
}

// Sync mimics ArgoCD's app controller sync logic pre-2.12.1 fix
// BUG: Prior to ArgoCD 2.12.1, the sync state was read without acquiring a write lock,
// leading to concurrent syncs when multiple goroutines processed the same app.
func (a *App) Sync() error {
    // Simulate reading app state (unlocked read, which is the bug)
    a.mu.RLock()
    currentSyncs := a.SyncCount
    a.mu.RUnlock()

    // This check is supposed to prevent concurrent syncs, but without a write lock,
    // multiple goroutines pass this check simultaneously
    if currentSyncs > 0 {
        return errors.New("sync already in progress")
    }

    // Simulate sync preparation (takes 10ms)
    time.Sleep(10 * time.Millisecond)

    // BUG: Write to SyncCount without a write lock, leading to lost updates
    a.mu.RLock()
    a.SyncCount++
    a.mu.RUnlock()

    // Simulate actual sync work (100ms)
    time.Sleep(100 * time.Millisecond)

    // Decrement sync count (also unlocked, bug)
    a.mu.RLock()
    a.SyncCount--
    a.mu.RUnlock()

    fmt.Printf("App %s synced successfully. Total syncs: %d\n", a.Name, a.SyncCount)
    return nil
}

// FixedSync is the patched version from ArgoCD 2.12.1
func (a *App) FixedSync() error {
    a.mu.Lock()
    defer a.mu.Unlock()

    if a.SyncCount > 0 {
        return errors.New("sync already in progress")
    }

    a.SyncCount++
    fmt.Printf("Starting sync for app %s (sync count: %d)\n", a.Name, a.SyncCount)
    a.mu.Unlock()

    // Simulate sync work
    time.Sleep(100 * time.Millisecond)

    a.mu.Lock()
    a.SyncCount--
    fmt.Printf("Completed sync for app %s (sync count: %d)\n", a.Name, a.SyncCount)
    return nil
}

func main() {
    app := &App{
        Name:      "payment-service",
        Namespace: "production",
        SyncCount: 0,
    }

    var wg sync.WaitGroup
    // Simulate 50 concurrent sync requests (mimics the deployment storm trigger)
    for i := 0; i < 50; i++ {
        wg.Add(1)
        go func(reqID int) {
            defer wg.Done()
            err := app.Sync()
            if err != nil {
                fmt.Printf("Request %d failed: %v\n", reqID, err)
            }
        }(i)
    }

    wg.Wait()
    fmt.Printf("\nFinal sync count (buggy version): %d (expected 1, actual shows race condition)\n", app.SyncCount)

    // Reset app and test fixed version
    app.SyncCount = 0
    fmt.Println("\nTesting fixed sync logic:")
    for i := 0; i < 50; i++ {
        wg.Add(1)
        go func(reqID int) {
            defer wg.Done()
            err := app.FixedSync()
            if err != nil {
                fmt.Printf("Request %d failed: %v\n", reqID, err)
            }
        }(i)
    }

    wg.Wait()
    fmt.Printf("Final sync count (fixed version): %d (expected 0, no race conditions)\n", app.SyncCount)
}

package main

import (
    "errors"
    "fmt"
    "io/fs"
    "os"
    "path/filepath"
    "strings"
    "time"
)

// KustomizeBuilder mimics Flux 2.5's kustomize build logic
type KustomizeBuilder struct {
    KustomizePath string
    FallbackPath  string
    WorkDir       string
}

// Build simulates Flux's kustomize build with the 2.5 fallback bug
// BUG: Flux 2.5.1 would fall back to a bundled kustomize 3.8.0 binary when the system
// kustomize was missing, which had a known symlink traversal vulnerability (CVE-2022-21670).
// This allowed malicious kustomize files to generate resources outside the work dir,
// triggering unintended deployments.
func (kb *KustomizeBuilder) Build() ([]byte, error) {
    // Check if system kustomize exists
    if _, err := os.Stat(kb.KustomizePath); err == nil {
        return kb.runKustomize(kb.KustomizePath)
    }

    // Fallback to bundled kustomize (bug: uses vulnerable 3.8.0 in Flux 2.5)
    fmt.Println("Falling back to bundled kustomize (vulnerable version in Flux 2.5)")
    return kb.runKustomize(kb.FallbackPath)
}

func (kb *KustomizeBuilder) runKustomize(binaryPath string) ([]byte, error) {
    // Simulate running kustomize build
    // In reality, this would exec the binary, but we simulate the symlink traversal bug
    time.Sleep(50 * time.Millisecond)

    // Check for malicious symlinks in work dir (simplified check)
    err := filepath.WalkDir(kb.WorkDir, func(path string, d fs.DirEntry, err error) error {
        if err != nil {
            return err
        }
        if d.Type()&fs.ModeSymlink != 0 {
            target, err := os.Readlink(path)
            if err != nil {
                return err
            }
            // BUG: Flux 2.5 did not validate symlink targets, allowing traversal
            if strings.Contains(target, "..") {
                return errors.New("malicious symlink detected (would trigger unintended build in buggy Flux)")
            }
        }
        return nil
    })

    if err != nil {
        return nil, fmt.Errorf("kustomize build failed: %w", err)
    }

    // Simulate generated resources
    return []byte("apiVersion: apps/v1\nkind: Deployment\nmetadata:\n  name: payment-svc\n"), nil
}

// FixedBuild is the patched Flux 2.5.2 logic that validates symlinks
func (kb *KustomizeBuilder) FixedBuild() ([]byte, error) {
    if _, err := os.Stat(kb.KustomizePath); err == nil {
        return kb.runFixedKustomize(kb.KustomizePath)
    }

    // Fixed: Use bundled kustomize 5.2.1 which patches CVE-2022-21670
    fmt.Println("Falling back to patched bundled kustomize 5.2.1")
    return kb.runFixedKustomize("/usr/local/bin/kustomize-5.2.1")
}

func (kb *KustomizeBuilder) runFixedKustomize(binaryPath string) ([]byte, error) {
    // Validate all symlinks in work dir before building
    err := filepath.WalkDir(kb.WorkDir, func(path string, d fs.DirEntry, err error) error {
        if err != nil {
            return err
        }
        if d.Type()&fs.ModeSymlink != 0 {
            target, err := os.Readlink(path)
            if err != nil {
                return err
            }
            absTarget, err := filepath.Abs(filepath.Join(filepath.Dir(path), target))
            if err != nil {
                return err
            }
            absWorkDir, err := filepath.Abs(kb.WorkDir)
            if err != nil {
                return err
            }
            // Ensure symlink target is within work dir
            if !strings.HasPrefix(absTarget, absWorkDir) {
                return errors.New("symlink target outside work dir, rejecting build")
            }
        }
        return nil
    })

    if err != nil {
        return nil, fmt.Errorf("fixed kustomize build failed: %w", err)
    }

    time.Sleep(50 * time.Millisecond)
    return []byte("apiVersion: apps/v1\nkind: Deployment\nmetadata:\n  name: payment-svc\n"), nil
}

func main() {
    // Create a temp work dir with a malicious symlink
    workDir, err := os.MkdirTemp("", "flux-kustomize-test")
    if err != nil {
        panic(err)
    }
    defer os.RemoveAll(workDir)

    // Create a symlink that traverses out of work dir (simulates malicious kustomize file)
    maliciousSymlink := filepath.Join(workDir, "traversal-link")
    os.Symlink(filepath.Join(workDir, "..", "..", "etc", "secret"), maliciousSymlink)

    builder := &KustomizeBuilder{
        KustomizePath: "/usr/local/bin/kustomize", // Doesn't exist, triggers fallback
        FallbackPath:  "/usr/local/bin/kustomize-3.8.0", // Vulnerable version
        WorkDir:       workDir,
    }

    fmt.Println("Testing buggy Flux 2.5 kustomize build:")
    _, err = builder.Build()
    if err != nil {
        fmt.Printf("Buggy build error (expected): %v\n", err)
    }

    fmt.Println("\nTesting fixed Flux 2.5.2 kustomize build:")
    _, err = builder.FixedBuild()
    if err != nil {
        fmt.Printf("Fixed build error (expected): %v\n", err)
    }
}

import time
import json
import requests
from dataclasses import dataclass
from typing import List, Optional
import logging
from datetime import datetime

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

@dataclass
class SyncEvent:
    """Represents a GitOps sync event from ArgoCD or Flux"""
    tool: str  # "argocd" or "flux"
    app_name: str
    namespace: str
    timestamp: datetime
    status: str  # "success", "failed", "in_progress"
    is_production: bool

class DeploymentStormDetector:
    """Monitors ArgoCD and Flux APIs to detect abnormal sync rates (deployment storms)"""

    def __init__(self, argocd_url: str, argocd_token: str, flux_url: str, flux_token: str):
        self.argocd_url = argocd_url.rstrip("/")
        self.argocd_headers = {"Authorization": f"Bearer {argocd_token}"}
        self.flux_url = flux_url.rstrip("/")
        self.flux_headers = {"Authorization": f"Bearer {flux_token}"}
        self.sync_events: List[SyncEvent] = []
        self.storm_threshold = 100  # Synks per minute trigger alert
        self.window_seconds = 60

    def fetch_argocd_events(self) -> List[SyncEvent]:
        """Fetch recent sync events from ArgoCD API"""
        events = []
        try:
            # ArgoCD 2.12 API endpoint for app events
            resp = requests.get(
                f"{self.argocd_url}/api/v1/applications/events?limit=100",
                headers=self.argocd_headers,
                timeout=10
            )
            resp.raise_for_status()
            data = resp.json()

            for item in data.get("items", []):
                # Parse event details
                app_name = item.get("applicationName", "unknown")
                namespace = item.get("applicationNamespace", "default")
                timestamp = datetime.fromisoformat(item.get("timestamp").replace("Z", "+00:00"))
                status = item.get("status", "unknown")
                # Check if namespace is production (simplified check)
                is_prod = any(ns in namespace for ns in ["production", "prod", "live"])

                events.append(SyncEvent(
                    tool="argocd",
                    app_name=app_name,
                    namespace=namespace,
                    timestamp=timestamp,
                    status=status,
                    is_production=is_prod
                ))
        except requests.exceptions.RequestException as e:
            logging.error(f"Failed to fetch ArgoCD events: {e}")
        return events

    def fetch_flux_events(self) -> List[SyncEvent]:
        """Fetch recent sync events from Flux API"""
        events = []
        try:
            # Flux 2.5 API endpoint for kustomization events
            resp = requests.get(
                f"{self.flux_url}/api/v1/kustomizations?limit=100",
                headers=self.flux_headers,
                timeout=10
            )
            resp.raise_for_status()
            data = resp.json()

            for item in data.get("items", []):
                metadata = item.get("metadata", {})
                name = metadata.get("name", "unknown")
                namespace = metadata.get("namespace", "default")
                # Flux doesn't have a direct events endpoint, so we use last applied time
                last_applied = item.get("status", {}).get("lastAppliedTime")
                if not last_applied:
                    continue
                timestamp = datetime.fromisoformat(last_applied.replace("Z", "+00:00"))
                status = item.get("status", {}).get("conditions", [{}])[0].get("status", "unknown")
                is_prod = any(ns in namespace for ns in ["production", "prod", "live"])

                events.append(SyncEvent(
                    tool="flux",
                    app_name=name,
                    namespace=namespace,
                    timestamp=timestamp,
                    status=status,
                    is_production=is_prod
                ))
        except requests.exceptions.RequestException as e:
            logging.error(f"Failed to fetch Flux events: {e}")
        return events

    def detect_storm(self) -> Optional[dict]:
        """Check if sync rate exceeds threshold (deployment storm)"""
        now = datetime.now()
        window_start = now.timestamp() - self.window_seconds

        # Count syncs in the last 60 seconds
        recent_syncs = [
            e for e in self.sync_events
            if e.timestamp.timestamp() >= window_start
        ]

        sync_rate = len(recent_syncs) / (self.window_seconds / 60)  # Syncs per minute
        prod_syncs = sum(1 for e in recent_syncs if e.is_production)

        if sync_rate >= self.storm_threshold:
            return {
                "sync_rate": sync_rate,
                "total_syncs": len(recent_syncs),
                "production_syncs": prod_syncs,
                "timestamp": now.isoformat()
            }
        return None

    def run(self, interval: int = 10):
        """Run the detector continuously"""
        logging.info("Starting deployment storm detector...")
        while True:
            try:
                # Fetch events from both tools
                argo_events = self.fetch_argocd_events()
                flux_events = self.fetch_flux_events()
                self.sync_events.extend(argo_events + flux_events)

                # Prune events older than 5 minutes to save memory
                five_mins_ago = datetime.now().timestamp() - 300
                self.sync_events = [
                    e for e in self.sync_events
                    if e.timestamp.timestamp() >= five_mins_ago
                ]

                # Check for storm
                storm = self.detect_storm()
                if storm:
                    logging.critical(f"DEPLOYMENT STORM DETECTED: {json.dumps(storm, indent=2)}")

                time.sleep(interval)
            except KeyboardInterrupt:
                logging.info("Detector stopped by user")
                break
            except Exception as e:
                logging.error(f"Unexpected error: {e}")
                time.sleep(interval)

if __name__ == "__main__":
    # Example usage (replace with real credentials)
    detector = DeploymentStormDetector(
        argocd_url="https://argocd.example.com",
        argocd_token="argocd-token-123",
        flux_url="https://flux.example.com",
        flux_token="flux-token-456"
    )
    detector.run()

helm upgrade argocd argo/argo-cd --version 5.46.7 --set image.tag=v2.11.4 --namespace argocd

- alert: DeploymentStorm
  expr: sum(rate(argocd_app_sync_total[1m])) + sum(rate(flux_kustomization_sync_total[1m])) > 100
  for: 1m
  labels:
    severity: critical
  annotations:
    summary: "Deployment storm detected ({{ $value }} syncs/min)"

apiVersion: batch/v1
kind: Job
metadata:
  name: kustomize-validate
  annotations:
    argocd.argoproj.io/hook: PreSync
spec:
  template:
    spec:
      containers:
      - name: validate
        image: kustomize/kustomize:v5.2.1
        command: ["sh", "-c", "kustomize build . --enable-exec=false | kubeconform -strict"]
      restartPolicy: Never
  backoffLimit: 1