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

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 0;
}

/*
 * 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) == 0)
			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;
	bool should_stop;

	complete(&local->io_thread_ready);

	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);
		should_stop = kthread_should_stop();
		if (!skb_queue_empty(&local->rx_queue) ||
		    !list_empty(&local->call_attend_q)) {
			__set_current_state(TASK_RUNNING);
			continue;
		}

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