[PATCH net v11 4/5] rxrpc: bound connection and OOB receive queues
Zihan Xi
zihanx at nebusec.ai
Wed Oct 21 00:37:15 PDT 2026
Packets dequeued by rxrpc_io_thread() can remain on conn->rx_queue or a
managed RxGK OOB queue after they leave local->rx_queue. UDP receive-memory
accounting alone therefore does not bound their aggregate size or lifetime.
Charge retained connection and OOB skbs to a per-local asynchronous
receive budget before releasing UDP ownership. Keep separate capacity for
connection-control and managed OOB packets. The OOB allowance is shared by
the local endpoint, so aggregate pressure can drop later challenges.
Release the charge when packets are processed, answered, freed or purged.
When a connection queue is purged, release any challenge reference still
held by a queued skb. If a challenge cannot be queued after its reference
is acquired, release that reference. Purge queued connection events if the
local endpoint is already closing, and keep conn->local alive until both
connection queues have been purged. Use an internal skb mark for managed
OOB challenges while returning RXRPC_OOB_CHALLENGE through the public OOB
API.
The issue was found during development and reproduced with a PoC against
an unpatched RxRPC baseline.
Fixes: 17926a79320a ("[AF_RXRPC]: Provide secure RxRPC sockets for use by userspace and kernel both")
Fixes: 5800b1cf3fd8 ("rxrpc: Allow CHALLENGEs to the passed to the app for a RESPONSE")
Cc: stable at vger.kernel.org
Link: https://lore.kernel.org/all/cover.1791994057.git.zihanx@nebusec.ai/
Reported-by: Vega <vega at nebusec.ai>
Assisted-by: LLM
Co-developed-by: Luxing Yin <root at tr0jan.top>
Signed-off-by: Luxing Yin <root at tr0jan.top>
Signed-off-by: Zihan Xi <zihanx at nebusec.ai>
---
changes in v11:
- Split the pre-existing CHALLENGE and SERVICE_CONN_SECURED
reference fixes into standalone patches.
- Name the parameters of the async queue accounting prototypes.
- Keep UDP rmem charged while skbs are held by call processing, and
transfer ownership only after the asynchronous budget is acquired.
- Return RXRPC_OOB_CHALLENGE directly and remove the unused local
variable reported by coccicheck.
- Document how the issue was found and reproduced.
- v10 Link: https://lore.kernel.org/all/cover.1792168462.git.zihanx@nebusec.ai/
changes in v10:
- Release the retained connection reference when an already-aborted
connection causes queued CHALLENGE processing to return early.
- Give queued OOB challenge skbs a dedicated internal skb mark, separate
from the public OOB type enum.
- v9 Link: https://lore.kernel.org/all/cover.1791994057.git.zihanx@nebusec.ai/
changes in v9:
- Split connection/OOB queue budgeting from DATA receive-queue
budgeting.
- Add the Fixes attribution for the RXGK challenge OOB path.
- Order receive-budget helper declarations per netdev reverse-xmas style.
- v8 Link: https://lore.kernel.org/all/cover.1791428529.git.zihanx@nebusec.ai/
---
net/rxrpc/ar-internal.h | 5 +++++
net/rxrpc/call_event.c | 2 ++
net/rxrpc/conn_event.c | 32 ++++++++++++++++++++++++++-----
net/rxrpc/conn_object.c | 31 ++++++++++++++++++++++++------
net/rxrpc/io_thread.c | 17 ++++++++++++-----
net/rxrpc/local_object.c | 3 +++
net/rxrpc/oob.c | 21 +++++++++++---------
net/rxrpc/recvmsg.c | 4 ++--
net/rxrpc/skbuff.c | 41 +++++++++++++++++++++++++++++++++++++++-
9 files changed, 128 insertions(+), 28 deletions(-)
diff --git a/net/rxrpc/ar-internal.h b/net/rxrpc/ar-internal.h
index b079ec98aaa74..e606145664e06 100644
--- a/net/rxrpc/ar-internal.h
+++ b/net/rxrpc/ar-internal.h
@@ -360,6 +360,7 @@ struct rxrpc_local {
struct sk_buff_head rx_delay_queue; /* Delay injection queue */
#endif
struct sk_buff_head rx_queue; /* Received packets */
+ atomic_long_t rx_async_queue_bytes; /* Outstanding async Rx bytes */
struct list_head conn_attend_q; /* Conns requiring immediate attention */
struct list_head call_attend_q; /* Calls requiring immediate attention */
@@ -1516,11 +1517,15 @@ void rxrpc_new_skb(struct sk_buff *, enum rxrpc_skb_trace);
void rxrpc_see_skb(struct sk_buff *, enum rxrpc_skb_trace);
void rxrpc_get_skb(struct sk_buff *, enum rxrpc_skb_trace);
void rxrpc_free_skb(struct sk_buff *, enum rxrpc_skb_trace);
+void rxrpc_skb_orphan_udp(struct sk_buff *skb);
+bool rxrpc_charge_async_queue(struct rxrpc_local *local, struct sk_buff *skb);
+void rxrpc_uncharge_async_queue(struct rxrpc_local *local, struct sk_buff *skb);
void rxrpc_purge_queue(struct sk_buff_head *);
/*
* stats.c
*/
+void rxrpc_purge_conn_queue(struct rxrpc_connection *);
int rxrpc_stats_show(struct seq_file *seq, void *v);
int rxrpc_stats_clear(struct file *file, char *buf, size_t size);
diff --git a/net/rxrpc/call_event.c b/net/rxrpc/call_event.c
index 21be9c86d7a74..17c3a2606216f 100644
--- a/net/rxrpc/call_event.c
+++ b/net/rxrpc/call_event.c
@@ -329,6 +329,7 @@ bool rxrpc_input_call_event(struct rxrpc_call *call)
if (__rxrpc_call_is_complete(call) ||
skb->mark == RXRPC_SKB_MARK_ERROR) {
+ rxrpc_skb_orphan_udp(skb);
rxrpc_free_skb(skb, rxrpc_skb_put_call_rx);
goto out;
}
@@ -336,6 +337,7 @@ bool rxrpc_input_call_event(struct rxrpc_call *call)
saw_ack |= sp->hdr.type == RXRPC_PACKET_TYPE_ACK;
rxrpc_input_call_packet(call, skb);
+ rxrpc_skb_orphan_udp(skb);
rxrpc_free_skb(skb, rxrpc_skb_put_call_rx);
did_receive = true;
}
diff --git a/net/rxrpc/conn_event.c b/net/rxrpc/conn_event.c
index 02e9b18ee534d..7d31d7fa9cbd3 100644
--- a/net/rxrpc/conn_event.c
+++ b/net/rxrpc/conn_event.c
@@ -392,9 +392,10 @@ static void rxrpc_do_process_connection(struct rxrpc_connection *conn)
* connection that each one has when we've finished with it */
while ((skb = skb_dequeue(&conn->rx_queue))) {
struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
-
rxrpc_see_skb(skb, rxrpc_skb_see_conn_work);
rxrpc_process_event(conn, skb);
+ if (skb->mark != RXRPC_SKB_MARK_SERVICE_CONN_SECURED)
+ rxrpc_uncharge_async_queue(conn->local, skb);
if (sp->hdr.type == RXRPC_PACKET_TYPE_CHALLENGE && sp->chall.conn) {
rxrpc_put_connection(sp->chall.conn,
rxrpc_conn_put_challenge_input);
@@ -414,6 +415,8 @@ void rxrpc_process_connection(struct work_struct *work)
if (__rxrpc_use_local(conn->local, rxrpc_local_use_conn_work)) {
rxrpc_do_process_connection(conn);
rxrpc_unuse_local(conn->local, rxrpc_local_unuse_conn_work);
+ } else {
+ rxrpc_purge_conn_queue(conn);
}
}
@@ -421,16 +424,30 @@ void rxrpc_process_connection(struct work_struct *work)
* post connection-level events to the connection
* - this includes challenges, responses, some aborts and call terminal packet
* retransmission.
+ * - bound the queued memory independently of the UDP rmem charge.
*/
-static void rxrpc_post_packet_to_conn(struct rxrpc_connection *conn,
+static bool rxrpc_post_packet_to_conn(struct rxrpc_connection *conn,
struct sk_buff *skb)
{
+ struct rxrpc_local *local = conn->local;
+
+ if (!rxrpc_charge_async_queue(local, skb))
+ return false;
+
_enter("%p,%p", conn, skb);
rxrpc_get_skb(skb, rxrpc_skb_get_conn_work);
skb_queue_tail(&conn->rx_queue, skb);
rxrpc_queue_conn(conn, rxrpc_conn_queue_rx_work);
+ spin_lock_bh(&local->rx_queue.lock);
+ if (unlikely(!atomic_read(&local->active_users))) {
+ spin_unlock_bh(&local->rx_queue.lock);
+ rxrpc_uncharge_async_queue(local, skb);
+ return false;
+ }
+ return true;
}
+ spin_unlock_bh(&local->rx_queue.lock);
/*
* Post a CHALLENGE packet to the socket of one of a connection's calls so that
@@ -449,7 +466,10 @@ static bool rxrpc_post_challenge(struct rxrpc_connection *conn,
rxrpc_get_connection(conn, rxrpc_conn_get_challenge_input);
if (!conn->security->challenge_to_recvmsg) {
- rxrpc_post_packet_to_conn(conn, skb);
+ if (!rxrpc_post_packet_to_conn(conn, skb)) {
+ rxrpc_put_connection(conn, rxrpc_conn_put_challenge_input);
+ sp->chall.conn = NULL;
+ }
return true;
}
@@ -487,8 +507,10 @@ static bool rxrpc_post_challenge(struct rxrpc_connection *conn,
return false;
}
- if (!call)
- rxrpc_post_packet_to_conn(conn, skb);
+ if (!call && !rxrpc_post_packet_to_conn(conn, skb)) {
+ rxrpc_put_connection(conn, rxrpc_conn_put_challenge_input);
+ sp->chall.conn = NULL;
+ }
return true;
}
diff --git a/net/rxrpc/conn_object.c b/net/rxrpc/conn_object.c
index 0ece717db0f85..73e60cbc8a138 100644
--- a/net/rxrpc/conn_object.c
+++ b/net/rxrpc/conn_object.c
@@ -300,6 +300,26 @@ static void rxrpc_rcu_free_connection(struct rcu_head *rcu)
wake_up_var(&rxnet->nr_conns);
}
+/*
+ * Purge a connection's receive queue, releasing accounting and challenge refs.
+ */
+void rxrpc_purge_conn_queue(struct rxrpc_connection *conn)
+{
+ struct sk_buff *skb;
+
+ while ((skb = skb_dequeue(&conn->rx_queue))) {
+ struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
+
+ rxrpc_uncharge_async_queue(conn->local, skb);
+ if (sp->hdr.type == RXRPC_PACKET_TYPE_CHALLENGE && sp->chall.conn) {
+ rxrpc_put_connection(sp->chall.conn,
+ rxrpc_conn_put_challenge_input);
+ sp->chall.conn = NULL;
+ }
+ rxrpc_free_skb(skb, rxrpc_skb_put_purge);
+ }
+}
+
/*
* Clean up a dead connection.
*/
@@ -329,7 +349,7 @@ static void rxrpc_clean_up_connection(struct work_struct *work)
conn->peer->pmtud_pending = true;
}
- rxrpc_purge_queue(&conn->rx_queue);
+ rxrpc_purge_conn_queue(conn);
rxrpc_free_skb(conn->tx_response, rxrpc_skb_put_response);
rxrpc_kill_client_conn(conn);
@@ -338,15 +358,14 @@ static void rxrpc_clean_up_connection(struct work_struct *work)
key_put(conn->key);
rxrpc_put_bundle(conn->bundle, rxrpc_bundle_put_conn);
rxrpc_put_peer(conn->peer, rxrpc_peer_put_conn);
- rxrpc_put_local(conn->local, rxrpc_local_put_kill_conn);
- /* Drain the Rx queue. Note that even though we've unpublished, an
- * incoming packet could still be being added to our Rx queue, so we
- * will need to drain it again in the RCU cleanup handler.
+ /* Drain the Rx queue again in case a packet was queued after the
+ * first purge while this connection was still published.
*/
- rxrpc_purge_queue(&conn->rx_queue);
+ rxrpc_purge_conn_queue(conn);
page_frag_cache_drain(&conn->tx_data_alloc);
+ rxrpc_put_local(conn->local, rxrpc_local_put_kill_conn);
call_rcu(&conn->rcu, rxrpc_rcu_free_connection);
}
diff --git a/net/rxrpc/io_thread.c b/net/rxrpc/io_thread.c
index 634a6e1cab87a..4699f4100ee4d 100644
--- a/net/rxrpc/io_thread.c
+++ b/net/rxrpc/io_thread.c
@@ -16,10 +16,10 @@ static int rxrpc_input_packet_on_conn(struct rxrpc_connection *conn,
struct sk_buff *skb);
/*
- * Drop UDP rmem ownership for packets charged in encap_rcv().
- * sk_forward_alloc is serialised by sk_receive_queue.lock.
+ * Release socket rmem ownership under the lock that serialises
+ * sk_forward_alloc updates.
*/
-static void rxrpc_skb_orphan_udp(struct sk_buff *skb)
+void rxrpc_skb_orphan_udp(struct sk_buff *skb)
{
struct sock *sk = skb->sk;
@@ -526,18 +526,23 @@ int rxrpc_io_thread(void *data)
while ((skb = __skb_dequeue(&rx_queue))) {
struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
- if (skb->mark == RXRPC_SKB_MARK_PACKET)
- rxrpc_skb_orphan_udp(skb);
+ if (skb->mark == RXRPC_SKB_MARK_SERVICE_CONN_SECURED)
+ rxrpc_uncharge_async_queue(sp->poke_conn->local, skb);
switch (skb->mark) {
case RXRPC_SKB_MARK_PACKET:
skb->priority = 0;
if (!rxrpc_input_packet(local, skb))
rxrpc_reject_packet(local, skb);
trace_rxrpc_rx_done(skb->mark, skb->priority);
+ /* Keep UDP accounting while the call queue holds the skb. */
+ if (!skb_shared(skb))
+ rxrpc_skb_orphan_udp(skb);
rxrpc_free_skb(skb, rxrpc_skb_put_input);
break;
case RXRPC_SKB_MARK_ERROR:
rxrpc_input_error(local, skb);
+ if (!skb_shared(skb))
+ rxrpc_skb_orphan_udp(skb);
rxrpc_free_skb(skb, rxrpc_skb_put_error_report);
break;
case RXRPC_SKB_MARK_SERVICE_CONN_SECURED:
@@ -547,6 +552,8 @@ int rxrpc_io_thread(void *data)
break;
default:
WARN_ON_ONCE(1);
+ if (!skb_shared(skb))
+ rxrpc_skb_orphan_udp(skb);
rxrpc_free_skb(skb, rxrpc_skb_put_unknown);
break;
}
diff --git a/net/rxrpc/local_object.c b/net/rxrpc/local_object.c
index f519a93059244..df1b4711082a1 100644
--- a/net/rxrpc/local_object.c
+++ b/net/rxrpc/local_object.c
@@ -137,6 +137,7 @@ static struct rxrpc_local *rxrpc_alloc_local(struct net *net,
skb_queue_head_init(&local->rx_delay_queue);
#endif
skb_queue_head_init(&local->rx_queue);
+ atomic_long_set(&local->rx_async_queue_bytes, 0);
INIT_LIST_HEAD(&local->conn_attend_q);
INIT_LIST_HEAD(&local->call_attend_q);
@@ -420,7 +421,9 @@ void rxrpc_unuse_local(struct rxrpc_local *local, enum rxrpc_local_trace why)
debug_id = local->debug_id;
r = refcount_read(&local->ref);
u = atomic_dec_return(&local->active_users);
+ spin_lock_bh(&local->rx_queue.lock);
trace_rxrpc_local(debug_id, why, r, u);
+ spin_unlock_bh(&local->rx_queue.lock);
if (u == 0)
kthread_stop(local->io_thread);
}
diff --git a/net/rxrpc/oob.c b/net/rxrpc/oob.c
index c80ee2487d094..0890b0d29112c 100644
--- a/net/rxrpc/oob.c
+++ b/net/rxrpc/oob.c
@@ -46,7 +46,9 @@ bool rxrpc_notify_socket_oob(struct rxrpc_call *call, struct sk_buff *skb)
sk = &rx->sk;
spin_lock_irq(&rx->recvmsg_lock);
- if (sk->sk_state < RXRPC_CLOSE) {
+ if (sk->sk_state < RXRPC_CLOSE &&
+ rxrpc_charge_async_queue(sp->chall.conn->local, skb)) {
+ skb->mark = RXRPC_SKB_MARK_CHALLENGE;
skb->skb_mstamp_ns = rx->oob_id_counter++;
rxrpc_get_skb(skb, rxrpc_skb_get_post_oob);
skb_queue_tail(&rx->recvmsg_oobq, skb);
@@ -194,7 +196,7 @@ static int rxrpc_respond_to_oob(struct rxrpc_sock *rx,
switch (p->command) {
case RXRPC_OOB_CMD_RESPOND:
ret = -EPROTO;
- if (skb->mark != RXRPC_OOB_CHALLENGE)
+ if (skb->mark != RXRPC_SKB_MARK_CHALLENGE)
break;
conn = sp->chall.conn;
ret = -EOPNOTSUPP;
@@ -214,7 +216,8 @@ static int rxrpc_respond_to_oob(struct rxrpc_sock *rx,
}
switch (skb->mark) {
- case RXRPC_OOB_CHALLENGE:
+ case RXRPC_SKB_MARK_CHALLENGE:
+ rxrpc_uncharge_async_queue(sp->chall.conn->local, skb);
rxrpc_put_connection(sp->chall.conn, rxrpc_conn_put_oob);
break;
}
@@ -275,10 +278,9 @@ enum rxrpc_oob_type rxrpc_kernel_query_oob(struct sk_buff *oob,
unsigned long *_peer_appdata)
{
struct rxrpc_skb_priv *sp = rxrpc_skb(oob);
- enum rxrpc_oob_type type = oob->mark;
- switch (type) {
- case RXRPC_OOB_CHALLENGE:
+ switch (oob->mark) {
+ case RXRPC_SKB_MARK_CHALLENGE:
*_peer = sp->chall.conn->peer;
*_peer_appdata = sp->chall.conn->peer->app_data;
break;
@@ -289,7 +291,7 @@ enum rxrpc_oob_type rxrpc_kernel_query_oob(struct sk_buff *oob,
break;
}
- return type;
+ return RXRPC_OOB_CHALLENGE;
}
EXPORT_SYMBOL(rxrpc_kernel_query_oob);
@@ -312,7 +314,7 @@ struct sk_buff *rxrpc_kernel_dequeue_oob(struct socket *sock,
oob = skb_dequeue(&rx->recvmsg_oobq);
if (oob)
- *_type = oob->mark;
+ *_type = RXRPC_OOB_CHALLENGE;
return oob;
}
EXPORT_SYMBOL(rxrpc_kernel_dequeue_oob);
@@ -328,7 +330,8 @@ void rxrpc_kernel_free_oob(struct sk_buff *oob)
struct rxrpc_skb_priv *sp = rxrpc_skb(oob);
switch (oob->mark) {
- case RXRPC_OOB_CHALLENGE:
+ case RXRPC_SKB_MARK_CHALLENGE:
+ rxrpc_uncharge_async_queue(sp->chall.conn->local, oob);
rxrpc_put_connection(sp->chall.conn, rxrpc_conn_put_oob);
break;
}
diff --git a/net/rxrpc/recvmsg.c b/net/rxrpc/recvmsg.c
index efcba4b2e74f0..172c72944cddd 100644
--- a/net/rxrpc/recvmsg.c
+++ b/net/rxrpc/recvmsg.c
@@ -248,8 +248,8 @@ static int rxrpc_recvmsg_oob(struct socket *sock, struct msghdr *msg,
if (ret < 0)
return ret;
- switch ((enum rxrpc_oob_type)skb->mark) {
- case RXRPC_OOB_CHALLENGE:
+ switch (skb->mark) {
+ case RXRPC_SKB_MARK_CHALLENGE:
need_response = true;
ret = rxrpc_recvmsg_challenge(sock, msg, skb, flags);
break;
diff --git a/net/rxrpc/skbuff.c b/net/rxrpc/skbuff.c
index d759715be02c1..357226741e97e 100644
--- a/net/rxrpc/skbuff.c
+++ b/net/rxrpc/skbuff.c
@@ -58,6 +58,41 @@ void rxrpc_free_skb(struct sk_buff *skb, enum rxrpc_skb_trace why)
}
}
+/*
+ * Transfer an skb to asynchronous receive-memory accounting.
+ */
+bool rxrpc_charge_async_queue(struct rxrpc_local *local, struct sk_buff *skb)
+{
+ struct sock *sk = local->socket->sk;
+ long rcvbuf;
+ long queued;
+
+ rcvbuf = READ_ONCE(sk->sk_rcvbuf);
+
+ for (;;) {
+ queued = atomic_long_read(&local->rx_async_queue_bytes);
+ if (queued &&
+ (queued >= rcvbuf ||
+ (long)skb->truesize >= rcvbuf - queued)) {
+ sk_drops_inc(sk);
+ return false;
+ }
+ if (atomic_long_cmpxchg(&local->rx_async_queue_bytes, queued,
+ queued + (long)skb->truesize) == queued) {
+ rxrpc_skb_orphan_udp(skb);
+ return true;
+ }
+ }
+}
+
+/*
+ * Release the asynchronous receive charge on an skb.
+ */
+void rxrpc_uncharge_async_queue(struct rxrpc_local *local, struct sk_buff *skb)
+{
+ atomic_long_sub(skb->truesize, &local->rx_async_queue_bytes);
+}
+
/*
* Clear a queue of socket buffers.
*/
@@ -66,12 +101,16 @@ void rxrpc_purge_queue(struct sk_buff_head *list)
struct sk_buff *skb;
while ((skb = skb_dequeue((list))) != NULL) {
- int n = atomic_dec_return(select_skb_count(skb));
+ int n;
+
if (skb->mark == RXRPC_SKB_MARK_SERVICE_CONN_SECURED) {
struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
+ rxrpc_uncharge_async_queue(sp->poke_conn->local, skb);
rxrpc_put_connection(sp->poke_conn, rxrpc_conn_put_poke);
}
+ rxrpc_skb_orphan_udp(skb);
+ n = atomic_dec_return(select_skb_count(skb));
trace_rxrpc_skb(skb, refcount_read(&skb->users), n,
rxrpc_skb_put_purge);
consume_skb(skb);
--
2.43.0
More information about the linux-afs
mailing list