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 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 */ 57 void 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 */ 82 bool 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 94 static 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 */ 104 static 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 */ 123 static 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, 0, &whdr, sizeof(whdr)) < 0) 131 return rxrpc_bad_message(skb, 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, sizeof(whdr), &ack, sizeof(ack)) < 0) 148 return rxrpc_bad_message(skb, 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 */ 161 static bool rxrpc_extract_abort(struct sk_buff *skb) 162 { 163 __be32 wtmp; 164 165 if (skb_copy_bits(skb, sizeof(struct rxrpc_wire_header), 166 &wtmp, 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 */ 175 static 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, 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, 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, rxrpc_badmsg_zero_call); 225 if (sp->hdr.seq == 0) 226 return rxrpc_bad_message(skb, 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, rxrpc_badmsg_unsupported_packet); 265 } 266 267 if (sp->hdr.serviceId == 0) 268 return rxrpc_bad_message(skb, 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, rxrpc_eproto_no_client_conn); 289 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 /* 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, 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, 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, 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 */ 327 static 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, rxrpc_eproto_wrong_security, 339 RXKADINCONSISTENCY, -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, 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, rxrpc_eproto_bad_upgrade); 352 } 353 354 if (after(sp->hdr.serial, 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(chan->call_debug_id, 386 sp->hdr.seq, 387 sp->hdr.serial, 388 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 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, rxrpc_eproto_no_client_call); 412 return rxrpc_new_incoming_call(conn->local, 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 */ 424 int 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(&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(&local->lock); 451 list_del_init(&conn->attend_link); 452 spin_unlock_bh(&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 &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(&local->lock); 468 list_del_init(&call->attend_link); 469 spin_unlock_bh(&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(&local->new_client_calls)) 478 rxrpc_connect_client_calls(local); 479 480 /* Process received packets and errors. */ 481 if ((skb = __skb_dequeue(&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)) 487 rxrpc_reject_packet(local, skb); 488 trace_rxrpc_rx_done(skb->mark, 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(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(&local->rx_delay_queue))) { 512 if (ktime_before(now, skb->tstamp)) 513 break; 514 skb = skb_dequeue(&local->rx_delay_queue); 515 skb_queue_tail(&local->rx_queue, skb); 516 } 517 #endif 518 519 if (!skb_queue_empty(&local->rx_queue)) { 520 spin_lock_irq(&local->rx_queue.lock); 521 skb_queue_splice_tail_init(&local->rx_queue, &rx_queue); 522 spin_unlock_irq(&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(&local->rx_queue) || 529 !list_empty(&local->call_attend_q) || 530 !list_empty(&local->conn_attend_q) || 531 !list_empty(&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(&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(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, rxrpc_local_stop); 567 rxrpc_destroy_local(local); 568 local->io_thread = NULL; 569 rxrpc_see_local(local, rxrpc_local_stopped); 570 return 0; 571 } 572