AI Model Security: Comparing Falco 0.40 vs Trivy 0.50 for Detecting Vulnerable LLM Dependencies

# falco_pytorch_rule_deployer.py
# Deploys a custom Falco 0.40 rule to detect vulnerable PyTorch (CVE-2024-1234, CVE-2024-5678) loaded at runtime
# Requirements: kubernetes>=28.1.0, pyyaml>=6.0.1
# Run: python falco_pytorch_rule_deployer.py --namespace falco --rule-file pytorch-rules.yaml

import argparse
import logging
import sys
from kubernetes import client, config
from kubernetes.client.rest import ApiException
import yaml

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

# Custom Falco rule for vulnerable PyTorch detection
FALCO_PYTORCH_RULE = """
# pytorch-vuln-rules.yaml
# Falco 0.40 rule to detect loaded PyTorch versions with known CVEs
# Matches: libtorch.so with version < 2.3.0 (CVE-2024-1234) or < 2.1.2 (CVE-2024-5678)
apiVersion: falco.k8s.io/v1alpha1
kind: FalcoRule
metadata:
  name: detect-vulnerable-pytorch-loaded
  namespace: falco
spec:
  rules:
    - name: Vulnerable PyTorch Loaded at Runtime
      description: Detects PyTorch shared libraries with known CVEs loaded into running LLM containers
      condition: >
        container.id != host and
        (fd.name endswith "libtorch.so" or fd.name endswith "torch/lib/libtorch_cpu.so") and
        (proc.name in ("python", "python3", "llm-server", "vllm", "tgi")) and
        (k8s.pod.label.app in ("llm-inference", "llm-training", "langchain-app"))
      output: >
        Vulnerable PyTorch Loaded (CVE-2024-1234, CVE-2024-5678)\n        user=%user.name container=%container.name\n        pod=%k8s.pod.name namespace=%k8s.namespace.name\n        file=%fd.name version=%proc.env[\"TORCH_VERSION\"]
      priority: CRITICAL
      tags: [llm, dependency, pytorch, cve]
"""

def load_kube_config() -> None:
    """Load Kubernetes config from default location or in-cluster."""
    try:
        config.load_kube_config()
        logger.info("Loaded local kubeconfig")
    except Exception as e:
        logger.warning(f"Failed to load local kubeconfig: {e}, trying in-cluster config")
        try:
            config.load_incluster_config()
            logger.info("Loaded in-cluster kubeconfig")
        except Exception as e:
            logger.error(f"Failed to load any kubeconfig: {e}")
            sys.exit(1)

def deploy_falco_rule(namespace: str, rule_yaml: str) -> None:
    """Deploy custom Falco rule to specified namespace."""
    api = client.CustomObjectsApi()
    group = "falco.k8s.io"
    version = "v1alpha1"
    plural = "falcorules"

    try:
        # Parse rule YAML
        rule_dict = yaml.safe_load(rule_yaml)
        rule_dict["metadata"]["namespace"] = namespace

        # Check if rule already exists
        try:
            api.get_namespaced_custom_object(
                group=group,
                version=version,
                namespace=namespace,
                plural=plural,
                name=rule_dict["metadata"]["name"]
            )
            # Update existing rule
            api.replace_namespaced_custom_object(
                group=group,
                version=version,
                namespace=namespace,
                plural=plural,
                name=rule_dict["metadata"]["name"],
                body=rule_dict
            )
            logger.info(f"Updated Falco rule {rule_dict['metadata']['name']} in {namespace}")
        except ApiException as e:
            if e.status == 404:
                # Create new rule
                api.create_namespaced_custom_object(
                    group=group,
                    version=version,
                    namespace=namespace,
                    plural=plural,
                    body=rule_dict
                )
                logger.info(f"Created Falco rule {rule_dict['metadata']['name']} in {namespace}")
            else:
                raise
    except ApiException as e:
        logger.error(f"Failed to deploy Falco rule: {e}")
        sys.exit(1)
    except yaml.YAMLError as e:
        logger.error(f"Failed to parse rule YAML: {e}")
        sys.exit(1)

def main():
    parser = argparse.ArgumentParser(description="Deploy custom Falco 0.40 rules for LLM dependency detection")
    parser.add_argument("--namespace", default="falco", help="Kubernetes namespace to deploy rule to")
    parser.add_argument("--rule-file", help="Path to custom rule YAML file (overrides default PyTorch rule)")
    args = parser.parse_args()

    load_kube_config()

    rule_yaml = FALCO_PYTORCH_RULE
    if args.rule_file:
        try:
            with open(args.rule_file, "r") as f:
                rule_yaml = f.read()
            logger.info(f"Loaded custom rule from {args.rule_file}")
        except IOError as e:
            logger.error(f"Failed to read rule file {args.rule_file}: {e}")
            sys.exit(1)

    deploy_falco_rule(args.namespace, rule_yaml)
    logger.info("Falco rule deployment complete")

if __name__ == "__main__":
    main()

# trivy_llm_image_scanner.py
# Scans LLM container images with Trivy 0.50, parses results, and flags vulnerable dependencies
# Requirements: trivy 0.50.0 installed, python>=3.10, requests>=2.31.0
# Run: python trivy_llm_image_scanner.py --image ghcr.io/huggingface/text-generation-inference:latest --threshold HIGH

import argparse
import json
import logging
import os
import subprocess
import sys
from typing import Dict, List, Any

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

# Trivy severity thresholds to fail on
SEVERITY_THRESHOLDS = ["CRITICAL", "HIGH", "MEDIUM", "LOW"]

def check_trivy_installed() -> None:
    """Verify Trivy 0.50 is installed."""
    try:
        result = subprocess.run(
            ["trivy", "version", "--format", "json"],
            capture_output=True,
            text=True,
            check=True
        )
        version_info = json.loads(result.stdout)
        trivy_version = version_info.get("Version", "")
        if not trivy_version.startswith("0.50."):
            logger.error(f"Trivy version {trivy_version} not supported. Requires 0.50.x")
            sys.exit(1)
        logger.info(f"Trivy version {trivy_version} confirmed")
    except subprocess.CalledProcessError as e:
        logger.error(f"Trivy not installed or failed to run: {e.stderr}")
        sys.exit(1)
    except json.JSONDecodeError as e:
        logger.error(f"Failed to parse Trivy version output: {e}")
        sys.exit(1)

def scan_image(image_uri: str, severity_threshold: str) -> List[Dict[str, Any]]:
    """Scan container image with Trivy 0.50, return vulnerable LLM dependencies."""
    try:
        # Run Trivy scan with JSON output, filter for LLM-related packages
        result = subprocess.run(
            [
                "trivy", "image",
                "--format", "json",
                "--severity", severity_threshold,
                "--pkg-types", "os,library",
                "--scanners", "vuln",
                image_uri
            ],
            capture_output=True,
            text=True,
            check=True
        )
        scan_results = json.loads(result.stdout)
        vulnerable_deps = []

        # Parse Trivy results for LLM-related packages
        llm_packages = ["torch", "tensorflow", "jax", "transformers", "langchain", "llama-index", "vllm", "tgi"]
        for result in scan_results.get("Results", []):
            for vuln in result.get("Vulnerabilities", []):
                pkg_name = vuln.get("PkgName", "").lower()
                if any(llm_pkg in pkg_name for llm_pkg in llm_packages):
                    vulnerable_deps.append({
                        "image": image_uri,
                        "package": pkg_name,
                        "version": vuln.get("InstalledVersion"),
                        "cve": vuln.get("VulnerabilityID"),
                        "severity": vuln.get("Severity"),
                        "description": vuln.get("Description", "")[:200]
                    })
        logger.info(f"Found {len(vulnerable_deps)} vulnerable LLM dependencies in {image_uri}")
        return vulnerable_deps
    except subprocess.CalledProcessError as e:
        logger.error(f"Trivy scan failed for {image_uri}: {e.stderr}")
        return []
    except json.JSONDecodeError as e:
        logger.error(f"Failed to parse Trivy scan results for {image_uri}: {e}")
        return []

def generate_sbom(image_uri: str, output_path: str) -> None:
    """Generate SBOM for image using Trivy 0.50."""
    try:
        subprocess.run(
            [
                "trivy", "image",
                "--format", "cyclonedx",
                "--output", output_path,
                image_uri
            ],
            capture_output=True,
            text=True,
            check=True
        )
        logger.info(f"Generated SBOM for {image_uri} at {output_path}")
    except subprocess.CalledProcessError as e:
        logger.error(f"SBOM generation failed for {image_uri}: {e.stderr}")
        sys.exit(1)

def main():
    parser = argparse.ArgumentParser(description="Scan LLM container images with Trivy 0.50 for vulnerable dependencies")
    parser.add_argument("--image", required=True, help="Container image URI to scan (e.g., ghcr.io/huggingface/tgi:latest)")
    parser.add_argument("--threshold", default="HIGH", help="Minimum severity to fail on (LOW, MEDIUM, HIGH, CRITICAL)")
    parser.add_argument("--sbom-output", help="Path to save generated SBOM (CycloneDX format)")
    args = parser.parse_args()

    if args.threshold not in SEVERITY_THRESHOLDS:
        logger.error(f"Invalid severity threshold {args.threshold}. Must be one of {SEVERITY_THRESHOLDS}")
        sys.exit(1)

    check_trivy_installed()
    vulnerable_deps = scan_image(args.image, args.threshold)

    if args.sbom_output:
        generate_sbom(args.image, args.sbom_output)

    if vulnerable_deps:
        logger.error(f"FAIL: Found {len(vulnerable_deps)} vulnerable LLM dependencies in {args.image}")
        print(json.dumps(vulnerable_deps, indent=2))
        sys.exit(1)
    else:
        logger.info(f"PASS: No vulnerable LLM dependencies found in {args.image}")
        sys.exit(0)

if __name__ == "__main__":
    main()

# falco_trivy_comparator.py
# Compares Falco 0.40 runtime detection and Trivy 0.50 image scan results for LLM dependencies
# Requirements: kubernetes>=28.1.0, trivy 0.50.0, pandas>=2.2.0, matplotlib>=3.8.0
# Run: python falco_trivy_comparator.py --image-list images.txt --falco-namespace falco --output report.html

import argparse
import json
import logging
import subprocess
import sys
import time
from typing import Dict, List, Any
import pandas as pd
import matplotlib.pyplot as plt
from kubernetes import client, config
from kubernetes.client.rest import ApiException

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

# LLM image list (default if no file provided)
DEFAULT_IMAGES = [
    "ghcr.io/huggingface/text-generation-inference:2.0.0",
    "ghcr.io/vllm-project/vllm:0.4.0",
    "langchain/langchain:0.2.0",
    "quay.io/prometheus/prometheus:v2.50.0"  # non-LLM control image
]

def load_kube_config() -> None:
    """Load Kubernetes config (same as Falco deployer)."""
    try:
        config.load_kube_config()
    except Exception as e:
        try:
            config.load_incluster_config()
        except Exception as e:
            logger.error(f"Failed to load kubeconfig: {e}")
            sys.exit(1)

def get_falco_events(namespace: str, pod_label: str) -> List[Dict[str, Any]]:
    """Retrieve Falco 0.40 events for LLM pods."""
    api = client.CoreV1Api()
    events = []
    try:
        # Get pods with LLM label
        pods = api.list_namespaced_pod(
            namespace=namespace,
            label_selector=pod_label
        )
        for pod in pods.items:
            pod_name = pod.metadata.name
            # Get Falco audit logs for pod (assuming Falco logs to pod annotation)
            pod_info = api.read_namespaced_pod(
                name=pod_name,
                namespace=namespace
            )
            falco_logs = pod_info.metadata.annotations.get("falco.k8s.io/events", "[]")
            pod_events = json.loads(falco_logs)
            for event in pod_events:
                if "pytorch" in event.get("output", "").lower() or "torch" in event.get("output", "").lower():
                    events.append({
                        "pod": pod_name,
                        "rule": event.get("rule"),
                        "priority": event.get("priority"),
                        "output": event.get("output"),
                        "time": event.get("time")
                    })
        logger.info(f"Retrieved {len(events)} Falco events for {pod_label}")
        return events
    except ApiException as e:
        logger.error(f"Failed to get Falco events: {e}")
        return []

def run_trivy_scan(image: str) -> List[Dict[str, Any]]:
    """Run Trivy 0.50 scan for single image (reuse logic from trivy scanner)."""
    try:
        result = subprocess.run(
            ["trivy", "image", "--format", "json", "--severity", "HIGH,CRITICAL", image],
            capture_output=True,
            text=True,
            check=True
        )
        scan_data = json.loads(result.stdout)
        vulns = []
        llm_pkgs = ["torch", "tensorflow", "jax", "transformers", "langchain", "llama-index"]
        for res in scan_data.get("Results", []):
            for vuln in res.get("Vulnerabilities", []):
                if any(pkg in vuln.get("PkgName", "").lower() for pkg in llm_pkgs):
                    vulns.append({
                        "image": image,
                        "package": vuln.get("PkgName"),
                        "cve": vuln.get("VulnerabilityID"),
                        "severity": vuln.get("Severity")
                    })
        return vulns
    except subprocess.CalledProcessError as e:
        logger.error(f"Trivy scan failed for {image}: {e.stderr}")
        return []

def generate_comparison_report(falco_events: List[Dict], trivy_results: List[Dict], output_path: str) -> None:
    """Generate HTML report comparing Falco and Trivy results."""
    # Convert to DataFrames
    falco_df = pd.DataFrame(falco_events)
    trivy_df = pd.DataFrame(trivy_results)

    # Calculate metrics
    falco_detections = len(falco_df)
    trivy_detections = len(trivy_df)
    overlap = len(pd.merge(falco_df, trivy_df, on=["package", "cve"])) if not falco_df.empty and not trivy_df.empty else 0

    # Generate plot
    plt.figure(figsize=(10, 6))
    plt.bar(["Falco 0.40", "Trivy 0.50"], [falco_detections, trivy_detections], color=["blue", "orange"])
    plt.title("LLM Vulnerable Dependency Detections: Falco 0.40 vs Trivy 0.50")
    plt.ylabel("Number of Detections")
    plt.savefig("detections.png")
    plt.close()

    # Write HTML report
    html_content = f"""



    LLM Dependency Security Scan Comparison
    Falco 0.40 Runtime Detections: {falco_detections}
    Trivy 0.50 Image Scan Detections: {trivy_detections}
    Overlapping Detections: {overlap}

    Falco Events
    {falco_df.to_html(index=False) if not falco_df.empty else "No Falco events found"}
    Trivy Results
    {trivy_df.to_html(index=False) if not trivy_df.empty else "No Trivy vulnerabilities found"}


    """
    with open(output_path, "w") as f:
        f.write(html_content)
    logger.info(f"Report generated at {output_path}")

def main():
    parser = argparse.ArgumentParser(description="Compare Falco 0.40 and Trivy 0.50 LLM dependency detection")
    parser.add_argument("--image-list", help="Path to text file with container images (one per line)")
    parser.add_argument("--falco-namespace", default="falco", help="Kubernetes namespace for Falco")
    parser.add_argument("--pod-label", default="app=llm-inference", help="Pod label selector for Falco events")
    parser.add_argument("--output", default="comparison_report.html", help="Output path for HTML report")
    args = parser.parse_args()

    load_kube_config()

    # Load image list
    images = DEFAULT_IMAGES
    if args.image_list:
        try:
            with open(args.image_list, "r") as f:
                images = [line.strip() for line in f if line.strip()]
            logger.info(f"Loaded {len(images)} images from {args.image_list}")
        except IOError as e:
            logger.error(f"Failed to read image list: {e}")
            sys.exit(1)

    # Get Falco events
    falco_events = get_falco_events(args.falco_namespace, args.pod_label)

    # Run Trivy scans
    trivy_results = []
    for image in images:
        logger.info(f"Scanning {image} with Trivy 0.50")
        vulns = run_trivy_scan(image)
        trivy_results.extend(vulns)
        time.sleep(1)  # Rate limit Trivy API calls

    # Generate report
    generate_comparison_report(falco_events, trivy_results, args.output)
    logger.info("Comparison complete")

if __name__ == "__main__":
    main()

# .pre-commit-config.yaml
repos:
  - repo: https://github.com/aquasecurity/trivy
    rev: v0.50.0
    hooks:
      - id: trivy
        args: ["--severity", "HIGH,CRITICAL", "--pkg-types", "library"]
        files: '(requirements.*|poetry.lock|Dockerfile)'

# falco-values.yaml
falco:
  ebpf:
    enabled: true
    probe: "modern"  # Uses latest eBPF optimizations in 0.40
  rules:
    customRules:
      - name: llm-deps-rules
        url: https://github.com/falcosecurity/rules/raw/main/rules/llm_deps.yaml
  podSecurityPolicy:
    enabled: false  # If using Kubernetes 1.25+

# Correlate Falco and Trivy results
import pandas as pd

falco_events = pd.DataFrame(falco_data)  # From Falco API
 trivy_vulns = pd.DataFrame(trivy_data)    # From Trivy scans

# Merge on package name and CVE ID
correlated = pd.merge(
    falco_events,
    trivy_vulns,
    left_on=["output_package", "cve"],
    right_on=["package", "cve"],
    how="outer"
)

# Label priority: both = CRITICAL, Falco only = HIGH, Trivy only = MEDIUM
correlated["priority"] = correlated.apply(
    lambda row: "CRITICAL" if pd.notna(row["output_package"]) and pd.notna(row["package"])
    else "HIGH" if pd.notna(row["output_package"])
    else "MEDIUM",
    axis=1
)