Merge branch 'master' into request-struct

tests/c-form/i32test.bpf.c

@@ -0,0 +1,15 @@
+#include <linux/bpf.h>
+#include <bpf/bpf_helpers.h>
+
+SEC("xdp")
+int print_xdp_data(struct xdp_md *ctx)
+{
+    // 'data' is a pointer to the start of packet data
+    long data = (long)ctx->data;
+
+    bpf_printk("ctx->data = %lld\n", data);
+
+    return XDP_PASS;
+}
+
+char LICENSE[] SEC("license") = "GPL";

tests/failing_tests/vmlinux/args_test.py

@@ -0,0 +1,30 @@
+import logging
+
+from pythonbpf import bpf, section, bpfglobal, compile_to_ir
+from pythonbpf import compile  # noqa: F401
+from vmlinux import TASK_COMM_LEN  # noqa: F401
+from vmlinux import struct_trace_event_raw_sys_enter  # noqa: F401
+from ctypes import c_int64, c_int32, c_void_p  # noqa: F401
+
+
+# from vmlinux import struct_uinput_device
+# from vmlinux import struct_blk_integrity_iter
+
+
+@bpf
+@section("tracepoint/syscalls/sys_enter_execve")
+def hello_world(ctx: struct_trace_event_raw_sys_enter) -> c_int64:
+    b = ctx.args
+    c = b[0]
+    print(f"This is context args field {c}")
+    return c_int64(0)
+
+
+@bpf
+@bpfglobal
+def LICENSE() -> str:
+    return "GPL"
+
+
+compile_to_ir("args_test.py", "args_test.ll", loglevel=logging.INFO)
+compile()

tests/passing_tests/hash_map_struct.py

@@ -0,0 +1,42 @@
+from pythonbpf import bpf, section, struct, bpfglobal, compile, map
+from pythonbpf.maps import HashMap
+from pythonbpf.helper import pid
+from ctypes import c_void_p, c_int64
+
+
+@bpf
+@struct
+class val_type:
+    counter: c_int64
+    shizzle: c_int64
+
+
+@bpf
+@map
+def last() -> HashMap:
+    return HashMap(key=val_type, value=c_int64, max_entries=16)
+
+
+@bpf
+@section("tracepoint/syscalls/sys_enter_clone")
+def hello_world(ctx: c_void_p) -> c_int64:
+    obj = val_type()
+    obj.counter, obj.shizzle = 42, 96
+    t = last.lookup(obj)
+    if t:
+        print(f"Found existing entry: counter={obj.counter}, pid={t}")
+        last.delete(obj)
+        return 0  # type: ignore [return-value]
+    val = pid()
+    last.update(obj, val)
+    print(f"Map updated!, {obj.counter}, {obj.shizzle}, {val}")
+    return 0  # type: ignore [return-value]
+
+
+@bpf
+@bpfglobal
+def LICENSE() -> str:
+    return "GPL"
+
+
+compile()

tests/passing_tests/helpers/bpf_probe_read.py

@@ -0,0 +1,29 @@
+from pythonbpf import bpf, section, bpfglobal, compile, struct
+from ctypes import c_void_p, c_int64, c_uint64, c_uint32
+from pythonbpf.helper import probe_read
+
+
+@bpf
+@struct
+class data_t:
+    pid: c_uint32
+    value: c_uint64
+
+
+@bpf
+@section("tracepoint/syscalls/sys_enter_execve")
+def test_probe_read(ctx: c_void_p) -> c_int64:
+    """Test bpf_probe_read helper function"""
+    data = data_t()
+    probe_read(data.value, 8, ctx)
+    probe_read(data.pid, 4, ctx)
+    return 0
+
+
+@bpf
+@bpfglobal
+def LICENSE() -> str:
+    return "GPL"
+
+
+compile()

tests/passing_tests/helpers/prandom.py

@@ -0,0 +1,25 @@
+from pythonbpf import bpf, bpfglobal, section, BPF, trace_pipe
+from ctypes import c_void_p, c_int64
+from pythonbpf.helper import random
+
+
+@bpf
+@section("tracepoint/syscalls/sys_enter_clone")
+def hello_world(ctx: c_void_p) -> c_int64:
+    r = random()
+    print(f"Hello, World!, {r}")
+    return 0  # type: ignore [return-value]
+
+
+@bpf
+@bpfglobal
+def LICENSE() -> str:
+    return "GPL"
+
+
+# Compile and load
+b = BPF()
+b.load()
+b.attach_all()
+
+trace_pipe()

tests/passing_tests/helpers/smp_processor_id.py

@@ -0,0 +1,40 @@
+from pythonbpf import bpf, section, bpfglobal, compile, struct
+from ctypes import c_void_p, c_int64, c_uint32, c_uint64
+from pythonbpf.helper import smp_processor_id, ktime
+
+
+@bpf
+@struct
+class cpu_event_t:
+    cpu_id: c_uint32
+    timestamp: c_uint64
+
+
+@bpf
+@section("tracepoint/syscalls/sys_enter_execve")
+def trace_with_cpu(ctx: c_void_p) -> c_int64:
+    """Test bpf_get_smp_processor_id helper function"""
+
+    # Get the current CPU ID
+    cpu = smp_processor_id()
+
+    # Print it
+    print(f"Running on CPU {cpu}")
+
+    # Use it in a struct
+    event = cpu_event_t()
+    event.cpu_id = smp_processor_id()
+    event.timestamp = ktime()
+
+    print(f"Event on CPU {event.cpu_id} at time {event.timestamp}")
+
+    return 0
+
+
+@bpf
+@bpfglobal
+def LICENSE() -> str:
+    return "GPL"
+
+
+compile()

tests/passing_tests/helpers/uid_gid.py

@@ -0,0 +1,31 @@
+from pythonbpf import bpf, section, bpfglobal, compile
+from ctypes import c_void_p, c_int64
+from pythonbpf.helper import uid, pid
+
+
+@bpf
+@section("tracepoint/syscalls/sys_enter_execve")
+def filter_by_user(ctx: c_void_p) -> c_int64:
+    """Filter events by specific user ID"""
+
+    current_uid = uid()
+
+    # Only trace root user (UID 0)
+    if current_uid == 0:
+        process_id = pid()
+        print(f"Root process {process_id} executed")
+
+    # Or trace specific user (e.g., UID 1000)
+    if current_uid == 1002:
+        print("User 1002 executed something")
+
+    return 0
+
+
+@bpf
+@bpfglobal
+def LICENSE() -> str:
+    return "GPL"
+
+
+compile()

tests/passing_tests/vmlinux/i32_test.py

@@ -0,0 +1,31 @@
+from ctypes import c_int64, c_void_p
+from pythonbpf import bpf, section, bpfglobal, compile_to_ir, compile
+from vmlinux import struct_xdp_md
+from vmlinux import XDP_PASS
+
+
+@bpf
+@section("xdp")
+def print_xdp_dat2a(ct2x: struct_xdp_md) -> c_int64:
+    data = ct2x.data  # 32-bit field: packet start pointer
+    print(f"ct2x->data = {data}")
+    return c_int64(XDP_PASS)
+
+
+@bpf
+@section("xdp")
+def print_xdp_data(ctx: struct_xdp_md) -> c_int64:
+    data = ctx.data  # 32-bit field: packet start pointer
+    something = c_void_p(data)
+    print(f"ctx->data = {something}")
+    return c_int64(XDP_PASS)
+
+
+@bpf
+@bpfglobal
+def LICENSE() -> str:
+    return "GPL"
+
+
+compile_to_ir("i32_test.py", "i32_test.ll")
+compile()

tests/passing_tests/vmlinux/i32_test_fail_1.py

@@ -0,0 +1,24 @@
+from ctypes import c_int64
+from pythonbpf import bpf, section, bpfglobal, compile
+from vmlinux import struct_xdp_md
+from vmlinux import XDP_PASS
+import logging
+
+
+@bpf
+@section("xdp")
+def print_xdp_data(ctx: struct_xdp_md) -> c_int64:
+    data = 0
+    data = ctx.data  # 32-bit field: packet start pointer
+    something = 2 + data
+    print(f"ctx->data = {something}")
+    return c_int64(XDP_PASS)
+
+
+@bpf
+@bpfglobal
+def LICENSE() -> str:
+    return "GPL"
+
+
+compile(logging.INFO)

tests/passing_tests/vmlinux/i32_test_fail_2.py

@@ -0,0 +1,24 @@
+from ctypes import c_int64
+from pythonbpf import bpf, section, bpfglobal, compile, compile_to_ir
+from vmlinux import struct_xdp_md
+from vmlinux import XDP_PASS
+import logging
+
+
+@bpf
+@section("xdp")
+def print_xdp_data(ctx: struct_xdp_md) -> c_int64:
+    data = c_int64(ctx.data)  # 32-bit field: packet start pointer
+    something = 2 + data
+    print(f"ctx->data = {something}")
+    return c_int64(XDP_PASS)
+
+
+@bpf
+@bpfglobal
+def LICENSE() -> str:
+    return "GPL"
+
+
+compile_to_ir("i32_test_fail_2.py", "i32_test_fail_2.ll")
+compile(logging.INFO)

tests/passing_tests/vmlinux/simple_struct_test.py

@@ -44,4 +44,4 @@ def LICENSE() -> str:
 
 
 compile_to_ir("simple_struct_test.py", "simple_struct_test.ll", loglevel=logging.DEBUG)
-# compile()
+compile()