try

let _ = Str.search_forward (Str.regexp_string substr) s 0 in

true

let test_cases = [

(* Basic exec() call *)

{|

fn main() -> i32 {

exec("./script.py")

return 0

}

|}, "basic exec call";

(* exec() with string variable *)

{|

fn main() -> i32 {

var script = "./analysis.py"

exec(script)

return 0

}

|}, "exec with variable";

(* exec() in conditional *)

{|

fn main() -> i32 {

if (condition) {

exec("./handler.py")

}

return 0

}

|}, "exec in conditional";

] in

List.iter (fun (code, name) ->

try

let _ = parse_string code in

() (* Successful parse *)

with

| e -> failwith (Printf.sprintf "%s: Parse error: %s" name (Printexc.to_string e))

(* Test that basic valid syntax parses *)

let valid_cases = [

{|

fn main() -> i32 {

exec("./script.py")

return 0

}

|}, "python file";

{|

fn main() -> i32 {

exec("./script.sh")

return 0

}

|}, "shell script"; (* Parser won't reject this, validation happens later *)

] in

List.iter (fun (code, name) ->

try

let _ = parse_string code in

() (* Parser accepts all string arguments *)

with

| e -> failwith (Printf.sprintf "%s: Unexpected parse error: %s" name (Printexc.to_string e))

(* Test that the Python wrapper template contains expected libbpf components *)

let wrapper_content = {|

import os

import ctypes

import ctypes.util

# Load libbpf for proper BPF operations

def find_libbpf():

"""Find libbpf library with fallback options"""

for lib_name in ['libbpf.so.1', 'libbpf.so.0', 'libbpf.so']:

try:

return ctypes.CDLL(lib_name)

except OSError:

continue

raise RuntimeError("libbpf not found")

libbpf = find_libbpf()

# Define libbpf function signatures

libbpf.bpf_map_lookup_elem.argtypes = [ctypes.c_int, ctypes.c_void_p, ctypes.c_void_p]

libbpf.bpf_map_lookup_elem.restype = ctypes.c_int

def _initialize_maps():

"""Initialize map objects from inherited file descriptors"""

map_fds_json = os.environ.get('KERNELSCRIPT_MAP_FDS')

if not map_fds_json:

return {}

try:

map_fds = json.loads(map_fds_json)

except json.JSONDecodeError as e:

raise RuntimeError(f"Invalid map FDs JSON: {e}")

maps = {}

for name, metadata in MAP_METADATA.items():

if name not in map_fds:

continue

fd = map_fds[name]

# Validate file descriptor

try:

import fcntl

fcntl.fcntl(fd, fcntl.F_GETFD)

except OSError as e:

print(f"ERROR: File descriptor {fd} for map '{name}' is invalid: {e}")

continue

maps[name] = fd

return maps

# Use .get() for robust map access

test_map = _maps.get('test_map')

|} in

(* Check for key components *)

let has_libbpf_loading = string_contains wrapper_content "find_libbpf()" in

let has_libbpf_functions = string_contains wrapper_content "libbpf.bpf_map_lookup_elem" in

let has_json_error_handling = string_contains wrapper_content "json.JSONDecodeError" in

let has_fd_validation = string_contains wrapper_content "fcntl.fcntl(fd, fcntl.F_GETFD)" in

let has_robust_access = string_contains wrapper_content "_maps.get(" in

check bool "libbpf loading mechanism" true has_libbpf_loading;

check bool "libbpf function bindings" true has_libbpf_functions;

check bool "JSON decode error handling" true has_json_error_handling;

check bool "file descriptor validation" true has_fd_validation;

(* Test that the C template contains expected FD_CLOEXEC clearing components *)

let c_content = {|

void exec_builtin(const char* python_script) {

// Create JSON with map name -> fd mapping for global maps

char map_fds_json[1024];

snprintf(map_fds_json, sizeof(map_fds_json), "{\"test_map\":%d}", test_map_fd);

setenv("KERNELSCRIPT_MAP_FDS", map_fds_json, 1);

// Clear FD_CLOEXEC flags to ensure file descriptors survive exec()

fcntl(test_map_fd, F_SETFD, fcntl(test_map_fd, F_GETFD) & ~FD_CLOEXEC);

// Execute Python - file descriptors automatically inherited!

char* args[] = {"python3", (char*)python_script, NULL};

execvp("python3", args);

perror("execvp failed");

exit(1);

}

|} in

(* Check for key FD_CLOEXEC components *)

let has_json_generation = string_contains c_content "snprintf(map_fds_json" in

let has_setenv = string_contains c_content "setenv(\"KERNELSCRIPT_MAP_FDS\"" in

let has_fcntl_call = string_contains c_content "fcntl(test_map_fd, F_SETFD" in

let has_fd_cloexec_mask = string_contains c_content "& ~FD_CLOEXEC" in

let has_execvp = string_contains c_content "execvp(\"python3\"" in

check bool "JSON generation for map FDs" true has_json_generation;

check bool "environment variable setup" true has_setenv;

check bool "fcntl call for FD clearing" true has_fcntl_call;

check bool "FD_CLOEXEC mask operation" true has_fd_cloexec_mask;

let code_with_exec = {|

fn main() -> i32 {

exec("./script.py")

return 0

}

|} in

let code_without_exec = {|

fn main() -> i32 {

print("Hello")

return 0

}

|} in

(* Test that both parse successfully *)

let test_parse code name =

try

let _ = parse_string code in

()

with

| e -> failwith (Printf.sprintf "%s failed: %s" name (Printexc.to_string e))

in

test_parse code_with_exec "code with exec";