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 rxrpc_channel *chan;
 159         struct rxrpc_call *call = NULL;
 160         struct rxrpc_skb_priv *sp;
 161         struct rxrpc_peer *peer = NULL;
 162         struct rxrpc_sock *rx = NULL;
 163         struct sk_buff *skb = *_skb;
 164         unsigned int channel;
 165
 166         if (skb->tstamp == 0)
 167                 skb->tstamp = ktime_get_real();
 168
 169         skb_pull(skb, sizeof(struct udphdr));
 170
 171         /* The UDP protocol already released all skb resources;
 172          * we are free to add our own data there.
 173          */
 174         sp = rxrpc_skb(skb);
 175
 176         /* dig out the RxRPC connection details */
 177         if (rxrpc_extract_header(sp, skb) < 0)
 178                 goto bad_message;
 179
 180         if (IS_ENABLED(CONFIG_AF_RXRPC_INJECT_LOSS)) {
 181                 static int lose;
 182                 if ((lose++ & 7) == 7) {
 183                         trace_rxrpc_rx_lose(sp);
 184                         return 0;
 185                 }
 186         }
 187
 188         if (skb->tstamp == 0)
 189                 skb->tstamp = ktime_get_real();
 190         trace_rxrpc_rx_packet(sp);
 191
 192         switch (sp->hdr.type) {
 193         case RXRPC_PACKET_TYPE_VERSION:
 194                 if (rxrpc_to_client(sp))
 195                         return 0;
 196                 rxrpc_post_packet_to_local(local, skb);
 197                 return 0;
 198
 199         case RXRPC_PACKET_TYPE_BUSY:
 200                 if (rxrpc_to_server(sp))
 201                         return 0;
 202                 fallthrough;
 203         case RXRPC_PACKET_TYPE_ACK:
 204         case RXRPC_PACKET_TYPE_ACKALL:
 205                 if (sp->hdr.callNumber == 0)
 206                         goto bad_message;
 207                 break;
 208         case RXRPC_PACKET_TYPE_ABORT:
 209                 if (!rxrpc_extract_abort(skb))
 210                         return 0; /* Just discard if malformed */
 211                 break;
 212
 213         case RXRPC_PACKET_TYPE_DATA:
 214                 if (sp->hdr.callNumber == 0 ||
 215                     sp->hdr.seq == 0)
 216                         goto bad_message;
 217
 218                 /* Unshare the packet so that it can be modified for in-place
 219                  * decryption.
 220                  */
 221                 if (sp->hdr.securityIndex != 0) {
 222                         skb = skb_unshare(skb, GFP_ATOMIC);
 223                         if (!skb) {
 224                                 rxrpc_eaten_skb(*_skb, rxrpc_skb_eaten_by_unshare_nomem);
 225                                 *_skb = NULL;
 226                                 return 0;
 227                         }
 228
 229                         if (skb != *_skb) {
 230                                 rxrpc_eaten_skb(*_skb, rxrpc_skb_eaten_by_unshare);
 231                                 *_skb = skb;
 232                                 rxrpc_new_skb(skb, rxrpc_skb_new_unshared);
 233                                 sp = rxrpc_skb(skb);
 234                         }
 235                 }
 236                 break;
 237
 238         case RXRPC_PACKET_TYPE_CHALLENGE:
 239                 if (rxrpc_to_server(sp))
 240                         return 0;
 241                 break;
 242         case RXRPC_PACKET_TYPE_RESPONSE:
 243                 if (rxrpc_to_client(sp))
 244                         return 0;
 245                 break;
 246
 247                 /* Packet types 9-11 should just be ignored. */
 248         case RXRPC_PACKET_TYPE_PARAMS:
 249         case RXRPC_PACKET_TYPE_10:
 250         case RXRPC_PACKET_TYPE_11:
 251                 return 0;
 252
 253         default:
 254                 goto bad_message;
 255         }
 256
 257         if (sp->hdr.serviceId == 0)
 258                 goto bad_message;
 259
 260         rcu_read_lock();
 261
 262         if (rxrpc_to_server(sp)) {
 263                 /* Weed out packets to services we're not offering.  Packets
 264                  * that would begin a call are explicitly rejected and the rest
 265                  * are just discarded.
 266                  */
 267                 rx = rcu_dereference(local->service);
 268                 if (!rx || (sp->hdr.serviceId != rx->srx.srx_service &&
 269                             sp->hdr.serviceId != rx->second_service)
 270                     ) {
 271                         rcu_read_unlock();
 272                         if (sp->hdr.type == RXRPC_PACKET_TYPE_DATA &&
 273                             sp->hdr.seq == 1)
 274                                 goto unsupported_service;
 275                         return 0;
 276                 }
 277         }
 278
 279         conn = rxrpc_find_connection_rcu(local, skb, &peer);
 280         if (conn) {
 281                 if (sp->hdr.securityIndex != conn->security_ix)
 282                         goto wrong_security;
 283
 284                 if (sp->hdr.serviceId != conn->service_id) {
 285                         int old_id;
 286
 287                         if (!test_bit(RXRPC_CONN_PROBING_FOR_UPGRADE, &conn->flags))
 288                                 goto reupgrade;
 289                         old_id = cmpxchg(&conn->service_id, conn->orig_service_id,
 290                                          sp->hdr.serviceId);
 291
 292                         if (old_id != conn->orig_service_id &&
 293                             old_id != sp->hdr.serviceId)
 294                                 goto reupgrade;
 295                 }
 296
 297                 if (sp->hdr.callNumber == 0) {
 298                         /* Connection-level packet */
 299                         _debug("CONN %p {%d}", conn, conn->debug_id);
 300                         conn = rxrpc_get_connection_maybe(conn, rxrpc_conn_get_conn_input);
 301                         rcu_read_unlock();
 302                         if (conn) {
 303                                 rxrpc_post_packet_to_conn(conn, skb);
 304                                 rxrpc_put_connection(conn, rxrpc_conn_put_conn_input);
 305                         }
 306                         return 0;
 307                 }
 308
 309                 if ((int)sp->hdr.serial - (int)conn->hi_serial > 0)
 310                         conn->hi_serial = sp->hdr.serial;
 311
 312                 /* Call-bound packets are routed by connection channel. */
 313                 channel = sp->hdr.cid & RXRPC_CHANNELMASK;
 314                 chan = &conn->channels[channel];
 315
 316                 /* Ignore really old calls */
 317                 if (sp->hdr.callNumber < chan->last_call) {
 318                         rcu_read_unlock();
 319                         return 0;
 320                 }
 321
 322                 if (sp->hdr.callNumber == chan->last_call) {
 323                         if (chan->call ||
 324                             sp->hdr.type == RXRPC_PACKET_TYPE_ABORT) {
 325                                 rcu_read_unlock();
 326                                 return 0;
 327                         }
 328
 329                         /* For the previous service call, if completed
 330                          * successfully, we discard all further packets.
 331                          */
 332                         if (rxrpc_conn_is_service(conn) &&
 333                             chan->last_type == RXRPC_PACKET_TYPE_ACK) {
 334                                 rcu_read_unlock();
 335                                 return 0;
 336                         }
 337
 338                         /* But otherwise we need to retransmit the final packet
 339                          * from data cached in the connection record.
 340                          */
 341                         if (sp->hdr.type == RXRPC_PACKET_TYPE_DATA)
 342                                 trace_rxrpc_rx_data(chan->call_debug_id,
 343                                                     sp->hdr.seq,
 344                                                     sp->hdr.serial,
 345                                                     sp->hdr.flags);
 346                         conn = rxrpc_get_connection_maybe(conn, rxrpc_conn_get_call_input);
 347                         rcu_read_unlock();
 348                         if (conn) {
 349                                 rxrpc_post_packet_to_conn(conn, skb);
 350                                 rxrpc_put_connection(conn, rxrpc_conn_put_call_input);
 351                         }
 352                         return 0;
 353                 }
 354
 355                 call = rcu_dereference(chan->call);
 356
 357                 if (sp->hdr.callNumber > chan->call_id) {
 358                         if (rxrpc_to_client(sp)) {
 359                                 rcu_read_unlock();
 360                                 goto reject_packet;
 361                         }
 362                         if (call) {
 363                                 rxrpc_input_implicit_end_call(conn, call);
 364                                 chan->call = NULL;
 365                                 call = NULL;
 366                         }
 367                 }
 368
 369                 if (call && !rxrpc_try_get_call(call, rxrpc_call_get_input))
 370                         call = NULL;
 371
 372                 if (call) {
 373                         if (sp->hdr.serviceId != call->dest_srx.srx_service)
 374                                 call->dest_srx.srx_service = sp->hdr.serviceId;
 375                         if ((int)sp->hdr.serial - (int)call->rx_serial > 0)
 376                                 call->rx_serial = sp->hdr.serial;
 377                         if (!test_bit(RXRPC_CALL_RX_HEARD, &call->flags))
 378                                 set_bit(RXRPC_CALL_RX_HEARD, &call->flags);
 379                 }
 380         }
 381
 382         if (!call) {
 383                 if (rxrpc_to_client(sp) ||
 384                     sp->hdr.type != RXRPC_PACKET_TYPE_DATA) {
 385                         rcu_read_unlock();
 386                         goto bad_message;
 387                 }
 388                 if (sp->hdr.seq != 1) {
 389                         rcu_read_unlock();
 390                         return 0;
 391                 }
 392                 call = rxrpc_new_incoming_call(local, rx, skb);
 393                 if (!call) {
 394                         rcu_read_unlock();
 395                         goto reject_packet;
 396                 }
 397         }
 398
 399         rcu_read_unlock();
 400
 401         /* Process a call packet. */
 402         rxrpc_input_call_event(call, skb);
 403         rxrpc_put_call(call, rxrpc_call_put_input);
 404         trace_rxrpc_rx_done(0, 0);
 405         return 0;
 406
 407 wrong_security:
 408         rcu_read_unlock();
 409         trace_rxrpc_abort(0, "SEC", sp->hdr.cid, sp->hdr.callNumber, sp->hdr.seq,
 410                           RXKADINCONSISTENCY, EBADMSG);
 411         skb->priority = RXKADINCONSISTENCY;
 412         goto post_abort;
 413
 414 unsupported_service:
 415         trace_rxrpc_abort(0, "INV", sp->hdr.cid, sp->hdr.callNumber, sp->hdr.seq,
 416                           RX_INVALID_OPERATION, EOPNOTSUPP);
 417         skb->priority = RX_INVALID_OPERATION;
 418         goto post_abort;
 419
 420 reupgrade:
 421         rcu_read_unlock();
 422         trace_rxrpc_abort(0, "UPG", sp->hdr.cid, sp->hdr.callNumber, sp->hdr.seq,
 423                           RX_PROTOCOL_ERROR, EBADMSG);
 424         goto protocol_error;
 425
 426 bad_message:
 427         trace_rxrpc_abort(0, "BAD", sp->hdr.cid, sp->hdr.callNumber, sp->hdr.seq,
 428                           RX_PROTOCOL_ERROR, EBADMSG);
 429 protocol_error:
 430         skb->priority = RX_PROTOCOL_ERROR;
 431 post_abort:
 432         skb->mark = RXRPC_SKB_MARK_REJECT_ABORT;
 433 reject_packet:
 434         rxrpc_reject_packet(local, skb);
 435         return 0;
 436 }
 437
 438 /*
 439  * I/O and event handling thread.
 440  */
 441 int rxrpc_io_thread(void *data)
 442 {
 443         struct sk_buff_head rx_queue;
 444         struct rxrpc_local *local = data;
 445         struct rxrpc_call *call;
 446         struct sk_buff *skb;
 447
 448         skb_queue_head_init(&rx_queue);
 449
 450         set_user_nice(current, MIN_NICE);
 451
 452         for (;;) {
 453                 rxrpc_inc_stat(local->rxnet, stat_io_loop);
 454
 455                 /* Deal with calls that want immediate attention. */
 456                 if ((call = list_first_entry_or_null(&local->call_attend_q,
 457                                                      struct rxrpc_call,
 458                                                      attend_link))) {
 459                         spin_lock_bh(&local->lock);
 460                         list_del_init(&call->attend_link);
 461                         spin_unlock_bh(&local->lock);
 462
 463                         trace_rxrpc_call_poked(call);
 464                         rxrpc_input_call_event(call, NULL);
 465                         rxrpc_put_call(call, rxrpc_call_put_poke);
 466                         continue;
 467                 }
 468
 469                 /* Process received packets and errors. */
 470                 if ((skb = __skb_dequeue(&rx_queue))) {
 471                         switch (skb->mark) {
 472                         case RXRPC_SKB_MARK_PACKET:
 473                                 skb->priority = 0;
 474                                 rxrpc_input_packet(local, &skb);
 475                                 trace_rxrpc_rx_done(skb->mark, skb->priority);
 476                                 rxrpc_free_skb(skb, rxrpc_skb_put_input);
 477                                 break;
 478                         case RXRPC_SKB_MARK_ERROR:
 479                                 rxrpc_input_error(local, skb);
 480                                 rxrpc_free_skb(skb, rxrpc_skb_put_error_report);
 481                                 break;
 482                         default:
 483                                 WARN_ON_ONCE(1);
 484                                 rxrpc_free_skb(skb, rxrpc_skb_put_unknown);
 485                                 break;
 486                         }
 487                         continue;
 488                 }
 489
 490                 if (!skb_queue_empty(&local->rx_queue)) {
 491                         spin_lock_irq(&local->rx_queue.lock);
 492                         skb_queue_splice_tail_init(&local->rx_queue, &rx_queue);
 493                         spin_unlock_irq(&local->rx_queue.lock);
 494                         continue;
 495                 }
 496
 497                 set_current_state(TASK_INTERRUPTIBLE);
 498                 if (!skb_queue_empty(&local->rx_queue) ||
 499                     !list_empty(&local->call_attend_q)) {
 500                         __set_current_state(TASK_RUNNING);
 501                         continue;
 502                 }
 503
 504                 if (kthread_should_stop())
 505                         break;
 506                 schedule();
 507         }
 508
 509         __set_current_state(TASK_RUNNING);
 510         rxrpc_see_local(local, rxrpc_local_stop);
 511         rxrpc_destroy_local(local);
 512         local->io_thread = NULL;
 513         rxrpc_see_local(local, rxrpc_local_stopped);
 514         return 0;
 515 }