net/rxrpc/io_thread.c

   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 /*
  13  * handle data received on the local endpoint
  14  * - may be called in interrupt context
  15  *
  16  * [!] Note that as this is called from the encap_rcv hook, the socket is not
  17  * held locked by the caller and nothing prevents sk_user_data on the UDP from
  18  * being cleared in the middle of processing this function.
  19  *
  20  * Called with the RCU read lock held from the IP layer via UDP.
  21  */
  22 int rxrpc_encap_rcv(struct sock *udp_sk, struct sk_buff *skb)
  23 {
  24         struct rxrpc_local *local = rcu_dereference_sk_user_data(udp_sk);
  25
  26         if (unlikely(!local)) {
  27                 kfree_skb(skb);
  28                 return 0;
  29         }
  30         if (skb->tstamp == 0)
  31                 skb->tstamp = ktime_get_real();
  32
  33         skb->mark = RXRPC_SKB_MARK_PACKET;
  34         rxrpc_new_skb(skb, rxrpc_skb_new_encap_rcv);
  35         skb_queue_tail(&local->rx_queue, skb);
  36         rxrpc_wake_up_io_thread(local);
  37         return 0;
  38 }
  39
  40 /*
  41  * Handle an error received on the local endpoint.
  42  */
  43 void rxrpc_error_report(struct sock *sk)
  44 {
  45         struct rxrpc_local *local;
  46         struct sk_buff *skb;
  47
  48         rcu_read_lock();
  49         local = rcu_dereference_sk_user_data(sk);
  50         if (unlikely(!local)) {
  51                 rcu_read_unlock();
  52                 return;
  53         }
  54
  55         while ((skb = skb_dequeue(&sk->sk_error_queue))) {
  56                 skb->mark = RXRPC_SKB_MARK_ERROR;
  57                 rxrpc_new_skb(skb, rxrpc_skb_new_error_report);
  58                 skb_queue_tail(&local->rx_queue, skb);
  59         }
  60
  61         rxrpc_wake_up_io_thread(local);
  62         rcu_read_unlock();
  63 }
  64
  65 /*
  66  * post connection-level events to the connection
  67  * - this includes challenges, responses, some aborts and call terminal packet
  68  *   retransmission.
  69  */
  70 static void rxrpc_post_packet_to_conn(struct rxrpc_connection *conn,
  71                                       struct sk_buff *skb)
  72 {
  73         _enter("%p,%p", conn, skb);
  74
  75         rxrpc_get_skb(skb, rxrpc_skb_get_conn_work);
  76         skb_queue_tail(&conn->rx_queue, skb);
  77         rxrpc_queue_conn(conn, rxrpc_conn_queue_rx_work);
  78 }
  79
  80 /*
  81  * post endpoint-level events to the local endpoint
  82  * - this includes debug and version messages
  83  */
  84 static void rxrpc_post_packet_to_local(struct rxrpc_local *local,
  85                                        struct sk_buff *skb)
  86 {
  87         _enter("%p,%p", local, skb);
  88
  89         if (rxrpc_get_local_maybe(local, rxrpc_local_get_queue)) {
  90                 rxrpc_get_skb(skb, rxrpc_skb_get_local_work);
  91                 skb_queue_tail(&local->event_queue, skb);
  92                 rxrpc_queue_local(local);
  93         }
  94 }
  95
  96 /*
  97  * put a packet up for transport-level abort
  98  */
  99 static void rxrpc_reject_packet(struct rxrpc_local *local, struct sk_buff *skb)
 100 {
 101         if (rxrpc_get_local_maybe(local, rxrpc_local_get_queue)) {
 102                 rxrpc_get_skb(skb, rxrpc_skb_get_reject_work);
 103                 skb_queue_tail(&local->reject_queue, skb);
 104                 rxrpc_queue_local(local);
 105         }
 106 }
 107
 108 /*
 109  * Extract the wire header from a packet and translate the byte order.
 110  */
 111 static noinline
 112 int rxrpc_extract_header(struct rxrpc_skb_priv *sp, struct sk_buff *skb)
 113 {
 114         struct rxrpc_wire_header whdr;
 115
 116         /* dig out the RxRPC connection details */
 117         if (skb_copy_bits(skb, 0, &whdr, sizeof(whdr)) < 0) {
 118                 trace_rxrpc_rx_eproto(NULL, sp->hdr.serial,
 119                                       tracepoint_string("bad_hdr"));
 120                 return -EBADMSG;
 121         }
 122
 123         memset(sp, 0, sizeof(*sp));
 124         sp->hdr.epoch           = ntohl(whdr.epoch);
 125         sp->hdr.cid             = ntohl(whdr.cid);
 126         sp->hdr.callNumber      = ntohl(whdr.callNumber);
 127         sp->hdr.seq             = ntohl(whdr.seq);
 128         sp->hdr.serial          = ntohl(whdr.serial);
 129         sp->hdr.flags           = whdr.flags;
 130         sp->hdr.type            = whdr.type;
 131         sp->hdr.userStatus      = whdr.userStatus;
 132         sp->hdr.securityIndex   = whdr.securityIndex;
 133         sp->hdr._rsvd           = ntohs(whdr._rsvd);
 134         sp->hdr.serviceId       = ntohs(whdr.serviceId);
 135         return 0;
 136 }
 137
 138 /*
 139  * Extract the abort code from an ABORT packet and stash it in skb->priority.
 140  */
 141 static bool rxrpc_extract_abort(struct sk_buff *skb)
 142 {
 143         __be32 wtmp;
 144
 145         if (skb_copy_bits(skb, sizeof(struct rxrpc_wire_header),
 146                           &wtmp, sizeof(wtmp)) < 0)
 147                 return false;
 148         skb->priority = ntohl(wtmp);
 149         return true;
 150 }
 151
 152 /*
 153  * Process packets received on the local endpoint
 154  */
 155 static int rxrpc_input_packet(struct rxrpc_local *local, struct sk_buff **_skb)
 156 {
 157         struct rxrpc_connection *conn;
 158         struct sockaddr_rxrpc peer_srx;
 159         struct rxrpc_channel *chan;
 160         struct rxrpc_call *call = NULL;
 161         struct rxrpc_skb_priv *sp;
 162         struct rxrpc_peer *peer = NULL;
 163         struct rxrpc_sock *rx = NULL;
 164         struct sk_buff *skb = *_skb;
 165         unsigned int channel;
 166
 167         if (skb->tstamp == 0)
 168                 skb->tstamp = ktime_get_real();
 169
 170         skb_pull(skb, sizeof(struct udphdr));
 171
 172         /* The UDP protocol already released all skb resources;
 173          * we are free to add our own data there.
 174          */
 175         sp = rxrpc_skb(skb);
 176
 177         /* dig out the RxRPC connection details */
 178         if (rxrpc_extract_header(sp, skb) < 0)
 179                 goto bad_message;
 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 0;
 186                 }
 187         }
 188
 189         if (skb->tstamp == 0)
 190                 skb->tstamp = ktime_get_real();
 191         trace_rxrpc_rx_packet(sp);
 192
 193         switch (sp->hdr.type) {
 194         case RXRPC_PACKET_TYPE_VERSION:
 195                 if (rxrpc_to_client(sp))
 196                         return 0;
 197                 rxrpc_post_packet_to_local(local, skb);
 198                 return 0;
 199
 200         case RXRPC_PACKET_TYPE_BUSY:
 201                 if (rxrpc_to_server(sp))
 202                         return 0;
 203                 fallthrough;
 204         case RXRPC_PACKET_TYPE_ACK:
 205         case RXRPC_PACKET_TYPE_ACKALL:
 206                 if (sp->hdr.callNumber == 0)
 207                         goto bad_message;
 208                 break;
 209         case RXRPC_PACKET_TYPE_ABORT:
 210                 if (!rxrpc_extract_abort(skb))
 211                         return 0; /* Just discard if malformed */
 212                 break;
 213
 214         case RXRPC_PACKET_TYPE_DATA:
 215                 if (sp->hdr.callNumber == 0 ||
 216                     sp->hdr.seq == 0)
 217                         goto bad_message;
 218
 219                 /* Unshare the packet so that it can be modified for in-place
 220                  * decryption.
 221                  */
 222                 if (sp->hdr.securityIndex != 0) {
 223                         skb = skb_unshare(skb, GFP_ATOMIC);
 224                         if (!skb) {
 225                                 rxrpc_eaten_skb(*_skb, rxrpc_skb_eaten_by_unshare_nomem);
 226                                 *_skb = NULL;
 227                                 return 0;
 228                         }
 229
 230                         if (skb != *_skb) {
 231                                 rxrpc_eaten_skb(*_skb, rxrpc_skb_eaten_by_unshare);
 232                                 *_skb = skb;
 233                                 rxrpc_new_skb(skb, rxrpc_skb_new_unshared);
 234                                 sp = rxrpc_skb(skb);
 235                         }
 236                 }
 237                 break;
 238
 239         case RXRPC_PACKET_TYPE_CHALLENGE:
 240                 if (rxrpc_to_server(sp))
 241                         return 0;
 242                 break;
 243         case RXRPC_PACKET_TYPE_RESPONSE:
 244                 if (rxrpc_to_client(sp))
 245                         return 0;
 246                 break;
 247
 248                 /* Packet types 9-11 should just be ignored. */
 249         case RXRPC_PACKET_TYPE_PARAMS:
 250         case RXRPC_PACKET_TYPE_10:
 251         case RXRPC_PACKET_TYPE_11:
 252                 return 0;
 253
 254         default:
 255                 goto bad_message;
 256         }
 257
 258         if (sp->hdr.serviceId == 0)
 259                 goto bad_message;
 260
 261         if (WARN_ON_ONCE(rxrpc_extract_addr_from_skb(&peer_srx, skb) < 0))
 262                 return 0; /* Unsupported address type - discard. */
 263
 264         if (peer_srx.transport.family != local->srx.transport.family &&
 265             (peer_srx.transport.family == AF_INET &&
 266              local->srx.transport.family != AF_INET6)) {
 267                 pr_warn_ratelimited("AF_RXRPC: Protocol mismatch %u not %u\n",
 268                                     peer_srx.transport.family,
 269                                     local->srx.transport.family);
 270                 return 0; /* Wrong address type - discard. */
 271         }
 272
 273         rcu_read_lock();
 274
 275         if (rxrpc_to_server(sp)) {
 276                 /* Weed out packets to services we're not offering.  Packets
 277                  * that would begin a call are explicitly rejected and the rest
 278                  * are just discarded.
 279                  */
 280                 rx = rcu_dereference(local->service);
 281                 if (!rx || (sp->hdr.serviceId != rx->srx.srx_service &&
 282                             sp->hdr.serviceId != rx->second_service)
 283                     ) {
 284                         rcu_read_unlock();
 285                         if (sp->hdr.type == RXRPC_PACKET_TYPE_DATA &&
 286                             sp->hdr.seq == 1)
 287                                 goto unsupported_service;
 288                         return 0;
 289                 }
 290         }
 291
 292         conn = rxrpc_find_connection_rcu(local, &peer_srx, skb, &peer);
 293         if (conn) {
 294                 if (sp->hdr.securityIndex != conn->security_ix)
 295                         goto wrong_security;
 296
 297                 if (sp->hdr.serviceId != conn->service_id) {
 298                         int old_id;
 299
 300                         if (!test_bit(RXRPC_CONN_PROBING_FOR_UPGRADE, &conn->flags))
 301                                 goto reupgrade;
 302                         old_id = cmpxchg(&conn->service_id, conn->orig_service_id,
 303                                          sp->hdr.serviceId);
 304
 305                         if (old_id != conn->orig_service_id &&
 306                             old_id != sp->hdr.serviceId)
 307                                 goto reupgrade;
 308                 }
 309
 310                 if (sp->hdr.callNumber == 0) {
 311                         /* Connection-level packet */
 312                         _debug("CONN %p {%d}", conn, conn->debug_id);
 313                         conn = rxrpc_get_connection_maybe(conn, rxrpc_conn_get_conn_input);
 314                         rcu_read_unlock();
 315                         if (conn) {
 316                                 rxrpc_post_packet_to_conn(conn, skb);
 317                                 rxrpc_put_connection(conn, rxrpc_conn_put_conn_input);
 318                         }
 319                         return 0;
 320                 }
 321
 322                 if ((int)sp->hdr.serial - (int)conn->hi_serial > 0)
 323                         conn->hi_serial = sp->hdr.serial;
 324
 325                 /* Call-bound packets are routed by connection channel. */
 326                 channel = sp->hdr.cid & RXRPC_CHANNELMASK;
 327                 chan = &conn->channels[channel];
 328
 329                 /* Ignore really old calls */
 330                 if (sp->hdr.callNumber < chan->last_call) {
 331                         rcu_read_unlock();
 332                         return 0;
 333                 }
 334
 335                 if (sp->hdr.callNumber == chan->last_call) {
 336                         if (chan->call ||
 337                             sp->hdr.type == RXRPC_PACKET_TYPE_ABORT) {
 338                                 rcu_read_unlock();
 339                                 return 0;
 340                         }
 341
 342                         /* For the previous service call, if completed
 343                          * successfully, we discard all further packets.
 344                          */
 345                         if (rxrpc_conn_is_service(conn) &&
 346                             chan->last_type == RXRPC_PACKET_TYPE_ACK) {
 347                                 rcu_read_unlock();
 348                                 return 0;
 349                         }
 350
 351                         /* But otherwise we need to retransmit the final packet
 352                          * from data cached in the connection record.
 353                          */
 354                         if (sp->hdr.type == RXRPC_PACKET_TYPE_DATA)
 355                                 trace_rxrpc_rx_data(chan->call_debug_id,
 356                                                     sp->hdr.seq,
 357                                                     sp->hdr.serial,
 358                                                     sp->hdr.flags);
 359                         conn = rxrpc_get_connection_maybe(conn, rxrpc_conn_get_call_input);
 360                         rcu_read_unlock();
 361                         if (conn) {
 362                                 rxrpc_post_packet_to_conn(conn, skb);
 363                                 rxrpc_put_connection(conn, rxrpc_conn_put_call_input);
 364                         }
 365                         return 0;
 366                 }
 367
 368                 call = rcu_dereference(chan->call);
 369
 370                 if (sp->hdr.callNumber > chan->call_id) {
 371                         if (rxrpc_to_client(sp)) {
 372                                 rcu_read_unlock();
 373                                 goto reject_packet;
 374                         }
 375                         if (call) {
 376                                 rxrpc_input_implicit_end_call(conn, call);
 377                                 chan->call = NULL;
 378                                 call = NULL;
 379                         }
 380                 }
 381
 382                 if (call && !rxrpc_try_get_call(call, rxrpc_call_get_input))
 383                         call = NULL;
 384
 385                 if (call) {
 386                         if (sp->hdr.serviceId != call->dest_srx.srx_service)
 387                                 call->dest_srx.srx_service = sp->hdr.serviceId;
 388                         if ((int)sp->hdr.serial - (int)call->rx_serial > 0)
 389                                 call->rx_serial = sp->hdr.serial;
 390                         if (!test_bit(RXRPC_CALL_RX_HEARD, &call->flags))
 391                                 set_bit(RXRPC_CALL_RX_HEARD, &call->flags);
 392                 }
 393         }
 394
 395         if (!call) {
 396                 if (rxrpc_to_client(sp) ||
 397                     sp->hdr.type != RXRPC_PACKET_TYPE_DATA) {
 398                         rcu_read_unlock();
 399                         goto bad_message;
 400                 }
 401                 if (sp->hdr.seq != 1) {
 402                         rcu_read_unlock();
 403                         return 0;
 404                 }
 405                 call = rxrpc_new_incoming_call(local, rx, &peer_srx, skb);
 406                 if (!call) {
 407                         rcu_read_unlock();
 408                         goto reject_packet;
 409                 }
 410         }
 411
 412         rcu_read_unlock();
 413
 414         /* Process a call packet. */
 415         rxrpc_input_call_event(call, skb);
 416         rxrpc_put_call(call, rxrpc_call_put_input);
 417         trace_rxrpc_rx_done(0, 0);
 418         return 0;
 419
 420 wrong_security:
 421         rcu_read_unlock();
 422         trace_rxrpc_abort(0, "SEC", sp->hdr.cid, sp->hdr.callNumber, sp->hdr.seq,
 423                           RXKADINCONSISTENCY, EBADMSG);
 424         skb->priority = RXKADINCONSISTENCY;
 425         goto post_abort;
 426
 427 unsupported_service:
 428         trace_rxrpc_abort(0, "INV", sp->hdr.cid, sp->hdr.callNumber, sp->hdr.seq,
 429                           RX_INVALID_OPERATION, EOPNOTSUPP);
 430         skb->priority = RX_INVALID_OPERATION;
 431         goto post_abort;
 432
 433 reupgrade:
 434         rcu_read_unlock();
 435         trace_rxrpc_abort(0, "UPG", sp->hdr.cid, sp->hdr.callNumber, sp->hdr.seq,
 436                           RX_PROTOCOL_ERROR, EBADMSG);
 437         goto protocol_error;
 438
 439 bad_message:
 440         trace_rxrpc_abort(0, "BAD", sp->hdr.cid, sp->hdr.callNumber, sp->hdr.seq,
 441                           RX_PROTOCOL_ERROR, EBADMSG);
 442 protocol_error:
 443         skb->priority = RX_PROTOCOL_ERROR;
 444 post_abort:
 445         skb->mark = RXRPC_SKB_MARK_REJECT_ABORT;
 446 reject_packet:
 447         rxrpc_reject_packet(local, skb);
 448         return 0;
 449 }
 450
 451 /*
 452  * I/O and event handling thread.
 453  */
 454 int rxrpc_io_thread(void *data)
 455 {
 456         struct sk_buff_head rx_queue;
 457         struct rxrpc_local *local = data;
 458         struct rxrpc_call *call;
 459         struct sk_buff *skb;
 460
 461         skb_queue_head_init(&rx_queue);
 462
 463         set_user_nice(current, MIN_NICE);
 464
 465         for (;;) {
 466                 rxrpc_inc_stat(local->rxnet, stat_io_loop);
 467
 468                 /* Deal with calls that want immediate attention. */
 469                 if ((call = list_first_entry_or_null(&local->call_attend_q,
 470                                                      struct rxrpc_call,
 471                                                      attend_link))) {
 472                         spin_lock_bh(&local->lock);
 473                         list_del_init(&call->attend_link);
 474                         spin_unlock_bh(&local->lock);
 475
 476                         trace_rxrpc_call_poked(call);
 477                         rxrpc_input_call_event(call, NULL);
 478                         rxrpc_put_call(call, rxrpc_call_put_poke);
 479                         continue;
 480                 }
 481
 482                 /* Process received packets and errors. */
 483                 if ((skb = __skb_dequeue(&rx_queue))) {
 484                         switch (skb->mark) {
 485                         case RXRPC_SKB_MARK_PACKET:
 486                                 skb->priority = 0;
 487                                 rxrpc_input_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                         default:
 496                                 WARN_ON_ONCE(1);
 497                                 rxrpc_free_skb(skb, rxrpc_skb_put_unknown);
 498                                 break;
 499                         }
 500                         continue;
 501                 }
 502
 503                 if (!skb_queue_empty(&local->rx_queue)) {
 504                         spin_lock_irq(&local->rx_queue.lock);
 505                         skb_queue_splice_tail_init(&local->rx_queue, &rx_queue);
 506                         spin_unlock_irq(&local->rx_queue.lock);
 507                         continue;
 508                 }
 509
 510                 set_current_state(TASK_INTERRUPTIBLE);
 511                 if (!skb_queue_empty(&local->rx_queue) ||
 512                     !list_empty(&local->call_attend_q)) {
 513                         __set_current_state(TASK_RUNNING);
 514                         continue;
 515                 }
 516
 517                 if (kthread_should_stop())
 518                         break;
 519                 schedule();
 520         }
 521
 522         __set_current_state(TASK_RUNNING);
 523         rxrpc_see_local(local, rxrpc_local_stop);
 524         rxrpc_destroy_local(local);
 525         local->io_thread = NULL;
 526         rxrpc_see_local(local, rxrpc_local_stopped);
 527         return 0;
 528 }