sbin/ggate/ggated/ggated.c

   1 /*-
   2  * SPDX-License-Identifier: BSD-2-Clause-FreeBSD
   3  *
   4  * Copyright (c) 2004 Pawel Jakub Dawidek <pjd@FreeBSD.org>
   5  * All rights reserved.
   6  *
   7  * Redistribution and use in source and binary forms, with or without
   8  * modification, are permitted provided that the following conditions
   9  * are met:
  10  * 1. Redistributions of source code must retain the above copyright
  11  *    notice, this list of conditions and the following disclaimer.
  12  * 2. Redistributions in binary form must reproduce the above copyright
  13  *    notice, this list of conditions and the following disclaimer in the
  14  *    documentation and/or other materials provided with the distribution.
  15  *
  16  * THIS SOFTWARE IS PROVIDED BY THE AUTHORS AND CONTRIBUTORS ``AS IS'' AND
  17  * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  18  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  19  * ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHORS OR CONTRIBUTORS BE LIABLE
  20  * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
  21  * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
  22  * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
  23  * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  24  * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
  25  * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
  26  * SUCH DAMAGE.
  27  *
  28  * $FreeBSD$
  29  */
  30
  31 #include <sys/param.h>
  32 #include <sys/bio.h>
  33 #include <sys/disk.h>
  34 #include <sys/endian.h>
  35 #include <sys/ioctl.h>
  36 #include <sys/queue.h>
  37 #include <sys/socket.h>
  38 #include <sys/stat.h>
  39 #include <sys/time.h>
  40 #include <arpa/inet.h>
  41 #include <netinet/in.h>
  42 #include <netinet/tcp.h>
  43 #include <assert.h>
  44 #include <err.h>
  45 #include <errno.h>
  46 #include <fcntl.h>
  47 #include <libgen.h>
  48 #include <libutil.h>
  49 #include <paths.h>
  50 #include <pthread.h>
  51 #include <signal.h>
  52 #include <stdarg.h>
  53 #include <stdio.h>
  54 #include <stdlib.h>
  55 #include <stdint.h>
  56 #include <string.h>
  57 #include <syslog.h>
  58 #include <unistd.h>
  59
  60 #include "ggate.h"
  61
  62
  63 #define GGATED_EXPORT_FILE      "/etc/gg.exports"
  64
  65 struct ggd_connection {
  66         off_t            c_mediasize;
  67         unsigned         c_sectorsize;
  68         unsigned         c_flags;       /* flags (RO/RW) */
  69         int              c_diskfd;
  70         int              c_sendfd;
  71         int              c_recvfd;
  72         time_t           c_birthtime;
  73         char            *c_path;
  74         uint64_t         c_token;
  75         in_addr_t        c_srcip;
  76         LIST_ENTRY(ggd_connection) c_next;
  77 };
  78
  79 struct ggd_request {
  80         struct g_gate_hdr        r_hdr;
  81         char                    *r_data;
  82         TAILQ_ENTRY(ggd_request) r_next;
  83 };
  84 #define r_cmd           r_hdr.gh_cmd
  85 #define r_offset        r_hdr.gh_offset
  86 #define r_length        r_hdr.gh_length
  87 #define r_error         r_hdr.gh_error
  88
  89 struct ggd_export {
  90         char            *e_path;        /* path to device/file */
  91         in_addr_t        e_ip;          /* remote IP address */
  92         in_addr_t        e_mask;        /* IP mask */
  93         unsigned         e_flags;       /* flags (RO/RW) */
  94         SLIST_ENTRY(ggd_export) e_next;
  95 };
  96
  97 static const char *exports_file = GGATED_EXPORT_FILE;
  98 static int got_sighup = 0;
  99 static in_addr_t bindaddr;
 100
 101 static TAILQ_HEAD(, ggd_request) inqueue = TAILQ_HEAD_INITIALIZER(inqueue);
 102 static TAILQ_HEAD(, ggd_request) outqueue = TAILQ_HEAD_INITIALIZER(outqueue);
 103 static pthread_mutex_t inqueue_mtx, outqueue_mtx;
 104 static pthread_cond_t inqueue_cond, outqueue_cond;
 105
 106 static SLIST_HEAD(, ggd_export) exports = SLIST_HEAD_INITIALIZER(exports);
 107 static LIST_HEAD(, ggd_connection) connections = LIST_HEAD_INITIALIZER(connections);
 108
 109 static void *recv_thread(void *arg);
 110 static void *disk_thread(void *arg);
 111 static void *send_thread(void *arg);
 112
 113 static void
 114 usage(void)
 115 {
 116
 117         fprintf(stderr, "usage: %s [-nv] [-a address] [-F pidfile] [-p port] "
 118             "[-R rcvbuf] [-S sndbuf] [exports file]\n", getprogname());
 119         exit(EXIT_FAILURE);
 120 }
 121
 122 static char *
 123 ip2str(in_addr_t ip)
 124 {
 125         static char sip[16];
 126
 127         snprintf(sip, sizeof(sip), "%u.%u.%u.%u",
 128             ((ip >> 24) & 0xff),
 129             ((ip >> 16) & 0xff),
 130             ((ip >> 8) & 0xff),
 131             (ip & 0xff));
 132         return (sip);
 133 }
 134
 135 static in_addr_t
 136 countmask(unsigned m)
 137 {
 138         in_addr_t mask;
 139
 140         if (m == 0) {
 141                 mask = 0x0;
 142         } else {
 143                 mask = 1 << (32 - m);
 144                 mask--;
 145                 mask = ~mask;
 146         }
 147         return (mask);
 148 }
 149
 150 static void
 151 line_parse(char *line, unsigned lineno)
 152 {
 153         struct ggd_export *ex;
 154         char *word, *path, *sflags;
 155         unsigned flags, i, vmask;
 156         in_addr_t ip, mask;
 157
 158         ip = mask = flags = vmask = 0;
 159         path = NULL;
 160         sflags = NULL;
 161
 162         for (i = 0, word = strtok(line, " \t"); word != NULL;
 163             i++, word = strtok(NULL, " \t")) {
 164                 switch (i) {
 165                 case 0: /* IP address or host name */
 166                         ip = g_gate_str2ip(strsep(&word, "/"));
 167                         if (ip == INADDR_NONE) {
 168                                 g_gate_xlog("Invalid IP/host name at line %u.",
 169                                     lineno);
 170                         }
 171                         ip = ntohl(ip);
 172                         if (word == NULL)
 173                                 vmask = 32;
 174                         else {
 175                                 errno = 0;
 176                                 vmask = strtoul(word, NULL, 10);
 177                                 if (vmask == 0 && errno != 0) {
 178                                         g_gate_xlog("Invalid IP mask value at "
 179                                             "line %u.", lineno);
 180                                 }
 181                                 if ((unsigned)vmask > 32) {
 182                                         g_gate_xlog("Invalid IP mask value at line %u.",
 183                                             lineno);
 184                                 }
 185                         }
 186                         mask = countmask(vmask);
 187                         break;
 188                 case 1: /* flags */
 189                         if (strcasecmp("rd", word) == 0 ||
 190                             strcasecmp("ro", word) == 0) {
 191                                 flags = O_RDONLY;
 192                         } else if (strcasecmp("wo", word) == 0) {
 193                                 flags = O_WRONLY;
 194                         } else if (strcasecmp("rw", word) == 0) {
 195                                 flags = O_RDWR;
 196                         } else {
 197                                 g_gate_xlog("Invalid value in flags field at "
 198                                     "line %u.", lineno);
 199                         }
 200                         sflags = word;
 201                         break;
 202                 case 2: /* path */
 203                         if (strlen(word) >= MAXPATHLEN) {
 204                                 g_gate_xlog("Path too long at line %u. ",
 205                                     lineno);
 206                         }
 207                         path = word;
 208                         break;
 209                 default:
 210                         g_gate_xlog("Too many arguments at line %u. ", lineno);
 211                 }
 212         }
 213         if (i != 3)
 214                 g_gate_xlog("Too few arguments at line %u.", lineno);
 215
 216         ex = malloc(sizeof(*ex));
 217         if (ex == NULL)
 218                 g_gate_xlog("Not enough memory.");
 219         ex->e_path = strdup(path);
 220         if (ex->e_path == NULL)
 221                 g_gate_xlog("Not enough memory.");
 222
 223         /* Made 'and' here. */
 224         ex->e_ip = (ip & mask);
 225         ex->e_mask = mask;
 226         ex->e_flags = flags;
 227
 228         SLIST_INSERT_HEAD(&exports, ex, e_next);
 229
 230         g_gate_log(LOG_DEBUG, "Added %s/%u %s %s to exports list.",
 231             ip2str(ex->e_ip), vmask, path, sflags);
 232 }
 233
 234 static void
 235 exports_clear(void)
 236 {
 237         struct ggd_export *ex;
 238
 239         while (!SLIST_EMPTY(&exports)) {
 240                 ex = SLIST_FIRST(&exports);
 241                 SLIST_REMOVE_HEAD(&exports, e_next);
 242                 free(ex);
 243         }
 244 }
 245
 246 #define EXPORTS_LINE_SIZE       2048
 247 static void
 248 exports_get(void)
 249 {
 250         char buf[EXPORTS_LINE_SIZE], *line;
 251         unsigned lineno = 0, objs = 0, len;
 252         FILE *fd;
 253
 254         exports_clear();
 255
 256         fd = fopen(exports_file, "r");
 257         if (fd == NULL) {
 258                 g_gate_xlog("Cannot open exports file (%s): %s.", exports_file,
 259                     strerror(errno));
 260         }
 261
 262         g_gate_log(LOG_INFO, "Reading exports file (%s).", exports_file);
 263
 264         for (;;) {
 265                 if (fgets(buf, sizeof(buf), fd) == NULL) {
 266                         if (feof(fd))
 267                                 break;
 268
 269                         g_gate_xlog("Error while reading exports file: %s.",
 270                             strerror(errno));
 271                 }
 272
 273                 /* Increase line count. */
 274                 lineno++;
 275
 276                 /* Skip spaces and tabs. */
 277                 for (line = buf; *line == ' ' || *line == '\t'; ++line)
 278                         ;
 279
 280                 /* Empty line, comment or empty line at the end of file. */
 281                 if (*line == '\n' || *line == '#' || *line == '\0')
 282                         continue;
 283
 284                 len = strlen(line);
 285                 if (line[len - 1] == '\n') {
 286                         /* Remove new line char. */
 287                         line[len - 1] = '\0';
 288                 } else {
 289                         if (!feof(fd))
 290                                 g_gate_xlog("Line %u too long.", lineno);
 291                 }
 292
 293                 line_parse(line, lineno);
 294                 objs++;
 295         }
 296
 297         fclose(fd);
 298
 299         if (objs == 0)
 300                 g_gate_xlog("There are no objects to export.");
 301
 302         g_gate_log(LOG_INFO, "Exporting %u object(s).", objs);
 303 }
 304
 305 static int
 306 exports_check(struct ggd_export *ex, struct g_gate_cinit *cinit,
 307     struct ggd_connection *conn)
 308 {
 309         char ipmask[32]; /* 32 == strlen("xxx.xxx.xxx.xxx/xxx.xxx.xxx.xxx")+1 */
 310         int error = 0, flags;
 311
 312         strlcpy(ipmask, ip2str(ex->e_ip), sizeof(ipmask));
 313         strlcat(ipmask, "/", sizeof(ipmask));
 314         strlcat(ipmask, ip2str(ex->e_mask), sizeof(ipmask));
 315         if ((cinit->gc_flags & GGATE_FLAG_RDONLY) != 0) {
 316                 if (ex->e_flags == O_WRONLY) {
 317                         g_gate_log(LOG_WARNING, "Read-only access requested, "
 318                             "but %s (%s) is exported write-only.", ex->e_path,
 319                             ipmask);
 320                         return (EPERM);
 321                 } else {
 322                         conn->c_flags |= GGATE_FLAG_RDONLY;
 323                 }
 324         } else if ((cinit->gc_flags & GGATE_FLAG_WRONLY) != 0) {
 325                 if (ex->e_flags == O_RDONLY) {
 326                         g_gate_log(LOG_WARNING, "Write-only access requested, "
 327                             "but %s (%s) is exported read-only.", ex->e_path,
 328                             ipmask);
 329                         return (EPERM);
 330                 } else {
 331                         conn->c_flags |= GGATE_FLAG_WRONLY;
 332                 }
 333         } else {
 334                 if (ex->e_flags == O_RDONLY) {
 335                         g_gate_log(LOG_WARNING, "Read-write access requested, "
 336                             "but %s (%s) is exported read-only.", ex->e_path,
 337                             ipmask);
 338                         return (EPERM);
 339                 } else if (ex->e_flags == O_WRONLY) {
 340                         g_gate_log(LOG_WARNING, "Read-write access requested, "
 341                             "but %s (%s) is exported write-only.", ex->e_path,
 342                             ipmask);
 343                         return (EPERM);
 344                 }
 345         }
 346         if ((conn->c_flags & GGATE_FLAG_RDONLY) != 0)
 347                 flags = O_RDONLY;
 348         else if ((conn->c_flags & GGATE_FLAG_WRONLY) != 0)
 349                 flags = O_WRONLY;
 350         else
 351                 flags = O_RDWR;
 352         if (conn->c_diskfd != -1) {
 353                 if (strcmp(conn->c_path, ex->e_path) != 0) {
 354                         g_gate_log(LOG_ERR, "old %s and new %s: "
 355                             "Path mismatch during handshakes.",
 356                             conn->c_path, ex->e_path);
 357                         return (EPERM);
 358                 }
 359                 return (0);
 360         }
 361
 362         conn->c_diskfd = open(ex->e_path, flags);
 363         if (conn->c_diskfd == -1) {
 364                 error = errno;
 365                 g_gate_log(LOG_ERR, "Cannot open %s: %s.", ex->e_path,
 366                     strerror(error));
 367                 return (error);
 368         }
 369         return (0);
 370 }
 371
 372 static struct ggd_export *
 373 exports_find(struct sockaddr *s, struct g_gate_cinit *cinit,
 374     struct ggd_connection *conn)
 375 {
 376         struct ggd_export *ex;
 377         in_addr_t ip;
 378         int error;
 379
 380         ip = htonl(((struct sockaddr_in *)(void *)s)->sin_addr.s_addr);
 381         SLIST_FOREACH(ex, &exports, e_next) {
 382                 if ((ip & ex->e_mask) != ex->e_ip) {
 383                         g_gate_log(LOG_DEBUG, "exports[%s]: IP mismatch.",
 384                             ex->e_path);
 385                         continue;
 386                 }
 387                 if (strcmp(cinit->gc_path, ex->e_path) != 0) {
 388                         g_gate_log(LOG_DEBUG, "exports[%s]: Path mismatch.",
 389                             ex->e_path);
 390                         continue;
 391                 }
 392                 error = exports_check(ex, cinit, conn);
 393                 if (error == 0)
 394                         return (ex);
 395                 else {
 396                         errno = error;
 397                         return (NULL);
 398                 }
 399         }
 400         g_gate_log(LOG_WARNING, "Unauthorized connection from: %s.",
 401             ip2str(ip));
 402         errno = EPERM;
 403         return (NULL);
 404 }
 405
 406 /*
 407  * Remove timed out connections.
 408  */
 409 static void
 410 connection_cleanups(void)
 411 {
 412         struct ggd_connection *conn, *tconn;
 413         time_t now;
 414
 415         time(&now);
 416         LIST_FOREACH_SAFE(conn, &connections, c_next, tconn) {
 417                 if (now - conn->c_birthtime > 10) {
 418                         LIST_REMOVE(conn, c_next);
 419                         g_gate_log(LOG_NOTICE,
 420                             "Connection from %s [%s] removed.",
 421                             ip2str(conn->c_srcip), conn->c_path);
 422                         close(conn->c_diskfd);
 423                         close(conn->c_sendfd);
 424                         close(conn->c_recvfd);
 425                         free(conn->c_path);
 426                         free(conn);
 427                 }
 428         }
 429 }
 430
 431 static struct ggd_connection *
 432 connection_find(struct g_gate_cinit *cinit)
 433 {
 434         struct ggd_connection *conn;
 435
 436         LIST_FOREACH(conn, &connections, c_next) {
 437                 if (conn->c_token == cinit->gc_token)
 438                         break;
 439         }
 440         return (conn);
 441 }
 442
 443 static struct ggd_connection *
 444 connection_new(struct g_gate_cinit *cinit, struct sockaddr *s, int sfd)
 445 {
 446         struct ggd_connection *conn;
 447         in_addr_t ip;
 448
 449         /*
 450          * First, look for old connections.
 451          * We probably should do it every X seconds, but what for?
 452          * It is only dangerous if an attacker wants to overload connections
 453          * queue, so here is a good place to do the cleanups.
 454          */
 455         connection_cleanups();
 456
 457         conn = malloc(sizeof(*conn));
 458         if (conn == NULL)
 459                 return (NULL);
 460         conn->c_path = strdup(cinit->gc_path);
 461         if (conn->c_path == NULL) {
 462                 free(conn);
 463                 return (NULL);
 464         }
 465         conn->c_token = cinit->gc_token;
 466         ip = htonl(((struct sockaddr_in *)(void *)s)->sin_addr.s_addr);
 467         conn->c_srcip = ip;
 468         conn->c_diskfd = conn->c_sendfd = conn->c_recvfd = -1;
 469         if ((cinit->gc_flags & GGATE_FLAG_SEND) != 0)
 470                 conn->c_sendfd = sfd;
 471         else
 472                 conn->c_recvfd = sfd;
 473         conn->c_mediasize = 0;
 474         conn->c_sectorsize = 0;
 475         time(&conn->c_birthtime);
 476         conn->c_flags = cinit->gc_flags;
 477         LIST_INSERT_HEAD(&connections, conn, c_next);
 478         g_gate_log(LOG_DEBUG, "Connection created [%s, %s].", ip2str(ip),
 479             conn->c_path);
 480         return (conn);
 481 }
 482
 483 static int
 484 connection_add(struct ggd_connection *conn, struct g_gate_cinit *cinit,
 485     struct sockaddr *s, int sfd)
 486 {
 487         in_addr_t ip;
 488
 489         ip = htonl(((struct sockaddr_in *)(void *)s)->sin_addr.s_addr);
 490         if ((cinit->gc_flags & GGATE_FLAG_SEND) != 0) {
 491                 if (conn->c_sendfd != -1) {
 492                         g_gate_log(LOG_WARNING,
 493                             "Send socket already exists [%s, %s].", ip2str(ip),
 494                             conn->c_path);
 495                         return (EEXIST);
 496                 }
 497                 conn->c_sendfd = sfd;
 498         } else {
 499                 if (conn->c_recvfd != -1) {
 500                         g_gate_log(LOG_WARNING,
 501                             "Receive socket already exists [%s, %s].",
 502                             ip2str(ip), conn->c_path);
 503                         return (EEXIST);
 504                 }
 505                 conn->c_recvfd = sfd;
 506         }
 507         g_gate_log(LOG_DEBUG, "Connection added [%s, %s].", ip2str(ip),
 508             conn->c_path);
 509         return (0);
 510 }
 511
 512 /*
 513  * Remove one socket from the given connection or the whole
 514  * connection if sfd == -1.
 515  */
 516 static void
 517 connection_remove(struct ggd_connection *conn)
 518 {
 519
 520         LIST_REMOVE(conn, c_next);
 521         g_gate_log(LOG_DEBUG, "Connection removed [%s %s].",
 522             ip2str(conn->c_srcip), conn->c_path);
 523         if (conn->c_diskfd != -1)
 524                 close(conn->c_diskfd);
 525         if (conn->c_sendfd != -1)
 526                 close(conn->c_sendfd);
 527         if (conn->c_recvfd != -1)
 528                 close(conn->c_recvfd);
 529         free(conn->c_path);
 530         free(conn);
 531 }
 532
 533 static int
 534 connection_ready(struct ggd_connection *conn)
 535 {
 536
 537         return (conn->c_sendfd != -1 && conn->c_recvfd != -1);
 538 }
 539
 540 static void
 541 connection_launch(struct ggd_connection *conn)
 542 {
 543         pthread_t td;
 544         int error, pid;
 545
 546         pid = fork();
 547         if (pid > 0)
 548                 return;
 549         else if (pid == -1) {
 550                 g_gate_log(LOG_ERR, "Cannot fork: %s.", strerror(errno));
 551                 return;
 552         }
 553         g_gate_log(LOG_DEBUG, "Process created [%s].", conn->c_path);
 554
 555         /*
 556          * Create condition variables and mutexes for in-queue and out-queue
 557          * synchronization.
 558          */
 559         error = pthread_mutex_init(&inqueue_mtx, NULL);
 560         if (error != 0) {
 561                 g_gate_xlog("pthread_mutex_init(inqueue_mtx): %s.",
 562                     strerror(error));
 563         }
 564         error = pthread_cond_init(&inqueue_cond, NULL);
 565         if (error != 0) {
 566                 g_gate_xlog("pthread_cond_init(inqueue_cond): %s.",
 567                     strerror(error));
 568         }
 569         error = pthread_mutex_init(&outqueue_mtx, NULL);
 570         if (error != 0) {
 571                 g_gate_xlog("pthread_mutex_init(outqueue_mtx): %s.",
 572                     strerror(error));
 573         }
 574         error = pthread_cond_init(&outqueue_cond, NULL);
 575         if (error != 0) {
 576                 g_gate_xlog("pthread_cond_init(outqueue_cond): %s.",
 577                     strerror(error));
 578         }
 579
 580         /*
 581          * Create threads:
 582          * recvtd - thread for receiving I/O request
 583          * diskio - thread for doing I/O request
 584          * sendtd - thread for sending I/O requests back
 585          */
 586         error = pthread_create(&td, NULL, send_thread, conn);
 587         if (error != 0) {
 588                 g_gate_xlog("pthread_create(send_thread): %s.",
 589                     strerror(error));
 590         }
 591         error = pthread_create(&td, NULL, recv_thread, conn);
 592         if (error != 0) {
 593                 g_gate_xlog("pthread_create(recv_thread): %s.",
 594                     strerror(error));
 595         }
 596         disk_thread(conn);
 597 }
 598
 599 static void
 600 sendfail(int sfd, int error, const char *fmt, ...)
 601 {
 602         struct g_gate_sinit sinit;
 603         va_list ap;
 604         ssize_t data;
 605
 606         memset(&sinit, 0, sizeof(sinit));
 607         sinit.gs_error = error;
 608         g_gate_swap2n_sinit(&sinit);
 609         data = g_gate_send(sfd, &sinit, sizeof(sinit), 0);
 610         g_gate_swap2h_sinit(&sinit);
 611         if (data != sizeof(sinit)) {
 612                 g_gate_log(LOG_WARNING, "Cannot send initial packet: %s.",
 613                     strerror(errno));
 614                 return;
 615         }
 616         if (fmt != NULL) {
 617                 va_start(ap, fmt);
 618                 g_gate_vlog(LOG_WARNING, fmt, ap);
 619                 va_end(ap);
 620         }
 621 }
 622
 623 static void *
 624 malloc_waitok(size_t size)
 625 {
 626         void *p;
 627
 628         while ((p = malloc(size)) == NULL) {
 629                 g_gate_log(LOG_DEBUG, "Cannot allocate %zu bytes.", size);
 630                 sleep(1);
 631         }
 632         return (p);
 633 }
 634
 635 static void *
 636 recv_thread(void *arg)
 637 {
 638         struct ggd_connection *conn;
 639         struct ggd_request *req;
 640         ssize_t data;
 641         int error, fd;
 642
 643         conn = arg;
 644         g_gate_log(LOG_NOTICE, "%s: started [%s]!", __func__, conn->c_path);
 645         fd = conn->c_recvfd;
 646         for (;;) {
 647                 /*
 648                  * Get header packet.
 649                  */
 650                 req = malloc_waitok(sizeof(*req));
 651                 data = g_gate_recv(fd, &req->r_hdr, sizeof(req->r_hdr),
 652                     MSG_WAITALL);
 653                 if (data == 0) {
 654                         g_gate_log(LOG_DEBUG, "Process %u exiting.", getpid());
 655                         exit(EXIT_SUCCESS);
 656                 } else if (data == -1) {
 657                         g_gate_xlog("Error while receiving hdr packet: %s.",
 658                             strerror(errno));
 659                 } else if (data != sizeof(req->r_hdr)) {
 660                         g_gate_xlog("Malformed hdr packet received.");
 661                 }
 662                 g_gate_log(LOG_DEBUG, "Received hdr packet.");
 663                 g_gate_swap2h_hdr(&req->r_hdr);
 664
 665                 g_gate_log(LOG_DEBUG, "%s: offset=%jd length=%u", __func__,
 666                     (intmax_t)req->r_offset, (unsigned)req->r_length);
 667
 668                 /*
 669                  * Allocate memory for data.
 670                  */
 671                 req->r_data = malloc_waitok(req->r_length);
 672
 673                 /*
 674                  * Receive data to write for WRITE request.
 675                  */
 676                 if (req->r_cmd == GGATE_CMD_WRITE) {
 677                         g_gate_log(LOG_DEBUG, "Waiting for %u bytes of data...",
 678                             req->r_length);
 679                         data = g_gate_recv(fd, req->r_data, req->r_length,
 680                             MSG_WAITALL);
 681                         if (data == -1) {
 682                                 g_gate_xlog("Error while receiving data: %s.",
 683                                     strerror(errno));
 684                         }
 685                 }
 686
 687                 /*
 688                  * Put the request onto the incoming queue.
 689                  */
 690                 error = pthread_mutex_lock(&inqueue_mtx);
 691                 assert(error == 0);
 692                 TAILQ_INSERT_TAIL(&inqueue, req, r_next);
 693                 error = pthread_cond_signal(&inqueue_cond);
 694                 assert(error == 0);
 695                 error = pthread_mutex_unlock(&inqueue_mtx);
 696                 assert(error == 0);
 697         }
 698 }
 699
 700 static void *
 701 disk_thread(void *arg)
 702 {
 703         struct ggd_connection *conn;
 704         struct ggd_request *req;
 705         ssize_t data;
 706         int error, fd;
 707
 708         conn = arg;
 709         g_gate_log(LOG_NOTICE, "%s: started [%s]!", __func__, conn->c_path);
 710         fd = conn->c_diskfd;
 711         for (;;) {
 712                 /*
 713                  * Get a request from the incoming queue.
 714                  */
 715                 error = pthread_mutex_lock(&inqueue_mtx);
 716                 assert(error == 0);
 717                 while ((req = TAILQ_FIRST(&inqueue)) == NULL) {
 718                         error = pthread_cond_wait(&inqueue_cond, &inqueue_mtx);
 719                         assert(error == 0);
 720                 }
 721                 TAILQ_REMOVE(&inqueue, req, r_next);
 722                 error = pthread_mutex_unlock(&inqueue_mtx);
 723                 assert(error == 0);
 724
 725                 /*
 726                  * Check the request.
 727                  */
 728                 assert(req->r_cmd == GGATE_CMD_READ || req->r_cmd == GGATE_CMD_WRITE);
 729                 assert(req->r_offset + req->r_length <= (uintmax_t)conn->c_mediasize);
 730                 assert((req->r_offset % conn->c_sectorsize) == 0);
 731                 assert((req->r_length % conn->c_sectorsize) == 0);
 732
 733                 g_gate_log(LOG_DEBUG, "%s: offset=%jd length=%u", __func__,
 734                     (intmax_t)req->r_offset, (unsigned)req->r_length);
 735
 736                 /*
 737                  * Do the request.
 738                  */
 739                 data = 0;
 740                 switch (req->r_cmd) {
 741                 case GGATE_CMD_READ:
 742                         data = pread(fd, req->r_data, req->r_length,
 743                             req->r_offset);
 744                         break;
 745                 case GGATE_CMD_WRITE:
 746                         data = pwrite(fd, req->r_data, req->r_length,
 747                             req->r_offset);
 748                         /* Free data memory here - better sooner. */
 749                         free(req->r_data);
 750                         req->r_data = NULL;
 751                         break;
 752                 }
 753                 if (data != (ssize_t)req->r_length) {
 754                         /* Report short reads/writes as I/O errors. */
 755                         if (errno == 0)
 756                                 errno = EIO;
 757                         g_gate_log(LOG_ERR, "Disk error: %s", strerror(errno));
 758                         req->r_error = errno;
 759                         if (req->r_data != NULL) {
 760                                 free(req->r_data);
 761                                 req->r_data = NULL;
 762                         }
 763                 }
 764
 765                 /*
 766                  * Put the request onto the outgoing queue.
 767                  */
 768                 error = pthread_mutex_lock(&outqueue_mtx);
 769                 assert(error == 0);
 770                 TAILQ_INSERT_TAIL(&outqueue, req, r_next);
 771                 error = pthread_cond_signal(&outqueue_cond);
 772                 assert(error == 0);
 773                 error = pthread_mutex_unlock(&outqueue_mtx);
 774                 assert(error == 0);
 775         }
 776
 777         /* NOTREACHED */
 778         return (NULL);
 779 }
 780
 781 static void *
 782 send_thread(void *arg)
 783 {
 784         struct ggd_connection *conn;
 785         struct ggd_request *req;
 786         ssize_t data;
 787         int error, fd;
 788
 789         conn = arg;
 790         g_gate_log(LOG_NOTICE, "%s: started [%s]!", __func__, conn->c_path);
 791         fd = conn->c_sendfd;
 792         for (;;) {
 793                 /*
 794                  * Get a request from the outgoing queue.
 795                  */
 796                 error = pthread_mutex_lock(&outqueue_mtx);
 797                 assert(error == 0);
 798                 while ((req = TAILQ_FIRST(&outqueue)) == NULL) {
 799                         error = pthread_cond_wait(&outqueue_cond,
 800                             &outqueue_mtx);
 801                         assert(error == 0);
 802                 }
 803                 TAILQ_REMOVE(&outqueue, req, r_next);
 804                 error = pthread_mutex_unlock(&outqueue_mtx);
 805                 assert(error == 0);
 806
 807                 g_gate_log(LOG_DEBUG, "%s: offset=%jd length=%u", __func__,
 808                     (intmax_t)req->r_offset, (unsigned)req->r_length);
 809
 810                 /*
 811                  * Send the request.
 812                  */
 813                 g_gate_swap2n_hdr(&req->r_hdr);
 814                 if (g_gate_send(fd, &req->r_hdr, sizeof(req->r_hdr), 0) == -1) {
 815                         g_gate_xlog("Error while sending hdr packet: %s.",
 816                             strerror(errno));
 817                 }
 818                 g_gate_log(LOG_DEBUG, "Sent hdr packet.");
 819                 g_gate_swap2h_hdr(&req->r_hdr);
 820                 if (req->r_data != NULL) {
 821                         data = g_gate_send(fd, req->r_data, req->r_length, 0);
 822                         if (data != (ssize_t)req->r_length) {
 823                                 g_gate_xlog("Error while sending data: %s.",
 824                                     strerror(errno));
 825                         }
 826                         g_gate_log(LOG_DEBUG,
 827                             "Sent %zd bytes (offset=%ju, size=%zu).", data,
 828                             (uintmax_t)req->r_offset, (size_t)req->r_length);
 829                         free(req->r_data);
 830                 }
 831                 free(req);
 832         }
 833
 834         /* NOTREACHED */
 835         return (NULL);
 836 }
 837
 838 static void
 839 log_connection(struct sockaddr *from)
 840 {
 841         in_addr_t ip;
 842
 843         ip = htonl(((struct sockaddr_in *)(void *)from)->sin_addr.s_addr);
 844         g_gate_log(LOG_INFO, "Connection from: %s.", ip2str(ip));
 845 }
 846
 847 static int
 848 handshake(struct sockaddr *from, int sfd)
 849 {
 850         struct g_gate_version ver;
 851         struct g_gate_cinit cinit;
 852         struct g_gate_sinit sinit;
 853         struct ggd_connection *conn;
 854         struct ggd_export *ex;
 855         ssize_t data;
 856
 857         log_connection(from);
 858         /*
 859          * Phase 1: Version verification.
 860          */
 861         g_gate_log(LOG_DEBUG, "Receiving version packet.");
 862         data = g_gate_recv(sfd, &ver, sizeof(ver), MSG_WAITALL);
 863         g_gate_swap2h_version(&ver);
 864         if (data != sizeof(ver)) {
 865                 g_gate_log(LOG_WARNING, "Malformed version packet.");
 866                 return (0);
 867         }
 868         g_gate_log(LOG_DEBUG, "Version packet received.");
 869         if (memcmp(ver.gv_magic, GGATE_MAGIC, strlen(GGATE_MAGIC)) != 0) {
 870                 g_gate_log(LOG_WARNING, "Invalid magic field.");
 871                 return (0);
 872         }
 873         if (ver.gv_version != GGATE_VERSION) {
 874                 g_gate_log(LOG_WARNING, "Version %u is not supported.",
 875                     ver.gv_version);
 876                 return (0);
 877         }
 878         ver.gv_error = 0;
 879         g_gate_swap2n_version(&ver);
 880         data = g_gate_send(sfd, &ver, sizeof(ver), 0);
 881         g_gate_swap2h_version(&ver);
 882         if (data == -1) {
 883                 sendfail(sfd, errno, "Error while sending version packet: %s.",
 884                     strerror(errno));
 885                 return (0);
 886         }
 887
 888         /*
 889          * Phase 2: Request verification.
 890          */
 891         g_gate_log(LOG_DEBUG, "Receiving initial packet.");
 892         data = g_gate_recv(sfd, &cinit, sizeof(cinit), MSG_WAITALL);
 893         g_gate_swap2h_cinit(&cinit);
 894         if (data != sizeof(cinit)) {
 895                 g_gate_log(LOG_WARNING, "Malformed initial packet.");
 896                 return (0);
 897         }
 898         g_gate_log(LOG_DEBUG, "Initial packet received.");
 899         conn = connection_find(&cinit);
 900         if (conn != NULL) {
 901                 /*
 902                  * Connection should already exists.
 903                  */
 904                 g_gate_log(LOG_DEBUG, "Found existing connection (token=%lu).",
 905                     (unsigned long)conn->c_token);
 906                 if (connection_add(conn, &cinit, from, sfd) == -1) {
 907                         connection_remove(conn);
 908                         return (0);
 909                 }
 910         } else {
 911                 /*
 912                  * New connection, allocate space.
 913                  */
 914                 conn = connection_new(&cinit, from, sfd);
 915                 if (conn == NULL) {
 916                         sendfail(sfd, ENOMEM,
 917                             "Cannot allocate new connection.");
 918                         return (0);
 919                 }
 920                 g_gate_log(LOG_DEBUG, "New connection created (token=%lu).",
 921                     (unsigned long)conn->c_token);
 922         }
 923
 924         ex = exports_find(from, &cinit, conn);
 925         if (ex == NULL) {
 926                 sendfail(sfd, errno, NULL);
 927                 connection_remove(conn);
 928                 return (0);
 929         }
 930         if (conn->c_mediasize == 0) {
 931                 conn->c_mediasize = g_gate_mediasize(conn->c_diskfd);
 932                 conn->c_sectorsize = g_gate_sectorsize(conn->c_diskfd);
 933         }
 934         sinit.gs_mediasize = conn->c_mediasize;
 935         sinit.gs_sectorsize = conn->c_sectorsize;
 936         sinit.gs_error = 0;
 937
 938         g_gate_log(LOG_DEBUG, "Sending initial packet.");
 939
 940         g_gate_swap2n_sinit(&sinit);
 941         data = g_gate_send(sfd, &sinit, sizeof(sinit), 0);
 942         g_gate_swap2h_sinit(&sinit);
 943         if (data == -1) {
 944                 sendfail(sfd, errno, "Error while sending initial packet: %s.",
 945                     strerror(errno));
 946                 return (0);
 947         }
 948
 949         if (connection_ready(conn)) {
 950                 connection_launch(conn);
 951                 connection_remove(conn);
 952         }
 953         return (1);
 954 }
 955
 956 static void
 957 huphandler(int sig __unused)
 958 {
 959
 960         got_sighup = 1;
 961 }
 962
 963 int
 964 main(int argc, char *argv[])
 965 {
 966         const char *ggated_pidfile = _PATH_VARRUN "/ggated.pid";
 967         struct pidfh *pfh;
 968         struct sockaddr_in serv;
 969         struct sockaddr from;
 970         socklen_t fromlen;
 971         pid_t otherpid;
 972         int ch, sfd, tmpsfd;
 973         unsigned port;
 974
 975         bindaddr = htonl(INADDR_ANY);
 976         port = G_GATE_PORT;
 977         while ((ch = getopt(argc, argv, "a:hnp:F:R:S:v")) != -1) {
 978                 switch (ch) {
 979                 case 'a':
 980                         bindaddr = g_gate_str2ip(optarg);
 981                         if (bindaddr == INADDR_NONE) {
 982                                 errx(EXIT_FAILURE,
 983                                     "Invalid IP/host name to bind to.");
 984                         }
 985                         break;
 986                 case 'F':
 987                         ggated_pidfile = optarg;
 988                         break;
 989                 case 'n':
 990                         nagle = 0;
 991                         break;
 992                 case 'p':
 993                         errno = 0;
 994                         port = strtoul(optarg, NULL, 10);
 995                         if (port == 0 && errno != 0)
 996                                 errx(EXIT_FAILURE, "Invalid port.");
 997                         break;
 998                 case 'R':
 999                         errno = 0;
1000                         rcvbuf = strtoul(optarg, NULL, 10);
1001                         if (rcvbuf == 0 && errno != 0)
1002                                 errx(EXIT_FAILURE, "Invalid rcvbuf.");
1003                         break;
1004                 case 'S':
1005                         errno = 0;
1006                         sndbuf = strtoul(optarg, NULL, 10);
1007                         if (sndbuf == 0 && errno != 0)
1008                                 errx(EXIT_FAILURE, "Invalid sndbuf.");
1009                         break;
1010                 case 'v':
1011                         g_gate_verbose++;
1012                         break;
1013                 case 'h':
1014                 default:
1015                         usage();
1016                 }
1017         }
1018         argc -= optind;
1019         argv += optind;
1020
1021         if (argv[0] != NULL)
1022                 exports_file = argv[0];
1023         exports_get();
1024
1025         pfh = pidfile_open(ggated_pidfile, 0600, &otherpid);
1026         if (pfh == NULL) {
1027                 if (errno == EEXIST) {
1028                         errx(EXIT_FAILURE, "Daemon already running, pid: %jd.",
1029                             (intmax_t)otherpid);
1030                 }
1031                 err(EXIT_FAILURE, "Cannot open/create pidfile");
1032         }
1033
1034         if (!g_gate_verbose) {
1035                 /* Run in daemon mode. */
1036                 if (daemon(0, 0) == -1)
1037                         g_gate_xlog("Cannot daemonize: %s", strerror(errno));
1038         }
1039
1040         pidfile_write(pfh);
1041
1042         signal(SIGCHLD, SIG_IGN);
1043
1044         sfd = socket(AF_INET, SOCK_STREAM, 0);
1045         if (sfd == -1)
1046                 g_gate_xlog("Cannot open stream socket: %s.", strerror(errno));
1047         bzero(&serv, sizeof(serv));
1048         serv.sin_family = AF_INET;
1049         serv.sin_addr.s_addr = bindaddr;
1050         serv.sin_port = htons(port);
1051
1052         g_gate_socket_settings(sfd);
1053
1054         if (bind(sfd, (struct sockaddr *)&serv, sizeof(serv)) == -1)
1055                 g_gate_xlog("bind(): %s.", strerror(errno));
1056         if (listen(sfd, 5) == -1)
1057                 g_gate_xlog("listen(): %s.", strerror(errno));
1058
1059         g_gate_log(LOG_INFO, "Listen on port: %d.", port);
1060
1061         signal(SIGHUP, huphandler);
1062
1063         for (;;) {
1064                 fromlen = sizeof(from);
1065                 tmpsfd = accept(sfd, &from, &fromlen);
1066                 if (tmpsfd == -1)
1067                         g_gate_xlog("accept(): %s.", strerror(errno));
1068
1069                 if (got_sighup) {
1070                         got_sighup = 0;
1071                         exports_get();
1072                 }
1073
1074                 if (!handshake(&from, tmpsfd))
1075                         close(tmpsfd);
1076         }
1077         close(sfd);
1078         pidfile_remove(pfh);
1079         exit(EXIT_SUCCESS);
1080 }