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 12 static 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 */ 26 int 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 */ 63 void 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 */ 88 bool 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 */ 103 bool 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 114 static 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 */ 124 static 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 */ 143 static 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 */ 181 static 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 */ 195 static 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 */ 347 static 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 rxrpc_put_call(call, rxrpc_call_put_input); 423 return rxrpc_protocol_error(skb, 424 rxrpc_eproto_unexpected_implicit_end); 425 } 426 427 if (call) { 428 rxrpc_implicit_end_call(call, skb); 429 rxrpc_put_call(call, rxrpc_call_put_input); 430 call = NULL; 431 } 432 } 433 434 if (!call) { 435 if (rxrpc_to_client(sp)) 436 return rxrpc_protocol_error(skb, rxrpc_eproto_no_client_call); 437 return rxrpc_new_incoming_call(conn->local, conn->peer, conn, 438 peer_srx, skb); 439 } 440 441 rxrpc_queue_rx_call_packet(call, skb); 442 rxrpc_put_call(call, rxrpc_call_put_input); 443 return true; 444 } 445 446 /* 447 * I/O and event handling thread. 448 */ 449 int rxrpc_io_thread(void *data) 450 { 451 struct rxrpc_connection *conn; 452 struct sk_buff_head rx_queue; 453 struct rxrpc_local *local = data; 454 struct rxrpc_call *call; 455 struct sk_buff *skb; 456 #ifdef CONFIG_AF_RXRPC_INJECT_RX_DELAY 457 ktime_t now; 458 #endif 459 bool should_stop; 460 LIST_HEAD(conn_attend_q); 461 LIST_HEAD(call_attend_q); 462 463 complete(&local->io_thread_ready); 464 465 skb_queue_head_init(&rx_queue); 466 467 set_user_nice(current, MIN_NICE); 468 469 for (;;) { 470 rxrpc_inc_stat(local->rxnet, stat_io_loop); 471 472 /* Inject a delay into packets if requested. */ 473 #ifdef CONFIG_AF_RXRPC_INJECT_RX_DELAY 474 now = ktime_get_real(); 475 while ((skb = skb_peek(&local->rx_delay_queue))) { 476 if (ktime_before(now, skb->tstamp)) 477 break; 478 skb = skb_dequeue(&local->rx_delay_queue); 479 skb_queue_tail(&local->rx_queue, skb); 480 } 481 #endif 482 483 if (!skb_queue_empty(&local->rx_queue)) { 484 spin_lock_irq(&local->rx_queue.lock); 485 skb_queue_splice_tail_init(&local->rx_queue, &rx_queue); 486 spin_unlock_irq(&local->rx_queue.lock); 487 trace_rxrpc_iothread_rx(local, skb_queue_len(&rx_queue)); 488 } 489 490 /* Distribute packets and errors. */ 491 while ((skb = __skb_dequeue(&rx_queue))) { 492 struct rxrpc_skb_priv *sp = rxrpc_skb(skb); 493 switch (skb->mark) { 494 case RXRPC_SKB_MARK_PACKET: 495 skb->priority = 0; 496 if (!rxrpc_input_packet(local, &skb)) 497 rxrpc_reject_packet(local, skb); 498 trace_rxrpc_rx_done(skb->mark, skb->priority); 499 rxrpc_free_skb(skb, rxrpc_skb_put_input); 500 break; 501 case RXRPC_SKB_MARK_ERROR: 502 rxrpc_input_error(local, skb); 503 rxrpc_free_skb(skb, rxrpc_skb_put_error_report); 504 break; 505 case RXRPC_SKB_MARK_SERVICE_CONN_SECURED: 506 rxrpc_input_conn_event(sp->poke_conn, skb); 507 rxrpc_put_connection(sp->poke_conn, rxrpc_conn_put_poke); 508 rxrpc_free_skb(skb, rxrpc_skb_put_conn_secured); 509 break; 510 default: 511 WARN_ON_ONCE(1); 512 rxrpc_free_skb(skb, rxrpc_skb_put_unknown); 513 break; 514 } 515 } 516 517 /* Deal with connections that want immediate attention. */ 518 if (!list_empty_careful(&local->conn_attend_q)) { 519 spin_lock_irq(&local->lock); 520 list_splice_tail_init(&local->conn_attend_q, &conn_attend_q); 521 spin_unlock_irq(&local->lock); 522 } 523 524 while ((conn = list_first_entry_or_null(&conn_attend_q, 525 struct rxrpc_connection, 526 attend_link))) { 527 spin_lock_irq(&local->lock); 528 list_del_init(&conn->attend_link); 529 spin_unlock_irq(&local->lock); 530 rxrpc_input_conn_event(conn, NULL); 531 rxrpc_put_connection(conn, rxrpc_conn_put_poke); 532 } 533 534 if (test_and_clear_bit(RXRPC_CLIENT_CONN_REAP_TIMER, 535 &local->client_conn_flags)) 536 rxrpc_discard_expired_client_conns(local); 537 538 /* Deal with calls that want immediate attention. */ 539 spin_lock_irq(&local->lock); 540 list_splice_tail_init(&local->call_attend_q, &call_attend_q); 541 spin_unlock_irq(&local->lock); 542 543 while ((call = list_first_entry_or_null(&call_attend_q, 544 struct rxrpc_call, 545 attend_link))) { 546 spin_lock_irq(&local->lock); 547 list_del_init(&call->attend_link); 548 spin_unlock_irq(&local->lock); 549 trace_rxrpc_call_poked(call); 550 rxrpc_input_call_event(call); 551 rxrpc_put_call(call, rxrpc_call_put_poke); 552 } 553 554 if (!list_empty(&local->new_client_calls)) 555 rxrpc_connect_client_calls(local); 556 557 set_current_state(TASK_INTERRUPTIBLE); 558 should_stop = kthread_should_stop(); 559 if (!skb_queue_empty(&local->rx_queue) || 560 !list_empty(&local->call_attend_q) || 561 !list_empty(&local->conn_attend_q) || 562 !list_empty(&local->new_client_calls) || 563 test_bit(RXRPC_CLIENT_CONN_REAP_TIMER, 564 &local->client_conn_flags)) { 565 __set_current_state(TASK_RUNNING); 566 continue; 567 } 568 569 if (should_stop) 570 break; 571 572 #ifdef CONFIG_AF_RXRPC_INJECT_RX_DELAY 573 skb = skb_peek(&local->rx_delay_queue); 574 if (skb) { 575 unsigned long timeout; 576 ktime_t tstamp = skb->tstamp; 577 ktime_t now = ktime_get_real(); 578 s64 delay_ns = ktime_to_ns(ktime_sub(tstamp, now)); 579 580 if (delay_ns <= 0) { 581 __set_current_state(TASK_RUNNING); 582 continue; 583 } 584 585 timeout = nsecs_to_jiffies(delay_ns); 586 timeout = umax(timeout, 1); 587 schedule_timeout(timeout); 588 __set_current_state(TASK_RUNNING); 589 continue; 590 } 591 #endif 592 593 schedule(); 594 } 595 596 __set_current_state(TASK_RUNNING); 597 rxrpc_see_local(local, rxrpc_local_stop); 598 rxrpc_destroy_local(local); 599 WRITE_ONCE(local->io_thread, NULL); 600 rxrpc_see_local(local, rxrpc_local_stopped); 601 return 0; 602 } 603