1 // SPDX-License-Identifier: GPL-2.0-or-later 2 /* RxRPC recvmsg() implementation 3 * 4 * Copyright (C) 2007 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 <linux/net.h> 11 #include <linux/skbuff.h> 12 #include <linux/export.h> 13 #include <linux/sched/signal.h> 14 15 #include <net/sock.h> 16 #include <net/af_rxrpc.h> 17 #include "ar-internal.h" 18 19 /* 20 * Post a call for attention by the socket or kernel service. Further 21 * notifications are suppressed by putting recvmsg_link on a dummy queue. 22 */ 23 void rxrpc_notify_socket(struct rxrpc_call *call) 24 { 25 struct rxrpc_sock *rx; 26 struct sock *sk; 27 28 _enter("%d", call->debug_id); 29 30 if (!list_empty(&call->recvmsg_link)) 31 return; 32 33 rcu_read_lock(); 34 35 rx = rcu_dereference(call->socket); 36 sk = &rx->sk; 37 if (rx && sk->sk_state < RXRPC_CLOSE) { 38 if (call->notify_rx) { 39 spin_lock(&call->notify_lock); 40 call->notify_rx(sk, call, call->user_call_ID); 41 spin_unlock(&call->notify_lock); 42 } else { 43 write_lock(&rx->recvmsg_lock); 44 if (list_empty(&call->recvmsg_link)) { 45 rxrpc_get_call(call, rxrpc_call_get_notify_socket); 46 list_add_tail(&call->recvmsg_link, &rx->recvmsg_q); 47 } 48 write_unlock(&rx->recvmsg_lock); 49 50 if (!sock_flag(sk, SOCK_DEAD)) { 51 _debug("call %ps", sk->sk_data_ready); 52 sk->sk_data_ready(sk); 53 } 54 } 55 } 56 57 rcu_read_unlock(); 58 _leave(""); 59 } 60 61 /* 62 * Pass a call terminating message to userspace. 63 */ 64 static int rxrpc_recvmsg_term(struct rxrpc_call *call, struct msghdr *msg) 65 { 66 u32 tmp = 0; 67 int ret; 68 69 switch (call->completion) { 70 case RXRPC_CALL_SUCCEEDED: 71 ret = 0; 72 if (rxrpc_is_service_call(call)) 73 ret = put_cmsg(msg, SOL_RXRPC, RXRPC_ACK, 0, &tmp); 74 break; 75 case RXRPC_CALL_REMOTELY_ABORTED: 76 tmp = call->abort_code; 77 ret = put_cmsg(msg, SOL_RXRPC, RXRPC_ABORT, 4, &tmp); 78 break; 79 case RXRPC_CALL_LOCALLY_ABORTED: 80 tmp = call->abort_code; 81 ret = put_cmsg(msg, SOL_RXRPC, RXRPC_ABORT, 4, &tmp); 82 break; 83 case RXRPC_CALL_NETWORK_ERROR: 84 tmp = -call->error; 85 ret = put_cmsg(msg, SOL_RXRPC, RXRPC_NET_ERROR, 4, &tmp); 86 break; 87 case RXRPC_CALL_LOCAL_ERROR: 88 tmp = -call->error; 89 ret = put_cmsg(msg, SOL_RXRPC, RXRPC_LOCAL_ERROR, 4, &tmp); 90 break; 91 default: 92 pr_err("Invalid terminal call state %u\n", call->completion); 93 BUG(); 94 break; 95 } 96 97 trace_rxrpc_recvdata(call, rxrpc_recvmsg_terminal, 98 lower_32_bits(atomic64_read(&call->ackr_window)) - 1, 99 call->rx_pkt_offset, call->rx_pkt_len, ret); 100 return ret; 101 } 102 103 /* 104 * Discard a packet we've used up and advance the Rx window by one. 105 */ 106 static void rxrpc_rotate_rx_window(struct rxrpc_call *call) 107 { 108 struct rxrpc_skb_priv *sp; 109 struct sk_buff *skb; 110 rxrpc_serial_t serial; 111 rxrpc_seq_t old_consumed = call->rx_consumed, tseq; 112 bool last; 113 int acked; 114 115 _enter("%d", call->debug_id); 116 117 skb = skb_dequeue(&call->recvmsg_queue); 118 rxrpc_see_skb(skb, rxrpc_skb_see_rotate); 119 120 sp = rxrpc_skb(skb); 121 tseq = sp->hdr.seq; 122 serial = sp->hdr.serial; 123 last = sp->hdr.flags & RXRPC_LAST_PACKET; 124 125 /* Barrier against rxrpc_input_data(). */ 126 if (after(tseq, call->rx_consumed)) 127 smp_store_release(&call->rx_consumed, tseq); 128 129 rxrpc_free_skb(skb, rxrpc_skb_put_rotate); 130 131 trace_rxrpc_receive(call, last ? rxrpc_receive_rotate_last : rxrpc_receive_rotate, 132 serial, call->rx_consumed); 133 134 if (last) 135 set_bit(RXRPC_CALL_RECVMSG_READ_ALL, &call->flags); 136 137 /* Check to see if there's an ACK that needs sending. */ 138 acked = atomic_add_return(call->rx_consumed - old_consumed, 139 &call->ackr_nr_consumed); 140 if (acked > 2 && 141 !test_and_set_bit(RXRPC_CALL_RX_IS_IDLE, &call->flags)) 142 rxrpc_poke_call(call, rxrpc_call_poke_idle); 143 } 144 145 /* 146 * Decrypt and verify a DATA packet. 147 */ 148 static int rxrpc_verify_data(struct rxrpc_call *call, struct sk_buff *skb) 149 { 150 struct rxrpc_skb_priv *sp = rxrpc_skb(skb); 151 152 if (sp->flags & RXRPC_RX_VERIFIED) 153 return 0; 154 return call->security->verify_packet(call, skb); 155 } 156 157 /* 158 * Deliver messages to a call. This keeps processing packets until the buffer 159 * is filled and we find either more DATA (returns 0) or the end of the DATA 160 * (returns 1). If more packets are required, it returns -EAGAIN and if the 161 * call has failed it returns -EIO. 162 */ 163 static int rxrpc_recvmsg_data(struct socket *sock, struct rxrpc_call *call, 164 struct msghdr *msg, struct iov_iter *iter, 165 size_t len, int flags, size_t *_offset) 166 { 167 struct rxrpc_skb_priv *sp; 168 struct sk_buff *skb; 169 rxrpc_seq_t seq = 0; 170 size_t remain; 171 unsigned int rx_pkt_offset, rx_pkt_len; 172 int copy, ret = -EAGAIN, ret2; 173 174 rx_pkt_offset = call->rx_pkt_offset; 175 rx_pkt_len = call->rx_pkt_len; 176 177 if (rxrpc_call_has_failed(call)) { 178 seq = lower_32_bits(atomic64_read(&call->ackr_window)) - 1; 179 ret = -EIO; 180 goto done; 181 } 182 183 if (test_bit(RXRPC_CALL_RECVMSG_READ_ALL, &call->flags)) { 184 seq = lower_32_bits(atomic64_read(&call->ackr_window)) - 1; 185 ret = 1; 186 goto done; 187 } 188 189 /* No one else can be removing stuff from the queue, so we shouldn't 190 * need the Rx lock to walk it. 191 */ 192 skb = skb_peek(&call->recvmsg_queue); 193 while (skb) { 194 rxrpc_see_skb(skb, rxrpc_skb_see_recvmsg); 195 sp = rxrpc_skb(skb); 196 seq = sp->hdr.seq; 197 198 if (!(flags & MSG_PEEK)) 199 trace_rxrpc_receive(call, rxrpc_receive_front, 200 sp->hdr.serial, seq); 201 202 if (msg) 203 sock_recv_timestamp(msg, sock->sk, skb); 204 205 if (rx_pkt_offset == 0) { 206 ret2 = rxrpc_verify_data(call, skb); 207 trace_rxrpc_recvdata(call, rxrpc_recvmsg_next, seq, 208 sp->offset, sp->len, ret2); 209 if (ret2 < 0) { 210 kdebug("verify = %d", ret2); 211 ret = ret2; 212 goto out; 213 } 214 rx_pkt_offset = sp->offset; 215 rx_pkt_len = sp->len; 216 } else { 217 trace_rxrpc_recvdata(call, rxrpc_recvmsg_cont, seq, 218 rx_pkt_offset, rx_pkt_len, 0); 219 } 220 221 /* We have to handle short, empty and used-up DATA packets. */ 222 remain = len - *_offset; 223 copy = rx_pkt_len; 224 if (copy > remain) 225 copy = remain; 226 if (copy > 0) { 227 ret2 = skb_copy_datagram_iter(skb, rx_pkt_offset, iter, 228 copy); 229 if (ret2 < 0) { 230 ret = ret2; 231 goto out; 232 } 233 234 /* handle piecemeal consumption of data packets */ 235 rx_pkt_offset += copy; 236 rx_pkt_len -= copy; 237 *_offset += copy; 238 } 239 240 if (rx_pkt_len > 0) { 241 trace_rxrpc_recvdata(call, rxrpc_recvmsg_full, seq, 242 rx_pkt_offset, rx_pkt_len, 0); 243 ASSERTCMP(*_offset, ==, len); 244 ret = 0; 245 break; 246 } 247 248 /* The whole packet has been transferred. */ 249 if (sp->hdr.flags & RXRPC_LAST_PACKET) 250 ret = 1; 251 rx_pkt_offset = 0; 252 rx_pkt_len = 0; 253 254 skb = skb_peek_next(skb, &call->recvmsg_queue); 255 256 if (!(flags & MSG_PEEK)) 257 rxrpc_rotate_rx_window(call); 258 } 259 260 out: 261 if (!(flags & MSG_PEEK)) { 262 call->rx_pkt_offset = rx_pkt_offset; 263 call->rx_pkt_len = rx_pkt_len; 264 } 265 done: 266 trace_rxrpc_recvdata(call, rxrpc_recvmsg_data_return, seq, 267 rx_pkt_offset, rx_pkt_len, ret); 268 if (ret == -EAGAIN) 269 set_bit(RXRPC_CALL_RX_IS_IDLE, &call->flags); 270 return ret; 271 } 272 273 /* 274 * Receive a message from an RxRPC socket 275 * - we need to be careful about two or more threads calling recvmsg 276 * simultaneously 277 */ 278 int rxrpc_recvmsg(struct socket *sock, struct msghdr *msg, size_t len, 279 int flags) 280 { 281 struct rxrpc_call *call; 282 struct rxrpc_sock *rx = rxrpc_sk(sock->sk); 283 struct list_head *l; 284 unsigned int call_debug_id = 0; 285 size_t copied = 0; 286 long timeo; 287 int ret; 288 289 DEFINE_WAIT(wait); 290 291 trace_rxrpc_recvmsg(0, rxrpc_recvmsg_enter, 0); 292 293 if (flags & (MSG_OOB | MSG_TRUNC)) 294 return -EOPNOTSUPP; 295 296 timeo = sock_rcvtimeo(&rx->sk, flags & MSG_DONTWAIT); 297 298 try_again: 299 lock_sock(&rx->sk); 300 301 /* Return immediately if a client socket has no outstanding calls */ 302 if (RB_EMPTY_ROOT(&rx->calls) && 303 list_empty(&rx->recvmsg_q) && 304 rx->sk.sk_state != RXRPC_SERVER_LISTENING) { 305 release_sock(&rx->sk); 306 return -EAGAIN; 307 } 308 309 if (list_empty(&rx->recvmsg_q)) { 310 ret = -EWOULDBLOCK; 311 if (timeo == 0) { 312 call = NULL; 313 goto error_no_call; 314 } 315 316 release_sock(&rx->sk); 317 318 /* Wait for something to happen */ 319 prepare_to_wait_exclusive(sk_sleep(&rx->sk), &wait, 320 TASK_INTERRUPTIBLE); 321 ret = sock_error(&rx->sk); 322 if (ret) 323 goto wait_error; 324 325 if (list_empty(&rx->recvmsg_q)) { 326 if (signal_pending(current)) 327 goto wait_interrupted; 328 trace_rxrpc_recvmsg(0, rxrpc_recvmsg_wait, 0); 329 timeo = schedule_timeout(timeo); 330 } 331 finish_wait(sk_sleep(&rx->sk), &wait); 332 goto try_again; 333 } 334 335 /* Find the next call and dequeue it if we're not just peeking. If we 336 * do dequeue it, that comes with a ref that we will need to release. 337 */ 338 write_lock(&rx->recvmsg_lock); 339 l = rx->recvmsg_q.next; 340 call = list_entry(l, struct rxrpc_call, recvmsg_link); 341 if (!(flags & MSG_PEEK)) 342 list_del_init(&call->recvmsg_link); 343 else 344 rxrpc_get_call(call, rxrpc_call_get_recvmsg); 345 write_unlock(&rx->recvmsg_lock); 346 347 call_debug_id = call->debug_id; 348 trace_rxrpc_recvmsg(call_debug_id, rxrpc_recvmsg_dequeue, 0); 349 350 /* We're going to drop the socket lock, so we need to lock the call 351 * against interference by sendmsg. 352 */ 353 if (!mutex_trylock(&call->user_mutex)) { 354 ret = -EWOULDBLOCK; 355 if (flags & MSG_DONTWAIT) 356 goto error_requeue_call; 357 ret = -ERESTARTSYS; 358 if (mutex_lock_interruptible(&call->user_mutex) < 0) 359 goto error_requeue_call; 360 } 361 362 release_sock(&rx->sk); 363 364 if (test_bit(RXRPC_CALL_RELEASED, &call->flags)) 365 BUG(); 366 367 if (test_bit(RXRPC_CALL_HAS_USERID, &call->flags)) { 368 if (flags & MSG_CMSG_COMPAT) { 369 unsigned int id32 = call->user_call_ID; 370 371 ret = put_cmsg(msg, SOL_RXRPC, RXRPC_USER_CALL_ID, 372 sizeof(unsigned int), &id32); 373 } else { 374 unsigned long idl = call->user_call_ID; 375 376 ret = put_cmsg(msg, SOL_RXRPC, RXRPC_USER_CALL_ID, 377 sizeof(unsigned long), &idl); 378 } 379 if (ret < 0) 380 goto error_unlock_call; 381 } 382 383 if (msg->msg_name && call->peer) { 384 size_t len = sizeof(call->dest_srx); 385 386 memcpy(msg->msg_name, &call->dest_srx, len); 387 msg->msg_namelen = len; 388 } 389 390 ret = rxrpc_recvmsg_data(sock, call, msg, &msg->msg_iter, len, 391 flags, &copied); 392 if (ret == -EAGAIN) 393 ret = 0; 394 if (ret == -EIO) 395 goto call_failed; 396 if (ret < 0) 397 goto error_unlock_call; 398 399 if (rxrpc_call_is_complete(call) && 400 skb_queue_empty(&call->recvmsg_queue)) 401 goto call_complete; 402 if (rxrpc_call_has_failed(call)) 403 goto call_failed; 404 405 rxrpc_notify_socket(call); 406 goto not_yet_complete; 407 408 call_failed: 409 rxrpc_purge_queue(&call->recvmsg_queue); 410 call_complete: 411 ret = rxrpc_recvmsg_term(call, msg); 412 if (ret < 0) 413 goto error_unlock_call; 414 if (!(flags & MSG_PEEK)) 415 rxrpc_release_call(rx, call); 416 msg->msg_flags |= MSG_EOR; 417 ret = 1; 418 419 not_yet_complete: 420 if (ret == 0) 421 msg->msg_flags |= MSG_MORE; 422 else 423 msg->msg_flags &= ~MSG_MORE; 424 ret = copied; 425 426 error_unlock_call: 427 mutex_unlock(&call->user_mutex); 428 rxrpc_put_call(call, rxrpc_call_put_recvmsg); 429 trace_rxrpc_recvmsg(call_debug_id, rxrpc_recvmsg_return, ret); 430 return ret; 431 432 error_requeue_call: 433 if (!(flags & MSG_PEEK)) { 434 write_lock(&rx->recvmsg_lock); 435 list_add(&call->recvmsg_link, &rx->recvmsg_q); 436 write_unlock(&rx->recvmsg_lock); 437 trace_rxrpc_recvmsg(call_debug_id, rxrpc_recvmsg_requeue, 0); 438 } else { 439 rxrpc_put_call(call, rxrpc_call_put_recvmsg); 440 } 441 error_no_call: 442 release_sock(&rx->sk); 443 error_trace: 444 trace_rxrpc_recvmsg(call_debug_id, rxrpc_recvmsg_return, ret); 445 return ret; 446 447 wait_interrupted: 448 ret = sock_intr_errno(timeo); 449 wait_error: 450 finish_wait(sk_sleep(&rx->sk), &wait); 451 call = NULL; 452 goto error_trace; 453 } 454 455 /** 456 * rxrpc_kernel_recv_data - Allow a kernel service to receive data/info 457 * @sock: The socket that the call exists on 458 * @call: The call to send data through 459 * @iter: The buffer to receive into 460 * @_len: The amount of data we want to receive (decreased on return) 461 * @want_more: True if more data is expected to be read 462 * @_abort: Where the abort code is stored if -ECONNABORTED is returned 463 * @_service: Where to store the actual service ID (may be upgraded) 464 * 465 * Allow a kernel service to receive data and pick up information about the 466 * state of a call. Returns 0 if got what was asked for and there's more 467 * available, 1 if we got what was asked for and we're at the end of the data 468 * and -EAGAIN if we need more data. 469 * 470 * Note that we may return -EAGAIN to drain empty packets at the end of the 471 * data, even if we've already copied over the requested data. 472 * 473 * *_abort should also be initialised to 0. 474 */ 475 int rxrpc_kernel_recv_data(struct socket *sock, struct rxrpc_call *call, 476 struct iov_iter *iter, size_t *_len, 477 bool want_more, u32 *_abort, u16 *_service) 478 { 479 size_t offset = 0; 480 int ret; 481 482 _enter("{%d},%zu,%d", call->debug_id, *_len, want_more); 483 484 mutex_lock(&call->user_mutex); 485 486 ret = rxrpc_recvmsg_data(sock, call, NULL, iter, *_len, 0, &offset); 487 *_len -= offset; 488 if (ret == -EIO) 489 goto call_failed; 490 if (ret < 0) 491 goto out; 492 493 /* We can only reach here with a partially full buffer if we have 494 * reached the end of the data. We must otherwise have a full buffer 495 * or have been given -EAGAIN. 496 */ 497 if (ret == 1) { 498 if (iov_iter_count(iter) > 0) 499 goto short_data; 500 if (!want_more) 501 goto read_phase_complete; 502 ret = 0; 503 goto out; 504 } 505 506 if (!want_more) 507 goto excess_data; 508 goto out; 509 510 read_phase_complete: 511 ret = 1; 512 out: 513 if (_service) 514 *_service = call->dest_srx.srx_service; 515 mutex_unlock(&call->user_mutex); 516 _leave(" = %d [%zu,%d]", ret, iov_iter_count(iter), *_abort); 517 return ret; 518 519 short_data: 520 trace_rxrpc_abort(call->debug_id, rxrpc_recvmsg_short_data, 521 call->cid, call->call_id, call->rx_consumed, 522 0, -EBADMSG); 523 ret = -EBADMSG; 524 goto out; 525 excess_data: 526 trace_rxrpc_abort(call->debug_id, rxrpc_recvmsg_excess_data, 527 call->cid, call->call_id, call->rx_consumed, 528 0, -EMSGSIZE); 529 ret = -EMSGSIZE; 530 goto out; 531 call_failed: 532 *_abort = call->abort_code; 533 ret = call->error; 534 if (call->completion == RXRPC_CALL_SUCCEEDED) { 535 ret = 1; 536 if (iov_iter_count(iter) > 0) 537 ret = -ECONNRESET; 538 } 539 goto out; 540 } 541 EXPORT_SYMBOL(rxrpc_kernel_recv_data); 542