io_thread.c

  1// SPDX-License-Identifier: GPL-2.0-or-later
  2/* RxRPC packet reception
  3 *
  4 * Copyright (C) 2007, 2016, 2022 Red Hat, Inc. All Rights Reserved.
  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
 12static int rxrpc_input_packet_on_conn(struct rxrpc_connection *conn,
 13				      struct sockaddr_rxrpc *peer_srx,
 14				      struct sk_buff *skb);
 15
 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 */
 26int rxrpc_encap_rcv(struct sock *udp_sk, struct sk_buff *skb)
 27{
 28	struct sk_buff_head *rx_queue;
 29	struct rxrpc_local *local = rcu_dereference_sk_user_data(udp_sk);
 30	struct task_struct *io_thread;
 31
 32	if (unlikely(!local)) {
 33		kfree_skb(skb);
 34		return 0;
 35	}
 36	io_thread = READ_ONCE(local->io_thread);
 37	if (!io_thread) {
 38		kfree_skb(skb);
 39		return 0;
 40	}
 41	if (skb->tstamp == 0)
 42		skb->tstamp = ktime_get_real();
 43
 44	skb->mark = RXRPC_SKB_MARK_PACKET;
 45	rxrpc_new_skb(skb, rxrpc_skb_new_encap_rcv);
 46	rx_queue = &local->rx_queue;
 47#ifdef CONFIG_AF_RXRPC_INJECT_RX_DELAY
 48	if (rxrpc_inject_rx_delay ||
 49	    !skb_queue_empty(&local->rx_delay_queue)) {
 50		skb->tstamp = ktime_add_ms(skb->tstamp, rxrpc_inject_rx_delay);
 51		rx_queue = &local->rx_delay_queue;
 52	}
 53#endif
 54
 55	skb_queue_tail(rx_queue, skb);
 56	wake_up_process(io_thread);
 57	return 0;
 58}
 59
 60/*
 61 * Handle an error received on the local endpoint.
 62 */
 63void rxrpc_error_report(struct sock *sk)
 64{
 65	struct rxrpc_local *local;
 66	struct sk_buff *skb;
 67
 68	rcu_read_lock();
 69	local = rcu_dereference_sk_user_data(sk);
 70	if (unlikely(!local)) {
 71		rcu_read_unlock();
 72		return;
 73	}
 74
 75	while ((skb = skb_dequeue(&sk->sk_error_queue))) {
 76		skb->mark = RXRPC_SKB_MARK_ERROR;
 77		rxrpc_new_skb(skb, rxrpc_skb_new_error_report);
 78		skb_queue_tail(&local->rx_queue, skb);
 79	}
 80
 81	rxrpc_wake_up_io_thread(local);
 82	rcu_read_unlock();
 83}
 84
 85/*
 86 * Directly produce an abort from a packet.
 87 */
 88bool rxrpc_direct_abort(struct sk_buff *skb, enum rxrpc_abort_reason why,
 89			s32 abort_code, int err)
 90{
 91	struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
 92
 93	trace_rxrpc_abort(0, why, sp->hdr.cid, sp->hdr.callNumber, sp->hdr.seq,
 94			  abort_code, err);
 95	skb->mark = RXRPC_SKB_MARK_REJECT_ABORT;
 96	skb->priority = abort_code;
 97	return false;
 98}
 99
100/*
101 * Directly produce a connection abort from a packet.
102 */
103bool rxrpc_direct_conn_abort(struct sk_buff *skb, enum rxrpc_abort_reason why,
104			     s32 abort_code, int err)
105{
106	struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
107
108	trace_rxrpc_abort(0, why, sp->hdr.cid, 0, sp->hdr.seq, abort_code, err);
109	skb->mark = RXRPC_SKB_MARK_REJECT_CONN_ABORT;
110	skb->priority = abort_code;
111	return false;
112}
113
114static bool rxrpc_bad_message(struct sk_buff *skb, enum rxrpc_abort_reason why)
115{
116	return rxrpc_direct_abort(skb, why, RX_PROTOCOL_ERROR, -EBADMSG);
117}
118
119#define just_discard true
120
121/*
122 * Process event packets targeted at a local endpoint.
123 */
124static bool rxrpc_input_version(struct rxrpc_local *local, struct sk_buff *skb)
125{
126	struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
127	char v;
128
129	_enter("");
130
131	rxrpc_see_skb(skb, rxrpc_skb_see_version);
132	if (skb_copy_bits(skb, sizeof(struct rxrpc_wire_header), &v, 1) >= 0) {
133		if (v == 0)
134			rxrpc_send_version_request(local, &sp->hdr, skb);
135	}
136
137	return true;
138}
139
140/*
141 * Extract the wire header from a packet and translate the byte order.
142 */
143static bool rxrpc_extract_header(struct rxrpc_skb_priv *sp,
144				 struct sk_buff *skb)
145{
146	struct rxrpc_wire_header whdr;
147	struct rxrpc_ackpacket ack;
148
149	/* dig out the RxRPC connection details */
150	if (skb_copy_bits(skb, 0, &whdr, sizeof(whdr)) < 0)
151		return rxrpc_bad_message(skb, rxrpc_badmsg_short_hdr);
152
153	memset(sp, 0, sizeof(*sp));
154	sp->hdr.epoch		= ntohl(whdr.epoch);
155	sp->hdr.cid		= ntohl(whdr.cid);
156	sp->hdr.callNumber	= ntohl(whdr.callNumber);
157	sp->hdr.seq		= ntohl(whdr.seq);
158	sp->hdr.serial		= ntohl(whdr.serial);
159	sp->hdr.flags		= whdr.flags;
160	sp->hdr.type		= whdr.type;
161	sp->hdr.userStatus	= whdr.userStatus;
162	sp->hdr.securityIndex	= whdr.securityIndex;
163	sp->hdr._rsvd		= ntohs(whdr._rsvd);
164	sp->hdr.serviceId	= ntohs(whdr.serviceId);
165
166	if (sp->hdr.type == RXRPC_PACKET_TYPE_ACK) {
167		if (skb_copy_bits(skb, sizeof(whdr), &ack, sizeof(ack)) < 0)
168			return rxrpc_bad_message(skb, rxrpc_badmsg_short_ack);
169		sp->ack.first_ack	= ntohl(ack.firstPacket);
170		sp->ack.prev_ack	= ntohl(ack.previousPacket);
171		sp->ack.acked_serial	= ntohl(ack.serial);
172		sp->ack.reason		= ack.reason;
173		sp->ack.nr_acks		= ack.nAcks;
174	}
175	return true;
176}
177
178/*
179 * Extract the abort code from an ABORT packet and stash it in skb->priority.
180 */
181static bool rxrpc_extract_abort(struct sk_buff *skb)
182{
183	__be32 wtmp;
184
185	if (skb_copy_bits(skb, sizeof(struct rxrpc_wire_header),
186			  &wtmp, sizeof(wtmp)) < 0)
187		return false;
188	skb->priority = ntohl(wtmp);
189	return true;
190}
191
192/*
193 * Process packets received on the local endpoint
194 */
195static bool rxrpc_input_packet(struct rxrpc_local *local, struct sk_buff **_skb)
196{
197	struct rxrpc_connection *conn;
198	struct sockaddr_rxrpc peer_srx;
199	struct rxrpc_skb_priv *sp;
200	struct rxrpc_peer *peer = NULL;
201	struct sk_buff *skb = *_skb;
202	bool ret = false;
203
204	skb_pull(skb, sizeof(struct udphdr));
205
206	sp = rxrpc_skb(skb);
207
208	/* dig out the RxRPC connection details */
209	if (!rxrpc_extract_header(sp, skb))
210		return just_discard;
211
212	if (IS_ENABLED(CONFIG_AF_RXRPC_INJECT_LOSS)) {
213		static int lose;
214		if ((lose++ & 7) == 7) {
215			trace_rxrpc_rx_lose(sp);
216			return just_discard;
217		}
218	}
219
220	trace_rxrpc_rx_packet(sp);
221
222	switch (sp->hdr.type) {
223	case RXRPC_PACKET_TYPE_VERSION:
224		if (rxrpc_to_client(sp))
225			return just_discard;
226		return rxrpc_input_version(local, skb);
227
228	case RXRPC_PACKET_TYPE_BUSY:
229		if (rxrpc_to_server(sp))
230			return just_discard;
231		fallthrough;
232	case RXRPC_PACKET_TYPE_ACK:
233	case RXRPC_PACKET_TYPE_ACKALL:
234		if (sp->hdr.callNumber == 0)
235			return rxrpc_bad_message(skb, rxrpc_badmsg_zero_call);
236		break;
237	case RXRPC_PACKET_TYPE_ABORT:
238		if (!rxrpc_extract_abort(skb))
239			return just_discard; /* Just discard if malformed */
240		break;
241
242	case RXRPC_PACKET_TYPE_DATA:
243		if (sp->hdr.callNumber == 0)
244			return rxrpc_bad_message(skb, rxrpc_badmsg_zero_call);
245		if (sp->hdr.seq == 0)
246			return rxrpc_bad_message(skb, rxrpc_badmsg_zero_seq);
247
248		/* Unshare the packet so that it can be modified for in-place
249		 * decryption.
250		 */
251		if (sp->hdr.securityIndex != 0) {
252			skb = skb_unshare(skb, GFP_ATOMIC);
253			if (!skb) {
254				rxrpc_eaten_skb(*_skb, rxrpc_skb_eaten_by_unshare_nomem);
255				*_skb = NULL;
256				return just_discard;
257			}
258
259			if (skb != *_skb) {
260				rxrpc_eaten_skb(*_skb, rxrpc_skb_eaten_by_unshare);
261				*_skb = skb;
262				rxrpc_new_skb(skb, rxrpc_skb_new_unshared);
263				sp = rxrpc_skb(skb);
264			}
265		}
266		break;
267
268	case RXRPC_PACKET_TYPE_CHALLENGE:
269		if (rxrpc_to_server(sp))
270			return just_discard;
271		break;
272	case RXRPC_PACKET_TYPE_RESPONSE:
273		if (rxrpc_to_client(sp))
274			return just_discard;
275		break;
276
277		/* Packet types 9-11 should just be ignored. */
278	case RXRPC_PACKET_TYPE_PARAMS:
279	case RXRPC_PACKET_TYPE_10:
280	case RXRPC_PACKET_TYPE_11:
281		return just_discard;
282
283	default:
284		return rxrpc_bad_message(skb, rxrpc_badmsg_unsupported_packet);
285	}
286
287	if (sp->hdr.serviceId == 0)
288		return rxrpc_bad_message(skb, rxrpc_badmsg_zero_service);
289
290	if (WARN_ON_ONCE(rxrpc_extract_addr_from_skb(&peer_srx, skb) < 0))
291		return just_discard; /* Unsupported address type. */
292
293	if (peer_srx.transport.family != local->srx.transport.family &&
294	    (peer_srx.transport.family == AF_INET &&
295	     local->srx.transport.family != AF_INET6)) {
296		pr_warn_ratelimited("AF_RXRPC: Protocol mismatch %u not %u\n",
297				    peer_srx.transport.family,
298				    local->srx.transport.family);
299		return just_discard; /* Wrong address type. */
300	}
301
302	if (rxrpc_to_client(sp)) {
303		rcu_read_lock();
304		conn = rxrpc_find_client_connection_rcu(local, &peer_srx, skb);
305		conn = rxrpc_get_connection_maybe(conn, rxrpc_conn_get_call_input);
306		rcu_read_unlock();
307		if (!conn)
308			return rxrpc_protocol_error(skb, rxrpc_eproto_no_client_conn);
309
310		ret = rxrpc_input_packet_on_conn(conn, &peer_srx, skb);
311		rxrpc_put_connection(conn, rxrpc_conn_put_call_input);
312		return ret;
313	}
314
315	/* We need to look up service connections by the full protocol
316	 * parameter set.  We look up the peer first as an intermediate step
317	 * and then the connection from the peer's tree.
318	 */
319	rcu_read_lock();
320
321	peer = rxrpc_lookup_peer_rcu(local, &peer_srx);
322	if (!peer) {
323		rcu_read_unlock();
324		return rxrpc_new_incoming_call(local, NULL, NULL, &peer_srx, skb);
325	}
326
327	conn = rxrpc_find_service_conn_rcu(peer, skb);
328	conn = rxrpc_get_connection_maybe(conn, rxrpc_conn_get_call_input);
329	if (conn) {
330		rcu_read_unlock();
331		ret = rxrpc_input_packet_on_conn(conn, &peer_srx, skb);
332		rxrpc_put_connection(conn, rxrpc_conn_put_call_input);
333		return ret;
334	}
335
336	peer = rxrpc_get_peer_maybe(peer, rxrpc_peer_get_input);
337	rcu_read_unlock();
338
339	ret = rxrpc_new_incoming_call(local, peer, NULL, &peer_srx, skb);
340	rxrpc_put_peer(peer, rxrpc_peer_put_input);
341	return ret;
342}
343
344/*
345 * Deal with a packet that's associated with an extant connection.
346 */
347static int rxrpc_input_packet_on_conn(struct rxrpc_connection *conn,
348				      struct sockaddr_rxrpc *peer_srx,
349				      struct sk_buff *skb)
350{
351	struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
352	struct rxrpc_channel *chan;
353	struct rxrpc_call *call = NULL;
354	unsigned int channel;
355
356	if (sp->hdr.securityIndex != conn->security_ix)
357		return rxrpc_direct_abort(skb, rxrpc_eproto_wrong_security,
358					  RXKADINCONSISTENCY, -EBADMSG);
359
360	if (sp->hdr.serviceId != conn->service_id) {
361		int old_id;
362
363		if (!test_bit(RXRPC_CONN_PROBING_FOR_UPGRADE, &conn->flags))
364			return rxrpc_protocol_error(skb, rxrpc_eproto_reupgrade);
365
366		old_id = cmpxchg(&conn->service_id, conn->orig_service_id,
367				 sp->hdr.serviceId);
368		if (old_id != conn->orig_service_id &&
369		    old_id != sp->hdr.serviceId)
370			return rxrpc_protocol_error(skb, rxrpc_eproto_bad_upgrade);
371	}
372
373	if (after(sp->hdr.serial, conn->hi_serial))
374		conn->hi_serial = sp->hdr.serial;
375
376	/* It's a connection-level packet if the call number is 0. */
377	if (sp->hdr.callNumber == 0)
378		return rxrpc_input_conn_packet(conn, skb);
379
380	/* Deal with path MTU discovery probing. */
381	if (sp->hdr.type == RXRPC_PACKET_TYPE_ACK &&
382	    conn->pmtud_probe &&
383	    after_eq(sp->ack.acked_serial, conn->pmtud_probe))
384		rxrpc_input_probe_for_pmtud(conn, sp->ack.acked_serial, false);
385
386	/* Call-bound packets are routed by connection channel. */
387	channel = sp->hdr.cid & RXRPC_CHANNELMASK;
388	chan = &conn->channels[channel];
389
390	/* Ignore really old calls */
391	if (sp->hdr.callNumber < chan->last_call)
392		return just_discard;
393
394	if (sp->hdr.callNumber == chan->last_call) {
395		if (chan->call ||
396		    sp->hdr.type == RXRPC_PACKET_TYPE_ABORT)
397			return just_discard;
398
399		/* For the previous service call, if completed successfully, we
400		 * discard all further packets.
401		 */
402		if (rxrpc_conn_is_service(conn) &&
403		    chan->last_type == RXRPC_PACKET_TYPE_ACK)
404			return just_discard;
405
406		/* But otherwise we need to retransmit the final packet from
407		 * data cached in the connection record.
408		 */
409		if (sp->hdr.type == RXRPC_PACKET_TYPE_DATA)
410			trace_rxrpc_rx_data(chan->call_debug_id,
411					    sp->hdr.seq,
412					    sp->hdr.serial,
413					    sp->hdr.flags);
414		rxrpc_conn_retransmit_call(conn, skb, channel);
415		return just_discard;
416	}
417
418	call = rxrpc_try_get_call(chan->call, rxrpc_call_get_input);
419
420	if (sp->hdr.callNumber > chan->call_id) {
421		if (rxrpc_to_client(sp)) {
422			if (call)
423				rxrpc_put_call(call, rxrpc_call_put_input);
424			return rxrpc_protocol_error(skb,
425						    rxrpc_eproto_unexpected_implicit_end);
426		}
427
428		if (call) {
429			rxrpc_implicit_end_call(call, skb);
430			rxrpc_put_call(call, rxrpc_call_put_input);
431			call = NULL;
432		}
433	}
434
435	if (!call) {
436		if (rxrpc_to_client(sp))
437			return rxrpc_protocol_error(skb, rxrpc_eproto_no_client_call);
438		return rxrpc_new_incoming_call(conn->local, conn->peer, conn,
439					       peer_srx, skb);
440	}
441
442	rxrpc_queue_rx_call_packet(call, skb);
443	rxrpc_put_call(call, rxrpc_call_put_input);
444	return true;
445}
446
447/*
448 * I/O and event handling thread.
449 */
450int rxrpc_io_thread(void *data)
451{
452	struct rxrpc_connection *conn;
453	struct sk_buff_head rx_queue;
454	struct rxrpc_local *local = data;
455	struct rxrpc_call *call;
456	struct sk_buff *skb;
457#ifdef CONFIG_AF_RXRPC_INJECT_RX_DELAY
458	ktime_t now;
459#endif
460	bool should_stop;
461	LIST_HEAD(conn_attend_q);
462	LIST_HEAD(call_attend_q);
463
464	complete(&local->io_thread_ready);
465
466	skb_queue_head_init(&rx_queue);
467
468	set_user_nice(current, MIN_NICE);
469
470	for (;;) {
471		rxrpc_inc_stat(local->rxnet, stat_io_loop);
472
473		/* Inject a delay into packets if requested. */
474#ifdef CONFIG_AF_RXRPC_INJECT_RX_DELAY
475		now = ktime_get_real();
476		while ((skb = skb_peek(&local->rx_delay_queue))) {
477			if (ktime_before(now, skb->tstamp))
478				break;
479			skb = skb_dequeue(&local->rx_delay_queue);
480			skb_queue_tail(&local->rx_queue, skb);
481		}
482#endif
483
484		if (!skb_queue_empty(&local->rx_queue)) {
485			spin_lock_irq(&local->rx_queue.lock);
486			skb_queue_splice_tail_init(&local->rx_queue, &rx_queue);
487			spin_unlock_irq(&local->rx_queue.lock);
488			trace_rxrpc_iothread_rx(local, skb_queue_len(&rx_queue));
489		}
490
491		/* Distribute packets and errors. */
492		while ((skb = __skb_dequeue(&rx_queue))) {
493			struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
494			switch (skb->mark) {
495			case RXRPC_SKB_MARK_PACKET:
496				skb->priority = 0;
497				if (!rxrpc_input_packet(local, &skb))
498					rxrpc_reject_packet(local, skb);
499				trace_rxrpc_rx_done(skb->mark, skb->priority);
500				rxrpc_free_skb(skb, rxrpc_skb_put_input);
501				break;
502			case RXRPC_SKB_MARK_ERROR:
503				rxrpc_input_error(local, skb);
504				rxrpc_free_skb(skb, rxrpc_skb_put_error_report);
505				break;
506			case RXRPC_SKB_MARK_SERVICE_CONN_SECURED:
507				rxrpc_input_conn_event(sp->poke_conn, skb);
508				rxrpc_put_connection(sp->poke_conn, rxrpc_conn_put_poke);
509				rxrpc_free_skb(skb, rxrpc_skb_put_conn_secured);
510				break;
511			default:
512				WARN_ON_ONCE(1);
513				rxrpc_free_skb(skb, rxrpc_skb_put_unknown);
514				break;
515			}
516		}
517
518		/* Deal with connections that want immediate attention. */
519		if (!list_empty_careful(&local->conn_attend_q)) {
520			spin_lock_irq(&local->lock);
521			list_splice_tail_init(&local->conn_attend_q, &conn_attend_q);
522			spin_unlock_irq(&local->lock);
523		}
524
525		while ((conn = list_first_entry_or_null(&conn_attend_q,
526							struct rxrpc_connection,
527							attend_link))) {
528			spin_lock_irq(&local->lock);
529			list_del_init(&conn->attend_link);
530			spin_unlock_irq(&local->lock);
531			rxrpc_input_conn_event(conn, NULL);
532			rxrpc_put_connection(conn, rxrpc_conn_put_poke);
533		}
534
535		if (test_and_clear_bit(RXRPC_CLIENT_CONN_REAP_TIMER,
536				       &local->client_conn_flags))
537			rxrpc_discard_expired_client_conns(local);
538
539		/* Deal with calls that want immediate attention. */
540		spin_lock_irq(&local->lock);
541		list_splice_tail_init(&local->call_attend_q, &call_attend_q);
542		spin_unlock_irq(&local->lock);
543
544		while ((call = list_first_entry_or_null(&call_attend_q,
545							struct rxrpc_call,
546							attend_link))) {
547			spin_lock_irq(&local->lock);
548			list_del_init(&call->attend_link);
549			spin_unlock_irq(&local->lock);
550			trace_rxrpc_call_poked(call);
551			rxrpc_input_call_event(call);
552			rxrpc_put_call(call, rxrpc_call_put_poke);
553		}
554
555		if (!list_empty(&local->new_client_calls))
556			rxrpc_connect_client_calls(local);
557
558		set_current_state(TASK_INTERRUPTIBLE);
559		should_stop = kthread_should_stop();
560		if (!skb_queue_empty(&local->rx_queue) ||
561		    !list_empty(&local->call_attend_q) ||
562		    !list_empty(&local->conn_attend_q) ||
563		    !list_empty(&local->new_client_calls) ||
564		    test_bit(RXRPC_CLIENT_CONN_REAP_TIMER,
565			     &local->client_conn_flags)) {
566			__set_current_state(TASK_RUNNING);
567			continue;
568		}
569
570		if (should_stop)
571			break;
572
573#ifdef CONFIG_AF_RXRPC_INJECT_RX_DELAY
574		skb = skb_peek(&local->rx_delay_queue);
575		if (skb) {
576			unsigned long timeout;
577			ktime_t tstamp = skb->tstamp;
578			ktime_t now = ktime_get_real();
579			s64 delay_ns = ktime_to_ns(ktime_sub(tstamp, now));
580
581			if (delay_ns <= 0) {
582				__set_current_state(TASK_RUNNING);
583				continue;
584			}
585
586			timeout = nsecs_to_jiffies(delay_ns);
587			timeout = umax(timeout, 1);
588			schedule_timeout(timeout);
589			__set_current_state(TASK_RUNNING);
590			continue;
591		}
592#endif
593
594		schedule();
595	}
596
597	__set_current_state(TASK_RUNNING);
598	rxrpc_see_local(local, rxrpc_local_stop);
599	rxrpc_destroy_local(local);
600	WRITE_ONCE(local->io_thread, NULL);
601	rxrpc_see_local(local, rxrpc_local_stopped);
602	return 0;
603}