Build an AI Runtime Anomaly Detector for Kubernetes with Claude API and eBPF

# TracingPolicy — capture process execs and network connections per pod
apiVersion: cilium.io/v1alpha1
kind: TracingPolicy
metadata:
  name: runtime-behavior-monitor
spec:
  kprobes:
    - call: "sys_execve"
      syscall: true
      args:
        - index: 0
          type: "string"
    - call: "tcp_connect"
      syscall: false
      args:
        - index: 0
          type: "sock"

import json
import subprocess
from collections import defaultdict
 
def stream_tetragon_events():
    """Consume the Tetragon gRPC/JSON event stream."""
    proc = subprocess.Popen(
        ["tetra", "getevents", "-o", "json"],
        stdout=subprocess.PIPE, text=True
    )
    for line in proc.stdout:
        yield json.loads(line)

def summarize_pod_behavior(pod_name: str, events: list[dict], window_seconds: int = 60) -> dict:
    """Turn a raw event stream into a compact behavioral summary."""
    process_execs = [e for e in events if e.get("process_exec")]
    network_connects = [e for e in events if e.get("process_connect")]
 
    return {
        "pod": pod_name,
        "window_seconds": window_seconds,
        "unique_processes_spawned": list(set(e["process_exec"]["binary"] for e in process_execs)),
        "network_destinations": list(set(e["process_connect"]["destination"] for e in network_connects)),
        "unusual_exec_count": len(process_execs),
        "sample_exec_chain": [e["process_exec"]["binary"] for e in process_execs[:20]],
    }

import anthropic
 
client = anthropic.Anthropic()
 
ANALYZE_PROMPT = """Analyze this container's runtime behavior for anomalies.
 
Pod: {pod}
Time window: {window_seconds}s
Processes spawned: {processes}
Network destinations contacted: {destinations}
Process exec chain sample: {exec_chain}
 
Baseline expected behavior for this pod (from its normal operating pattern
over the last 7 days): {baseline}
 
Flag as suspicious or malicious if you see patterns like:
- Process spawning a shell (sh, bash) when the app normally never does
- Network connections to destinations outside the known service mesh/allowlist
- Binaries in /tmp or unusual paths being executed
- Cryptomining indicators (sustained high CPU exec chains to mining-pool-like destinations)
- Reconnaissance patterns (whoami, id, cat /etc/passwd, uname -a in sequence)
 
Respond with ONLY valid JSON:
{{"verdict": "normal"|"suspicious"|"malicious", "confidence": "high"|"medium"|"low",
  "evidence": "specific reasoning", "recommended_action": "..."}}"""
 
 
def analyze_behavior(summary: dict, baseline: dict) -> dict:
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=500,
        messages=[{
            "role": "user",
            "content": ANALYZE_PROMPT.format(
                pod=summary["pod"], window_seconds=summary["window_seconds"],
                processes=summary["unique_processes_spawned"],
                destinations=summary["network_destinations"],
                exec_chain=summary["sample_exec_chain"],
                baseline=baseline,
            )
        }]
    )
    text = response.content[0].text.strip()
    if text.startswith("```"):
        text = text.split("```")[1].replace("json", "", 1).strip()
    import json
    return json.loads(text)

def monitor_and_respond():
    events_by_pod = defaultdict(list)
 
    for event in stream_tetragon_events():
        pod = event.get("pod_name", "unknown")
        events_by_pod[pod].append(event)
 
        if len(events_by_pod[pod]) >= 50:    # Batch, don't analyze every single event
            summary = summarize_pod_behavior(pod, events_by_pod[pod])
            baseline = get_pod_baseline(pod)
            verdict = analyze_behavior(summary, baseline)
 
            if verdict["verdict"] == "malicious" and verdict["confidence"] == "high":
                isolate_pod(pod)    # NetworkPolicy that blocks all egress
                page_security_oncall(pod, verdict["evidence"])
            elif verdict["verdict"] in ("suspicious", "malicious"):
                notify_security_channel(pod, verdict)
 
            events_by_pod[pod] = []    # Reset window

def build_baseline(pod_label: str, days: int = 7) -> dict:
    """Run this on a schedule against historical event data to keep
    baselines current as legitimate app behavior evolves."""
    historical_events = query_historical_events(pod_label, days=days)
    return {
        "normal_processes": extract_common_processes(historical_events),
        "normal_destinations": extract_common_destinations(historical_events),
    }