1676084f21dfad95b764c74cef4997af46cc303d
[deb_libnfs.git] / lib / socket.c
1 /*
2 Copyright (C) 2010 by Ronnie Sahlberg <ronniesahlberg@gmail.com>
3
4 This program is free software; you can redistribute it and/or modify
5 it under the terms of the GNU Lesser General Public License as published by
6 the Free Software Foundation; either version 2.1 of the License, or
7 (at your option) any later version.
8
9 This program is distributed in the hope that it will be useful,
10 but WITHOUT ANY WARRANTY; without even the implied warranty of
11 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 GNU Lesser General Public License for more details.
13
14 You should have received a copy of the GNU Lesser General Public License
15 along with this program; if not, see <http://www.gnu.org/licenses/>.
16 */
17 #ifdef HAVE_CONFIG_H
18 #include "config.h"
19 #endif
20
21 #ifdef AROS
22 #include "aros_compat.h"
23 #endif
24
25 #ifdef WIN32
26 #include "win32_compat.h"
27 #endif
28
29 #ifdef HAVE_ARPA_INET_H
30 #include <arpa/inet.h>
31 #endif
32
33 #ifdef HAVE_POLL_H
34 #include <poll.h>
35 #endif
36
37 #ifdef HAVE_UNISTD_H
38 #include <unistd.h>
39 #endif
40
41 #ifdef HAVE_SYS_IOCTL_H
42 #include <sys/ioctl.h>
43 #endif
44
45 #ifdef HAVE_SYS_SOCKET_H
46 #include <sys/socket.h>
47 #endif
48
49 #ifdef HAVE_NETINET_TCP_H
50 #include <netinet/tcp.h>
51 #endif
52
53 #ifdef HAVE_NETDB_H
54 #include <netdb.h>
55 #endif
56
57 #ifdef HAVE_SYS_FILIO_H
58 #include <sys/filio.h>
59 #endif
60
61 #ifdef HAVE_SYS_SOCKIO_H
62 #include <sys/sockio.h>
63 #endif
64
65 #include <stdio.h>
66 #include <stdlib.h>
67 #include <assert.h>
68 #include <fcntl.h>
69 #include <string.h>
70 #include <errno.h>
71 #include <sys/types.h>
72 #include "libnfs-zdr.h"
73 #include "libnfs.h"
74 #include "libnfs-raw.h"
75 #include "libnfs-private.h"
76 #include "slist.h"
77
78 #ifdef WIN32
79 //has to be included after stdlib!!
80 #include "win32_errnowrapper.h"
81 #endif
82
83
84 static int rpc_reconnect_requeue(struct rpc_context *rpc);
85 static int rpc_connect_sockaddr_async(struct rpc_context *rpc, struct sockaddr_storage *s);
86
87 static void set_nonblocking(int fd)
88 {
89 int v = 0;
90 #if defined(WIN32)
91 long nonblocking=1;
92 v = ioctl(fd, FIONBIO, &nonblocking);
93 #else
94 v = fcntl(fd, F_GETFL, 0);
95 fcntl(fd, F_SETFL, v | O_NONBLOCK);
96 #endif //FIXME
97 }
98
99 #ifdef HAVE_NETINET_TCP_H
100 int set_tcp_sockopt(int sockfd, int optname, int value)
101 {
102 int level;
103
104 #if defined(__FreeBSD__) || defined(__sun) || (defined(__APPLE__) && defined(__MACH__))
105 struct protoent *buf;
106
107 if ((buf = getprotobyname("tcp")) != NULL)
108 level = buf->p_proto;
109 else
110 return -1;
111 #else
112 level = SOL_TCP;
113 #endif
114
115 return setsockopt(sockfd, level, optname, (char *)&value, sizeof(value));
116 }
117 #endif
118
119 int rpc_get_fd(struct rpc_context *rpc)
120 {
121 assert(rpc->magic == RPC_CONTEXT_MAGIC);
122
123 return rpc->fd;
124 }
125
126 int rpc_which_events(struct rpc_context *rpc)
127 {
128 int events;
129
130 assert(rpc->magic == RPC_CONTEXT_MAGIC);
131
132 events = rpc->is_connected ? POLLIN : POLLOUT;
133
134 if (rpc->is_udp != 0) {
135 /* for udp sockets we only wait for pollin */
136 return POLLIN;
137 }
138
139 if (rpc->outqueue) {
140 events |= POLLOUT;
141 }
142 return events;
143 }
144
145 static int rpc_write_to_socket(struct rpc_context *rpc)
146 {
147 int32_t count;
148
149 assert(rpc->magic == RPC_CONTEXT_MAGIC);
150
151 if (rpc->fd == -1) {
152 rpc_set_error(rpc, "trying to write but not connected");
153 return -1;
154 }
155
156 while (rpc->outqueue != NULL) {
157 int64_t total;
158
159 total = rpc->outqueue->outdata.size;
160
161 count = send(rpc->fd, rpc->outqueue->outdata.data + rpc->outqueue->written, total - rpc->outqueue->written, 0);
162 if (count == -1) {
163 if (errno == EAGAIN || errno == EWOULDBLOCK) {
164 return 0;
165 }
166 rpc_set_error(rpc, "Error when writing to socket :%s(%d)", strerror(errno), errno);
167 return -1;
168 }
169
170 rpc->outqueue->written += count;
171 if (rpc->outqueue->written == total) {
172 struct rpc_pdu *pdu = rpc->outqueue;
173
174 SLIST_REMOVE(&rpc->outqueue, pdu);
175 SLIST_ADD_END(&rpc->waitpdu, pdu);
176 }
177 }
178 return 0;
179 }
180
181 static int rpc_read_from_socket(struct rpc_context *rpc)
182 {
183 int available;
184 int size;
185 int pdu_size;
186 int32_t count;
187
188 assert(rpc->magic == RPC_CONTEXT_MAGIC);
189
190 if (ioctl(rpc->fd, FIONREAD, &available) != 0) {
191 rpc_set_error(rpc, "Ioctl FIONREAD returned error : %d. Closing socket.", errno);
192 return -1;
193 }
194
195 if (available == 0) {
196 rpc_set_error(rpc, "Socket has been closed");
197 return -1;
198 }
199
200 if (rpc->is_udp) {
201 char *buf;
202 socklen_t socklen = sizeof(rpc->udp_src);
203
204 buf = malloc(available);
205 if (buf == NULL) {
206 rpc_set_error(rpc, "Failed to malloc buffer for recvfrom");
207 return -1;
208 }
209 count = recvfrom(rpc->fd, buf, available, MSG_DONTWAIT, (struct sockaddr *)&rpc->udp_src, &socklen);
210 if (count < 0) {
211 rpc_set_error(rpc, "Failed recvfrom: %s", strerror(errno));
212 free(buf);
213 }
214 if (rpc_process_pdu(rpc, buf, count) != 0) {
215 rpc_set_error(rpc, "Invalid/garbage pdu received from server. Ignoring PDU");
216 free(buf);
217 return -1;
218 }
219 free(buf);
220 return 0;
221 }
222
223 /* read record marker, 4 bytes at the beginning of every pdu */
224 if (rpc->inbuf == NULL) {
225 rpc->insize = 4;
226 rpc->inbuf = malloc(rpc->insize);
227 if (rpc->inbuf == NULL) {
228 rpc_set_error(rpc, "Failed to allocate buffer for record marker, errno:%d. Closing socket.", errno);
229 return -1;
230 }
231 }
232 if (rpc->inpos < 4) {
233 size = 4 - rpc->inpos;
234
235 count = recv(rpc->fd, rpc->inbuf + rpc->inpos, size, 0);
236 if (count == -1) {
237 if (errno == EINTR) {
238 return 0;
239 }
240 rpc_set_error(rpc, "Read from socket failed, errno:%d. Closing socket.", errno);
241 return -1;
242 }
243 available -= count;
244 rpc->inpos += count;
245 }
246
247 if (available == 0) {
248 return 0;
249 }
250
251 pdu_size = rpc_get_pdu_size(rpc->inbuf);
252 if (rpc->insize < pdu_size) {
253 unsigned char *buf;
254
255 buf = malloc(pdu_size);
256 if (buf == NULL) {
257 rpc_set_error(rpc, "Failed to allocate buffer of %d bytes for pdu, errno:%d. Closing socket.", pdu_size, errno);
258 return -1;
259 }
260 memcpy(buf, rpc->inbuf, rpc->insize);
261 free(rpc->inbuf);
262 rpc->inbuf = buf;
263 rpc->insize = rpc_get_pdu_size(rpc->inbuf);
264 }
265
266 size = available;
267 if (size > rpc->insize - rpc->inpos) {
268 size = rpc->insize - rpc->inpos;
269 }
270
271 count = recv(rpc->fd, rpc->inbuf + rpc->inpos, size, 0);
272 if (count == -1) {
273 if (errno == EINTR) {
274 return 0;
275 }
276 rpc_set_error(rpc, "Read from socket failed, errno:%d. Closing socket.", errno);
277 return -1;
278 }
279 available -= count;
280 rpc->inpos += count;
281
282 if (rpc->inpos == rpc->insize) {
283 char *buf = rpc->inbuf;
284
285 rpc->inbuf = NULL;
286 rpc->insize = 0;
287 rpc->inpos = 0;
288
289 if (rpc_process_pdu(rpc, buf, pdu_size) != 0) {
290 rpc_set_error(rpc, "Invalid/garbage pdu received from server. Closing socket");
291 return -1;
292 }
293 free(buf);
294 }
295
296 return 0;
297 }
298
299
300
301 int rpc_service(struct rpc_context *rpc, int revents)
302 {
303 assert(rpc->magic == RPC_CONTEXT_MAGIC);
304
305 if (revents & POLLERR) {
306 #ifdef WIN32
307 char err = 0;
308 #else
309 int err = 0;
310 #endif
311 socklen_t err_size = sizeof(err);
312
313 if (getsockopt(rpc->fd, SOL_SOCKET, SO_ERROR,
314 (char *)&err, &err_size) != 0 || err != 0) {
315 if (err == 0) {
316 err = errno;
317 }
318 rpc_set_error(rpc, "rpc_service: socket error "
319 "%s(%d).",
320 strerror(err), err);
321 } else {
322 rpc_set_error(rpc, "rpc_service: POLLERR, "
323 "Unknown socket error.");
324 }
325 if (rpc->connect_cb != NULL) {
326 rpc->connect_cb(rpc, RPC_STATUS_ERROR, rpc->error_string, rpc->connect_data);
327 }
328 return -1;
329 }
330 if (revents & POLLHUP) {
331 rpc_set_error(rpc, "Socket failed with POLLHUP");
332 if (rpc->connect_cb != NULL) {
333 rpc->connect_cb(rpc, RPC_STATUS_ERROR, rpc->error_string, rpc->connect_data);
334 }
335 return -1;
336 }
337
338 if (rpc->is_connected == 0 && rpc->fd != -1 && revents&POLLOUT) {
339 int err = 0;
340 socklen_t err_size = sizeof(err);
341
342 if (getsockopt(rpc->fd, SOL_SOCKET, SO_ERROR,
343 (char *)&err, &err_size) != 0 || err != 0) {
344 if (err == 0) {
345 err = errno;
346 }
347 rpc_set_error(rpc, "rpc_service: socket error "
348 "%s(%d) while connecting.",
349 strerror(err), err);
350 if (rpc->connect_cb != NULL) {
351 rpc->connect_cb(rpc, RPC_STATUS_ERROR,
352 NULL, rpc->connect_data);
353 }
354 return -1;
355 }
356
357 rpc->is_connected = 1;
358 if (rpc->connect_cb != NULL) {
359 rpc->connect_cb(rpc, RPC_STATUS_SUCCESS, NULL, rpc->connect_data);
360 }
361 return 0;
362 }
363
364 if (revents & POLLIN) {
365 if (rpc_read_from_socket(rpc) != 0) {
366 rpc_reconnect_requeue(rpc);
367 return 0;
368 }
369 }
370
371 if (revents & POLLOUT && rpc->outqueue != NULL) {
372 if (rpc_write_to_socket(rpc) != 0) {
373 rpc_set_error(rpc, "write to socket failed");
374 return -1;
375 }
376 }
377
378 return 0;
379 }
380
381 void rpc_set_autoreconnect(struct rpc_context *rpc)
382 {
383 assert(rpc->magic == RPC_CONTEXT_MAGIC);
384
385 rpc->auto_reconnect = 1;
386 }
387
388 void rpc_unset_autoreconnect(struct rpc_context *rpc)
389 {
390 assert(rpc->magic == RPC_CONTEXT_MAGIC);
391
392 rpc->auto_reconnect = 0;
393 }
394
395 void rpc_set_tcp_syncnt(struct rpc_context *rpc, int v)
396 {
397 assert(rpc->magic == RPC_CONTEXT_MAGIC);
398
399 rpc->tcp_syncnt = v;
400 }
401
402 static int rpc_connect_sockaddr_async(struct rpc_context *rpc, struct sockaddr_storage *s)
403 {
404 int socksize;
405
406 assert(rpc->magic == RPC_CONTEXT_MAGIC);
407
408 switch (s->ss_family) {
409 case AF_INET:
410 socksize = sizeof(struct sockaddr_in);
411 rpc->fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
412 #ifdef HAVE_NETINET_TCP_H
413 if (rpc->tcp_syncnt != RPC_PARAM_UNDEFINED) {
414 set_tcp_sockopt(rpc->fd, TCP_SYNCNT, rpc->tcp_syncnt);
415 }
416 #endif
417 break;
418 default:
419 rpc_set_error(rpc, "Can not handle AF_FAMILY:%d", s->ss_family);
420 return -1;
421 }
422
423 if (rpc->fd == -1) {
424 rpc_set_error(rpc, "Failed to open socket");
425 return -1;
426 }
427
428 /* Some systems allow you to set capabilities on an executable
429 * to allow the file to be executed with privilege to bind to
430 * privileged system ports, even if the user is not root.
431 *
432 * Opportunistically try to bind the socket to a low numbered
433 * system port in the hope that the user is either root or the
434 * executable has the CAP_NET_BIND_SERVICE.
435 *
436 * As soon as we fail the bind() with EACCES we know we will never
437 * be able to bind to a system port so we terminate the loop.
438 *
439 * On linux, use
440 * sudo setcap 'cap_net_bind_service=+ep' /path/executable
441 * to make the executable able to bind to a system port.
442 *
443 * On Windows, there is no concept of privileged ports. Thus
444 * binding will usually succeed.
445 */
446 {
447 struct sockaddr_in sin;
448 static int portOfs = 0;
449 const int firstPort = 512; /* >= 512 according to Sun docs */
450 const int portCount = IPPORT_RESERVED - firstPort;
451 int startOfs, port, rc;
452
453 if (portOfs == 0) {
454 portOfs = time(NULL) % 400;
455 }
456 startOfs = portOfs;
457 do {
458 rc = -1;
459 port = htons(firstPort + portOfs);
460 portOfs = (portOfs + 1) % portCount;
461
462 /* skip well-known ports */
463 if (!getservbyport(port, "tcp")) {
464 memset(&sin, 0, sizeof(sin));
465 sin.sin_port = port;
466 sin.sin_family = AF_INET;
467 sin.sin_addr.s_addr = 0;
468
469 rc = bind(rpc->fd, (struct sockaddr *)&sin, sizeof(struct sockaddr_in));
470 #if !defined(WIN32)
471 /* we got EACCES, so don't try again */
472 if (rc != 0 && errno == EACCES)
473 break;
474 #endif
475 }
476 } while (rc != 0 && portOfs != startOfs);
477 }
478
479 set_nonblocking(rpc->fd);
480
481 if (connect(rpc->fd, (struct sockaddr *)s, socksize) != 0 && errno != EINPROGRESS) {
482 rpc_set_error(rpc, "connect() to server failed. %s(%d)", strerror(errno), errno);
483 return -1;
484 }
485
486 return 0;
487 }
488
489 int rpc_connect_async(struct rpc_context *rpc, const char *server, int port, rpc_cb cb, void *private_data)
490 {
491 struct sockaddr_in *sin = (struct sockaddr_in *)&rpc->s;
492
493 assert(rpc->magic == RPC_CONTEXT_MAGIC);
494
495 if (rpc->fd != -1) {
496 rpc_set_error(rpc, "Trying to connect while already connected");
497 return -1;
498 }
499
500 if (rpc->is_udp != 0) {
501 rpc_set_error(rpc, "Trying to connect on UDP socket");
502 return -1;
503 }
504
505 rpc->auto_reconnect = 0;
506
507 sin->sin_family = AF_INET;
508 sin->sin_port = htons(port);
509 if (inet_pton(AF_INET, server, &sin->sin_addr) != 1) {
510 rpc_set_error(rpc, "Not a valid server ip address");
511 return -1;
512 }
513
514
515 switch (rpc->s.ss_family) {
516 case AF_INET:
517 #ifdef HAVE_SOCKADDR_LEN
518 sin->sin_len = sizeof(struct sockaddr_in);
519 #endif
520 break;
521 }
522
523 rpc->connect_cb = cb;
524 rpc->connect_data = private_data;
525
526 if (rpc_connect_sockaddr_async(rpc, &rpc->s) != 0) {
527 return -1;
528 }
529
530 return 0;
531 }
532
533 int rpc_disconnect(struct rpc_context *rpc, char *error)
534 {
535 assert(rpc->magic == RPC_CONTEXT_MAGIC);
536
537 rpc_unset_autoreconnect(rpc);
538
539 if (rpc->fd != -1) {
540 close(rpc->fd);
541 }
542 rpc->fd = -1;
543
544 rpc->is_connected = 0;
545
546 rpc_error_all_pdus(rpc, error);
547
548 return 0;
549 }
550
551 static void reconnect_cb(struct rpc_context *rpc, int status, void *data _U_, void *private_data)
552 {
553 assert(rpc->magic == RPC_CONTEXT_MAGIC);
554
555 if (status != RPC_STATUS_SUCCESS) {
556 rpc_error_all_pdus(rpc, "RPC ERROR: Failed to reconnect async");
557 return;
558 }
559
560 rpc->is_connected = 1;
561 rpc->connect_cb = NULL;
562 }
563
564 /* disconnect but do not error all PDUs, just move pdus in-flight back to the outqueue and reconnect */
565 static int rpc_reconnect_requeue(struct rpc_context *rpc)
566 {
567 struct rpc_pdu *pdu;
568
569 assert(rpc->magic == RPC_CONTEXT_MAGIC);
570
571 if (rpc->fd != -1) {
572 close(rpc->fd);
573 }
574 rpc->fd = -1;
575
576 rpc->is_connected = 0;
577
578 /* socket is closed so we will not get any replies to any commands
579 * in flight. Move them all over from the waitpdu queue back to the out queue
580 */
581 for (pdu=rpc->waitpdu; pdu; pdu=pdu->next) {
582 SLIST_REMOVE(&rpc->waitpdu, pdu);
583 SLIST_ADD(&rpc->outqueue, pdu);
584 /* we have to re-send the whole pdu again */
585 pdu->written = 0;
586 }
587
588 if (rpc->auto_reconnect != 0) {
589 rpc->connect_cb = reconnect_cb;
590
591 if (rpc_connect_sockaddr_async(rpc, &rpc->s) != 0) {
592 rpc_error_all_pdus(rpc, "RPC ERROR: Failed to reconnect async");
593 return -1;
594 }
595 }
596
597 return 0;
598 }
599
600
601 int rpc_bind_udp(struct rpc_context *rpc, char *addr, int port)
602 {
603 struct addrinfo *ai = NULL;
604 char service[6];
605
606 assert(rpc->magic == RPC_CONTEXT_MAGIC);
607
608 if (rpc->is_udp == 0) {
609 rpc_set_error(rpc, "Cant not bind UDP. Not UDP context");
610 return -1;
611 }
612
613 sprintf(service, "%d", port);
614 if (getaddrinfo(addr, service, NULL, &ai) != 0) {
615 rpc_set_error(rpc, "Invalid address:%s. "
616 "Can not resolv into IPv4/v6 structure.");
617 return -1;
618 }
619
620 switch(ai->ai_family) {
621 case AF_INET:
622 rpc->fd = socket(ai->ai_family, SOCK_DGRAM, 0);
623 if (rpc->fd == -1) {
624 rpc_set_error(rpc, "Failed to create UDP socket: %s", strerror(errno));
625 freeaddrinfo(ai);
626 return -1;
627 }
628
629 if (bind(rpc->fd, (struct sockaddr *)ai->ai_addr, sizeof(struct sockaddr_in)) != 0) {
630 rpc_set_error(rpc, "Failed to bind to UDP socket: %s",strerror(errno));
631 freeaddrinfo(ai);
632 return -1;
633 }
634 break;
635 default:
636 rpc_set_error(rpc, "Can not handle UPD sockets of family %d yet", ai->ai_family);
637 freeaddrinfo(ai);
638 return -1;
639 }
640
641 freeaddrinfo(ai);
642
643 return 0;
644 }
645
646 int rpc_set_udp_destination(struct rpc_context *rpc, char *addr, int port, int is_broadcast)
647 {
648 struct addrinfo *ai = NULL;
649 char service[6];
650
651 assert(rpc->magic == RPC_CONTEXT_MAGIC);
652
653 if (rpc->is_udp == 0) {
654 rpc_set_error(rpc, "Can not set destination sockaddr. Not UDP context");
655 return -1;
656 }
657
658 sprintf(service, "%d", port);
659 if (getaddrinfo(addr, service, NULL, &ai) != 0) {
660 rpc_set_error(rpc, "Invalid address:%s. "
661 "Can not resolv into IPv4/v6 structure.");
662 return -1;
663 }
664
665 if (rpc->udp_dest) {
666 free(rpc->udp_dest);
667 rpc->udp_dest = NULL;
668 }
669 rpc->udp_dest = malloc(ai->ai_addrlen);
670 if (rpc->udp_dest == NULL) {
671 rpc_set_error(rpc, "Out of memory. Failed to allocate sockaddr structure");
672 freeaddrinfo(ai);
673 return -1;
674 }
675 memcpy(rpc->udp_dest, ai->ai_addr, ai->ai_addrlen);
676 freeaddrinfo(ai);
677
678 rpc->is_broadcast = is_broadcast;
679 setsockopt(rpc->fd, SOL_SOCKET, SO_BROADCAST, (char *)&is_broadcast, sizeof(is_broadcast));
680
681 return 0;
682 }
683
684 struct sockaddr *rpc_get_recv_sockaddr(struct rpc_context *rpc)
685 {
686 assert(rpc->magic == RPC_CONTEXT_MAGIC);
687
688 return (struct sockaddr *)&rpc->udp_src;
689 }
690
691 int rpc_queue_length(struct rpc_context *rpc)
692 {
693 int i=0;
694 struct rpc_pdu *pdu;
695
696 assert(rpc->magic == RPC_CONTEXT_MAGIC);
697
698 for(pdu = rpc->outqueue; pdu; pdu = pdu->next) {
699 i++;
700 }
701 for(pdu = rpc->waitpdu; pdu; pdu = pdu->next) {
702 i++;
703 }
704 return i;
705 }