#!/usr/bin/python3 -I """iso_exec.py - unprivilegierter Isolationskaefig (Landlock ABI>=4 + seccomp + rlimits). Aufruf: iso_exec.py CFG.json -- /abs/prog arg... Herkunft: abgeleitet aus scratchpad/codex-iso/iso_exec.py (Isolationsvorschlag ยง4). Gegenueber dem Original behoben (Review befunde.md): M2 memfd_create wird verboten, execveat(AT_EMPTY_PATH) wird verboten -> der memfd+execveat-Ausbruch der EXEC-Allowlist ist geschlossen. M3 RLIMIT_AS und RLIMIT_CPU (plus FSIZE/NOFILE, optional NPROC) werden im Wrapper selbst gesetzt, nicht nur vom aeusseren systemd-Scope. N1 vererbte FDs werden VOR restrict_self/seccomp geschlossen (3..high). N2 umask 077 und cwd werden gesetzt. N3 das Landlock-FS-Handled-Set wird aus der GEMELDETEN ABI abgeleitet (ABI5+ ergaenzt IOCTL_DEV); ABI < erwartet -> fail-closed. N4 Symlinks in Konfig-Pfaden werden abgelehnt (lstat + O_NOFOLLOW). CFG-Schluessel (alle Pfade absolut und KEINE Symlinks; fehlender Pfad = harter Abbruch = fail-closed): ro : dirs/files -> READ_FILE|READ_DIR rx : dirs -> READ_FILE|READ_DIR|EXECUTE rw : dirs -> read/write/create/remove (keine Geraete/Sockets/FIFOs) file_ro : files -> READ_FILE file_rw : files -> READ_FILE|WRITE_FILE|TRUNCATE (in place, kein rename/create) file_x : files -> READ_FILE|EXECUTE no_x : files -> EXECUTE wird NICHT gewaehrt, obwohl der Baum rx/x ist (z. B. 'claude': ein Start wuerde das Telegram-Plugin toeten) tcp_connect : [ports] -> Schluessel VORHANDEN => NET wird restringiert, nur diese Ziel-TCP-Ports (leere Liste => alle TCP gesperrt). Schluessel FEHLT => NET unrestringiert (TCP/Bind frei). min_abi : int -> geforderte Mindest-ABI (Default 4). cwd : dir -> Arbeitsverzeichnis des Ziels (Default '/'). env : {k: v} -> exakte Umgebung des Ziels (nichts wird geerbt). rlimits : {"AS":bytes,"CPU":sec,"FSIZE":bytes,"NOFILE":n,"NPROC":n,"CORE":n} Exit-Code des Wrappers selbst: 70 = Setup-Fehler (Ziel laeuft NIE ungeschuetzt). """ import ctypes import ctypes.util import json import os import resource import stat import sys FD_CLOSE_HIGH = 65536 def die(msg): sys.stderr.write(f"iso_exec: SETUP-FEHLER: {msg}\n") os._exit(70) libc = ctypes.CDLL(None, use_errno=True) libc.syscall.restype = ctypes.c_long libc.prctl.argtypes = [ctypes.c_int, ctypes.c_ulong, ctypes.c_ulong, ctypes.c_ulong, ctypes.c_ulong] # ---- Landlock ----------------------------------------------------------- SYS_create, SYS_add_rule, SYS_restrict = 444, 445, 446 EXECUTE, WRITE_FILE, READ_FILE, READ_DIR = 1 << 0, 1 << 1, 1 << 2, 1 << 3 REMOVE_DIR, REMOVE_FILE, MAKE_CHAR, MAKE_DIR = 1 << 4, 1 << 5, 1 << 6, 1 << 7 MAKE_REG, MAKE_SOCK, MAKE_FIFO, MAKE_BLOCK = 1 << 8, 1 << 9, 1 << 10, 1 << 11 MAKE_SYM, REFER, TRUNCATE, IOCTL_DEV = 1 << 12, 1 << 13, 1 << 14, 1 << 15 NET_BIND_TCP, NET_CONNECT_TCP = 1 << 0, 1 << 1 RO = READ_FILE | READ_DIR RX = RO | EXECUTE RW = RO | WRITE_FILE | REMOVE_DIR | REMOVE_FILE | MAKE_DIR | MAKE_REG | MAKE_SYM | REFER | TRUNCATE F_RO, F_RW, F_X = READ_FILE, READ_FILE | WRITE_FILE | TRUNCATE, READ_FILE | EXECUTE def fs_handled_for_abi(abi): """N3: Handled-FS-Set aus der gemeldeten ABI ableiten (fail-closed).""" mask = (1 << 15) - 1 # ABI 1..4: Bits 0..14 (bis TRUNCATE) if abi >= 5: mask |= IOCTL_DEV # ABI 5 (Kernel >= 6.10): Geraete-ioctls mediieren return mask class RulesetAttr(ctypes.Structure): _fields_ = [("fs", ctypes.c_uint64), ("net", ctypes.c_uint64)] class PathBeneath(ctypes.Structure): _pack_ = 1 _fields_ = [("allowed", ctypes.c_uint64), ("parent_fd", ctypes.c_int32)] class NetPort(ctypes.Structure): _fields_ = [("allowed", ctypes.c_uint64), ("port", ctypes.c_uint64)] def _no_symlink(p): """N4: ein als Konfig-Pfad angegebener Symlink wird abgelehnt.""" try: st = os.lstat(p) except OSError as e: die(f"Pfad nicht lstat-bar: {p} ({e})") if stat.S_ISLNK(st.st_mode): die(f"Symlink als Konfig-Pfad abgelehnt (N4): {p}") def landlock(cfg): min_abi = int(cfg.get("min_abi", 4)) abi = libc.syscall(ctypes.c_long(SYS_create), None, ctypes.c_long(0), ctypes.c_long(1)) if abi < min_abi: die(f"Landlock ABI {abi} < erforderlich {min_abi} (fail-closed)") restrict_net = "tcp_connect" in cfg net_mask = (NET_BIND_TCP | NET_CONNECT_TCP) if restrict_net else 0 attr = RulesetAttr(fs_handled_for_abi(abi), net_mask) rfd = libc.syscall(ctypes.c_long(SYS_create), ctypes.byref(attr), ctypes.c_long(ctypes.sizeof(attr)), ctypes.c_long(0)) if rfd < 0: die(f"landlock_create_ruleset errno={ctypes.get_errno()}") def add_path(p, acc, want_dir): if not os.path.isabs(p): die(f"Konfig-Pfad nicht absolut: {p}") _no_symlink(p) if not os.path.exists(p): die(f"Pfad fehlt: {p}") if want_dir is True and not os.path.isdir(p): die(f"kein Verzeichnis: {p}") if want_dir is False and os.path.isdir(p): die(f"keine Datei: {p}") fd = os.open(p, os.O_PATH | os.O_CLOEXEC | os.O_NOFOLLOW) pb = PathBeneath(acc, fd) r = libc.syscall(ctypes.c_long(SYS_add_rule), ctypes.c_long(rfd), ctypes.c_long(1), ctypes.byref(pb), ctypes.c_long(0)) e = ctypes.get_errno() os.close(fd) if r != 0: die(f"landlock_add_rule {p} errno={e}") no_x = {os.path.abspath(p) for p in cfg.get("no_x", [])} for p in cfg.get("ro", []): add_path(p, RO if os.path.isdir(p) else F_RO, None) for p in cfg.get("rx", []): add_path(p, RX, True) for p in cfg.get("rw", []): add_path(p, RW, True) for p in cfg.get("file_ro", []): add_path(p, F_RO, False) for p in cfg.get("file_rw", []): add_path(p, F_RW, False) for p in cfg.get("file_x", []): # no_x entzieht EXECUTE, selbst wenn file_x gewuenscht war (Schutz gegen claude-Start) add_path(p, (F_X & ~EXECUTE) if os.path.abspath(p) in no_x else F_X, False) if restrict_net: for port in cfg.get("tcp_connect", []): np_ = NetPort(NET_CONNECT_TCP, int(port)) if libc.syscall(ctypes.c_long(SYS_add_rule), ctypes.c_long(rfd), ctypes.c_long(2), ctypes.byref(np_), ctypes.c_long(0)) != 0: die(f"landlock net rule {port} errno={ctypes.get_errno()}") return rfd # ---- seccomp via libseccomp --------------------------------------------- class ArgCmp(ctypes.Structure): _fields_ = [("arg", ctypes.c_uint), ("op", ctypes.c_int), ("a", ctypes.c_uint64), ("b", ctypes.c_uint64)] CMP_NE, CMP_EQ, CMP_GT, CMP_MASKED_EQ = 1, 4, 6, 7 ACT_ALLOW = 0x7FFF0000 def ACT_ERRNO(e): return 0x00050000 | (e & 0xFFFF) EPERM, ENOSYS, EAFNOSUPPORT = 1, 38, 97 AT_EMPTY_PATH = 0x1000 DENY_ALWAYS = """ptrace process_vm_readv process_vm_writev unshare setns mount umount2 pivot_root chroot keyctl add_key request_key bpf perf_event_open userfaultfd io_uring_setup io_uring_enter io_uring_register open_tree move_mount fsopen fsconfig fsmount fspick mount_setattr kexec_load kexec_file_load init_module finit_module delete_module acct swapon swapoff reboot syslog pidfd_send_signal pidfd_getfd tkill quotactl name_to_handle_at open_by_handle_at personality vhangup memfd_create memfd_secret process_madvise""".split() NS_FLAGS = [0x00020000, 0x02000000, 0x04000000, 0x08000000, 0x10000000, 0x20000000, 0x40000000, 0x00000080] # fcntl-Kommandos, die Signale an eine (fremde) PID lenken koennen (SIGIO/async I/O). Review #8b: # F_SETOWN/F_SETOWN_EX setzen den Signal-Empfaenger, F_SETSIG waehlt das Signal -> zusammen eine # Signalzustellung an fremde Prozesse OHNE kill/tgkill (die bereits gefiltert sind). F_SETSIG, F_SETOWN, F_SETOWN_EX = 10, 8, 15 def seccomp_prepare(mypid): """Filter bauen, solange Bibliotheken noch lesbar sind; spaeter (nach Landlock) laden.""" path = ctypes.util.find_library("seccomp") or "libseccomp.so.2" ls = ctypes.CDLL(path, use_errno=True) ls.seccomp_init.restype = ctypes.c_void_p ls.seccomp_init.argtypes = [ctypes.c_uint32] ls.seccomp_rule_add_array.argtypes = [ctypes.c_void_p, ctypes.c_uint32, ctypes.c_int, ctypes.c_uint, ctypes.c_void_p] ls.seccomp_syscall_resolve_name.argtypes = [ctypes.c_char_p] ls.seccomp_load.argtypes = [ctypes.c_void_p] ctx = ls.seccomp_init(ACT_ALLOW) if not ctx: die("seccomp_init") def nr(name): n = ls.seccomp_syscall_resolve_name(name.encode()) if n < 0: die(f"unbekannter syscall {name}") return n def rule(action, name, *cmps): arr = (ArgCmp * max(1, len(cmps)))(*cmps) if cmps else None r = ls.seccomp_rule_add_array(ctx, action, nr(name), len(cmps), arr) if r != 0: die(f"seccomp_rule_add {name} rc={r}") for s in DENY_ALWAYS: if ls.seccomp_syscall_resolve_name(s.encode()) < 0: continue # auf dieser Arch unbekannt (z. B. memfd_secret) -> nicht vorhanden, ignorieren rule(ACT_ERRNO(EPERM), s) # M2: execveat(AT_EMPTY_PATH) verbieten (memfd-Re-Exec); normales execveat per Pfad bleibt erlaubt. rule(ACT_ERRNO(EPERM), "execveat", ArgCmp(4, CMP_MASKED_EQ, AT_EMPTY_PATH, AT_EMPTY_PATH)) rule(ACT_ERRNO(ENOSYS), "clone3") # erzwingt libc-Fallback auf clone(), dessen Flags wir pruefen for f in NS_FLAGS: rule(ACT_ERRNO(EPERM), "clone", ArgCmp(0, CMP_MASKED_EQ, f, f)) # socket(): nur AF_INET(2)/AF_INET6(10); alles andere (AF_UNIX=1, AF_NETLINK=16, ...) abgelehnt for dom in [0, 1, 3, 4, 5, 6, 7, 8, 9]: rule(ACT_ERRNO(EAFNOSUPPORT), "socket", ArgCmp(0, CMP_EQ, dom, 0)) rule(ACT_ERRNO(EAFNOSUPPORT), "socket", ArgCmp(0, CMP_GT, 10, 0)) # Signale nur an uns selbst (gleiche tgid); keine Signale an fremde aria-Prozesse for s in ["kill", "tgkill", "rt_sigqueueinfo", "rt_tgsigqueueinfo"]: rule(ACT_ERRNO(EPERM), s, ArgCmp(0, CMP_NE, mypid, 0)) # Review #8b: F_SETOWN/F_SETOWN_EX/F_SETSIG verbieten (SIGIO-Umleitung an fremde PID). Arg-Filter # auf das fcntl-Kommando (arg1), kein Pointer-Deref noetig. for cmd in (F_SETOWN, F_SETOWN_EX, F_SETSIG): rule(ACT_ERRNO(EPERM), "fcntl", ArgCmp(1, CMP_EQ, cmd, 0)) # Review #8c: prlimit64 nur fuer die EIGENE PID (pid==0); fremde PID -> EPERM (kein DoS an aria). rule(ACT_ERRNO(EPERM), "prlimit64", ArgCmp(0, CMP_NE, 0, 0)) # tty-Injektion for req in [0x5412, 0x541C]: # TIOCSTI, TIOCLINUX (Kernel schneidet cmd auf 32 Bit -> masked compare) rule(ACT_ERRNO(EPERM), "ioctl", ArgCmp(1, CMP_MASKED_EQ, 0xFFFFFFFF, req)) return ls, ctx def seccomp_load(ls, ctx): r = ls.seccomp_load(ctx) if r != 0: die(f"seccomp_load rc={r}") def set_rlimits(lim): # CORE immer 0 (keine Core-Dumps mit evtl. Secret-Speicher) resource.setrlimit(resource.RLIMIT_CORE, (int(lim.get("CORE", 0)), int(lim.get("CORE", 0)))) # M3: harter Boden fuer Speicher und CPU im Wrapper selbst if "AS" in lim: resource.setrlimit(resource.RLIMIT_AS, (int(lim["AS"]), int(lim["AS"]))) if "CPU" in lim: resource.setrlimit(resource.RLIMIT_CPU, (int(lim["CPU"]), int(lim["CPU"]))) if "FSIZE" in lim: resource.setrlimit(resource.RLIMIT_FSIZE, (int(lim["FSIZE"]), int(lim["FSIZE"]))) if "NOFILE" in lim: resource.setrlimit(resource.RLIMIT_NOFILE, (int(lim["NOFILE"]), int(lim["NOFILE"]))) # NPROC ist bewusst OPTIONAL: RLIMIT_NPROC zaehlt pro UID (alle aria-Threads), # nicht pro Sandbox -> unbrauchbar als Sandbox-Grenze. TasksMax im systemd-Scope # ist die korrekte prozess-/threadbezogene Grenze. Nur setzen, wenn ausdruecklich gewuenscht. if "NPROC" in lim: resource.setrlimit(resource.RLIMIT_NPROC, (int(lim["NPROC"]), int(lim["NPROC"]))) def main(): if len(sys.argv) < 4 or sys.argv[2] != "--": die("usage: iso_exec.py CFG.json -- /abs/prog args...") with open(sys.argv[1]) as fh: cfg = json.load(fh) argv = sys.argv[3:] if not os.path.isabs(argv[0]): die("Programm muss absolut angegeben werden") env = {str(k): str(v) for k, v in cfg.get("env", {}).items()} # N1: vererbte FDs schliessen, BEVOR irgendetwas Weiteres passiert (vor restrict/seccomp). os.closerange(3, FD_CLOSE_HIGH) set_rlimits(cfg.get("rlimits", {})) # N2: umask und cwd setzen os.umask(0o077) cwd = cfg.get("cwd", "/") try: os.chdir(cwd) except OSError as e: die(f"chdir {cwd}: {e}") if libc.prctl(38, 1, 0, 0, 0) != 0: # PR_SET_NO_NEW_PRIVS die("PR_SET_NO_NEW_PRIVS") ls, ctx = seccomp_prepare(os.getpid()) rfd = landlock(cfg) if libc.syscall(ctypes.c_long(SYS_restrict), ctypes.c_long(rfd), ctypes.c_long(0)) != 0: die(f"landlock_restrict_self errno={ctypes.get_errno()}") os.close(rfd) seccomp_load(ls, ctx) os.closerange(3, FD_CLOSE_HIGH) # belt-and-suspenders direkt vor execve try: os.execve(argv[0], argv, env) except OSError as e: die(f"execve {argv[0]}: {e}") if __name__ == "__main__": try: main() except SystemExit: raise except BaseException as e: # jeder Wrapper-Fehler ist fail-closed mit festem Code die(f"{type(e).__name__}: {e}")