[linux-block.git] / net / rxrpc / io_thread.c

// 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"

/*
 * 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();
}

/*
 * post connection-level events to the connection
 * - this includes challenges, responses, some aborts and call terminal packet
 *   retransmission.
 */
static void rxrpc_post_packet_to_conn(struct rxrpc_connection *conn,
				      struct sk_buff *skb)
{
	_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);
}

/*
 * post endpoint-level events to the local endpoint
 * - this includes debug and version messages
 */
static void rxrpc_post_packet_to_local(struct rxrpc_local *local,
				       struct sk_buff *skb)
{
	_enter("%p,%p", local, skb);

	if (rxrpc_get_local_maybe(local, rxrpc_local_get_queue)) {
		rxrpc_get_skb(skb, rxrpc_skb_get_local_work);
		skb_queue_tail(&local->event_queue, skb);
		rxrpc_queue_local(local);
	}
}

/*
 * put a packet up for transport-level abort
 */
static void rxrpc_reject_packet(struct rxrpc_local *local, struct sk_buff *skb)
{
	if (rxrpc_get_local_maybe(local, rxrpc_local_get_queue)) {
		rxrpc_get_skb(skb, rxrpc_skb_get_reject_work);
		skb_queue_tail(&local->reject_queue, skb);
		rxrpc_queue_local(local);
	}
}

/*
 * 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_channel *chan;
	struct rxrpc_call *call = NULL;
	struct rxrpc_skb_priv *sp;
	struct rxrpc_peer *peer = NULL;
	struct rxrpc_sock *rx = NULL;
	struct sk_buff *skb = *_skb;
	unsigned int channel;

	if (skb->tstamp == 0)
		skb->tstamp = ktime_get_real();

	skb_pull(skb, sizeof(struct udphdr));

	/* The UDP protocol already released all skb resources;
	 * we are free to add our own data there.
	 */
	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;
		}
	}

	if (skb->tstamp == 0)
		skb->tstamp = ktime_get_real();
	trace_rxrpc_rx_packet(sp);

	switch (sp->hdr.type) {
	case RXRPC_PACKET_TYPE_VERSION:
		if (rxrpc_to_client(sp))
			return 0;
		rxrpc_post_packet_to_local(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 0; /* 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 0; /* Wrong address type - discard. */
	}

	rcu_read_lock();

	if (rxrpc_to_server(sp)) {
		/* Weed out packets to services we're not offering.  Packets
		 * that would begin a call are explicitly rejected and the rest
		 * are just discarded.
		 */
		rx = rcu_dereference(local->service);
		if (!rx || (sp->hdr.serviceId != rx->srx.srx_service &&
			    sp->hdr.serviceId != rx->second_service)
		    ) {
			rcu_read_unlock();
			if (sp->hdr.type == RXRPC_PACKET_TYPE_DATA &&
			    sp->hdr.seq == 1)
				goto unsupported_service;
			return 0;
		}
	}

	conn = rxrpc_find_connection_rcu(local, &peer_srx, skb, &peer);
	if (conn) {
		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 (sp->hdr.callNumber == 0) {
			/* Connection-level packet */
			_debug("CONN %p {%d}", conn, conn->debug_id);
			conn = rxrpc_get_connection_maybe(conn, rxrpc_conn_get_conn_input);
			rcu_read_unlock();
			if (conn) {
				rxrpc_post_packet_to_conn(conn, skb);
				rxrpc_put_connection(conn, rxrpc_conn_put_conn_input);
			}
			return 0;
		}

		if ((int)sp->hdr.serial - (int)conn->hi_serial > 0)
			conn->hi_serial = sp->hdr.serial;

		/* 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) {
			rcu_read_unlock();
			return 0;
		}

		if (sp->hdr.callNumber == chan->last_call) {
			if (chan->call ||
			    sp->hdr.type == RXRPC_PACKET_TYPE_ABORT) {
				rcu_read_unlock();
				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) {
				rcu_read_unlock();
				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);
			conn = rxrpc_get_connection_maybe(conn, rxrpc_conn_get_call_input);
			rcu_read_unlock();
			if (conn) {
				rxrpc_post_packet_to_conn(conn, skb);
				rxrpc_put_connection(conn, rxrpc_conn_put_call_input);
			}
			return 0;
		}

		call = rcu_dereference(chan->call);

		if (sp->hdr.callNumber > chan->call_id) {
			if (rxrpc_to_client(sp)) {
				rcu_read_unlock();
				goto reject_packet;
			}
			if (call) {
				rxrpc_input_implicit_end_call(conn, call);
				chan->call = NULL;
				call = NULL;
			}
		}

		if (call && !rxrpc_try_get_call(call, rxrpc_call_get_input))
			call = NULL;

		if (call) {
			if (sp->hdr.serviceId != call->dest_srx.srx_service)
				call->dest_srx.srx_service = sp->hdr.serviceId;
			if ((int)sp->hdr.serial - (int)call->rx_serial > 0)
				call->rx_serial = sp->hdr.serial;
			if (!test_bit(RXRPC_CALL_RX_HEARD, &call->flags))
				set_bit(RXRPC_CALL_RX_HEARD, &call->flags);
		}
	}

	if (!call) {
		if (rxrpc_to_client(sp) ||
		    sp->hdr.type != RXRPC_PACKET_TYPE_DATA) {
			rcu_read_unlock();
			goto bad_message;
		}
		if (sp->hdr.seq != 1) {
			rcu_read_unlock();
			return 0;
		}
		call = rxrpc_new_incoming_call(local, rx, &peer_srx, skb);
		if (!call) {
			rcu_read_unlock();
			goto reject_packet;
		}
	}

	rcu_read_unlock();

	/* Process a call packet. */
	rxrpc_input_call_event(call, skb);
	rxrpc_put_call(call, rxrpc_call_put_input);
	trace_rxrpc_rx_done(0, 0);
	return 0;

wrong_security:
	rcu_read_unlock();
	trace_rxrpc_abort(0, "SEC", sp->hdr.cid, sp->hdr.callNumber, sp->hdr.seq,
			  RXKADINCONSISTENCY, EBADMSG);
	skb->priority = RXKADINCONSISTENCY;
	goto post_abort;

unsupported_service:
	trace_rxrpc_abort(0, "INV", sp->hdr.cid, sp->hdr.callNumber, sp->hdr.seq,
			  RX_INVALID_OPERATION, EOPNOTSUPP);
	skb->priority = RX_INVALID_OPERATION;
	goto post_abort;

reupgrade:
	rcu_read_unlock();
	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(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;
}
Commit	Line	Data
96b2d69b DH	1	// SPDX-License-Identifier: GPL-2.0-or-later
	2	/* RxRPC packet reception
	3	*
a275da62	4	* Copyright (C) 2007, 2016, 2022 Red Hat, Inc. All Rights Reserved.
96b2d69b DH	5	* Written by David Howells (dhowells@redhat.com)
	6	*/
	7
	8	#define pr_fmt(fmt) KBUILD_MODNAME ": " fmt
	9
	10	#include "ar-internal.h"
	11
446b3e14 DH	12	/*
	13	* handle data received on the local endpoint
	14	* - may be called in interrupt context
	15	*
	16	* [!] Note that as this is called from the encap_rcv hook, the socket is not
	17	* held locked by the caller and nothing prevents sk_user_data on the UDP from
	18	* being cleared in the middle of processing this function.
	19	*
	20	* Called with the RCU read lock held from the IP layer via UDP.
	21	*/
	22	int rxrpc_encap_rcv(struct sock udp_sk, struct sk_buff skb)
	23	{
	24	struct rxrpc_local *local = rcu_dereference_sk_user_data(udp_sk);
	25
	26	if (unlikely(!local)) {
	27	kfree_skb(skb);
	28	return 0;
	29	}
	30	if (skb->tstamp == 0)
	31	skb->tstamp = ktime_get_real();
	32
	33	skb->mark = RXRPC_SKB_MARK_PACKET;
	34	rxrpc_new_skb(skb, rxrpc_skb_new_encap_rcv);
	35	skb_queue_tail(&local->rx_queue, skb);
	36	rxrpc_wake_up_io_thread(local);
	37	return 0;
	38	}
	39
ff734825 DH	40	/*
	41	* Handle an error received on the local endpoint.
	42	*/
	43	void rxrpc_error_report(struct sock *sk)
	44	{
	45	struct rxrpc_local *local;
	46	struct sk_buff *skb;
	47
	48	rcu_read_lock();
	49	local = rcu_dereference_sk_user_data(sk);
	50	if (unlikely(!local)) {
	51	rcu_read_unlock();
	52	return;
	53	}
	54
	55	while ((skb = skb_dequeue(&sk->sk_error_queue))) {
	56	skb->mark = RXRPC_SKB_MARK_ERROR;
	57	rxrpc_new_skb(skb, rxrpc_skb_new_error_report);
	58	skb_queue_tail(&local->rx_queue, skb);
	59	}
	60
	61	rxrpc_wake_up_io_thread(local);
	62	rcu_read_unlock();
	63	}
	64
96b2d69b DH	65	/*
	66	* post connection-level events to the connection
	67	* - this includes challenges, responses, some aborts and call terminal packet
	68	* retransmission.
	69	*/
	70	static void rxrpc_post_packet_to_conn(struct rxrpc_connection *conn,
	71	struct sk_buff *skb)
	72	{
	73	_enter("%p,%p", conn, skb);
	74
2d1faf7a	75	rxrpc_get_skb(skb, rxrpc_skb_get_conn_work);
96b2d69b DH	76	skb_queue_tail(&conn->rx_queue, skb);
	77	rxrpc_queue_conn(conn, rxrpc_conn_queue_rx_work);
	78	}
	79
	80	/*
	81	* post endpoint-level events to the local endpoint
	82	* - this includes debug and version messages
	83	*/
	84	static void rxrpc_post_packet_to_local(struct rxrpc_local *local,
	85	struct sk_buff *skb)
	86	{
	87	_enter("%p,%p", local, skb);
	88
	89	if (rxrpc_get_local_maybe(local, rxrpc_local_get_queue)) {
2d1faf7a	90	rxrpc_get_skb(skb, rxrpc_skb_get_local_work);
96b2d69b DH	91	skb_queue_tail(&local->event_queue, skb);
96b2d69b DH	92	rxrpc_queue_local(local);
96b2d69b DH	93	}
	94	}
	95
	96	/*
	97	* put a packet up for transport-level abort
	98	*/
	99	static void rxrpc_reject_packet(struct rxrpc_local local, struct sk_buff skb)
	100	{
	101	if (rxrpc_get_local_maybe(local, rxrpc_local_get_queue)) {
2d1faf7a	102	rxrpc_get_skb(skb, rxrpc_skb_get_reject_work);
96b2d69b DH	103	skb_queue_tail(&local->reject_queue, skb);
96b2d69b DH	104	rxrpc_queue_local(local);
96b2d69b DH	105	}
	106	}
	107
	108	/*
	109	* Extract the wire header from a packet and translate the byte order.
	110	*/
	111	static noinline
	112	int rxrpc_extract_header(struct rxrpc_skb_priv sp, struct sk_buff skb)
	113	{
	114	struct rxrpc_wire_header whdr;
	115
	116	/* dig out the RxRPC connection details */
	117	if (skb_copy_bits(skb, 0, &whdr, sizeof(whdr)) < 0) {
	118	trace_rxrpc_rx_eproto(NULL, sp->hdr.serial,
	119	tracepoint_string("bad_hdr"));
	120	return -EBADMSG;
	121	}
	122
	123	memset(sp, 0, sizeof(*sp));
	124	sp->hdr.epoch = ntohl(whdr.epoch);
	125	sp->hdr.cid = ntohl(whdr.cid);
	126	sp->hdr.callNumber = ntohl(whdr.callNumber);
	127	sp->hdr.seq = ntohl(whdr.seq);
	128	sp->hdr.serial = ntohl(whdr.serial);
	129	sp->hdr.flags = whdr.flags;
	130	sp->hdr.type = whdr.type;
	131	sp->hdr.userStatus = whdr.userStatus;
	132	sp->hdr.securityIndex = whdr.securityIndex;
	133	sp->hdr._rsvd = ntohs(whdr._rsvd);
	134	sp->hdr.serviceId = ntohs(whdr.serviceId);
	135	return 0;
	136	}
	137
	138	/*
	139	* Extract the abort code from an ABORT packet and stash it in skb->priority.
	140	*/
	141	static bool rxrpc_extract_abort(struct sk_buff *skb)
	142	{
	143	__be32 wtmp;
	144
	145	if (skb_copy_bits(skb, sizeof(struct rxrpc_wire_header),
	146	&wtmp, sizeof(wtmp)) < 0)
	147	return false;
	148	skb->priority = ntohl(wtmp);
	149	return true;
	150	}
	151
	152	/*
446b3e14	153	* Process packets received on the local endpoint
96b2d69b	154	*/
2d1faf7a	155	static int rxrpc_input_packet(struct rxrpc_local local, struct sk_buff *_skb)
96b2d69b	156	{
96b2d69b	157	struct rxrpc_connection *conn;
393a2a20	158	struct sockaddr_rxrpc peer_srx;
96b2d69b DH	159	struct rxrpc_channel *chan;
	160	struct rxrpc_call *call = NULL;
	161	struct rxrpc_skb_priv *sp;
	162	struct rxrpc_peer *peer = NULL;
	163	struct rxrpc_sock *rx = NULL;
2d1faf7a	164	struct sk_buff skb = _skb;
96b2d69b DH	165	unsigned int channel;
96b2d69b DH	166
96b2d69b DH	167	if (skb->tstamp == 0)
	168	skb->tstamp = ktime_get_real();
	169
96b2d69b DH	170	skb_pull(skb, sizeof(struct udphdr));
	171
	172	/* The UDP protocol already released all skb resources;
	173	* we are free to add our own data there.
	174	*/
	175	sp = rxrpc_skb(skb);
	176
	177	/* dig out the RxRPC connection details */
	178	if (rxrpc_extract_header(sp, skb) < 0)
	179	goto bad_message;
	180
	181	if (IS_ENABLED(CONFIG_AF_RXRPC_INJECT_LOSS)) {
	182	static int lose;
	183	if ((lose++ & 7) == 7) {
	184	trace_rxrpc_rx_lose(sp);
96b2d69b DH	185	return 0;
	186	}
	187	}
	188
	189	if (skb->tstamp == 0)
	190	skb->tstamp = ktime_get_real();
	191	trace_rxrpc_rx_packet(sp);
	192
	193	switch (sp->hdr.type) {
	194	case RXRPC_PACKET_TYPE_VERSION:
	195	if (rxrpc_to_client(sp))
2d1faf7a	196	return 0;
96b2d69b	197	rxrpc_post_packet_to_local(local, skb);
2d1faf7a	198	return 0;
96b2d69b DH	199
	200	case RXRPC_PACKET_TYPE_BUSY:
	201	if (rxrpc_to_server(sp))
2d1faf7a	202	return 0;
96b2d69b DH	203	fallthrough;
	204	case RXRPC_PACKET_TYPE_ACK:
	205	case RXRPC_PACKET_TYPE_ACKALL:
	206	if (sp->hdr.callNumber == 0)
	207	goto bad_message;
	208	break;
	209	case RXRPC_PACKET_TYPE_ABORT:
	210	if (!rxrpc_extract_abort(skb))
2d1faf7a	211	return 0; /* Just discard if malformed */
96b2d69b DH	212	break;
	213
	214	case RXRPC_PACKET_TYPE_DATA:
	215	if (sp->hdr.callNumber == 0 \|\|
	216	sp->hdr.seq == 0)
	217	goto bad_message;
	218
	219	/* Unshare the packet so that it can be modified for in-place
	220	* decryption.
	221	*/
	222	if (sp->hdr.securityIndex != 0) {
2d1faf7a DH	223	skb = skb_unshare(skb, GFP_ATOMIC);
	224	if (!skb) {
	225	rxrpc_eaten_skb(*_skb, rxrpc_skb_eaten_by_unshare_nomem);
	226	*_skb = NULL;
	227	return 0;
96b2d69b DH	228	}
96b2d69b DH	229
2d1faf7a DH	230	if (skb != *_skb) {
	231	rxrpc_eaten_skb(*_skb, rxrpc_skb_eaten_by_unshare);
	232	*_skb = skb;
96b2d69b DH	233	rxrpc_new_skb(skb, rxrpc_skb_new_unshared);
	234	sp = rxrpc_skb(skb);
	235	}
	236	}
	237	break;
	238
	239	case RXRPC_PACKET_TYPE_CHALLENGE:
	240	if (rxrpc_to_server(sp))
2d1faf7a	241	return 0;
96b2d69b DH	242	break;
	243	case RXRPC_PACKET_TYPE_RESPONSE:
	244	if (rxrpc_to_client(sp))
2d1faf7a	245	return 0;
96b2d69b DH	246	break;
	247
	248	/* Packet types 9-11 should just be ignored. */
	249	case RXRPC_PACKET_TYPE_PARAMS:
	250	case RXRPC_PACKET_TYPE_10:
	251	case RXRPC_PACKET_TYPE_11:
2d1faf7a	252	return 0;
96b2d69b DH	253
	254	default:
	255	goto bad_message;
	256	}
	257
	258	if (sp->hdr.serviceId == 0)
	259	goto bad_message;
	260
393a2a20 DH	261	if (WARN_ON_ONCE(rxrpc_extract_addr_from_skb(&peer_srx, skb) < 0))
	262	return 0; /* Unsupported address type - discard. */
	263
	264	if (peer_srx.transport.family != local->srx.transport.family &&
	265	(peer_srx.transport.family == AF_INET &&
	266	local->srx.transport.family != AF_INET6)) {
	267	pr_warn_ratelimited("AF_RXRPC: Protocol mismatch %u not %u\n",
	268	peer_srx.transport.family,
	269	local->srx.transport.family);
	270	return 0; /* Wrong address type - discard. */
	271	}
	272
cd21effb DH	273	rcu_read_lock();
cd21effb DH	274
96b2d69b DH	275	if (rxrpc_to_server(sp)) {
	276	/* Weed out packets to services we're not offering. Packets
	277	* that would begin a call are explicitly rejected and the rest
	278	* are just discarded.
	279	*/
	280	rx = rcu_dereference(local->service);
	281	if (!rx \|\| (sp->hdr.serviceId != rx->srx.srx_service &&
cd21effb DH	282	sp->hdr.serviceId != rx->second_service)
	283	) {
	284	rcu_read_unlock();
96b2d69b DH	285	if (sp->hdr.type == RXRPC_PACKET_TYPE_DATA &&
	286	sp->hdr.seq == 1)
	287	goto unsupported_service;
2d1faf7a	288	return 0;
96b2d69b DH	289	}
	290	}
	291
393a2a20	292	conn = rxrpc_find_connection_rcu(local, &peer_srx, skb, &peer);
96b2d69b DH	293	if (conn) {
	294	if (sp->hdr.securityIndex != conn->security_ix)
	295	goto wrong_security;
	296
	297	if (sp->hdr.serviceId != conn->service_id) {
	298	int old_id;
	299
	300	if (!test_bit(RXRPC_CONN_PROBING_FOR_UPGRADE, &conn->flags))
	301	goto reupgrade;
	302	old_id = cmpxchg(&conn->service_id, conn->orig_service_id,
	303	sp->hdr.serviceId);
	304
	305	if (old_id != conn->orig_service_id &&
	306	old_id != sp->hdr.serviceId)
	307	goto reupgrade;
	308	}
	309
	310	if (sp->hdr.callNumber == 0) {
	311	/* Connection-level packet */
	312	_debug("CONN %p {%d}", conn, conn->debug_id);
cd21effb DH	313	conn = rxrpc_get_connection_maybe(conn, rxrpc_conn_get_conn_input);
	314	rcu_read_unlock();
	315	if (conn) {
	316	rxrpc_post_packet_to_conn(conn, skb);
	317	rxrpc_put_connection(conn, rxrpc_conn_put_conn_input);
	318	}
2d1faf7a	319	return 0;
96b2d69b DH	320	}
	321
	322	if ((int)sp->hdr.serial - (int)conn->hi_serial > 0)
	323	conn->hi_serial = sp->hdr.serial;
	324
	325	/* Call-bound packets are routed by connection channel. */
	326	channel = sp->hdr.cid & RXRPC_CHANNELMASK;
	327	chan = &conn->channels[channel];
	328
	329	/* Ignore really old calls */
cd21effb DH	330	if (sp->hdr.callNumber < chan->last_call) {
cd21effb DH	331	rcu_read_unlock();
2d1faf7a	332	return 0;
cd21effb	333	}
96b2d69b DH	334
	335	if (sp->hdr.callNumber == chan->last_call) {
	336	if (chan->call \|\|
cd21effb DH	337	sp->hdr.type == RXRPC_PACKET_TYPE_ABORT) {
cd21effb DH	338	rcu_read_unlock();
2d1faf7a	339	return 0;
cd21effb	340	}
96b2d69b DH	341
	342	/* For the previous service call, if completed
	343	* successfully, we discard all further packets.
	344	*/
	345	if (rxrpc_conn_is_service(conn) &&
cd21effb DH	346	chan->last_type == RXRPC_PACKET_TYPE_ACK) {
cd21effb DH	347	rcu_read_unlock();
2d1faf7a	348	return 0;
cd21effb	349	}
96b2d69b DH	350
	351	/* But otherwise we need to retransmit the final packet
	352	* from data cached in the connection record.
	353	*/
	354	if (sp->hdr.type == RXRPC_PACKET_TYPE_DATA)
	355	trace_rxrpc_rx_data(chan->call_debug_id,
	356	sp->hdr.seq,
	357	sp->hdr.serial,
	358	sp->hdr.flags);
cd21effb DH	359	conn = rxrpc_get_connection_maybe(conn, rxrpc_conn_get_call_input);
	360	rcu_read_unlock();
	361	if (conn) {
	362	rxrpc_post_packet_to_conn(conn, skb);
	363	rxrpc_put_connection(conn, rxrpc_conn_put_call_input);
	364	}
2d1faf7a	365	return 0;
96b2d69b DH	366	}
	367
	368	call = rcu_dereference(chan->call);
	369
	370	if (sp->hdr.callNumber > chan->call_id) {
cd21effb DH	371	if (rxrpc_to_client(sp)) {
cd21effb DH	372	rcu_read_unlock();
96b2d69b	373	goto reject_packet;
cd21effb DH	374	}
	375	if (call) {
	376	rxrpc_input_implicit_end_call(conn, call);
	377	chan->call = NULL;
	378	call = NULL;
	379	}
96b2d69b DH	380	}
96b2d69b DH	381
cd21effb DH	382	if (call && !rxrpc_try_get_call(call, rxrpc_call_get_input))
	383	call = NULL;
	384
96b2d69b	385	if (call) {
f3441d41 DH	386	if (sp->hdr.serviceId != call->dest_srx.srx_service)
f3441d41 DH	387	call->dest_srx.srx_service = sp->hdr.serviceId;
96b2d69b DH	388	if ((int)sp->hdr.serial - (int)call->rx_serial > 0)
	389	call->rx_serial = sp->hdr.serial;
	390	if (!test_bit(RXRPC_CALL_RX_HEARD, &call->flags))
	391	set_bit(RXRPC_CALL_RX_HEARD, &call->flags);
	392	}
	393	}
	394
cd21effb	395	if (!call) {
96b2d69b	396	if (rxrpc_to_client(sp) \|\|
cd21effb DH	397	sp->hdr.type != RXRPC_PACKET_TYPE_DATA) {
cd21effb DH	398	rcu_read_unlock();
96b2d69b	399	goto bad_message;
cd21effb DH	400	}
	401	if (sp->hdr.seq != 1) {
	402	rcu_read_unlock();
2d1faf7a	403	return 0;
cd21effb	404	}
393a2a20	405	call = rxrpc_new_incoming_call(local, rx, &peer_srx, skb);
cd21effb DH	406	if (!call) {
cd21effb DH	407	rcu_read_unlock();
96b2d69b	408	goto reject_packet;
cd21effb	409	}
96b2d69b DH	410	}
96b2d69b DH	411
cd21effb DH	412	rcu_read_unlock();
cd21effb DH	413
2d1faf7a	414	/* Process a call packet. */
15f661dc	415	rxrpc_input_call_event(call, skb);
cd21effb	416	rxrpc_put_call(call, rxrpc_call_put_input);
96b2d69b DH	417	trace_rxrpc_rx_done(0, 0);
	418	return 0;
	419
	420	wrong_security:
cd21effb	421	rcu_read_unlock();
96b2d69b DH	422	trace_rxrpc_abort(0, "SEC", sp->hdr.cid, sp->hdr.callNumber, sp->hdr.seq,
	423	RXKADINCONSISTENCY, EBADMSG);
	424	skb->priority = RXKADINCONSISTENCY;
	425	goto post_abort;
	426
	427	unsupported_service:
	428	trace_rxrpc_abort(0, "INV", sp->hdr.cid, sp->hdr.callNumber, sp->hdr.seq,
	429	RX_INVALID_OPERATION, EOPNOTSUPP);
	430	skb->priority = RX_INVALID_OPERATION;
	431	goto post_abort;
	432
	433	reupgrade:
cd21effb	434	rcu_read_unlock();
96b2d69b DH	435	trace_rxrpc_abort(0, "UPG", sp->hdr.cid, sp->hdr.callNumber, sp->hdr.seq,
	436	RX_PROTOCOL_ERROR, EBADMSG);
	437	goto protocol_error;
	438
	439	bad_message:
	440	trace_rxrpc_abort(0, "BAD", sp->hdr.cid, sp->hdr.callNumber, sp->hdr.seq,
	441	RX_PROTOCOL_ERROR, EBADMSG);
	442	protocol_error:
	443	skb->priority = RX_PROTOCOL_ERROR;
	444	post_abort:
	445	skb->mark = RXRPC_SKB_MARK_REJECT_ABORT;
	446	reject_packet:
96b2d69b	447	rxrpc_reject_packet(local, skb);
96b2d69b DH	448	return 0;
96b2d69b DH	449	}
a275da62 DH	450
	451	/*
	452	* I/O and event handling thread.
	453	*/
	454	int rxrpc_io_thread(void *data)
	455	{
	456	struct sk_buff_head rx_queue;
	457	struct rxrpc_local *local = data;
15f661dc	458	struct rxrpc_call *call;
a275da62 DH	459	struct sk_buff *skb;
	460
	461	skb_queue_head_init(&rx_queue);
	462
	463	set_user_nice(current, MIN_NICE);
	464
	465	for (;;) {
	466	rxrpc_inc_stat(local->rxnet, stat_io_loop);
	467
15f661dc DH	468	/* Deal with calls that want immediate attention. */
	469	if ((call = list_first_entry_or_null(&local->call_attend_q,
	470	struct rxrpc_call,
	471	attend_link))) {
	472	spin_lock_bh(&local->lock);
	473	list_del_init(&call->attend_link);
	474	spin_unlock_bh(&local->lock);
	475
	476	trace_rxrpc_call_poked(call);
	477	rxrpc_input_call_event(call, NULL);
	478	rxrpc_put_call(call, rxrpc_call_put_poke);
	479	continue;
	480	}
	481
a275da62 DH	482	/* Process received packets and errors. */
a275da62 DH	483	if ((skb = __skb_dequeue(&rx_queue))) {
446b3e14 DH	484	switch (skb->mark) {
446b3e14 DH	485	case RXRPC_SKB_MARK_PACKET:
2d1faf7a	486	skb->priority = 0;
2d1faf7a	487	rxrpc_input_packet(local, &skb);
2d1faf7a DH	488	trace_rxrpc_rx_done(skb->mark, skb->priority);
2d1faf7a DH	489	rxrpc_free_skb(skb, rxrpc_skb_put_input);
446b3e14	490	break;
ff734825 DH	491	case RXRPC_SKB_MARK_ERROR:
	492	rxrpc_input_error(local, skb);
	493	rxrpc_free_skb(skb, rxrpc_skb_put_error_report);
	494	break;
446b3e14 DH	495	default:
	496	WARN_ON_ONCE(1);
	497	rxrpc_free_skb(skb, rxrpc_skb_put_unknown);
	498	break;
	499	}
a275da62 DH	500	continue;
	501	}
	502
	503	if (!skb_queue_empty(&local->rx_queue)) {
	504	spin_lock_irq(&local->rx_queue.lock);
	505	skb_queue_splice_tail_init(&local->rx_queue, &rx_queue);
	506	spin_unlock_irq(&local->rx_queue.lock);
	507	continue;
	508	}
	509
	510	set_current_state(TASK_INTERRUPTIBLE);
15f661dc DH	511	if (!skb_queue_empty(&local->rx_queue) \|\|
15f661dc DH	512	!list_empty(&local->call_attend_q)) {
a275da62 DH	513	__set_current_state(TASK_RUNNING);
	514	continue;
	515	}
	516
	517	if (kthread_should_stop())
	518	break;
	519	schedule();
	520	}
	521
	522	__set_current_state(TASK_RUNNING);
	523	rxrpc_see_local(local, rxrpc_local_stop);
	524	rxrpc_destroy_local(local);
	525	local->io_thread = NULL;
	526	rxrpc_see_local(local, rxrpc_local_stopped);
	527	return 0;
	528	}