Retrospective: Implementing Shift-Left Security with Falco 0.38 and Tetragon 1.0 for Kubernetes 1.34

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "os"
    "time"

    // Import Falco 0.38 official Go client
    // Canonical repo: https://github.com/falcosecurity/falco
    falco "github.com/falcosecurity/falco/client/go/v0_38"
)

// FalcoEvent represents a parsed Falco security event
type FalcoEvent struct {
    Timestamp   time.Time `json:"timestamp"`
    Rule        string    `json:"rule"`
    Priority    string    `json:"priority"`
    Source      string    `json:"source"`
    ContainerID string    `json:"container_id,omitempty"`
    Pod         string    `json:"pod,omitempty"`
    Namespace   string    `json:"namespace,omitempty"`
    Output      string    `json:"output"`
}

func main() {
    // Initialize Falco client with default config path for 0.38
    clientCfg := falco.DefaultConfig()
    clientCfg.SocketPath = "/run/falco/falco.sock" // Default Falco 0.38 socket path
    clientCfg.Timeout = 30 * time.Second

    // Create new Falco client with error handling
    client, err := falco.NewClient(clientCfg)
    if err != nil {
        log.Fatalf("failed to initialize Falco client: %v", err)
    }
    defer func() {
        if closeErr := client.Close(); closeErr != nil {
            log.Printf("warning: failed to close Falco client: %v", closeErr)
        }
    }()

    // Subscribe to high and critical priority events only
    subCfg := falco.SubscriptionConfig{
        Priorities: []string{"High", "Critical"},
        // Filter to only K8s 1.34 related events
        Filter: "k8s.version = '1.34'",
    }

    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    eventChan, err := client.Subscribe(ctx, subCfg)
    if err != nil {
        log.Fatalf("failed to subscribe to Falco events: %v", err)
    }

    fmt.Println("Listening for Falco 0.38 events on K8s 1.34 clusters...")

    // Process events with retry logic for transient errors
    for {
        select {
        case event, ok := <-eventChan:
            if !ok {
                log.Println("event channel closed, shutting down")
                return
            }
            // Parse event into structured type
            var falcoEvent FalcoEvent
            if err := json.Unmarshal(event.Payload, &falcoEvent); err != nil {
                log.Printf("failed to unmarshal event: %v", err)
                continue
            }
            // Log critical events to stderr for alerting
            if falcoEvent.Priority == "Critical" {
                fmt.Fprintf(os.Stderr, "[CRITICAL] %s: %s\n", falcoEvent.Rule, falcoEvent.Output)
            } else {
                fmt.Printf("[HIGH] %s: %s\n", falcoEvent.Rule, falcoEvent.Output)
            }
        case <-ctx.Done():
            log.Println("context cancelled, exiting event loop")
            return
        }
    }
}

package main

import (
    "context"
    "fmt"
    "log"
    "time"

    // Import Tetragon 1.0 official Go client
    // Canonical repo: https://github.com/cilium/tetragon
    tetragon "github.com/cilium/tetragon/pkg/client/v1"
    "github.com/cilium/tetragon/pkg/events"
)

// TetragonPolicyEnforcer validates and applies Tetragon 1.0 policies to K8s 1.34 clusters
type TetragonPolicyEnforcer struct {
    client     *tetragon.Client
    k8sVersion string
}

// NewTetragonPolicyEnforcer initializes a new enforcer for target K8s version
func NewTetragonPolicyEnforcer(socketPath, k8sVersion string) (*TetragonPolicyEnforcer, error) {
    cfg := tetragon.Config{
        SocketPath: socketPath,
        Timeout:    30 * time.Second,
    }
    client, err := tetragon.NewClient(cfg)
    if err != nil {
        return nil, fmt.Errorf("failed to create Tetragon client: %w", err)
    }
    return &TetragonPolicyEnforcer{
        client:     client,
        k8sVersion: k8sVersion,
    }, nil
}

// ApplyPolicy validates a Tetragon policy and applies it to the cluster
func (e *TetragonPolicyEnforcer) ApplyPolicy(ctx context.Context, policyPath string) error {
    // Read policy file
    policy, err := tetragon.LoadPolicy(policyPath)
    if err != nil {
        return fmt.Errorf("failed to load policy from %s: %w", policyPath, err)
    }

    // Validate policy is compatible with Tetragon 1.0 and K8s 1.34
    if err := policy.Validate(); err != nil {
        return fmt.Errorf("invalid policy: %w", err)
    }
    if policy.Spec.K8sVersionConstraint != "" {
        // Check if policy supports K8s 1.34
        matches, err := policy.MatchesK8sVersion(e.k8sVersion)
        if err != nil {
            return fmt.Errorf("failed to check K8s version compatibility: %w", err)
        }
        if !matches {
            return fmt.Errorf("policy does not support K8s version %s", e.k8sVersion)
        }
    }

    // Apply policy with retry logic for transient API errors
    var applyErr error
    for retry := 0; retry < 3; retry++ {
        applyErr = e.client.ApplyPolicy(ctx, policy)
        if applyErr == nil {
            log.Printf("successfully applied policy %s to K8s %s cluster", policy.Name, e.k8sVersion)
            return nil
        }
        log.Printf("retry %d: failed to apply policy: %v", retry+1, applyErr)
        time.Sleep(time.Second * time.Duration(retry+1))
    }
    return fmt.Errorf("failed to apply policy after 3 retries: %w", applyErr)
}

func main() {
    enforcer, err := NewTetragonPolicyEnforcer("/run/tetragon/tetragon.sock", "1.34")
    if err != nil {
        log.Fatalf("failed to initialize enforcer: %v", err)
    }
    defer func() {
        if closeErr := enforcer.client.Close(); closeErr != nil {
            log.Printf("warning: failed to close Tetragon client: %v", closeErr)
        }
    }()

    ctx := context.Background()
    if err := enforcer.ApplyPolicy(ctx, "/etc/tetragon/policies/no-privileged-containers.yaml"); err != nil {
        log.Fatalf("failed to apply policy: %v", err)
    }
}

#!/usr/bin/env python3
"""
Falco 0.38 and Tetragon 1.0 event aggregator for K8s 1.34 clusters.
Ships normalized events to Elasticsearch for long-term storage and alerting.
"""

import json
import logging
import socket
import time
from contextlib import contextmanager
from dataclasses import dataclass
from typing import Optional

import requests
from requests.exceptions import RequestException

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

# Canonical repos:
# Falco: https://github.com/falcosecurity/falco
# Tetragon: https://github.com/cilium/tetragon

@dataclass
class NormalizedSecurityEvent:
    """Unified event format for Falco and Tetragon events"""
    event_id: str
    timestamp: float
    source_tool: str  # "falco" or "tetragon"
    tool_version: str
    k8s_version: str
    priority: str
    rule_name: str
    namespace: str
    pod_name: str
    container_id: str
    raw_output: str

class EventAggregator:
    """Aggregates security events from Falco and Tetragon"""

    def __init__(self, es_url: str, k8s_version: str = "1.34"):
        self.es_url = es_url
        self.k8s_version = k8s_version
        self.session = requests.Session()
        self.session.headers.update({"Content-Type": "application/json"})

    @contextmanager
    def _es_connection(self):
        """Handle Elasticsearch connection errors"""
        try:
            yield self.session
        except RequestException as e:
            logger.error(f"Elasticsearch connection error: {e}")
            raise

    def ship_event(self, event: NormalizedSecurityEvent) -> bool:
        """Ship normalized event to Elasticsearch with retry logic"""
        payload = {
            "event_id": event.event_id,
            "timestamp": event.timestamp,
            "source_tool": event.source_tool,
            "tool_version": event.tool_version,
            "k8s_version": event.k8s_version,
            "priority": event.priority,
            "rule_name": event.rule_name,
            "namespace": event.namespace,
            "pod_name": event.pod_name,
            "container_id": event.container_id,
            "raw_output": event.raw_output,
            "@timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(event.timestamp))
        }

        for retry in range(3):
            try:
                with self._es_connection() as s:
                    resp = s.post(
                        f"{self.es_url}/security-events/_doc/{event.event_id}",
                        data=json.dumps(payload),
                        timeout=10
                    )
                    resp.raise_for_status()
                    logger.info(f"Shipped event {event.event_id} to ES")
                    return True
            except RequestException as e:
                logger.warning(f"Retry {retry+1} failed to ship event: {e}")
                time.sleep(retry + 1)
        logger.error(f"Failed to ship event {event.event_id} after 3 retries")
        return False

def parse_falco_event(raw_event: dict) -> Optional[NormalizedSecurityEvent]:
    """Parse raw Falco 0.38 event into normalized format"""
    try:
        return NormalizedSecurityEvent(
            event_id=raw_event["event_id"],
            timestamp=raw_event["timestamp"],
            source_tool="falco",
            tool_version="0.38",
            k8s_version=raw_event.get("k8s_version", "1.34"),
            priority=raw_event["priority"],
            rule_name=raw_event["rule"],
            namespace=raw_event.get("namespace", ""),
            pod_name=raw_event.get("pod", ""),
            container_id=raw_event.get("container_id", ""),
            raw_output=raw_event["output"]
        )
    except KeyError as e:
        logger.error(f"Missing key in Falco event: {e}")
        return None

if __name__ == "__main__":
    aggregator = EventAggregator(es_url="http://elasticsearch:9200")
    # Example raw Falco event (truncated for brevity)
    sample_falco_event = {
        "event_id": "falco-12345",
        "timestamp": time.time(),
        "priority": "Critical",
        "rule": "Privileged Container Spawned",
        "namespace": "prod",
        "pod": "api-deploy-123",
        "container_id": "abc123",
        "output": "Privileged container spawned in prod namespace"
    }
    normalized = parse_falco_event(sample_falco_event)
    if normalized:
        aggregator.ship_event(normalized)

# .pre-commit-config.yaml
repos:
  - repo: https://github.com/falcosecurity/falco
    rev: 0.38.0
    hooks:
      - id: validate-falco-rules
        args: ["--k8s-version", "1.34", "--priority-check"]
        files: \.yaml$
        exclude: ^helm/

# test_tetragon_policy.py
import pytest
from tetragon.testing import PolicyTestClient

def test_no_privileged_containers_policy():
    client = PolicyTestClient(policy_path="/etc/tetragon/policies/no-privileged.yaml", k8s_version="1.34")
    # Test 1: Privileged container should be blocked
    assert client.simulate_pod_spawn(privileged=True).action == "BLOCK"
    # Test 2: Non-privileged container should be allowed
    assert client.simulate_pod_spawn(privileged=False).action == "ALLOW"
    # Test 3: Ephemeral container (K8s 1.34+) should follow same rules
    assert client.simulate_ephemeral_container(privileged=True).action == "BLOCK"

# Prometheus query for p99 Falco + Tetragon event response time
histogram_quantile(0.99, 
  sum(rate(falco_event_processing_seconds_bucket[5m])) by (le) 
  + 
  sum(rate(tetragon_policy_decision_seconds_bucket[5m])) by (le)
)