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