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