1 // SPDX-License-Identifier: GPL-2.0
2
3 #define _GNU_SOURCE
4
5 #include <errno.h>
6 #include <limits.h>
7 #include <fcntl.h>
8 #include <string.h>
9 #include <stdarg.h>
10 #include <stdbool.h>
11 #include <stdint.h>
12 #include <stdio.h>
13 #include <stdlib.h>
14 #include <strings.h>
15 #include <signal.h>
16 #include <unistd.h>
17 #include <time.h>
18
19 #include <sys/ioctl.h>
20 #include <sys/poll.h>
21 #include <sys/random.h>
22 #include <sys/sendfile.h>
23 #include <sys/stat.h>
24 #include <sys/socket.h>
25 #include <sys/types.h>
26 #include <sys/mman.h>
27
28 #include <netdb.h>
29 #include <netinet/in.h>
30
31 #include <linux/tcp.h>
32 #include <linux/time_types.h>
33 #include <linux/sockios.h>
34
35 extern int optind;
36
37 #ifndef IPPROTO_MPTCP
38 #define IPPROTO_MPTCP 262
39 #endif
40 #ifndef TCP_ULP
41 #define TCP_ULP 31
42 #endif
43
44 static int poll_timeout = 10 * 1000;
45 static bool listen_mode;
46 static bool quit;
47
48 enum cfg_mode {
49 CFG_MODE_POLL,
50 CFG_MODE_MMAP,
51 CFG_MODE_SENDFILE,
52 };
53
54 enum cfg_peek {
55 CFG_NONE_PEEK,
56 CFG_WITH_PEEK,
57 CFG_AFTER_PEEK,
58 };
59
60 static enum cfg_mode cfg_mode = CFG_MODE_POLL;
61 static enum cfg_peek cfg_peek = CFG_NONE_PEEK;
62 static const char *cfg_host;
63 static const char *cfg_port = "12000";
64 static int cfg_sock_proto = IPPROTO_MPTCP;
65 static int pf = AF_INET;
66 static int cfg_sndbuf;
67 static int cfg_rcvbuf;
68 static bool cfg_join;
69 static bool cfg_remove;
70 static unsigned int cfg_time;
71 static unsigned int cfg_do_w;
72 static int cfg_wait;
73 static uint32_t cfg_mark;
74 static char *cfg_input;
75 static int cfg_repeat = 1;
76 static int cfg_truncate;
77 static int cfg_rcv_trunc;
78
79 struct cfg_cmsg_types {
80 unsigned int cmsg_enabled:1;
81 unsigned int timestampns:1;
82 unsigned int tcp_inq:1;
83 };
84
85 struct cfg_sockopt_types {
86 unsigned int transparent:1;
87 };
88
89 struct tcp_inq_state {
90 unsigned int last;
91 bool expect_eof;
92 };
93
94 static struct tcp_inq_state tcp_inq;
95
96 static struct cfg_cmsg_types cfg_cmsg_types;
97 static struct cfg_sockopt_types cfg_sockopt_types;
98
die_usage(void)99 static void die_usage(void)
100 {
101 fprintf(stderr, "Usage: mptcp_connect [-6] [-c cmsg] [-f offset] [-i file] [-I num] [-j] [-l] "
102 "[-m mode] [-M mark] [-o option] [-p port] [-P mode] [-r num] [-R num] "
103 "[-s MPTCP|TCP] [-S num] [-t num] [-T num] [-w sec] connect_address\n");
104 fprintf(stderr, "\t-6 use ipv6\n");
105 fprintf(stderr, "\t-c cmsg -- test cmsg type <cmsg>\n");
106 fprintf(stderr, "\t-f offset -- stop the I/O after receiving and sending the specified amount "
107 "of bytes. If there are unread bytes in the receive queue, that will cause a MPTCP "
108 "fastclose at close/shutdown. If offset is negative, expect the peer to close before "
109 "all the local data as been sent, thus toleration errors on write and EPIPE signals\n");
110 fprintf(stderr, "\t-i file -- read the data to send from the given file instead of stdin");
111 fprintf(stderr, "\t-I num -- repeat the transfer 'num' times. In listen mode accepts num "
112 "incoming connections, in client mode, disconnect and reconnect to the server\n");
113 fprintf(stderr, "\t-j -- add additional sleep at connection start and tear down "
114 "-- for MPJ tests\n");
115 fprintf(stderr, "\t-l -- listens mode, accepts incoming connection\n");
116 fprintf(stderr, "\t-m [poll|mmap|sendfile] -- use poll(default)/mmap+write/sendfile\n");
117 fprintf(stderr, "\t-M mark -- set socket packet mark\n");
118 fprintf(stderr, "\t-o option -- test sockopt <option>\n");
119 fprintf(stderr, "\t-p num -- use port num\n");
120 fprintf(stderr,
121 "\t-P [saveWithPeek|saveAfterPeek] -- save data with/after MSG_PEEK form tcp socket\n");
122 fprintf(stderr, "\t-r num -- enable slow mode, limiting each write to num bytes "
123 "-- for remove addr tests\n");
124 fprintf(stderr, "\t-R num -- set SO_RCVBUF to num\n");
125 fprintf(stderr, "\t-s [MPTCP|TCP] -- use mptcp(default) or tcp sockets\n");
126 fprintf(stderr, "\t-S num -- set SO_SNDBUF to num\n");
127 fprintf(stderr, "\t-t num -- set poll timeout to num\n");
128 fprintf(stderr, "\t-T num -- set expected runtime to num ms\n");
129 fprintf(stderr, "\t-w num -- wait num sec before closing the socket\n");
130 exit(1);
131 }
132
xerror(const char * fmt,...)133 static void xerror(const char *fmt, ...)
134 {
135 va_list ap;
136
137 va_start(ap, fmt);
138 vfprintf(stderr, fmt, ap);
139 va_end(ap);
140 exit(1);
141 }
142
handle_signal(int nr)143 static void handle_signal(int nr)
144 {
145 quit = true;
146 }
147
getxinfo_strerr(int err)148 static const char *getxinfo_strerr(int err)
149 {
150 if (err == EAI_SYSTEM)
151 return strerror(errno);
152
153 return gai_strerror(err);
154 }
155
xgetnameinfo(const struct sockaddr * addr,socklen_t addrlen,char * host,socklen_t hostlen,char * serv,socklen_t servlen)156 static void xgetnameinfo(const struct sockaddr *addr, socklen_t addrlen,
157 char *host, socklen_t hostlen,
158 char *serv, socklen_t servlen)
159 {
160 int flags = NI_NUMERICHOST | NI_NUMERICSERV;
161 int err = getnameinfo(addr, addrlen, host, hostlen, serv, servlen,
162 flags);
163
164 if (err) {
165 const char *errstr = getxinfo_strerr(err);
166
167 fprintf(stderr, "Fatal: getnameinfo: %s\n", errstr);
168 exit(1);
169 }
170 }
171
xgetaddrinfo(const char * node,const char * service,const struct addrinfo * hints,struct addrinfo ** res)172 static void xgetaddrinfo(const char *node, const char *service,
173 const struct addrinfo *hints,
174 struct addrinfo **res)
175 {
176 int err = getaddrinfo(node, service, hints, res);
177
178 if (err) {
179 const char *errstr = getxinfo_strerr(err);
180
181 fprintf(stderr, "Fatal: getaddrinfo(%s:%s): %s\n",
182 node ? node : "", service ? service : "", errstr);
183 exit(1);
184 }
185 }
186
set_rcvbuf(int fd,unsigned int size)187 static void set_rcvbuf(int fd, unsigned int size)
188 {
189 int err;
190
191 err = setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &size, sizeof(size));
192 if (err) {
193 perror("set SO_RCVBUF");
194 exit(1);
195 }
196 }
197
set_sndbuf(int fd,unsigned int size)198 static void set_sndbuf(int fd, unsigned int size)
199 {
200 int err;
201
202 err = setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &size, sizeof(size));
203 if (err) {
204 perror("set SO_SNDBUF");
205 exit(1);
206 }
207 }
208
set_mark(int fd,uint32_t mark)209 static void set_mark(int fd, uint32_t mark)
210 {
211 int err;
212
213 err = setsockopt(fd, SOL_SOCKET, SO_MARK, &mark, sizeof(mark));
214 if (err) {
215 perror("set SO_MARK");
216 exit(1);
217 }
218 }
219
set_transparent(int fd,int pf)220 static void set_transparent(int fd, int pf)
221 {
222 int one = 1;
223
224 switch (pf) {
225 case AF_INET:
226 if (-1 == setsockopt(fd, SOL_IP, IP_TRANSPARENT, &one, sizeof(one)))
227 perror("IP_TRANSPARENT");
228 break;
229 case AF_INET6:
230 if (-1 == setsockopt(fd, IPPROTO_IPV6, IPV6_TRANSPARENT, &one, sizeof(one)))
231 perror("IPV6_TRANSPARENT");
232 break;
233 }
234 }
235
do_ulp_so(int sock,const char * name)236 static int do_ulp_so(int sock, const char *name)
237 {
238 return setsockopt(sock, IPPROTO_TCP, TCP_ULP, name, strlen(name));
239 }
240
241 #define X(m) xerror("%s:%u: %s: failed for proto %d at line %u", __FILE__, __LINE__, (m), proto, line)
sock_test_tcpulp(int sock,int proto,unsigned int line)242 static void sock_test_tcpulp(int sock, int proto, unsigned int line)
243 {
244 socklen_t buflen = 8;
245 char buf[8] = "";
246 int ret = getsockopt(sock, IPPROTO_TCP, TCP_ULP, buf, &buflen);
247
248 if (ret != 0)
249 X("getsockopt");
250
251 if (buflen > 0) {
252 if (strcmp(buf, "mptcp") != 0)
253 xerror("unexpected ULP '%s' for proto %d at line %u", buf, proto, line);
254 ret = do_ulp_so(sock, "tls");
255 if (ret == 0)
256 X("setsockopt");
257 } else if (proto == IPPROTO_MPTCP) {
258 ret = do_ulp_so(sock, "tls");
259 if (ret != -1)
260 X("setsockopt");
261 }
262
263 ret = do_ulp_so(sock, "mptcp");
264 if (ret != -1)
265 X("setsockopt");
266
267 #undef X
268 }
269
270 #define SOCK_TEST_TCPULP(s, p) sock_test_tcpulp((s), (p), __LINE__)
271
sock_listen_mptcp(const char * const listenaddr,const char * const port)272 static int sock_listen_mptcp(const char * const listenaddr,
273 const char * const port)
274 {
275 int sock = -1;
276 struct addrinfo hints = {
277 .ai_protocol = IPPROTO_TCP,
278 .ai_socktype = SOCK_STREAM,
279 .ai_flags = AI_PASSIVE | AI_NUMERICHOST
280 };
281
282 hints.ai_family = pf;
283
284 struct addrinfo *a, *addr;
285 int one = 1;
286
287 xgetaddrinfo(listenaddr, port, &hints, &addr);
288 hints.ai_family = pf;
289
290 for (a = addr; a; a = a->ai_next) {
291 sock = socket(a->ai_family, a->ai_socktype, cfg_sock_proto);
292 if (sock < 0)
293 continue;
294
295 SOCK_TEST_TCPULP(sock, cfg_sock_proto);
296
297 if (-1 == setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &one,
298 sizeof(one)))
299 perror("setsockopt");
300
301 if (cfg_sockopt_types.transparent)
302 set_transparent(sock, pf);
303
304 if (bind(sock, a->ai_addr, a->ai_addrlen) == 0)
305 break; /* success */
306
307 perror("bind");
308 close(sock);
309 sock = -1;
310 }
311
312 freeaddrinfo(addr);
313
314 if (sock < 0) {
315 fprintf(stderr, "Could not create listen socket\n");
316 return sock;
317 }
318
319 SOCK_TEST_TCPULP(sock, cfg_sock_proto);
320
321 if (listen(sock, 20)) {
322 perror("listen");
323 close(sock);
324 return -1;
325 }
326
327 SOCK_TEST_TCPULP(sock, cfg_sock_proto);
328
329 return sock;
330 }
331
sock_connect_mptcp(const char * const remoteaddr,const char * const port,int proto,struct addrinfo ** peer)332 static int sock_connect_mptcp(const char * const remoteaddr,
333 const char * const port, int proto,
334 struct addrinfo **peer)
335 {
336 struct addrinfo hints = {
337 .ai_protocol = IPPROTO_TCP,
338 .ai_socktype = SOCK_STREAM,
339 };
340 struct addrinfo *a, *addr;
341 int sock = -1;
342
343 hints.ai_family = pf;
344
345 xgetaddrinfo(remoteaddr, port, &hints, &addr);
346 for (a = addr; a; a = a->ai_next) {
347 sock = socket(a->ai_family, a->ai_socktype, proto);
348 if (sock < 0) {
349 perror("socket");
350 continue;
351 }
352
353 SOCK_TEST_TCPULP(sock, proto);
354
355 if (cfg_mark)
356 set_mark(sock, cfg_mark);
357
358 if (connect(sock, a->ai_addr, a->ai_addrlen) == 0) {
359 *peer = a;
360 break; /* success */
361 }
362
363 perror("connect()");
364 close(sock);
365 sock = -1;
366 }
367
368 freeaddrinfo(addr);
369 if (sock != -1)
370 SOCK_TEST_TCPULP(sock, proto);
371 return sock;
372 }
373
do_rnd_write(const int fd,char * buf,const size_t len)374 static size_t do_rnd_write(const int fd, char *buf, const size_t len)
375 {
376 static bool first = true;
377 unsigned int do_w;
378 ssize_t bw;
379
380 do_w = rand() & 0xffff;
381 if (do_w == 0 || do_w > len)
382 do_w = len;
383
384 if (cfg_join && first && do_w > 100)
385 do_w = 100;
386
387 if (cfg_remove && do_w > cfg_do_w)
388 do_w = cfg_do_w;
389
390 bw = write(fd, buf, do_w);
391 if (bw < 0)
392 return bw;
393
394 /* let the join handshake complete, before going on */
395 if (cfg_join && first) {
396 usleep(200000);
397 first = false;
398 }
399
400 if (cfg_remove)
401 usleep(200000);
402
403 return bw;
404 }
405
do_write(const int fd,char * buf,const size_t len)406 static size_t do_write(const int fd, char *buf, const size_t len)
407 {
408 size_t offset = 0;
409
410 while (offset < len) {
411 size_t written;
412 ssize_t bw;
413
414 bw = write(fd, buf + offset, len - offset);
415 if (bw < 0) {
416 perror("write");
417 return 0;
418 }
419
420 written = (size_t)bw;
421 offset += written;
422 }
423
424 return offset;
425 }
426
process_cmsg(struct msghdr * msgh)427 static void process_cmsg(struct msghdr *msgh)
428 {
429 struct __kernel_timespec ts;
430 bool inq_found = false;
431 bool ts_found = false;
432 unsigned int inq = 0;
433 struct cmsghdr *cmsg;
434
435 for (cmsg = CMSG_FIRSTHDR(msgh); cmsg ; cmsg = CMSG_NXTHDR(msgh, cmsg)) {
436 if (cmsg->cmsg_level == SOL_SOCKET && cmsg->cmsg_type == SO_TIMESTAMPNS_NEW) {
437 memcpy(&ts, CMSG_DATA(cmsg), sizeof(ts));
438 ts_found = true;
439 continue;
440 }
441 if (cmsg->cmsg_level == IPPROTO_TCP && cmsg->cmsg_type == TCP_CM_INQ) {
442 memcpy(&inq, CMSG_DATA(cmsg), sizeof(inq));
443 inq_found = true;
444 continue;
445 }
446
447 }
448
449 if (cfg_cmsg_types.timestampns) {
450 if (!ts_found)
451 xerror("TIMESTAMPNS not present\n");
452 }
453
454 if (cfg_cmsg_types.tcp_inq) {
455 if (!inq_found)
456 xerror("TCP_INQ not present\n");
457
458 if (inq > 1024)
459 xerror("tcp_inq %u is larger than one kbyte\n", inq);
460 tcp_inq.last = inq;
461 }
462 }
463
do_recvmsg_cmsg(const int fd,char * buf,const size_t len)464 static ssize_t do_recvmsg_cmsg(const int fd, char *buf, const size_t len)
465 {
466 char msg_buf[8192];
467 struct iovec iov = {
468 .iov_base = buf,
469 .iov_len = len,
470 };
471 struct msghdr msg = {
472 .msg_iov = &iov,
473 .msg_iovlen = 1,
474 .msg_control = msg_buf,
475 .msg_controllen = sizeof(msg_buf),
476 };
477 int flags = 0;
478 unsigned int last_hint = tcp_inq.last;
479 int ret = recvmsg(fd, &msg, flags);
480
481 if (ret <= 0) {
482 if (ret == 0 && tcp_inq.expect_eof)
483 return ret;
484
485 if (ret == 0 && cfg_cmsg_types.tcp_inq)
486 if (last_hint != 1 && last_hint != 0)
487 xerror("EOF but last tcp_inq hint was %u\n", last_hint);
488
489 return ret;
490 }
491
492 if (tcp_inq.expect_eof)
493 xerror("expected EOF, last_hint %u, now %u\n",
494 last_hint, tcp_inq.last);
495
496 if (msg.msg_controllen && !cfg_cmsg_types.cmsg_enabled)
497 xerror("got %lu bytes of cmsg data, expected 0\n",
498 (unsigned long)msg.msg_controllen);
499
500 if (msg.msg_controllen == 0 && cfg_cmsg_types.cmsg_enabled)
501 xerror("%s\n", "got no cmsg data");
502
503 if (msg.msg_controllen)
504 process_cmsg(&msg);
505
506 if (cfg_cmsg_types.tcp_inq) {
507 if ((size_t)ret < len && last_hint > (unsigned int)ret) {
508 if (ret + 1 != (int)last_hint) {
509 int next = read(fd, msg_buf, sizeof(msg_buf));
510
511 xerror("read %u of %u, last_hint was %u tcp_inq hint now %u next_read returned %d/%m\n",
512 ret, (unsigned int)len, last_hint, tcp_inq.last, next);
513 } else {
514 tcp_inq.expect_eof = true;
515 }
516 }
517 }
518
519 return ret;
520 }
521
do_rnd_read(const int fd,char * buf,const size_t len)522 static ssize_t do_rnd_read(const int fd, char *buf, const size_t len)
523 {
524 int ret = 0;
525 char tmp[16384];
526 size_t cap = rand();
527
528 cap &= 0xffff;
529
530 if (cap == 0)
531 cap = 1;
532 else if (cap > len)
533 cap = len;
534
535 if (cfg_peek == CFG_WITH_PEEK) {
536 ret = recv(fd, buf, cap, MSG_PEEK);
537 ret = (ret < 0) ? ret : read(fd, tmp, ret);
538 } else if (cfg_peek == CFG_AFTER_PEEK) {
539 ret = recv(fd, buf, cap, MSG_PEEK);
540 ret = (ret < 0) ? ret : read(fd, buf, cap);
541 } else if (cfg_cmsg_types.cmsg_enabled) {
542 ret = do_recvmsg_cmsg(fd, buf, cap);
543 } else {
544 ret = read(fd, buf, cap);
545 }
546
547 return ret;
548 }
549
set_nonblock(int fd,bool nonblock)550 static void set_nonblock(int fd, bool nonblock)
551 {
552 int flags = fcntl(fd, F_GETFL);
553
554 if (flags == -1)
555 return;
556
557 if (nonblock)
558 fcntl(fd, F_SETFL, flags | O_NONBLOCK);
559 else
560 fcntl(fd, F_SETFL, flags & ~O_NONBLOCK);
561 }
562
shut_wr(int fd)563 static void shut_wr(int fd)
564 {
565 /* Close our write side, ev. give some time
566 * for address notification and/or checking
567 * the current status
568 */
569 if (cfg_wait)
570 usleep(cfg_wait);
571
572 shutdown(fd, SHUT_WR);
573 }
574
copyfd_io_poll(int infd,int peerfd,int outfd,bool * in_closed_after_out)575 static int copyfd_io_poll(int infd, int peerfd, int outfd, bool *in_closed_after_out)
576 {
577 struct pollfd fds = {
578 .fd = peerfd,
579 .events = POLLIN | POLLOUT,
580 };
581 unsigned int woff = 0, wlen = 0, total_wlen = 0, total_rlen = 0;
582 char wbuf[8192];
583
584 set_nonblock(peerfd, true);
585
586 for (;;) {
587 char rbuf[8192];
588 ssize_t len;
589
590 if (fds.events == 0)
591 break;
592
593 switch (poll(&fds, 1, poll_timeout)) {
594 case -1:
595 if (errno == EINTR)
596 continue;
597 perror("poll");
598 return 1;
599 case 0:
600 fprintf(stderr, "%s: poll timed out (events: "
601 "POLLIN %u, POLLOUT %u)\n", __func__,
602 fds.events & POLLIN, fds.events & POLLOUT);
603 return 2;
604 }
605
606 if (fds.revents & POLLIN) {
607 ssize_t rb = sizeof(rbuf);
608
609 /* limit the total amount of read data to the trunc value*/
610 if (cfg_truncate > 0) {
611 if (rb + total_rlen > cfg_truncate)
612 rb = cfg_truncate - total_rlen;
613 len = read(peerfd, rbuf, rb);
614 } else {
615 len = do_rnd_read(peerfd, rbuf, sizeof(rbuf));
616 }
617 if (len == 0) {
618 /* no more data to receive:
619 * peer has closed its write side
620 */
621 fds.events &= ~POLLIN;
622
623 if ((fds.events & POLLOUT) == 0) {
624 *in_closed_after_out = true;
625 /* and nothing more to send */
626 break;
627 }
628
629 /* Else, still have data to transmit */
630 } else if (len < 0) {
631 if (cfg_rcv_trunc)
632 return 0;
633 perror("read");
634 return 3;
635 }
636
637 total_rlen += len;
638 do_write(outfd, rbuf, len);
639 }
640
641 if (fds.revents & POLLOUT) {
642 if (wlen == 0) {
643 woff = 0;
644 wlen = read(infd, wbuf, sizeof(wbuf));
645 }
646
647 if (wlen > 0) {
648 ssize_t bw;
649
650 /* limit the total amount of written data to the trunc value */
651 if (cfg_truncate > 0 && wlen + total_wlen > cfg_truncate)
652 wlen = cfg_truncate - total_wlen;
653
654 bw = do_rnd_write(peerfd, wbuf + woff, wlen);
655 if (bw < 0) {
656 if (cfg_rcv_trunc)
657 return 0;
658 perror("write");
659 return 111;
660 }
661
662 woff += bw;
663 wlen -= bw;
664 total_wlen += bw;
665 } else if (wlen == 0) {
666 /* We have no more data to send. */
667 fds.events &= ~POLLOUT;
668
669 if ((fds.events & POLLIN) == 0)
670 /* ... and peer also closed already */
671 break;
672
673 shut_wr(peerfd);
674 } else {
675 if (errno == EINTR)
676 continue;
677 perror("read");
678 return 4;
679 }
680 }
681
682 if (fds.revents & (POLLERR | POLLNVAL)) {
683 if (cfg_rcv_trunc)
684 return 0;
685 fprintf(stderr, "Unexpected revents: "
686 "POLLERR/POLLNVAL(%x)\n", fds.revents);
687 return 5;
688 }
689
690 if (cfg_truncate > 0 && total_wlen >= cfg_truncate &&
691 total_rlen >= cfg_truncate)
692 break;
693 }
694
695 /* leave some time for late join/announce */
696 if (cfg_remove)
697 usleep(cfg_wait);
698
699 return 0;
700 }
701
do_recvfile(int infd,int outfd)702 static int do_recvfile(int infd, int outfd)
703 {
704 ssize_t r;
705
706 do {
707 char buf[16384];
708
709 r = do_rnd_read(infd, buf, sizeof(buf));
710 if (r > 0) {
711 if (write(outfd, buf, r) != r)
712 break;
713 } else if (r < 0) {
714 perror("read");
715 }
716 } while (r > 0);
717
718 return (int)r;
719 }
720
do_mmap(int infd,int outfd,unsigned int size)721 static int do_mmap(int infd, int outfd, unsigned int size)
722 {
723 char *inbuf = mmap(NULL, size, PROT_READ, MAP_SHARED, infd, 0);
724 ssize_t ret = 0, off = 0;
725 size_t rem;
726
727 if (inbuf == MAP_FAILED) {
728 perror("mmap");
729 return 1;
730 }
731
732 rem = size;
733
734 while (rem > 0) {
735 ret = write(outfd, inbuf + off, rem);
736
737 if (ret < 0) {
738 perror("write");
739 break;
740 }
741
742 off += ret;
743 rem -= ret;
744 }
745
746 munmap(inbuf, size);
747 return rem;
748 }
749
get_infd_size(int fd)750 static int get_infd_size(int fd)
751 {
752 struct stat sb;
753 ssize_t count;
754 int err;
755
756 err = fstat(fd, &sb);
757 if (err < 0) {
758 perror("fstat");
759 return -1;
760 }
761
762 if ((sb.st_mode & S_IFMT) != S_IFREG) {
763 fprintf(stderr, "%s: stdin is not a regular file\n", __func__);
764 return -2;
765 }
766
767 count = sb.st_size;
768 if (count > INT_MAX) {
769 fprintf(stderr, "File too large: %zu\n", count);
770 return -3;
771 }
772
773 return (int)count;
774 }
775
do_sendfile(int infd,int outfd,unsigned int count)776 static int do_sendfile(int infd, int outfd, unsigned int count)
777 {
778 while (count > 0) {
779 ssize_t r;
780
781 r = sendfile(outfd, infd, NULL, count);
782 if (r < 0) {
783 perror("sendfile");
784 return 3;
785 }
786
787 count -= r;
788 }
789
790 return 0;
791 }
792
copyfd_io_mmap(int infd,int peerfd,int outfd,unsigned int size,bool * in_closed_after_out)793 static int copyfd_io_mmap(int infd, int peerfd, int outfd,
794 unsigned int size, bool *in_closed_after_out)
795 {
796 int err;
797
798 if (listen_mode) {
799 err = do_recvfile(peerfd, outfd);
800 if (err)
801 return err;
802
803 err = do_mmap(infd, peerfd, size);
804 } else {
805 err = do_mmap(infd, peerfd, size);
806 if (err)
807 return err;
808
809 shut_wr(peerfd);
810
811 err = do_recvfile(peerfd, outfd);
812 *in_closed_after_out = true;
813 }
814
815 return err;
816 }
817
copyfd_io_sendfile(int infd,int peerfd,int outfd,unsigned int size,bool * in_closed_after_out)818 static int copyfd_io_sendfile(int infd, int peerfd, int outfd,
819 unsigned int size, bool *in_closed_after_out)
820 {
821 int err;
822
823 if (listen_mode) {
824 err = do_recvfile(peerfd, outfd);
825 if (err)
826 return err;
827
828 err = do_sendfile(infd, peerfd, size);
829 } else {
830 err = do_sendfile(infd, peerfd, size);
831 if (err)
832 return err;
833
834 shut_wr(peerfd);
835
836 err = do_recvfile(peerfd, outfd);
837 *in_closed_after_out = true;
838 }
839
840 return err;
841 }
842
copyfd_io(int infd,int peerfd,int outfd,bool close_peerfd)843 static int copyfd_io(int infd, int peerfd, int outfd, bool close_peerfd)
844 {
845 bool in_closed_after_out = false;
846 struct timespec start, end;
847 int file_size;
848 int ret;
849
850 if (cfg_time && (clock_gettime(CLOCK_MONOTONIC, &start) < 0))
851 xerror("can not fetch start time %d", errno);
852
853 switch (cfg_mode) {
854 case CFG_MODE_POLL:
855 ret = copyfd_io_poll(infd, peerfd, outfd, &in_closed_after_out);
856 break;
857
858 case CFG_MODE_MMAP:
859 file_size = get_infd_size(infd);
860 if (file_size < 0)
861 return file_size;
862 ret = copyfd_io_mmap(infd, peerfd, outfd, file_size, &in_closed_after_out);
863 break;
864
865 case CFG_MODE_SENDFILE:
866 file_size = get_infd_size(infd);
867 if (file_size < 0)
868 return file_size;
869 ret = copyfd_io_sendfile(infd, peerfd, outfd, file_size, &in_closed_after_out);
870 break;
871
872 default:
873 fprintf(stderr, "Invalid mode %d\n", cfg_mode);
874
875 die_usage();
876 return 1;
877 }
878
879 if (ret)
880 return ret;
881
882 if (close_peerfd)
883 close(peerfd);
884
885 if (cfg_time) {
886 unsigned int delta_ms;
887
888 if (clock_gettime(CLOCK_MONOTONIC, &end) < 0)
889 xerror("can not fetch end time %d", errno);
890 delta_ms = (end.tv_sec - start.tv_sec) * 1000 + (end.tv_nsec - start.tv_nsec) / 1000000;
891 if (delta_ms > cfg_time) {
892 xerror("transfer slower than expected! runtime %d ms, expected %d ms",
893 delta_ms, cfg_time);
894 }
895
896 /* show the runtime only if this end shutdown(wr) before receiving the EOF,
897 * (that is, if this end got the longer runtime)
898 */
899 if (in_closed_after_out)
900 fprintf(stderr, "%d", delta_ms);
901 }
902
903 return 0;
904 }
905
check_sockaddr(int pf,struct sockaddr_storage * ss,socklen_t salen)906 static void check_sockaddr(int pf, struct sockaddr_storage *ss,
907 socklen_t salen)
908 {
909 struct sockaddr_in6 *sin6;
910 struct sockaddr_in *sin;
911 socklen_t wanted_size = 0;
912
913 switch (pf) {
914 case AF_INET:
915 wanted_size = sizeof(*sin);
916 sin = (void *)ss;
917 if (!sin->sin_port)
918 fprintf(stderr, "accept: something wrong: ip connection from port 0");
919 break;
920 case AF_INET6:
921 wanted_size = sizeof(*sin6);
922 sin6 = (void *)ss;
923 if (!sin6->sin6_port)
924 fprintf(stderr, "accept: something wrong: ipv6 connection from port 0");
925 break;
926 default:
927 fprintf(stderr, "accept: Unknown pf %d, salen %u\n", pf, salen);
928 return;
929 }
930
931 if (salen != wanted_size)
932 fprintf(stderr, "accept: size mismatch, got %d expected %d\n",
933 (int)salen, wanted_size);
934
935 if (ss->ss_family != pf)
936 fprintf(stderr, "accept: pf mismatch, expect %d, ss_family is %d\n",
937 (int)ss->ss_family, pf);
938 }
939
check_getpeername(int fd,struct sockaddr_storage * ss,socklen_t salen)940 static void check_getpeername(int fd, struct sockaddr_storage *ss, socklen_t salen)
941 {
942 struct sockaddr_storage peerss;
943 socklen_t peersalen = sizeof(peerss);
944
945 if (getpeername(fd, (struct sockaddr *)&peerss, &peersalen) < 0) {
946 perror("getpeername");
947 return;
948 }
949
950 if (peersalen != salen) {
951 fprintf(stderr, "%s: %d vs %d\n", __func__, peersalen, salen);
952 return;
953 }
954
955 if (memcmp(ss, &peerss, peersalen)) {
956 char a[INET6_ADDRSTRLEN];
957 char b[INET6_ADDRSTRLEN];
958 char c[INET6_ADDRSTRLEN];
959 char d[INET6_ADDRSTRLEN];
960
961 xgetnameinfo((struct sockaddr *)ss, salen,
962 a, sizeof(a), b, sizeof(b));
963
964 xgetnameinfo((struct sockaddr *)&peerss, peersalen,
965 c, sizeof(c), d, sizeof(d));
966
967 fprintf(stderr, "%s: memcmp failure: accept %s vs peername %s, %s vs %s salen %d vs %d\n",
968 __func__, a, c, b, d, peersalen, salen);
969 }
970 }
971
check_getpeername_connect(int fd)972 static void check_getpeername_connect(int fd)
973 {
974 struct sockaddr_storage ss;
975 socklen_t salen = sizeof(ss);
976 char a[INET6_ADDRSTRLEN];
977 char b[INET6_ADDRSTRLEN];
978
979 if (getpeername(fd, (struct sockaddr *)&ss, &salen) < 0) {
980 perror("getpeername");
981 return;
982 }
983
984 xgetnameinfo((struct sockaddr *)&ss, salen,
985 a, sizeof(a), b, sizeof(b));
986
987 if (strcmp(cfg_host, a) || strcmp(cfg_port, b))
988 fprintf(stderr, "%s: %s vs %s, %s vs %s\n", __func__,
989 cfg_host, a, cfg_port, b);
990 }
991
maybe_close(int fd)992 static void maybe_close(int fd)
993 {
994 unsigned int r = rand();
995
996 if (!(cfg_join || cfg_remove || cfg_repeat > 1) && (r & 1))
997 close(fd);
998 }
999
main_loop_s(int listensock)1000 int main_loop_s(int listensock)
1001 {
1002 struct sockaddr_storage ss;
1003 struct pollfd polls;
1004 socklen_t salen;
1005 int remotesock;
1006 int fd = 0;
1007
1008 again:
1009 polls.fd = listensock;
1010 polls.events = POLLIN;
1011
1012 switch (poll(&polls, 1, poll_timeout)) {
1013 case -1:
1014 perror("poll");
1015 return 1;
1016 case 0:
1017 fprintf(stderr, "%s: timed out\n", __func__);
1018 close(listensock);
1019 return 2;
1020 }
1021
1022 salen = sizeof(ss);
1023 remotesock = accept(listensock, (struct sockaddr *)&ss, &salen);
1024 if (remotesock >= 0) {
1025 maybe_close(listensock);
1026 check_sockaddr(pf, &ss, salen);
1027 check_getpeername(remotesock, &ss, salen);
1028
1029 if (cfg_input) {
1030 fd = open(cfg_input, O_RDONLY);
1031 if (fd < 0)
1032 xerror("can't open %s: %d", cfg_input, errno);
1033 }
1034
1035 SOCK_TEST_TCPULP(remotesock, 0);
1036
1037 copyfd_io(fd, remotesock, 1, true);
1038 } else {
1039 perror("accept");
1040 return 1;
1041 }
1042
1043 if (--cfg_repeat > 0) {
1044 if (cfg_input)
1045 close(fd);
1046 goto again;
1047 }
1048
1049 return 0;
1050 }
1051
init_rng(void)1052 static void init_rng(void)
1053 {
1054 unsigned int foo;
1055
1056 if (getrandom(&foo, sizeof(foo), 0) == -1) {
1057 perror("getrandom");
1058 exit(1);
1059 }
1060
1061 srand(foo);
1062 }
1063
xsetsockopt(int fd,int level,int optname,const void * optval,socklen_t optlen)1064 static void xsetsockopt(int fd, int level, int optname, const void *optval, socklen_t optlen)
1065 {
1066 int err;
1067
1068 err = setsockopt(fd, level, optname, optval, optlen);
1069 if (err) {
1070 perror("setsockopt");
1071 exit(1);
1072 }
1073 }
1074
apply_cmsg_types(int fd,const struct cfg_cmsg_types * cmsg)1075 static void apply_cmsg_types(int fd, const struct cfg_cmsg_types *cmsg)
1076 {
1077 static const unsigned int on = 1;
1078
1079 if (cmsg->timestampns)
1080 xsetsockopt(fd, SOL_SOCKET, SO_TIMESTAMPNS_NEW, &on, sizeof(on));
1081 if (cmsg->tcp_inq)
1082 xsetsockopt(fd, IPPROTO_TCP, TCP_INQ, &on, sizeof(on));
1083 }
1084
parse_cmsg_types(const char * type)1085 static void parse_cmsg_types(const char *type)
1086 {
1087 char *next = strchr(type, ',');
1088 unsigned int len = 0;
1089
1090 cfg_cmsg_types.cmsg_enabled = 1;
1091
1092 if (next) {
1093 parse_cmsg_types(next + 1);
1094 len = next - type;
1095 } else {
1096 len = strlen(type);
1097 }
1098
1099 if (strncmp(type, "TIMESTAMPNS", len) == 0) {
1100 cfg_cmsg_types.timestampns = 1;
1101 return;
1102 }
1103
1104 if (strncmp(type, "TCPINQ", len) == 0) {
1105 cfg_cmsg_types.tcp_inq = 1;
1106 return;
1107 }
1108
1109 fprintf(stderr, "Unrecognized cmsg option %s\n", type);
1110 exit(1);
1111 }
1112
parse_setsock_options(const char * name)1113 static void parse_setsock_options(const char *name)
1114 {
1115 char *next = strchr(name, ',');
1116 unsigned int len = 0;
1117
1118 if (next) {
1119 parse_setsock_options(next + 1);
1120 len = next - name;
1121 } else {
1122 len = strlen(name);
1123 }
1124
1125 if (strncmp(name, "TRANSPARENT", len) == 0) {
1126 cfg_sockopt_types.transparent = 1;
1127 return;
1128 }
1129
1130 fprintf(stderr, "Unrecognized setsockopt option %s\n", name);
1131 exit(1);
1132 }
1133
xdisconnect(int fd,int addrlen)1134 void xdisconnect(int fd, int addrlen)
1135 {
1136 struct sockaddr_storage empty;
1137 int msec_sleep = 10;
1138 int queued = 1;
1139 int i;
1140
1141 shutdown(fd, SHUT_WR);
1142
1143 /* while until the pending data is completely flushed, the later
1144 * disconnect will bypass/ignore/drop any pending data.
1145 */
1146 for (i = 0; ; i += msec_sleep) {
1147 if (ioctl(fd, SIOCOUTQ, &queued) < 0)
1148 xerror("can't query out socket queue: %d", errno);
1149
1150 if (!queued)
1151 break;
1152
1153 if (i > poll_timeout)
1154 xerror("timeout while waiting for spool to complete");
1155 usleep(msec_sleep * 1000);
1156 }
1157
1158 memset(&empty, 0, sizeof(empty));
1159 empty.ss_family = AF_UNSPEC;
1160 if (connect(fd, (struct sockaddr *)&empty, addrlen) < 0)
1161 xerror("can't disconnect: %d", errno);
1162 }
1163
main_loop(void)1164 int main_loop(void)
1165 {
1166 int fd, ret, fd_in = 0;
1167 struct addrinfo *peer;
1168
1169 /* listener is ready. */
1170 fd = sock_connect_mptcp(cfg_host, cfg_port, cfg_sock_proto, &peer);
1171 if (fd < 0)
1172 return 2;
1173
1174 again:
1175 check_getpeername_connect(fd);
1176
1177 SOCK_TEST_TCPULP(fd, cfg_sock_proto);
1178
1179 if (cfg_rcvbuf)
1180 set_rcvbuf(fd, cfg_rcvbuf);
1181 if (cfg_sndbuf)
1182 set_sndbuf(fd, cfg_sndbuf);
1183 if (cfg_cmsg_types.cmsg_enabled)
1184 apply_cmsg_types(fd, &cfg_cmsg_types);
1185
1186 if (cfg_input) {
1187 fd_in = open(cfg_input, O_RDONLY);
1188 if (fd < 0)
1189 xerror("can't open %s:%d", cfg_input, errno);
1190 }
1191
1192 /* close the client socket open only if we are not going to reconnect */
1193 ret = copyfd_io(fd_in, fd, 1, 0);
1194 if (ret)
1195 return ret;
1196
1197 if (cfg_truncate > 0) {
1198 xdisconnect(fd, peer->ai_addrlen);
1199 } else if (--cfg_repeat > 0) {
1200 xdisconnect(fd, peer->ai_addrlen);
1201
1202 /* the socket could be unblocking at this point, we need the
1203 * connect to be blocking
1204 */
1205 set_nonblock(fd, false);
1206 if (connect(fd, peer->ai_addr, peer->ai_addrlen))
1207 xerror("can't reconnect: %d", errno);
1208 if (cfg_input)
1209 close(fd_in);
1210 goto again;
1211 } else {
1212 close(fd);
1213 }
1214
1215 return 0;
1216 }
1217
parse_proto(const char * proto)1218 int parse_proto(const char *proto)
1219 {
1220 if (!strcasecmp(proto, "MPTCP"))
1221 return IPPROTO_MPTCP;
1222 if (!strcasecmp(proto, "TCP"))
1223 return IPPROTO_TCP;
1224
1225 fprintf(stderr, "Unknown protocol: %s\n.", proto);
1226 die_usage();
1227
1228 /* silence compiler warning */
1229 return 0;
1230 }
1231
parse_mode(const char * mode)1232 int parse_mode(const char *mode)
1233 {
1234 if (!strcasecmp(mode, "poll"))
1235 return CFG_MODE_POLL;
1236 if (!strcasecmp(mode, "mmap"))
1237 return CFG_MODE_MMAP;
1238 if (!strcasecmp(mode, "sendfile"))
1239 return CFG_MODE_SENDFILE;
1240
1241 fprintf(stderr, "Unknown test mode: %s\n", mode);
1242 fprintf(stderr, "Supported modes are:\n");
1243 fprintf(stderr, "\t\t\"poll\" - interleaved read/write using poll()\n");
1244 fprintf(stderr, "\t\t\"mmap\" - send entire input file (mmap+write), then read response (-l will read input first)\n");
1245 fprintf(stderr, "\t\t\"sendfile\" - send entire input file (sendfile), then read response (-l will read input first)\n");
1246
1247 die_usage();
1248
1249 /* silence compiler warning */
1250 return 0;
1251 }
1252
parse_peek(const char * mode)1253 int parse_peek(const char *mode)
1254 {
1255 if (!strcasecmp(mode, "saveWithPeek"))
1256 return CFG_WITH_PEEK;
1257 if (!strcasecmp(mode, "saveAfterPeek"))
1258 return CFG_AFTER_PEEK;
1259
1260 fprintf(stderr, "Unknown: %s\n", mode);
1261 fprintf(stderr, "Supported MSG_PEEK mode are:\n");
1262 fprintf(stderr,
1263 "\t\t\"saveWithPeek\" - recv data with flags 'MSG_PEEK' and save the peek data into file\n");
1264 fprintf(stderr,
1265 "\t\t\"saveAfterPeek\" - read and save data into file after recv with flags 'MSG_PEEK'\n");
1266
1267 die_usage();
1268
1269 /* silence compiler warning */
1270 return 0;
1271 }
1272
parse_int(const char * size)1273 static int parse_int(const char *size)
1274 {
1275 unsigned long s;
1276
1277 errno = 0;
1278
1279 s = strtoul(size, NULL, 0);
1280
1281 if (errno) {
1282 fprintf(stderr, "Invalid sndbuf size %s (%s)\n",
1283 size, strerror(errno));
1284 die_usage();
1285 }
1286
1287 if (s > INT_MAX) {
1288 fprintf(stderr, "Invalid sndbuf size %s (%s)\n",
1289 size, strerror(ERANGE));
1290 die_usage();
1291 }
1292
1293 return (int)s;
1294 }
1295
parse_opts(int argc,char ** argv)1296 static void parse_opts(int argc, char **argv)
1297 {
1298 int c;
1299
1300 while ((c = getopt(argc, argv, "6c:f:hi:I:jlm:M:o:p:P:r:R:s:S:t:T:w:")) != -1) {
1301 switch (c) {
1302 case 'f':
1303 cfg_truncate = atoi(optarg);
1304
1305 /* when receiving a fastclose, ignore PIPE signals and
1306 * all the I/O errors later in the code
1307 */
1308 if (cfg_truncate < 0) {
1309 cfg_rcv_trunc = true;
1310 signal(SIGPIPE, handle_signal);
1311 }
1312 break;
1313 case 'j':
1314 cfg_join = true;
1315 cfg_mode = CFG_MODE_POLL;
1316 break;
1317 case 'r':
1318 cfg_remove = true;
1319 cfg_mode = CFG_MODE_POLL;
1320 cfg_wait = 400000;
1321 cfg_do_w = atoi(optarg);
1322 if (cfg_do_w <= 0)
1323 cfg_do_w = 50;
1324 break;
1325 case 'i':
1326 cfg_input = optarg;
1327 break;
1328 case 'I':
1329 cfg_repeat = atoi(optarg);
1330 break;
1331 case 'l':
1332 listen_mode = true;
1333 break;
1334 case 'p':
1335 cfg_port = optarg;
1336 break;
1337 case 's':
1338 cfg_sock_proto = parse_proto(optarg);
1339 break;
1340 case 'h':
1341 die_usage();
1342 break;
1343 case '6':
1344 pf = AF_INET6;
1345 break;
1346 case 't':
1347 poll_timeout = atoi(optarg) * 1000;
1348 if (poll_timeout <= 0)
1349 poll_timeout = -1;
1350 break;
1351 case 'T':
1352 cfg_time = atoi(optarg);
1353 break;
1354 case 'm':
1355 cfg_mode = parse_mode(optarg);
1356 break;
1357 case 'S':
1358 cfg_sndbuf = parse_int(optarg);
1359 break;
1360 case 'R':
1361 cfg_rcvbuf = parse_int(optarg);
1362 break;
1363 case 'w':
1364 cfg_wait = atoi(optarg)*1000000;
1365 break;
1366 case 'M':
1367 cfg_mark = strtol(optarg, NULL, 0);
1368 break;
1369 case 'P':
1370 cfg_peek = parse_peek(optarg);
1371 break;
1372 case 'c':
1373 parse_cmsg_types(optarg);
1374 break;
1375 case 'o':
1376 parse_setsock_options(optarg);
1377 break;
1378 }
1379 }
1380
1381 if (optind + 1 != argc)
1382 die_usage();
1383 cfg_host = argv[optind];
1384
1385 if (strchr(cfg_host, ':'))
1386 pf = AF_INET6;
1387 }
1388
main(int argc,char * argv[])1389 int main(int argc, char *argv[])
1390 {
1391 init_rng();
1392
1393 signal(SIGUSR1, handle_signal);
1394 parse_opts(argc, argv);
1395
1396 if (listen_mode) {
1397 int fd = sock_listen_mptcp(cfg_host, cfg_port);
1398
1399 if (fd < 0)
1400 return 1;
1401
1402 if (cfg_rcvbuf)
1403 set_rcvbuf(fd, cfg_rcvbuf);
1404 if (cfg_sndbuf)
1405 set_sndbuf(fd, cfg_sndbuf);
1406 if (cfg_mark)
1407 set_mark(fd, cfg_mark);
1408 if (cfg_cmsg_types.cmsg_enabled)
1409 apply_cmsg_types(fd, &cfg_cmsg_types);
1410
1411 return main_loop_s(fd);
1412 }
1413
1414 return main_loop();
1415 }
1416