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

source code of linux/net/rxrpc/io_thread.c