#!/usr/bin/python3 -I
"""Wird NUR von Tests INNERHALB des Kaefigs (iso_exec.py) via python3 gestartet, nie gegen codex.

Aufruf: cage_probe.py <check> [args...]; exit 0, wenn das erwartete Kaefig-Verhalten eintritt,
sonst exit 1 mit Meldung auf stderr. So bleiben die Tests rc-basiert und deterministisch.
Gibt NIEMALS Dateiinhalte aus.
"""
import ctypes
import errno
import fcntl
import os
import resource
import socket
import struct
import sys

# x86_64 syscall-Nummern
NR_memfd_create = 319
NR_execveat = 322
NR_keyctl = 250
NR_io_uring_setup = 425
NR_unshare = 272
NR_ptrace = 101
NR_prlimit64 = 302
NR_pidfd_open = 434
NR_process_madvise = 440
NR_kcmp = 312
NR_sched_setaffinity = 203


def _ok():
    sys.exit(0)


def _fail(msg):
    sys.stderr.write("PROBE-FAIL: " + msg + "\n")
    sys.exit(1)


def _denied(exc):
    return isinstance(exc, OSError) and exc.errno in (errno.EACCES, errno.EPERM, errno.EAFNOSUPPORT)


def main():
    check = sys.argv[1]
    a = sys.argv[2:]
    libc = ctypes.CDLL(None, use_errno=True)
    libc.syscall.restype = ctypes.c_long

    def sc(nr, *args):
        ctypes.set_errno(0)
        r = libc.syscall(ctypes.c_long(nr), *[ctypes.c_long(x) for x in args])
        return r, ctypes.get_errno()

    if check == "read-denied":
        try:
            open(a[0], "rb").read(1)
        except OSError as e:
            _ok() if _denied(e) else _fail(f"read {a[0]} errno={e.errno}")
        _fail(f"read {a[0]} LEAK (succeeded)")
    elif check == "read-ok":
        open(a[0], "rb").read(1)
        _ok()
    elif check == "list-denied":
        try:
            os.listdir(a[0])
        except OSError as e:
            _ok() if _denied(e) else _fail(f"list {a[0]} errno={e.errno}")
        _fail(f"list {a[0]} LEAK")
    elif check == "write-ok":
        p = a[0]
        with open(p, "wb") as fh:
            fh.write(b"x")
        _ok()
    elif check == "write-denied":
        try:
            fd = os.open(a[0], os.O_WRONLY | os.O_TRUNC)
            os.close(fd)
        except OSError as e:
            _ok() if _denied(e) else _fail(f"write {a[0]} errno={e.errno}")
        _fail(f"write {a[0]} LEAK")
    elif check == "ro-write-denied-rw-ok":
        # Re-Review 2: in EINEM Lauf - zwei read-only-Ziele (ro-Dotfile, cron-Ziel) duerfen NICHT
        # beschreibbar sein, ein Workspace-Ziel schon. O_WRONLY OHNE O_TRUNC -> kein Datenverlust an
        # echten Dateien, selbst wenn (fehlerhaft) geoeffnet werden koennte. Letztes Arg = rw-Ziel.
        ro_targets, rw_target = a[:-1], a[-1]
        for ro in ro_targets:
            try:
                fd = os.open(ro, os.O_WRONLY | os.O_NOFOLLOW)
                os.close(fd)
                _fail(f"ro target writable LEAK: {ro}")
            except OSError as e:
                if not _denied(e):
                    _fail(f"ro {ro} unexpected errno={e.errno}")
        with open(rw_target, "wb") as fh:          # Workspace-Schreiben muss im selben Lauf gelingen
            fh.write(b"x")
        _ok()
    elif check == "ro-content-immutable":
        # Re-Review 2 (Landlock-Grenze): eine read-only-Datei ist INHALTLICH unveraenderbar
        # (open O_WRONLY und O_WRONLY|O_TRUNC -> EACCES => keine Nutzlast einschleusbar = Kernschutz).
        # utimensat (touch) ist von Landlock NICHT mediiert und darf gelingen; Groesse/Inhalt MUSS
        # gleich bleiben. So ist die Grenze als Test festgehalten, nicht nur in Prosa.
        p = a[0]
        before = os.path.getsize(p)
        for flags in (os.O_WRONLY | os.O_NOFOLLOW, os.O_WRONLY | os.O_TRUNC | os.O_NOFOLLOW):
            try:
                fd = os.open(p, flags)
                os.close(fd)
                _fail(f"content-write LEAK (open flags={flags} succeeded)")
            except OSError as e:
                if not _denied(e):
                    _fail(f"open {p} errno={e.errno}")
        try:
            os.utime(p, None)                 # utimensat; darf (dokumentiert) gelingen
        except OSError:
            pass
        _ok() if os.path.getsize(p) == before else _fail("size changed (content tampered)")
    elif check == "write-matrix":
        # E44 Option 2: mehrere Erwartungen in EINEM Kaefig-Lauf, damit jede Verweigerung neben einer
        # gelungenen Positivkontrolle steht. Nicht zerstoerend: Verweigerungen werden mit O_APPEND
        # (kein O_TRUNC, nichts geschrieben) bzw. O_CREAT|O_EXCL geprueft; bei unerwartetem Erfolg wird
        # nichts geschrieben und eine neu angelegte Datei sofort wieder entfernt.
        #   write-ok:<p>       Datei schreiben (anlegen) muss gelingen
        #   append-denied:<p>  vorhandene Datei zum Schreiben oeffnen muss scheitern (EACCES/EPERM)
        #   create-denied:<p>  neue Datei anlegen muss scheitern
        #   read-ok:<p>        lesen muss gelingen
        #   read-denied:<p>    lesen muss scheitern (EACCES/EPERM)
        for token in a:
            kind, _, p = token.partition(":")
            if kind == "write-ok":
                try:
                    with open(p, "wb") as fh:
                        fh.write(b"x")
                except OSError as e:
                    _fail(f"{token} errno={e.errno}")
            elif kind == "append-denied":
                try:
                    fd = os.open(p, os.O_WRONLY | os.O_APPEND | os.O_NOFOLLOW)
                except OSError as e:
                    if not _denied(e):
                        _fail(f"{token} unexpected errno={e.errno}")
                    continue
                os.close(fd)
                _fail(f"{token} LEAK (zum Schreiben geoeffnet)")
            elif kind == "create-denied":
                try:
                    fd = os.open(p, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
                except OSError as e:
                    if not _denied(e):
                        _fail(f"{token} unexpected errno={e.errno}")
                    continue
                os.close(fd)
                try:
                    os.unlink(p)
                except OSError:
                    pass
                _fail(f"{token} LEAK (Datei liess sich anlegen)")
            elif kind == "read-ok":
                try:
                    open(p, "rb").read(1)
                except OSError as e:
                    _fail(f"{token} errno={e.errno}")
            elif kind == "read-denied":
                try:
                    open(p, "rb").read(1)
                except OSError as e:
                    if not _denied(e):
                        _fail(f"{token} unexpected errno={e.errno}")
                    continue
                _fail(f"{token} LEAK (lesbar)")
            else:
                _fail(f"unknown write-matrix token {token}")
        _ok()
    elif check == "create-denied":
        # Latente Datei (z. B. ~/.bash_aliases): Neuanlage unter home_root muss scheitern
        # (home_root hat keine MAKE_REG-Regel). O_CREAT|O_EXCL; bei Erfolg Datei wieder entfernen.
        try:
            fd = os.open(a[0], os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
        except OSError as e:
            _ok() if _denied(e) else _fail(f"create {a[0]} errno={e.errno}")
        os.close(fd)
        try:
            os.unlink(a[0])
        except OSError:
            pass
        _fail(f"create {a[0]} LEAK (Datei liess sich anlegen)")
    elif check == "exec-denied":
        pid = os.fork()
        if pid == 0:
            try:
                os.execv(a[0], [a[0], "--version"])
            except OSError as e:
                os._exit(100 + (e.errno % 100))
            os._exit(99)
        _, st = os.waitpid(pid, 0)
        code = os.waitstatus_to_exitcode(st)
        _ok() if code == 100 + errno.EACCES else _fail(f"exec {a[0]} exit={code} (expected EACCES)")
    elif check == "exec-ok":
        pid = os.fork()
        if pid == 0:
            try:
                os.execv(a[0], [a[0]])
            except OSError:
                os._exit(50)
            os._exit(99)
        _, st = os.waitpid(pid, 0)
        code = os.waitstatus_to_exitcode(st)
        _ok() if code not in (50, 99) and code < 100 else _fail(f"exec {a[0]} exit={code}")
    elif check == "memfd-denied":
        r, e = sc(NR_memfd_create, 0, 0)
        _ok() if r < 0 and e == errno.EPERM else _fail(f"memfd_create r={r} errno={e}")
    elif check == "execveat-empty-denied":
        # execveat(dirfd=-1, "", NULL, NULL, AT_EMPTY_PATH) -> muss EPERM sein (M2)
        r, e = sc(NR_execveat, -1, 0, 0, 0, 0x1000)
        _ok() if r < 0 and e == errno.EPERM else _fail(f"execveat(AT_EMPTY_PATH) r={r} errno={e}")
    elif check == "unshare-denied":
        r, e = sc(NR_unshare, 0x10000000)   # CLONE_NEWUSER
        _ok() if r < 0 and e == errno.EPERM else _fail(f"unshare r={r} errno={e}")
    elif check == "keyctl-denied":
        r, e = sc(NR_keyctl, 0, -4, 0)
        _ok() if r < 0 and e == errno.EPERM else _fail(f"keyctl r={r} errno={e}")
    elif check == "iouring-denied":
        buf = ctypes.create_string_buffer(120)
        r, e = sc(NR_io_uring_setup, 1, ctypes.addressof(buf))
        _ok() if r < 0 and e == errno.EPERM else _fail(f"io_uring_setup r={r} errno={e}")
    elif check == "ptrace-denied":
        r, e = sc(NR_ptrace, 0, 0, 0, 0)    # PTRACE_TRACEME
        _ok() if r < 0 and e == errno.EPERM else _fail(f"ptrace r={r} errno={e}")
    elif check == "kill-denied":
        try:
            os.kill(int(a[0]), 0)
        except OSError as e:
            _ok() if e.errno == errno.EPERM else _fail(f"kill errno={e.errno}")
        _fail("kill foreign pid LEAK")
    elif check == "af-unix-denied":
        try:
            s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
            s.settimeout(2)
            s.connect(a[0])
        except OSError as e:
            _ok() if _denied(e) else _fail(f"af_unix errno={e.errno}")
        _fail("af_unix connect LEAK")
    elif check == "sendto-unix-denied":
        # Review #8a: ein frischer AF_UNIX-Datagramm-Socket darf gar nicht erst entstehen
        # (socket(AF_UNIX) ist per seccomp gesperrt) -> kein sendto an einen fremden Unix-Dienst.
        try:
            s = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
            s.sendto(b"x", a[0])
        except OSError as e:
            _ok() if _denied(e) else _fail(f"sendto-unix errno={e.errno}")
        _fail("sendto-unix LEAK (fremder Unix-Dienst erreichbar)")
    elif check == "socketpair-afunix-ok":
        # socketpair(AF_UNIX) bleibt ERLAUBT (prozessintern, kein externer Kanal); V8/libuv nutzt es.
        try:
            x, y = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM)
            x.close()
            y.close()
            _ok()
        except OSError as e:
            _fail(f"socketpair(AF_UNIX) unerwartet gesperrt errno={e.errno}")
    elif check == "fcntl-setown-denied":
        # Review #8b: F_SETOWN auf eine FREMDE PID (SIGIO-Umleitung) muss EPERM sein.
        r, w = os.pipe()
        try:
            fcntl.fcntl(r, fcntl.F_SETOWN, int(a[0]))
        except OSError as e:
            _ok() if e.errno == errno.EPERM else _fail(f"F_SETOWN errno={e.errno}")
        finally:
            os.close(r)
            os.close(w)
        _fail("F_SETOWN auf fremde PID LEAK (SIGIO-Umleitung moeglich)")
    elif check == "fcntl-setsig-denied":
        r, w = os.pipe()
        try:
            fcntl.fcntl(r, 10, int(a[0]) if a else 0)      # F_SETSIG=10
        except OSError as e:
            os.close(r); os.close(w)
            _ok() if e.errno == errno.EPERM else _fail(f"F_SETSIG errno={e.errno}")
        os.close(r); os.close(w)
        _fail("F_SETSIG LEAK (Signalwahl bei async I/O)")
    elif check == "prlimit-foreign-denied":
        # Review #8c: prlimit64 mit pid != 0 auf einen fremden Prozess muss EPERM sein.
        buf = ctypes.create_string_buffer(16)             # struct rlimit {cur, max}
        r, e = sc(NR_prlimit64, int(a[0]), resource.RLIMIT_CPU, 0, ctypes.addressof(buf))
        if r < 0 and e == errno.EPERM:
            _ok()
        _fail(f"prlimit64(pid={a[0]}) r={r} errno={e} LEAK (fremder Prozess manipulierbar)")
    elif check == "prlimit-self-ok":
        buf = ctypes.create_string_buffer(16)
        r, e = sc(NR_prlimit64, 0, resource.RLIMIT_CPU, 0, ctypes.addressof(buf))
        _ok() if r == 0 else _fail(f"prlimit64(pid=0, self) unerwartet r={r} errno={e}")
    elif check == "process-madvise-denied":
        # pidfd_open(fremd) + process_madvise -> muss scheitern (EPERM/ENOSYS).
        pidfd, e = sc(NR_pidfd_open, int(a[0]), 0)
        if pidfd < 0:
            _ok()                                         # schon pidfd_open gescheitert
        iov = struct.pack("QQ", 0x1000, 0x1000)
        buf = ctypes.create_string_buffer(iov)
        MADV_DONTNEED = 4
        r, e = sc(NR_process_madvise, pidfd, ctypes.addressof(buf), 1, MADV_DONTNEED, 0)
        os.close(pidfd)
        _ok() if (r < 0 and e in (errno.EPERM, errno.ENOSYS)) else _fail(
            f"process_madvise r={r} errno={e} LEAK (fremder Speicher beeinflussbar)")
    elif check == "kcmp-denied":
        r, e = sc(NR_kcmp, int(a[0]), os.getpid(), 0, 0, 0)   # KCMP_FILE=0
        _ok() if (r < 0 and e == errno.EPERM) else _fail(f"kcmp r={r} errno={e} LEAK")
    elif check == "env-clean":
        bad = [k for k in os.environ if k.startswith((
            "TELEGRAM", "SUPABASE", "ANTHROPIC", "OPENAI", "GITHUB", "GH_", "AWS", "CLAUDE", "SSH_", "DBUS", "XDG"))]
        _ok() if not bad else _fail("env leak: " + ",".join(bad))
    elif check == "env-exact":
        got = set(os.environ)
        want = set(a[0].split(","))
        _ok() if got <= want else _fail("unexpected env: " + ",".join(sorted(got - want)))
    elif check == "umask-077":
        m = os.umask(0)
        _ok() if m == 0o077 else _fail(f"umask={oct(m)}")
    elif check == "rlimit":
        name, want = a[0], int(a[1])
        soft, _hard = resource.getrlimit(getattr(resource, "RLIMIT_" + name))
        _ok() if soft == want else _fail(f"RLIMIT_{name} soft={soft} want={want}")
    elif check == "cwd":
        _ok() if os.getcwd() == a[0] else _fail(f"cwd={os.getcwd()} want={a[0]}")
    elif check == "fd-closed":
        # N1: ein vom Launcher geerbter fd muss VOR restrict geschlossen sein -> EBADF.
        try:
            os.read(int(a[0]), 1)
        except OSError as e:
            _ok() if e.errno == errno.EBADF else _fail(f"fd {a[0]} errno={e.errno} (expected EBADF)")
        _fail(f"fd {a[0]} still readable (inherited fd LEAK)")
    else:
        _fail("unknown check " + check)


if __name__ == "__main__":
    main()