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