"""Execute a command whose process tree cannot read the eval harness. Runner E2E starts the product from a repository checkout that also holds the scenarios, fixtures and graders. This helper applies a Linux Landlock domain that grants filesystem access everywhere except the given hidden paths, then replaces itself with the command (same PID, argv and environment). Every descendant inherits the domain, including an agent's shell commands. Usage: agent-filesystem-sandbox.py [--required] [--unset NAME]... [--set NAME=VALUE]... --hide PATH [--hide PATH]... -- COMMAND [ARG]... Without Landlock, --required exits 125 instead of running the command. """ import ctypes import os import stat import struct import sys SYS_LANDLOCK_CREATE_RULESET = 444 SYS_LANDLOCK_ADD_RULE = 445 SYS_LANDLOCK_RESTRICT_SELF = 446 LANDLOCK_CREATE_RULESET_VERSION = 1 LANDLOCK_RULE_PATH_BENEATH = 1 PR_SET_NO_NEW_PRIVS = 38 FILE_RIGHTS = (1 << 0) | (1 << 1) | (1 << 2) # execute, write, read ABI_1_RIGHTS = (1 << 13) - 1 REFER = 1 << 13 TRUNCATE = 1 << 14 IOCTL_DEV = 1 << 15 class RulesetAttr(ctypes.Structure): _fields_ = [("handled_access_fs", ctypes.c_uint64)] def fail(message, code=125): sys.stderr.write("agent_filesystem_sandbox: %s\n" % message) sys.stderr.flush() os._exit(code) def parse(argv): required = False hidden = [] unset = [] assignments = [] index = 0 while index < len(argv): token = argv[index] if token == "--": command = argv[index + 1:] if not command: fail("missing command", 2) return required, hidden, unset, assignments, command if token == "--required": required = True elif token in ("--hide", "--unset", "--set") and index + 1 < len(argv): index += 1 value = argv[index] if token == "--hide": if not os.path.isabs(value): fail("hidden path must be absolute: %s" % value, 2) hidden.append(value) elif token == "--unset": unset.append(value) else: name, separator, assigned = value.partition("=") if not separator or not name: fail("--set requires NAME=VALUE", 2) assignments.append((name, assigned)) else: fail("unexpected argument: %s" % token, 2) index += 1 fail("missing -- before the command", 2) def within(candidate, root): return candidate == root or candidate.startswith(root.rstrip("/") + "/") def landlock(): libc = ctypes.CDLL(None, use_errno=True) libc.syscall.restype = ctypes.c_long def syscall(*args): result = libc.syscall(*args) if result < 0: raise OSError(ctypes.get_errno(), os.strerror(ctypes.get_errno())) return result return libc, syscall def restrict(hidden): libc, syscall = landlock() try: abi = syscall( ctypes.c_long(SYS_LANDLOCK_CREATE_RULESET), ctypes.c_void_p(None), ctypes.c_long(0), ctypes.c_long(LANDLOCK_CREATE_RULESET_VERSION), ) except OSError as error: raise RuntimeError("Landlock is unavailable on this kernel (%s)" % error) handled = ABI_1_RIGHTS if abi >= 2: handled |= REFER if abi >= 3: handled |= TRUNCATE if abi >= 5: handled |= IOCTL_DEV attr = RulesetAttr(handled) ruleset = syscall( ctypes.c_long(SYS_LANDLOCK_CREATE_RULESET), ctypes.byref(attr), ctypes.c_long(ctypes.sizeof(attr)), ctypes.c_long(0), ) def allow(path): try: descriptor = os.open(path, os.O_PATH | os.O_CLOEXEC) except OSError: return try: mode = os.fstat(descriptor).st_mode access = handled if stat.S_ISDIR(mode) else handled & (FILE_RIGHTS | TRUNCATE | IOCTL_DEV) # struct landlock_path_beneath_attr is packed: u64 allowed_access, s32 parent_fd. rule = ctypes.create_string_buffer(struct.pack("=Qi", access, descriptor), 12) syscall( ctypes.c_long(SYS_LANDLOCK_ADD_RULE), ctypes.c_long(ruleset), ctypes.c_long(LANDLOCK_RULE_PATH_BENEATH), rule, ctypes.c_long(0), ) finally: os.close(descriptor) # Landlock only grants. Grant every entry except the hidden paths, walking # down only through their ancestors; ancestors themselves get no rule. def grant(directory): try: names = sorted(os.listdir(directory)) except OSError: return for name in names: child = os.path.join(directory, name) if any(within(child, path) for path in hidden): continue try: is_link = os.path.islink(child) real = os.path.realpath(child) except OSError: continue if any(within(real, path) or within(path, real) for path in hidden): if not is_link and any(within(path, child) for path in hidden): grant(child) # A link into, or onto an ancestor of, a hidden path stays ungranted. continue allow(child) grant("/") if libc.prctl(ctypes.c_int(PR_SET_NO_NEW_PRIVS), ctypes.c_ulong(1), ctypes.c_ulong(0), ctypes.c_ulong(0), ctypes.c_ulong(0)) != 0: raise RuntimeError("prctl(PR_SET_NO_NEW_PRIVS) failed") syscall(ctypes.c_long(SYS_LANDLOCK_RESTRICT_SELF), ctypes.c_long(ruleset), ctypes.c_long(0)) os.close(ruleset) # Fail closed unless this process (which becomes the command) is denied. for path in hidden: try: if os.path.isdir(path): os.listdir(path) else: os.close(os.open(path, os.O_RDONLY | os.O_CLOEXEC)) except PermissionError: continue except OSError: continue raise RuntimeError("sandbox ineffective: %s is still readable" % path) return abi def main(): required, hidden_inputs, unset, assignments, command = parse(sys.argv[1:]) hidden = [] for path in hidden_inputs: # Keep paths that do not exist yet: their ancestors stay ungranted, so # a later mkdir there is unreachable too. real = os.path.realpath(path) if real == "/": fail("refusing to hide the filesystem root", 2) hidden.append(real) for name in unset: os.environ.pop(name, None) for name, value in assignments: os.environ[name] = value if hidden: if not sys.platform.startswith("linux"): if required: fail("Landlock requires Linux") else: try: restrict(hidden) except (OSError, RuntimeError) as error: if required: fail(str(error)) sys.stderr.write("agent_filesystem_sandbox: continuing without isolation: %s\n" % error) try: if os.path.isabs(command[0]): os.execv(command[0], command) os.execvp(command[0], command) except OSError as error: fail("cannot execute %s: %s" % (command[0], error), 127) if __name__ == "__main__": main()