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