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