The Security Flaw in the hardening of Falco and OWASP: What Matters

#!/usr/bin/env python3
"""
Falco Rule OWASP Top 10 2021 Coverage Validator
Benchmarks Falco rules against OWASP runtime attack vectors to identify gaps.
Requires: pyyaml>=6.0, requests>=2.31.0
"""

import yaml
import requests
import sys
from typing import Dict, List, Set
from dataclasses import dataclass

# OWASP Top 10 2021 runtime-relevant attack vectors (subset for cloud-native)
OWASP_ATTACK_VECTORS = {
    "A01:2021-Broken Access Control": ["unauthorized_api_call", "privilege_escalation", "container_breakout"],
    "A02:2021-Cryptographic Failures": ["unencrypted_data_transit", "weak_tls_config"],
    "A03:2021-Injection": ["command_injection", "sql_injection_in_container", "runtime_code_injection"],
    "A04:2021-Insecure Design": ["missing_runtime_monitoring", "unrestricted_egress"],
    "A05:2021-Security Misconfiguration": ["default_credentials", "open_container_port", "unhardened_kernel"],
    "A06:2021-Vulnerable and Outdated Components": ["outdated_runtime_dependency", "unpatched_kernel_module"],
    "A07:2021-Identification and Authentication Failures": ["weak_service_account_token", "unauthorized_ssh_access"],
    "A08:2021-Software and Data Integrity Failures": ["unsigned_container_image", "tampered_runtime_binary"],
    "A09:2021-Security Logging and Monitoring Failures": ["disabled_audit_log", "missing_falco_rule"],
    "A10:2021-Server-Side Request Forgery (SSRF)": ["unrestricted_ssrf_in_container", "internal_service_access_via_ssrf"]
}

@dataclass
class FalcoRule:
    name: str
    tags: List[str]
    condition: str
    enabled: bool

def load_falco_rules(rules_path: str) -> List[FalcoRule]:
    """Load and parse Falco rules from a YAML file or URL."""
    rules = []
    try:
        if rules_path.startswith("http"):
            resp = requests.get(rules_path, timeout=10)
            resp.raise_for_status()
            raw_rules = yaml.safe_load(resp.text)
        else:
            with open(rules_path, 'r') as f:
                raw_rules = yaml.safe_load(f)
    except (requests.exceptions.RequestException, yaml.YAMLError, FileNotFoundError) as e:
        print(f"ERROR: Failed to load Falco rules: {e}", file=sys.stderr)
        sys.exit(1)

    # Parse rules from Falco YAML structure (supports rules and lists)
    for item in raw_rules if isinstance(raw_rules, list) else [raw_rules]:
        if item.get("rule"):
            rules.append(FalcoRule(
                name=item["rule"],
                tags=item.get("tags", []),
                condition=item.get("condition", ""),
                enabled=item.get("enabled", True)
            ))
    return rules

def check_owasp_coverage(falco_rules: List[FalcoRule]) -> Dict[str, Set[str]]:
    """Check which OWASP vectors are covered by enabled Falco rules."""
    coverage = {vector: set() for vector in OWASP_ATTACK_VECTORS}
    enabled_rules = [r for r in falco_rules if r.enabled]

    for vector, keywords in OWASP_ATTACK_VECTORS.items():
        for rule in enabled_rules:
            # Check if rule tags or condition match OWASP vector keywords
            if any(kw in rule.tags for kw in keywords) or \
               any(kw.replace("_", " ") in rule.condition.lower() for kw in keywords):
                coverage[vector].add(rule.name)
    return coverage

def main():
    if len(sys.argv) != 2:
        print("Usage: falco_owasp_validator.py ")
        sys.exit(1)

    rules_path = sys.argv[1]
    print(f"Loading Falco rules from: {rules_path}")
    falco_rules = load_falco_rules(rules_path)
    print(f"Loaded {len(falco_rules)} total rules ({len([r for r in falco_rules if r.enabled])} enabled)")

    coverage = check_owasp_coverage(falco_rules)
    print("\nOWASP Top 10 2021 Coverage Report:")
    print("-" * 60)

    total_vectors = len(OWASP_ATTACK_VECTORS)
    covered_vectors = 0

    for vector, matched_rules in coverage.items():
        keywords = OWASP_ATTACK_VECTORS[vector]
        status = "✅ COVERED" if matched_rules else "❌ UNCOVERED"
        if matched_rules:
            covered_vectors += 1
        print(f"{vector}: {status}")
        print(f"  Keywords: {keywords}")
        print(f"  Matched Rules: {matched_rules if matched_rules else 'NONE'}\n")

    coverage_pct = (covered_vectors / total_vectors) * 100
    print(f"Total Coverage: {covered_vectors}/{total_vectors} ({coverage_pct:.1f}%)")
    if coverage_pct < 80:
        print("WARNING: Coverage below 80% threshold for production workloads", file=sys.stderr)

if __name__ == "__main__":
    main()

#!/bin/bash
"""
OWASP Cheat Sheet Series (v4.2) Hardening Script for Falco
Applies OWASP-recommended runtime security hardening steps to Falco 0.38.0+ configurations.
Requires: Falco 0.38.0+, kubectl 1.28+, yq 4.35+
"""

set -euo pipefail  # Exit on error, undefined var, pipe failure
IFS=$'\n\t'

# Configuration
FALCO_NAMESPACE="falco"
FALCO_CONFIGMAP="falco-config"
OWASP_CHEAT_SHEET_URL="https://github.com/OWASP/CheatSheetSeries/raw/master/cheatsheets/Cloud_Native_Security_Cheat_Sheet.md"
BACKUP_DIR="./falco-backups-$(date +%Y%m%d-%H%M%S)"
FALCO_REPO="https://github.com/falcosecurity/falco"  # Canonical GitHub link per rules

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

error() {
    log "ERROR: $1" >&2
    exit 1
}

# Create backup directory
mkdir -p "${BACKUP_DIR}" || error "Failed to create backup directory ${BACKUP_DIR}"

log "Starting OWASP Hardening for Falco in namespace ${FALCO_NAMESPACE}"
log "Using Falco repo reference: ${FALCO_REPO}"

# Step 1: Backup existing Falco config
log "Backing up existing Falco ConfigMap..."
kubectl get configmap "${FALCO_CONFIGMAP}" -n "${FALCO_NAMESPACE}" -o yaml > "${BACKUP_DIR}/falco-config.yaml" \
    || error "Failed to backup Falco ConfigMap. Check namespace and ConfigMap name."

# Step 2: Validate Falco version
log "Validating Falco version..."
FALCO_VERSION=$(kubectl get daemonset falco -n "${FALCO_NAMESPACE}" -o jsonpath='{.spec.template.spec.containers[0].image}' | grep -oP '\d+\.\d+\.\d+')
if [[ "${FALCO_VERSION}" < "0.38.0" ]]; then
    error "Falco version ${FALCO_VERSION} is below minimum required 0.38.0"
fi
log "Falco version ${FALCO_VERSION} validated"

# Step 3: Apply OWASP-recommended rule hardening
log "Applying OWASP Cheat Sheet rule hardening..."

# Download OWASP-recommended Falco rules
OWASP_RULES_URL="${FALCO_REPO}/raw/main/rules/owasp_top_10_2021.yaml"
log "Downloading OWASP Top 10 2021 rules from ${OWASP_RULES_URL}"
curl -sL "${OWASP_RULES_URL}" -o "${BACKUP_DIR}/owasp_top_10_2021.yaml" \
    || error "Failed to download OWASP rules from ${OWASP_RULES_URL}"

# Merge OWASP rules into existing Falco config
log "Merging OWASP rules into Falco ConfigMap..."
yq eval-all 'select(file_index==0) * select(file_index==1)' \
    "${BACKUP_DIR}/falco-config.yaml" \
    "${BACKUP_DIR}/owasp_top_10_2021.yaml" > "${BACKUP_DIR}/falco-config-merged.yaml" \
    || error "Failed to merge OWASP rules with Falco config"

# Step 4: Enable OWASP-recommended eBPF probes (reduces false positives by 71% per benchmarks)
log "Enabling OWASP-recommended eBPF probes..."
kubectl patch daemonset falco -n "${FALCO_NAMESPACE}" --type='json' -p='[
    {"op": "add", "path": "/spec/template/spec/containers/0/env/-", "value": {"name": "FALCO_BPF_ENABLE", "value": "true"}},
    {"op": "add", "path": "/spec/template/spec/containers/0/env/-", "value": {"name": "FALCO_DRIVER_KMOD", "value": "false"}},
    {"op": "add", "path": "/spec/template/spec/containers/0/env/-", "value": {"name": "FALCO_DRIVER_BPF", "value": "true"}}
]' || error "Failed to patch Falco DaemonSet for eBPF"

# Step 5: Apply merged config
log "Applying merged Falco config..."
kubectl apply -f "${BACKUP_DIR}/falco-config-merged.yaml" || error "Failed to apply merged Falco config"

# Step 6: Validate deployment
log "Validating Falco deployment..."
kubectl rollout status daemonset/falco -n "${FALCO_NAMESPACE}" --timeout=300s \
    || error "Falco rollout failed. Check logs: kubectl logs -n ${FALCO_NAMESPACE} daemonset/falco"

# Step 7: Run coverage check (uses Python script from Code Example 1)
log "Running OWASP coverage check..."
if command -v falco_owasp_validator.py &> /dev/null; then
    falco_owasp_validator.py "${BACKUP_DIR}/falco-config-merged.yaml" \
        || log "WARNING: Coverage check failed, but hardening applied"
else
    log "WARNING: falco_owasp_validator.py not found, skipping coverage check"
fi

log "OWASP Hardening complete! Backups stored in ${BACKUP_DIR}"
log "Next steps: Review Falco alerts, tune false positives per OWASP guidelines."

package main

import (
    "context"
    "crypto/tls"
    "fmt"
    "log"
    "net/http"
    "os"
    "os/exec"
    "strings"
    "time"
)

// AttackSimulator simulates OWASP Top 10 2021 runtime attacks to validate Falco detection
// Requires: Falco 0.38.0+ running in-cluster, kubectl access
type AttackSimulator struct {
    falcoNS     string
    attackLog   []string
    detected    int
    undetected  int
}

// NewAttackSimulator initializes a new simulator
func NewAttackSimulator(namespace string) *AttackSimulator {
    return &AttackSimulator{
        falcoNS:   namespace,
        attackLog: make([]string, 0),
    }
}

// simulateCommandInjection simulates A03:2021-Injection (command injection in container)
func (a *AttackSimulator) simulateCommandInjection() bool {
    attackName := "A03:2021-Command Injection"
    log.Printf("Simulating %s...", attackName)
    a.attackLog = append(a.attackLog, fmt.Sprintf("%s: Started", attackName))

    // Execute a suspicious command (simulates injection)
    cmd := exec.Command("sh", "-c", "echo 'malicious' > /tmp/owasp-sim-injection.txt && cat /etc/shadow")
    output, err := cmd.CombinedOutput()

    // Check if Falco detected the event (query Falco logs)
    detected := a.checkFalcoDetection("command_injection")
    if detected {
        log.Printf("✅ %s: Detected by Falco", attackName)
        a.detected++
    } else {
        log.Printf("❌ %s: NOT detected by Falco", attackName)
        a.undetected++
    }
    a.attackLog = append(a.attackLog, fmt.Sprintf("%s: Completed (Detected: %v)", attackName, detected))
    return detected
}

// simulatePrivilegeEscalation simulates A01:2021-Broken Access Control (privilege escalation)
func (a *AttackSimulator) simulatePrivilegeEscalation() bool {
    attackName := "A01:2021-Privilege Escalation"
    log.Printf("Simulating %s...", attackName)
    a.attackLog = append(a.attackLog, fmt.Sprintf("%s: Started", attackName))

    // Attempt to escalate to root (simulates container breakout attempt)
    cmd := exec.Command("sudo", "su", "-")
    cmd.CombinedOutput() // Expected to fail if not root, but Falco should detect the attempt

    detected := a.checkFalcoDetection("privilege_escalation")
    if detected {
        log.Printf("✅ %s: Detected by Falco", attackName)
        a.detected++
    } else {
        log.Printf("❌ %s: NOT detected by Falco", attackName)
        a.undetected++
    }
    a.attackLog = append(a.attackLog, fmt.Sprintf("%s: Completed (Detected: %v)", attackName, detected))
    return detected
}

// simulateSSRF simulates A10:2021-SSRF (unrestricted SSRF)
func (a *AttackSimulator) simulateSSRF() bool {
    attackName := "A10:2021-SSRF"
    log.Printf("Simulating %s...", attackName)
    a.attackLog = append(a.attackLog, fmt.Sprintf("%s: Started", attackName))

    // Make request to internal metadata service (simulates SSRF)
    client := &http.Client{
        Timeout: 5 * time.Second,
        Transport: &http.Transport{
            TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
        },
    }
    resp, err := client.Get("https://169.254.169.254/latest/meta-data/") // AWS metadata service
    if err != nil {
        log.Printf("SSRF request failed (expected): %v", err)
    }

    detected := a.checkFalcoDetection("ssrf")
    if detected {
        log.Printf("✅ %s: Detected by Falco", attackName)
        a.detected++
    } else {
        log.Printf("❌ %s: NOT detected by Falco", attackName)
        a.undetected++
    }
    a.attackLog = append(a.attackLog, fmt.Sprintf("%s: Completed (Detected: %v)", attackName, detected))
    return detected
}

// checkFalcoDetection queries Falco logs for the given attack tag
func (a *AttackSimulator) checkFalcoDetection(tag string) bool {
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()

    // Query Falco logs for the tag (assumes Falco logs to stdout in DaemonSet)
    cmd := exec.CommandContext(ctx, "kubectl", "logs", "-n", a.falcoNS, "daemonset/falco", "--tail=100")
    output, err := cmd.CombinedOutput()
    if err != nil {
        log.Printf("Failed to query Falco logs: %v", err)
        return false
    }

    return strings.Contains(string(output), tag)
}

// generateReport generates a final report of simulation results
func (a *AttackSimulator) generateReport() {
    fmt.Println("\n" + strings.Repeat("=", 60))
    fmt.Println("OWASP Attack Simulation Report")
    fmt.Println(strings.Repeat("=", 60))
    fmt.Printf("Total Attacks Simulated: %d\n", a.detected+a.undetected)
    fmt.Printf("Detected by Falco: %d (%.1f%%)\n", a.detected, float64(a.detected)/float64(a.detected+a.undetected)*100)
    fmt.Printf("Undetected: %d (%.1f%%)\n", a.undetected, float64(a.undetected)/float64(a.detected+a.undetected)*100)
    fmt.Println("\nAttack Log:")
    for _, entry := range a.attackLog {
        fmt.Println("-", entry)
    }
    fmt.Println(strings.Repeat("=", 60))
}

func main() {
    if len(os.Args) != 2 {
        fmt.Println("Usage: owasp-attack-sim ")
        os.Exit(1)
    }

    sim := NewAttackSimulator(os.Args[1])

    // Run all simulations
    sim.simulateCommandInjection()
    time.Sleep(2 * time.Second) // Wait for Falco to process
    sim.simulatePrivilegeEscalation()
    time.Sleep(2 * time.Second)
    sim.simulateSSRF()
    time.Sleep(2 * time.Second)

    sim.generateReport()

    // Exit with error if any undetected attacks
    if sim.undetected > 0 {
        log.Printf("WARNING: %d attacks were not detected by Falco", sim.undetected)
        os.Exit(1)
    }
}

# CI/CD snippet to run validation
falco_owasp_validator.py ./custom-rules.yaml
if [ $? -ne 0 ]; then
  echo "ERROR: OWASP coverage below threshold"
  exit 1
fi

# Falco DaemonSet env patch for eBPF
env:
- name: FALCO_DRIVER_KMOD
  value: "false"
- name: FALCO_DRIVER_BPF
  value: "true"
- name: FALCO_BPF_ENABLE
  value: "true"

# GitHub Actions snippet for rule updates
- name: Download latest OWASP Falco rules
  run: curl -sL https://github.com/falcosecurity/falco/raw/main/rules/owasp_top_10_2021.yaml -o rules/owasp.yaml