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

[Maciej Fijalkowski <maciej.fijalkowski@intel.com>]

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?