1 files changed, 496 insertions, 0 deletions

diff --git a/net/rxrpc/io_thread.c b/net/rxrpc/io_thread.c
new file mode 100644
index 000000000000..d83ae3193032
--- /dev/null
+++ b/net/rxrpc/io_thread.c
@@ -0,0 +1,496 @@
+// SPDX-License-Identifier: GPL-2.0-or-later
+/* RxRPC packet reception
+ *
+ * Copyright (C) 2007, 2016, 2022 Red Hat, Inc. All Rights Reserved.
+ * Written by David Howells (dhowells@redhat.com)
+ */
+
+#define pr_fmt(fmt) KBUILD_MODNAME ": " fmt
+
+#include "ar-internal.h"
+
+static int rxrpc_input_packet_on_conn(struct rxrpc_connection *conn,
+				      struct sockaddr_rxrpc *peer_srx,
+				      struct sk_buff *skb);
+
+/*
+ * handle data received on the local endpoint
+ * - may be called in interrupt context
+ *
+ * [!] Note that as this is called from the encap_rcv hook, the socket is not
+ * held locked by the caller and nothing prevents sk_user_data on the UDP from
+ * being cleared in the middle of processing this function.
+ *
+ * Called with the RCU read lock held from the IP layer via UDP.
+ */
+int rxrpc_encap_rcv(struct sock *udp_sk, struct sk_buff *skb)
+{
+	struct rxrpc_local *local = rcu_dereference_sk_user_data(udp_sk);
+
+	if (unlikely(!local)) {
+		kfree_skb(skb);
+		return 0;
+	}
+	if (skb->tstamp == 0)
+		skb->tstamp = ktime_get_real();
+
+	skb->mark = RXRPC_SKB_MARK_PACKET;
+	rxrpc_new_skb(skb, rxrpc_skb_new_encap_rcv);
+	skb_queue_tail(&local->rx_queue, skb);
+	rxrpc_wake_up_io_thread(local);
+	return 0;
+}
+
+/*
+ * Handle an error received on the local endpoint.
+ */
+void rxrpc_error_report(struct sock *sk)
+{
+	struct rxrpc_local *local;
+	struct sk_buff *skb;
+
+	rcu_read_lock();
+	local = rcu_dereference_sk_user_data(sk);
+	if (unlikely(!local)) {
+		rcu_read_unlock();
+		return;
+	}
+
+	while ((skb = skb_dequeue(&sk->sk_error_queue))) {
+		skb->mark = RXRPC_SKB_MARK_ERROR;
+		rxrpc_new_skb(skb, rxrpc_skb_new_error_report);
+		skb_queue_tail(&local->rx_queue, skb);
+	}
+
+	rxrpc_wake_up_io_thread(local);
+	rcu_read_unlock();
+}
+
+/*
+ * Process event packets targeted at a local endpoint.
+ */
+static void rxrpc_input_version(struct rxrpc_local *local, struct sk_buff *skb)
+{
+	struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
+	char v;
+
+	_enter("");
+
+	rxrpc_see_skb(skb, rxrpc_skb_see_version);
+	if (skb_copy_bits(skb, sizeof(struct rxrpc_wire_header), &v, 1) >= 0) {
+		if (v == 0)
+			rxrpc_send_version_request(local, &sp->hdr, skb);
+	}
+}
+
+/*
+ * Extract the wire header from a packet and translate the byte order.
+ */
+static noinline
+int rxrpc_extract_header(struct rxrpc_skb_priv *sp, struct sk_buff *skb)
+{
+	struct rxrpc_wire_header whdr;
+
+	/* dig out the RxRPC connection details */
+	if (skb_copy_bits(skb, 0, &whdr, sizeof(whdr)) < 0) {
+		trace_rxrpc_rx_eproto(NULL, sp->hdr.serial,
+				      tracepoint_string("bad_hdr"));
+		return -EBADMSG;
+	}
+
+	memset(sp, 0, sizeof(*sp));
+	sp->hdr.epoch		= ntohl(whdr.epoch);
+	sp->hdr.cid		= ntohl(whdr.cid);
+	sp->hdr.callNumber	= ntohl(whdr.callNumber);
+	sp->hdr.seq		= ntohl(whdr.seq);
+	sp->hdr.serial		= ntohl(whdr.serial);
+	sp->hdr.flags		= whdr.flags;
+	sp->hdr.type		= whdr.type;
+	sp->hdr.userStatus	= whdr.userStatus;
+	sp->hdr.securityIndex	= whdr.securityIndex;
+	sp->hdr._rsvd		= ntohs(whdr._rsvd);
+	sp->hdr.serviceId	= ntohs(whdr.serviceId);
+	return 0;
+}
+
+/*
+ * Extract the abort code from an ABORT packet and stash it in skb->priority.
+ */
+static bool rxrpc_extract_abort(struct sk_buff *skb)
+{
+	__be32 wtmp;
+
+	if (skb_copy_bits(skb, sizeof(struct rxrpc_wire_header),
+			  &wtmp, sizeof(wtmp)) < 0)
+		return false;
+	skb->priority = ntohl(wtmp);
+	return true;
+}
+
+/*
+ * Process packets received on the local endpoint
+ */
+static int rxrpc_input_packet(struct rxrpc_local *local, struct sk_buff **_skb)
+{
+	struct rxrpc_connection *conn;
+	struct sockaddr_rxrpc peer_srx;
+	struct rxrpc_skb_priv *sp;
+	struct rxrpc_peer *peer = NULL;
+	struct sk_buff *skb = *_skb;
+	int ret = 0;
+
+	skb_pull(skb, sizeof(struct udphdr));
+
+	sp = rxrpc_skb(skb);
+
+	/* dig out the RxRPC connection details */
+	if (rxrpc_extract_header(sp, skb) < 0)
+		goto bad_message;
+
+	if (IS_ENABLED(CONFIG_AF_RXRPC_INJECT_LOSS)) {
+		static int lose;
+		if ((lose++ & 7) == 7) {
+			trace_rxrpc_rx_lose(sp);
+			return 0;
+		}
+	}
+
+	trace_rxrpc_rx_packet(sp);
+
+	switch (sp->hdr.type) {
+	case RXRPC_PACKET_TYPE_VERSION:
+		if (rxrpc_to_client(sp))
+			return 0;
+		rxrpc_input_version(local, skb);
+		return 0;
+
+	case RXRPC_PACKET_TYPE_BUSY:
+		if (rxrpc_to_server(sp))
+			return 0;
+		fallthrough;
+	case RXRPC_PACKET_TYPE_ACK:
+	case RXRPC_PACKET_TYPE_ACKALL:
+		if (sp->hdr.callNumber == 0)
+			goto bad_message;
+		break;
+	case RXRPC_PACKET_TYPE_ABORT:
+		if (!rxrpc_extract_abort(skb))
+			return 0; /* Just discard if malformed */
+		break;
+
+	case RXRPC_PACKET_TYPE_DATA:
+		if (sp->hdr.callNumber == 0 ||
+		    sp->hdr.seq == 0)
+			goto bad_message;
+
+		/* Unshare the packet so that it can be modified for in-place
+		 * decryption.
+		 */
+		if (sp->hdr.securityIndex != 0) {
+			skb = skb_unshare(skb, GFP_ATOMIC);
+			if (!skb) {
+				rxrpc_eaten_skb(*_skb, rxrpc_skb_eaten_by_unshare_nomem);
+				*_skb = NULL;
+				return 0;
+			}
+
+			if (skb != *_skb) {
+				rxrpc_eaten_skb(*_skb, rxrpc_skb_eaten_by_unshare);
+				*_skb = skb;
+				rxrpc_new_skb(skb, rxrpc_skb_new_unshared);
+				sp = rxrpc_skb(skb);
+			}
+		}
+		break;
+
+	case RXRPC_PACKET_TYPE_CHALLENGE:
+		if (rxrpc_to_server(sp))
+			return 0;
+		break;
+	case RXRPC_PACKET_TYPE_RESPONSE:
+		if (rxrpc_to_client(sp))
+			return 0;
+		break;
+
+		/* Packet types 9-11 should just be ignored. */
+	case RXRPC_PACKET_TYPE_PARAMS:
+	case RXRPC_PACKET_TYPE_10:
+	case RXRPC_PACKET_TYPE_11:
+		return 0;
+
+	default:
+		goto bad_message;
+	}
+
+	if (sp->hdr.serviceId == 0)
+		goto bad_message;
+
+	if (WARN_ON_ONCE(rxrpc_extract_addr_from_skb(&peer_srx, skb) < 0))
+		return true; /* Unsupported address type - discard. */
+
+	if (peer_srx.transport.family != local->srx.transport.family &&
+	    (peer_srx.transport.family == AF_INET &&
+	     local->srx.transport.family != AF_INET6)) {
+		pr_warn_ratelimited("AF_RXRPC: Protocol mismatch %u not %u\n",
+				    peer_srx.transport.family,
+				    local->srx.transport.family);
+		return true; /* Wrong address type - discard. */
+	}
+
+	if (rxrpc_to_client(sp)) {
+		rcu_read_lock();
+		conn = rxrpc_find_client_connection_rcu(local, &peer_srx, skb);
+		conn = rxrpc_get_connection_maybe(conn, rxrpc_conn_get_call_input);
+		rcu_read_unlock();
+		if (!conn) {
+			trace_rxrpc_abort(0, "NCC", sp->hdr.cid,
+					  sp->hdr.callNumber, sp->hdr.seq,
+					  RXKADINCONSISTENCY, EBADMSG);
+			goto protocol_error;
+		}
+
+		ret = rxrpc_input_packet_on_conn(conn, &peer_srx, skb);
+		rxrpc_put_connection(conn, rxrpc_conn_put_call_input);
+		return ret;
+	}
+
+	/* We need to look up service connections by the full protocol
+	 * parameter set.  We look up the peer first as an intermediate step
+	 * and then the connection from the peer's tree.
+	 */
+	rcu_read_lock();
+
+	peer = rxrpc_lookup_peer_rcu(local, &peer_srx);
+	if (!peer) {
+		rcu_read_unlock();
+		return rxrpc_new_incoming_call(local, NULL, NULL, &peer_srx, skb);
+	}
+
+	conn = rxrpc_find_service_conn_rcu(peer, skb);
+	conn = rxrpc_get_connection_maybe(conn, rxrpc_conn_get_call_input);
+	if (conn) {
+		rcu_read_unlock();
+		ret = rxrpc_input_packet_on_conn(conn, &peer_srx, skb);
+		rxrpc_put_connection(conn, rxrpc_conn_put_call_input);
+		return ret;
+	}
+
+	peer = rxrpc_get_peer_maybe(peer, rxrpc_peer_get_input);
+	rcu_read_unlock();
+
+	ret = rxrpc_new_incoming_call(local, peer, NULL, &peer_srx, skb);
+	rxrpc_put_peer(peer, rxrpc_peer_put_input);
+	if (ret < 0)
+		goto reject_packet;
+	return 0;
+
+bad_message:
+	trace_rxrpc_abort(0, "BAD", sp->hdr.cid, sp->hdr.callNumber, sp->hdr.seq,
+			  RX_PROTOCOL_ERROR, EBADMSG);
+protocol_error:
+	skb->priority = RX_PROTOCOL_ERROR;
+	skb->mark = RXRPC_SKB_MARK_REJECT_ABORT;
+reject_packet:
+	rxrpc_reject_packet(local, skb);
+	return ret;
+}
+
+/*
+ * Deal with a packet that's associated with an extant connection.
+ */
+static int rxrpc_input_packet_on_conn(struct rxrpc_connection *conn,
+				      struct sockaddr_rxrpc *peer_srx,
+				      struct sk_buff *skb)
+{
+	struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
+	struct rxrpc_channel *chan;
+	struct rxrpc_call *call = NULL;
+	unsigned int channel;
+
+	if (sp->hdr.securityIndex != conn->security_ix)
+		goto wrong_security;
+
+	if (sp->hdr.serviceId != conn->service_id) {
+		int old_id;
+
+		if (!test_bit(RXRPC_CONN_PROBING_FOR_UPGRADE, &conn->flags))
+			goto reupgrade;
+		old_id = cmpxchg(&conn->service_id, conn->orig_service_id,
+				 sp->hdr.serviceId);
+
+		if (old_id != conn->orig_service_id &&
+		    old_id != sp->hdr.serviceId)
+			goto reupgrade;
+	}
+
+	if (after(sp->hdr.serial, conn->hi_serial))
+		conn->hi_serial = sp->hdr.serial;
+
+	/* It's a connection-level packet if the call number is 0. */
+	if (sp->hdr.callNumber == 0)
+		return rxrpc_input_conn_packet(conn, skb);
+
+	/* Call-bound packets are routed by connection channel. */
+	channel = sp->hdr.cid & RXRPC_CHANNELMASK;
+	chan = &conn->channels[channel];
+
+	/* Ignore really old calls */
+	if (sp->hdr.callNumber < chan->last_call)
+		return 0;
+
+	if (sp->hdr.callNumber == chan->last_call) {
+		if (chan->call ||
+		    sp->hdr.type == RXRPC_PACKET_TYPE_ABORT)
+			return 0;
+
+		/* For the previous service call, if completed successfully, we
+		 * discard all further packets.
+		 */
+		if (rxrpc_conn_is_service(conn) &&
+		    chan->last_type == RXRPC_PACKET_TYPE_ACK)
+			return 0;
+
+		/* But otherwise we need to retransmit the final packet from
+		 * data cached in the connection record.
+		 */
+		if (sp->hdr.type == RXRPC_PACKET_TYPE_DATA)
+			trace_rxrpc_rx_data(chan->call_debug_id,
+					    sp->hdr.seq,
+					    sp->hdr.serial,
+					    sp->hdr.flags);
+		rxrpc_input_conn_packet(conn, skb);
+		return 0;
+	}
+
+	rcu_read_lock();
+	call = rxrpc_try_get_call(rcu_dereference(chan->call),
+				  rxrpc_call_get_input);
+	rcu_read_unlock();
+
+	if (sp->hdr.callNumber > chan->call_id) {
+		if (rxrpc_to_client(sp)) {
+			rxrpc_put_call(call, rxrpc_call_put_input);
+			goto reject_packet;
+		}
+
+		if (call) {
+			rxrpc_implicit_end_call(call, skb);
+			rxrpc_put_call(call, rxrpc_call_put_input);
+			call = NULL;
+		}
+	}
+
+	if (!call) {
+		if (rxrpc_to_client(sp))
+			goto bad_message;
+		if (rxrpc_new_incoming_call(conn->local, conn->peer, conn,
+					    peer_srx, skb))
+			return 0;
+		goto reject_packet;
+	}
+
+	rxrpc_input_call_event(call, skb);
+	rxrpc_put_call(call, rxrpc_call_put_input);
+	return 0;
+
+wrong_security:
+	trace_rxrpc_abort(0, "SEC", sp->hdr.cid, sp->hdr.callNumber, sp->hdr.seq,
+			  RXKADINCONSISTENCY, EBADMSG);
+	skb->priority = RXKADINCONSISTENCY;
+	goto post_abort;
+
+reupgrade:
+	trace_rxrpc_abort(0, "UPG", sp->hdr.cid, sp->hdr.callNumber, sp->hdr.seq,
+			  RX_PROTOCOL_ERROR, EBADMSG);
+	goto protocol_error;
+
+bad_message:
+	trace_rxrpc_abort(0, "BAD", sp->hdr.cid, sp->hdr.callNumber, sp->hdr.seq,
+			  RX_PROTOCOL_ERROR, EBADMSG);
+protocol_error:
+	skb->priority = RX_PROTOCOL_ERROR;
+post_abort:
+	skb->mark = RXRPC_SKB_MARK_REJECT_ABORT;
+reject_packet:
+	rxrpc_reject_packet(conn->local, skb);
+	return 0;
+}
+
+/*
+ * I/O and event handling thread.
+ */
+int rxrpc_io_thread(void *data)
+{
+	struct sk_buff_head rx_queue;
+	struct rxrpc_local *local = data;
+	struct rxrpc_call *call;
+	struct sk_buff *skb;
+
+	skb_queue_head_init(&rx_queue);
+
+	set_user_nice(current, MIN_NICE);
+
+	for (;;) {
+		rxrpc_inc_stat(local->rxnet, stat_io_loop);
+
+		/* Deal with calls that want immediate attention. */
+		if ((call = list_first_entry_or_null(&local->call_attend_q,
+						     struct rxrpc_call,
+						     attend_link))) {
+			spin_lock_bh(&local->lock);
+			list_del_init(&call->attend_link);
+			spin_unlock_bh(&local->lock);
+
+			trace_rxrpc_call_poked(call);
+			rxrpc_input_call_event(call, NULL);
+			rxrpc_put_call(call, rxrpc_call_put_poke);
+			continue;
+		}
+
+		/* Process received packets and errors. */
+		if ((skb = __skb_dequeue(&rx_queue))) {
+			switch (skb->mark) {
+			case RXRPC_SKB_MARK_PACKET:
+				skb->priority = 0;
+				rxrpc_input_packet(local, &skb);
+				trace_rxrpc_rx_done(skb->mark, skb->priority);
+				rxrpc_free_skb(skb, rxrpc_skb_put_input);
+				break;
+			case RXRPC_SKB_MARK_ERROR:
+				rxrpc_input_error(local, skb);
+				rxrpc_free_skb(skb, rxrpc_skb_put_error_report);
+				break;
+			default:
+				WARN_ON_ONCE(1);
+				rxrpc_free_skb(skb, rxrpc_skb_put_unknown);
+				break;
+			}
+			continue;
+		}
+
+		if (!skb_queue_empty(&local->rx_queue)) {
+			spin_lock_irq(&local->rx_queue.lock);
+			skb_queue_splice_tail_init(&local->rx_queue, &rx_queue);
+			spin_unlock_irq(&local->rx_queue.lock);
+			continue;
+		}
+
+		set_current_state(TASK_INTERRUPTIBLE);
+		if (!skb_queue_empty(&local->rx_queue) ||
+		    !list_empty(&local->call_attend_q)) {
+			__set_current_state(TASK_RUNNING);
+			continue;
+		}
+
+		if (kthread_should_stop())
+			break;
+		schedule();
+	}
+
+	__set_current_state(TASK_RUNNING);
+	rxrpc_see_local(local, rxrpc_local_stop);
+	rxrpc_destroy_local(local);
+	local->io_thread = NULL;
+	rxrpc_see_local(local, rxrpc_local_stopped);
+	return 0;
+}