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