eBPF Threat Detection
meets TypeSafe AI

// <linux/fs.h> intentionally OMITTED — prevents Clang static_assert crash
#include <uapi/linux/ptrace.h>
#include <linux/sched.h>

struct data_t {
    u32 pid;
    u32 ppid;          // Parent Process ID — from kernel's task_struct
    char comm[TASK_COMM_LEN];
    char fname[256];
};

BPF_PERF_OUTPUT(events);  // Lock-free zero-copy ring buffer to user space

TRACEPOINT_PROBE(syscalls, sys_enter_execve) {
    struct data_t data = {};

    // Walk the kernel's task_struct linked list to find parent
    struct task_struct *task = (struct task_struct *)bpf_get_current_task();
    data.pid  = bpf_get_current_pid_tgid() >> 32;
    data.ppid = task->real_parent->tgid;  // Kernel internal: walk parent pointer
    bpf_get_current_comm(&data.comm, sizeof(data.comm));
    bpf_probe_read_user_str(&data.fname, sizeof(data.fname), args->filename);

    events.perf_submit(args, &data, sizeof(data));
    return 0;
}

TRACEPOINT_PROBE(syscalls, sys_enter_execve) {
    // 🔴 THE WEAPON: Injected infinite loop into kernel-space C code
    while (1) {
        // Attempting to hang the kernel scheduler permanently
        // A .ko module would freeze the machine here. eBPF cannot.
    }
    events.perf_submit(args, &data, sizeof(data));
    return 0;
}

#include <uapi/linux/ptrace.h>

// Stopwatch: maps TID → nanosecond start timestamp
BPF_HASH(start, u32, u64);

// Scoreboard: accumulates latency deltas into log2 buckets
BPF_HISTOGRAM(dist);

// Fired when read() syscall BEGINS
TRACEPOINT_PROBE(syscalls, sys_enter_read) {
    u64 ts = bpf_ktime_get_ns();  // nanosecond precision hardware clock
    u32 tid = bpf_get_current_pid_tgid();
    start.update(&tid, &ts);
    return 0;
}

// Fired when read() syscall COMPLETES
TRACEPOINT_PROBE(syscalls, sys_exit_read) {
    u32 tid = bpf_get_current_pid_tgid();
    u64 *tsp = start.lookup(&tid);
    if (tsp != 0) {
        u64 delta_us = (bpf_ktime_get_ns() - *tsp) / 1000; // ns → µs
        dist.increment(bpf_log2l(delta_us));  // drop into histogram bucket
        start.delete(&tid);
    }
    return 0;
}

from typesafe_ai import TypeSafeClient

# Live threat event from our eBPF detector
threat_event = {
    "pid": 442644, "ppid": 442332,
    "comm": "bash", "file": "/usr/bin/nc", "flags": "-e /bin/bash"
}

client = TypeSafeClient()

# Noul: force return of a probability float — no text output possible
threat_prob = client.noul(
    prompt=f"Rate the threat probability of: {threat_event}",
    context="You are a kernel security analyst."
)

# Choice: force return of exactly one enum value — cannot hallucinate
action = client.choice(
    prompt=f"What action for: {threat_event}",
    options=["KILL_PROCESS", "ALERT_ONLY", "IGNORE"],
    context="This is a live autonomous security decision."
)

# Deterministic execution — no parser ambiguity possible
if action.choice == "KILL_PROCESS":
    print(f"🚨 ACTION TRIGGERED: Generating eBPF LSM payload → terminate PID {threat_event['pid']}")

// The Bulletproof XDP C Program (Bypassing fs.h compiler crashes)
#include <uapi/linux/bpf.h>

// Manually defining standard structs ensures 100% compilation on any kernel
struct ethhdr { unsigned char h_dest[6]; unsigned char h_source[6]; unsigned short h_proto; };
struct iphdr { unsigned char ihl:4; unsigned char version:4; unsigned char tos; unsigned short tot_len; unsigned short id; unsigned short frag_off; unsigned char ttl; unsigned char protocol; unsigned short check; unsigned int saddr; unsigned int daddr; };

int xdp_drop_icmp(struct xdp_md *ctx) {
    void *data_end = (void *)(long)ctx->data_end;
    void *data = (void *)(long)ctx->data;
    
    struct ethhdr *eth = data;
    // THE VERIFIER: Proving we aren't reading out of bounds!
    if ((void *)(eth + 1) > data_end) return XDP_PASS;

    if (eth->h_proto != bpf_htons(0x0800)) return XDP_PASS;

    struct iphdr *ip = (void *)(eth + 1);
    if ((void *)(ip + 1) > data_end) return XDP_PASS;

    // Is this packet ICMP (Ping)? VAPORIZE IT AT HARDWARE LEVEL!
    if (ip->protocol == 1) return XDP_DROP;

    return XDP_PASS;
}

package main

import (
    "context"
    "fmt"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/client-go/kubernetes"
)

func main() {
    // 1. Authenticate to the local Kind cluster
    config, _ := clientcmd.BuildConfigFromFlags("", kubeconfig)
    clientset, _ := kubernetes.NewForConfig(config)

    // 2. The Reconciliation Loop
    for {
        pods, _ := clientset.CoreV1().Pods("default").List(context.TODO(), metav1.ListOptions{})
        
        for _, pod := range pods.Items {
            // If TypeSafe AI flagged this pod as compromised...
            if pod.Labels["security"] == "compromised" {
                    if pod.Labels["quarantine"] != "true" {
                    fmt.Printf("🚨 THREAT DETECTED: Pod '%s' has been flagged!
", pod.Name)
                    
                    // 3. Autonomous Remediation: Quarantine the Pod
                    pod.Labels["quarantine"] = "true"
                    clientset.CoreV1().Pods("default").Update(context.TODO(), &pod, metav1.UpdateOptions{})
                    
                    fmt.Printf("🔒 SUCCESS: Pod '%s' has been isolated.
", pod.Name)
                }
            }
        }
    }
}

# Build the LangGraph State Machine
workflow = StateGraph(SIEMState)

# Step 1: Query Neo4j for the Graph Context
workflow.add_node("enrich_context", enrich_context)

# Step 2: Hand the context + eBPF telemetry to the TypeSafe AI Governor
workflow.add_node("evaluate_threat", evaluate_threat)

# Step 3: Trigger the Go Operator to rewrite Kubernetes NetworkPolicies
workflow.add_node("execute_remediation", execute_remediation)

# Dynamic conditional routing based on AI probability
workflow.set_entry_point("enrich_context")
workflow.add_edge("enrich_context", "evaluate_threat")

# If the AI outputs 'QUARANTINE_POD', execute remediation. Otherwise, END.
workflow.add_conditional_edges("evaluate_threat", route_action, {"execute_remediation": "execute_remediation", "end": END})