After rxrpc_io_thread() dequeues a packet, the UDP receive-memory charge is released. Connection-level packets can remain queued on conn->rx_queue, while managed RXGK challenges can remain on the socket's OOB queue and pending-response set. Account both paths against a per-local receive budget. Release the charge when a packet is processed, answered, freed or purged, and retain it when a secured service packet is returned to the local I/O queue. 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@vger.kernel.org Link: https://lore.kernel.org/all/cover.1791994057.git.zihanx@nebusec.ai/ Reported-by: Vega Assisted-by: LLM Co-developed-by: Luxing Yin Signed-off-by: Luxing Yin Signed-off-by: Zihan Xi --- 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 | 3 +++ net/rxrpc/conn_event.c | 29 +++++++++++++++++++++++---- net/rxrpc/conn_object.c | 19 +++++++++++++++--- net/rxrpc/io_thread.c | 2 ++ net/rxrpc/local_object.c | 1 + net/rxrpc/oob.c | 20 +++++++++++-------- net/rxrpc/recvmsg.c | 4 ++-- net/rxrpc/skbuff.c | 43 +++++++++++++++++++++++++++++++++++++++- 8 files changed, 103 insertions(+), 18 deletions(-) diff --git a/net/rxrpc/ar-internal.h b/net/rxrpc/ar-internal.h index b079ec98aaa74..36fc4609181e1 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,6 +1517,8 @@ 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); +bool rxrpc_charge_async_queue(struct rxrpc_local *, struct sk_buff *); +void rxrpc_uncharge_async_queue(struct rxrpc_local *, struct sk_buff *); void rxrpc_purge_queue(struct sk_buff_head *); /* diff --git a/net/rxrpc/conn_event.c b/net/rxrpc/conn_event.c index 611c790bc6d0c..d37cef2523997 100644 --- a/net/rxrpc/conn_event.c +++ b/net/rxrpc/conn_event.c @@ -391,8 +391,17 @@ static void rxrpc_do_process_connection(struct rxrpc_connection *conn) /* go through the conn-level event packets, releasing the ref on this * 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); + sp->chall.conn = NULL; + } rxrpc_free_skb(skb, rxrpc_skb_put_conn_work); } } @@ -414,15 +423,22 @@ 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); + return true; } /* @@ -442,7 +458,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; } @@ -480,8 +499,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..d85345e014a9f 100644 --- a/net/rxrpc/conn_object.c +++ b/net/rxrpc/conn_object.c @@ -300,6 +300,19 @@ static void rxrpc_rcu_free_connection(struct rcu_head *rcu) wake_up_var(&rxnet->nr_conns); } +/* + * Purge a connection's receive queue, updating the local queue accounting. + */ +static void rxrpc_purge_conn_queue(struct rxrpc_connection *conn) +{ + struct sk_buff *skb; + + while ((skb = skb_dequeue(&conn->rx_queue))) { + rxrpc_uncharge_async_queue(conn->local, skb); + rxrpc_free_skb(skb, rxrpc_skb_put_purge); + } +} + /* * Clean up a dead connection. */ @@ -329,7 +342,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 +351,15 @@ 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. */ - 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..8679402c03e1b 100644 --- a/net/rxrpc/io_thread.c +++ b/net/rxrpc/io_thread.c @@ -528,6 +528,8 @@ int rxrpc_io_thread(void *data) if (skb->mark == RXRPC_SKB_MARK_PACKET) rxrpc_skb_orphan_udp(skb); + else 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; diff --git a/net/rxrpc/local_object.c b/net/rxrpc/local_object.c index f519a93059244..8e43f77d43683 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); diff --git a/net/rxrpc/oob.c b/net/rxrpc/oob.c index c80ee2487d094..e3d135b12d22d 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,10 @@ 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; + enum rxrpc_oob_type type = RXRPC_OOB_CHALLENGE; - 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; @@ -312,7 +315,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 +331,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 e2169d1a14b5f..a5113a284456e 100644 --- a/net/rxrpc/skbuff.c +++ b/net/rxrpc/skbuff.c @@ -58,6 +58,39 @@ void rxrpc_free_skb(struct sk_buff *skb, enum rxrpc_skb_trace why) } } +/* + * Charge an skb retained outside UDP 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) + 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,7 +99,15 @@ 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); + } + 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