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 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 rxrpc_local *local = rcu_dereference_sk_user_data(udp_sk);
  29
  30         if (unlikely(!local)) {
  31                 kfree_skb(skb);
  32                 return 0;
  33         }
  34         if (skb->tstamp == 0)
  35                 skb->tstamp = ktime_get_real();
  36
  37         skb->mark = RXRPC_SKB_MARK_PACKET;
  38         rxrpc_new_skb(skb, rxrpc_skb_new_encap_rcv);
  39         skb_queue_tail(&local->rx_queue, skb);
  40         rxrpc_wake_up_io_thread(local);
  41         return 0;
  42 }
  43
  44 /*
  45  * Handle an error received on the local endpoint.
  46  */
  47 void rxrpc_error_report(struct sock *sk)
  48 {
  49         struct rxrpc_local *local;
  50         struct sk_buff *skb;
  51
  52         rcu_read_lock();
  53         local = rcu_dereference_sk_user_data(sk);
  54         if (unlikely(!local)) {
  55                 rcu_read_unlock();
  56                 return;
  57         }
  58
  59         while ((skb = skb_dequeue(&sk->sk_error_queue))) {
  60                 skb->mark = RXRPC_SKB_MARK_ERROR;
  61                 rxrpc_new_skb(skb, rxrpc_skb_new_error_report);
  62                 skb_queue_tail(&local->rx_queue, skb);
  63         }
  64
  65         rxrpc_wake_up_io_thread(local);
  66         rcu_read_unlock();
  67 }
  68
  69 /*
  70  * Directly produce an abort from a packet.
  71  */
  72 bool rxrpc_direct_abort(struct sk_buff *skb, enum rxrpc_abort_reason why,
  73                         s32 abort_code, int err)
  74 {
  75         struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
  76
  77         trace_rxrpc_abort(0, why, sp->hdr.cid, sp->hdr.callNumber, sp->hdr.seq,
  78                           abort_code, err);
  79         skb->mark = RXRPC_SKB_MARK_REJECT_ABORT;
  80         skb->priority = abort_code;
  81         return false;
  82 }
  83
  84 static bool rxrpc_bad_message(struct sk_buff *skb, enum rxrpc_abort_reason why)
  85 {
  86         return rxrpc_direct_abort(skb, why, RX_PROTOCOL_ERROR, -EBADMSG);
  87 }
  88
  89 #define just_discard true
  90
  91 /*
  92  * Process event packets targeted at a local endpoint.
  93  */
  94 static bool rxrpc_input_version(struct rxrpc_local *local, struct sk_buff *skb)
  95 {
  96         struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
  97         char v;
  98
  99         _enter("");
 100
 101         rxrpc_see_skb(skb, rxrpc_skb_see_version);
 102         if (skb_copy_bits(skb, sizeof(struct rxrpc_wire_header), &v, 1) >= 0) {
 103                 if (v == 0)
 104                         rxrpc_send_version_request(local, &sp->hdr, skb);
 105         }
 106
 107         return true;
 108 }
 109
 110 /*
 111  * Extract the wire header from a packet and translate the byte order.
 112  */
 113 static bool rxrpc_extract_header(struct rxrpc_skb_priv *sp,
 114                                  struct sk_buff *skb)
 115 {
 116         struct rxrpc_wire_header whdr;
 117
 118         /* dig out the RxRPC connection details */
 119         if (skb_copy_bits(skb, 0, &whdr, sizeof(whdr)) < 0)
 120                 return rxrpc_bad_message(skb, rxrpc_badmsg_short_hdr);
 121
 122         memset(sp, 0, sizeof(*sp));
 123         sp->hdr.epoch           = ntohl(whdr.epoch);
 124         sp->hdr.cid             = ntohl(whdr.cid);
 125         sp->hdr.callNumber      = ntohl(whdr.callNumber);
 126         sp->hdr.seq             = ntohl(whdr.seq);
 127         sp->hdr.serial          = ntohl(whdr.serial);
 128         sp->hdr.flags           = whdr.flags;
 129         sp->hdr.type            = whdr.type;
 130         sp->hdr.userStatus      = whdr.userStatus;
 131         sp->hdr.securityIndex   = whdr.securityIndex;
 132         sp->hdr._rsvd           = ntohs(whdr._rsvd);
 133         sp->hdr.serviceId       = ntohs(whdr.serviceId);
 134         return true;
 135 }
 136
 137 /*
 138  * Extract the abort code from an ABORT packet and stash it in skb->priority.
 139  */
 140 static bool rxrpc_extract_abort(struct sk_buff *skb)
 141 {
 142         __be32 wtmp;
 143
 144         if (skb_copy_bits(skb, sizeof(struct rxrpc_wire_header),
 145                           &wtmp, sizeof(wtmp)) < 0)
 146                 return false;
 147         skb->priority = ntohl(wtmp);
 148         return true;
 149 }
 150
 151 /*
 152  * Process packets received on the local endpoint
 153  */
 154 static bool rxrpc_input_packet(struct rxrpc_local *local, struct sk_buff **_skb)
 155 {
 156         struct rxrpc_connection *conn;
 157         struct sockaddr_rxrpc peer_srx;
 158         struct rxrpc_skb_priv *sp;
 159         struct rxrpc_peer *peer = NULL;
 160         struct sk_buff *skb = *_skb;
 161         bool ret = false;
 162
 163         skb_pull(skb, sizeof(struct udphdr));
 164
 165         sp = rxrpc_skb(skb);
 166
 167         /* dig out the RxRPC connection details */
 168         if (!rxrpc_extract_header(sp, skb))
 169                 return just_discard;
 170
 171         if (IS_ENABLED(CONFIG_AF_RXRPC_INJECT_LOSS)) {
 172                 static int lose;
 173                 if ((lose++ & 7) == 7) {
 174                         trace_rxrpc_rx_lose(sp);
 175                         return just_discard;
 176                 }
 177         }
 178
 179         trace_rxrpc_rx_packet(sp);
 180
 181         switch (sp->hdr.type) {
 182         case RXRPC_PACKET_TYPE_VERSION:
 183                 if (rxrpc_to_client(sp))
 184                         return just_discard;
 185                 return rxrpc_input_version(local, skb);
 186
 187         case RXRPC_PACKET_TYPE_BUSY:
 188                 if (rxrpc_to_server(sp))
 189                         return just_discard;
 190                 fallthrough;
 191         case RXRPC_PACKET_TYPE_ACK:
 192         case RXRPC_PACKET_TYPE_ACKALL:
 193                 if (sp->hdr.callNumber == 0)
 194                         return rxrpc_bad_message(skb, rxrpc_badmsg_zero_call);
 195                 break;
 196         case RXRPC_PACKET_TYPE_ABORT:
 197                 if (!rxrpc_extract_abort(skb))
 198                         return just_discard; /* Just discard if malformed */
 199                 break;
 200
 201         case RXRPC_PACKET_TYPE_DATA:
 202                 if (sp->hdr.callNumber == 0)
 203                         return rxrpc_bad_message(skb, rxrpc_badmsg_zero_call);
 204                 if (sp->hdr.seq == 0)
 205                         return rxrpc_bad_message(skb, rxrpc_badmsg_zero_seq);
 206
 207                 /* Unshare the packet so that it can be modified for in-place
 208                  * decryption.
 209                  */
 210                 if (sp->hdr.securityIndex != 0) {
 211                         skb = skb_unshare(skb, GFP_ATOMIC);
 212                         if (!skb) {
 213                                 rxrpc_eaten_skb(*_skb, rxrpc_skb_eaten_by_unshare_nomem);
 214                                 *_skb = NULL;
 215                                 return just_discard;
 216                         }
 217
 218                         if (skb != *_skb) {
 219                                 rxrpc_eaten_skb(*_skb, rxrpc_skb_eaten_by_unshare);
 220                                 *_skb = skb;
 221                                 rxrpc_new_skb(skb, rxrpc_skb_new_unshared);
 222                                 sp = rxrpc_skb(skb);
 223                         }
 224                 }
 225                 break;
 226
 227         case RXRPC_PACKET_TYPE_CHALLENGE:
 228                 if (rxrpc_to_server(sp))
 229                         return just_discard;
 230                 break;
 231         case RXRPC_PACKET_TYPE_RESPONSE:
 232                 if (rxrpc_to_client(sp))
 233                         return just_discard;
 234                 break;
 235
 236                 /* Packet types 9-11 should just be ignored. */
 237         case RXRPC_PACKET_TYPE_PARAMS:
 238         case RXRPC_PACKET_TYPE_10:
 239         case RXRPC_PACKET_TYPE_11:
 240                 return just_discard;
 241
 242         default:
 243                 return rxrpc_bad_message(skb, rxrpc_badmsg_unsupported_packet);
 244         }
 245
 246         if (sp->hdr.serviceId == 0)
 247                 return rxrpc_bad_message(skb, rxrpc_badmsg_zero_service);
 248
 249         if (WARN_ON_ONCE(rxrpc_extract_addr_from_skb(&peer_srx, skb) < 0))
 250                 return just_discard; /* Unsupported address type. */
 251
 252         if (peer_srx.transport.family != local->srx.transport.family &&
 253             (peer_srx.transport.family == AF_INET &&
 254              local->srx.transport.family != AF_INET6)) {
 255                 pr_warn_ratelimited("AF_RXRPC: Protocol mismatch %u not %u\n",
 256                                     peer_srx.transport.family,
 257                                     local->srx.transport.family);
 258                 return just_discard; /* Wrong address type. */
 259         }
 260
 261         if (rxrpc_to_client(sp)) {
 262                 rcu_read_lock();
 263                 conn = rxrpc_find_client_connection_rcu(local, &peer_srx, skb);
 264                 conn = rxrpc_get_connection_maybe(conn, rxrpc_conn_get_call_input);
 265                 rcu_read_unlock();
 266                 if (!conn)
 267                         return rxrpc_protocol_error(skb, rxrpc_eproto_no_client_conn);
 268
 269                 ret = rxrpc_input_packet_on_conn(conn, &peer_srx, skb);
 270                 rxrpc_put_connection(conn, rxrpc_conn_put_call_input);
 271                 return ret;
 272         }
 273
 274         /* We need to look up service connections by the full protocol
 275          * parameter set.  We look up the peer first as an intermediate step
 276          * and then the connection from the peer's tree.
 277          */
 278         rcu_read_lock();
 279
 280         peer = rxrpc_lookup_peer_rcu(local, &peer_srx);
 281         if (!peer) {
 282                 rcu_read_unlock();
 283                 return rxrpc_new_incoming_call(local, NULL, NULL, &peer_srx, skb);
 284         }
 285
 286         conn = rxrpc_find_service_conn_rcu(peer, skb);
 287         conn = rxrpc_get_connection_maybe(conn, rxrpc_conn_get_call_input);
 288         if (conn) {
 289                 rcu_read_unlock();
 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         peer = rxrpc_get_peer_maybe(peer, rxrpc_peer_get_input);
 296         rcu_read_unlock();
 297
 298         ret = rxrpc_new_incoming_call(local, peer, NULL, &peer_srx, skb);
 299         rxrpc_put_peer(peer, rxrpc_peer_put_input);
 300         return ret;
 301 }
 302
 303 /*
 304  * Deal with a packet that's associated with an extant connection.
 305  */
 306 static int rxrpc_input_packet_on_conn(struct rxrpc_connection *conn,
 307                                       struct sockaddr_rxrpc *peer_srx,
 308                                       struct sk_buff *skb)
 309 {
 310         struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
 311         struct rxrpc_channel *chan;
 312         struct rxrpc_call *call = NULL;
 313         unsigned int channel;
 314         bool ret;
 315
 316         if (sp->hdr.securityIndex != conn->security_ix)
 317                 return rxrpc_direct_abort(skb, rxrpc_eproto_wrong_security,
 318                                           RXKADINCONSISTENCY, -EBADMSG);
 319
 320         if (sp->hdr.serviceId != conn->service_id) {
 321                 int old_id;
 322
 323                 if (!test_bit(RXRPC_CONN_PROBING_FOR_UPGRADE, &conn->flags))
 324                         return rxrpc_protocol_error(skb, rxrpc_eproto_reupgrade);
 325
 326                 old_id = cmpxchg(&conn->service_id, conn->orig_service_id,
 327                                  sp->hdr.serviceId);
 328                 if (old_id != conn->orig_service_id &&
 329                     old_id != sp->hdr.serviceId)
 330                         return rxrpc_protocol_error(skb, rxrpc_eproto_bad_upgrade);
 331         }
 332
 333         if (after(sp->hdr.serial, conn->hi_serial))
 334                 conn->hi_serial = sp->hdr.serial;
 335
 336         /* It's a connection-level packet if the call number is 0. */
 337         if (sp->hdr.callNumber == 0)
 338                 return rxrpc_input_conn_packet(conn, skb);
 339
 340         /* Call-bound packets are routed by connection channel. */
 341         channel = sp->hdr.cid & RXRPC_CHANNELMASK;
 342         chan = &conn->channels[channel];
 343
 344         /* Ignore really old calls */
 345         if (sp->hdr.callNumber < chan->last_call)
 346                 return just_discard;
 347
 348         if (sp->hdr.callNumber == chan->last_call) {
 349                 if (chan->call ||
 350                     sp->hdr.type == RXRPC_PACKET_TYPE_ABORT)
 351                         return just_discard;
 352
 353                 /* For the previous service call, if completed successfully, we
 354                  * discard all further packets.
 355                  */
 356                 if (rxrpc_conn_is_service(conn) &&
 357                     chan->last_type == RXRPC_PACKET_TYPE_ACK)
 358                         return just_discard;
 359
 360                 /* But otherwise we need to retransmit the final packet from
 361                  * data cached in the connection record.
 362                  */
 363                 if (sp->hdr.type == RXRPC_PACKET_TYPE_DATA)
 364                         trace_rxrpc_rx_data(chan->call_debug_id,
 365                                             sp->hdr.seq,
 366                                             sp->hdr.serial,
 367                                             sp->hdr.flags);
 368                 rxrpc_conn_retransmit_call(conn, skb, channel);
 369                 return just_discard;
 370         }
 371
 372         rcu_read_lock();
 373         call = rxrpc_try_get_call(rcu_dereference(chan->call),
 374                                   rxrpc_call_get_input);
 375         rcu_read_unlock();
 376
 377         if (sp->hdr.callNumber > chan->call_id) {
 378                 if (rxrpc_to_client(sp)) {
 379                         rxrpc_put_call(call, rxrpc_call_put_input);
 380                         return rxrpc_protocol_error(skb,
 381                                                     rxrpc_eproto_unexpected_implicit_end);
 382                 }
 383
 384                 if (call) {
 385                         rxrpc_implicit_end_call(call, skb);
 386                         rxrpc_put_call(call, rxrpc_call_put_input);
 387                         call = NULL;
 388                 }
 389         }
 390
 391         if (!call) {
 392                 if (rxrpc_to_client(sp))
 393                         return rxrpc_protocol_error(skb, rxrpc_eproto_no_client_call);
 394                 return rxrpc_new_incoming_call(conn->local, conn->peer, conn,
 395                                                peer_srx, skb);
 396         }
 397
 398         ret = rxrpc_input_call_event(call, skb);
 399         rxrpc_put_call(call, rxrpc_call_put_input);
 400         return ret;
 401 }
 402
 403 /*
 404  * I/O and event handling thread.
 405  */
 406 int rxrpc_io_thread(void *data)
 407 {
 408         struct rxrpc_connection *conn;
 409         struct sk_buff_head rx_queue;
 410         struct rxrpc_local *local = data;
 411         struct rxrpc_call *call;
 412         struct sk_buff *skb;
 413         bool should_stop;
 414
 415         complete(&local->io_thread_ready);
 416
 417         skb_queue_head_init(&rx_queue);
 418
 419         set_user_nice(current, MIN_NICE);
 420
 421         for (;;) {
 422                 rxrpc_inc_stat(local->rxnet, stat_io_loop);
 423
 424                 /* Deal with connections that want immediate attention. */
 425                 conn = list_first_entry_or_null(&local->conn_attend_q,
 426                                                 struct rxrpc_connection,
 427                                                 attend_link);
 428                 if (conn) {
 429                         spin_lock_bh(&local->lock);
 430                         list_del_init(&conn->attend_link);
 431                         spin_unlock_bh(&local->lock);
 432
 433                         rxrpc_input_conn_event(conn, NULL);
 434                         rxrpc_put_connection(conn, rxrpc_conn_put_poke);
 435                         continue;
 436                 }
 437
 438                 /* Deal with calls that want immediate attention. */
 439                 if ((call = list_first_entry_or_null(&local->call_attend_q,
 440                                                      struct rxrpc_call,
 441                                                      attend_link))) {
 442                         spin_lock_bh(&local->lock);
 443                         list_del_init(&call->attend_link);
 444                         spin_unlock_bh(&local->lock);
 445
 446                         trace_rxrpc_call_poked(call);
 447                         rxrpc_input_call_event(call, NULL);
 448                         rxrpc_put_call(call, rxrpc_call_put_poke);
 449                         continue;
 450                 }
 451
 452                 /* Process received packets and errors. */
 453                 if ((skb = __skb_dequeue(&rx_queue))) {
 454                         switch (skb->mark) {
 455                         case RXRPC_SKB_MARK_PACKET:
 456                                 skb->priority = 0;
 457                                 if (!rxrpc_input_packet(local, &skb))
 458                                         rxrpc_reject_packet(local, skb);
 459                                 trace_rxrpc_rx_done(skb->mark, skb->priority);
 460                                 rxrpc_free_skb(skb, rxrpc_skb_put_input);
 461                                 break;
 462                         case RXRPC_SKB_MARK_ERROR:
 463                                 rxrpc_input_error(local, skb);
 464                                 rxrpc_free_skb(skb, rxrpc_skb_put_error_report);
 465                                 break;
 466                                 break;
 467                         default:
 468                                 WARN_ON_ONCE(1);
 469                                 rxrpc_free_skb(skb, rxrpc_skb_put_unknown);
 470                                 break;
 471                         }
 472                         continue;
 473                 }
 474
 475                 if (!skb_queue_empty(&local->rx_queue)) {
 476                         spin_lock_irq(&local->rx_queue.lock);
 477                         skb_queue_splice_tail_init(&local->rx_queue, &rx_queue);
 478                         spin_unlock_irq(&local->rx_queue.lock);
 479                         continue;
 480                 }
 481
 482                 set_current_state(TASK_INTERRUPTIBLE);
 483                 should_stop = kthread_should_stop();
 484                 if (!skb_queue_empty(&local->rx_queue) ||
 485                     !list_empty(&local->call_attend_q) ||
 486                     !list_empty(&local->conn_attend_q)) {
 487                         __set_current_state(TASK_RUNNING);
 488                         continue;
 489                 }
 490
 491                 if (should_stop)
 492                         break;
 493                 schedule();
 494         }
 495
 496         __set_current_state(TASK_RUNNING);
 497         rxrpc_see_local(local, rxrpc_local_stop);
 498         rxrpc_destroy_local(local);
 499         local->io_thread = NULL;
 500         rxrpc_see_local(local, rxrpc_local_stopped);
 501         return 0;
 502 }