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