src/socket-proxy/socket-proxyd.c

   1 /*-*- Mode: C; c-basic-offset: 8; indent-tabs-mode: nil -*-*/
   2
   3 /***
   4   This file is part of systemd.
   5
   6   Copyright 2013 David Strauss
   7
   8   systemd is free software; you can redistribute it and/or modify it
   9   under the terms of the GNU Lesser General Public License as published by
  10   the Free Software Foundation; either version 2.1 of the License, or
  11   (at your option) any later version.
  12
  13   systemd is distributed in the hope that it will be useful, but
  14   WITHOUT ANY WARRANTY; without even the implied warranty of
  15   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  16   Lesser General Public License for more details.
  17
  18   You should have received a copy of the GNU Lesser General Public License
  19   along with systemd; If not, see <http://www.gnu.org/licenses/>.
  20  ***/
  21
  22 #include <arpa/inet.h>
  23 #include <errno.h>
  24 #include <getopt.h>
  25 #include <stdio.h>
  26 #include <stdlib.h>
  27 #include <string.h>
  28 #include <netdb.h>
  29 #include <sys/fcntl.h>
  30 #include <sys/socket.h>
  31 #include <sys/un.h>
  32 #include <unistd.h>
  33
  34 #include "sd-daemon.h"
  35 #include "sd-event.h"
  36 #include "log.h"
  37 #include "socket-util.h"
  38 #include "util.h"
  39 #include "event-util.h"
  40
  41 #define BUFFER_SIZE 16384
  42 #define _cleanup_freeaddrinfo_ _cleanup_(freeaddrinfop)
  43
  44 unsigned int total_clients = 0;
  45
  46 DEFINE_TRIVIAL_CLEANUP_FUNC(struct addrinfo *, freeaddrinfo);
  47
  48 struct proxy {
  49         int listen_fd;
  50         bool ignore_env;
  51         bool remote_is_inet;
  52         const char *remote_host;
  53         const char *remote_service;
  54 };
  55
  56 struct connection {
  57         int fd;
  58         uint32_t events;
  59         sd_event_source *w;
  60         struct connection *c_destination;
  61         size_t buffer_filled_len;
  62         size_t buffer_sent_len;
  63         char buffer[BUFFER_SIZE];
  64 };
  65
  66 static void free_connection(struct connection *c) {
  67         if (c != NULL) {
  68                 log_debug("Freeing fd=%d (conn %p).", c->fd, c);
  69                 sd_event_source_unref(c->w);
  70                 if (c->fd > 0)
  71                         close_nointr_nofail(c->fd);
  72                 free(c);
  73         }
  74 }
  75
  76 static int add_event_to_connection(struct connection *c, uint32_t events) {
  77         int r;
  78
  79         log_debug("Have revents=%d. Adding revents=%d.", c->events, events);
  80
  81         c->events |= events;
  82
  83         r = sd_event_source_set_io_events(c->w, c->events);
  84         if (r < 0) {
  85                 log_error("Error %d setting revents: %s", r, strerror(-r));
  86                 return r;
  87         }
  88
  89         r = sd_event_source_set_enabled(c->w, SD_EVENT_ON);
  90         if (r < 0) {
  91                 log_error("Error %d enabling source: %s", r, strerror(-r));
  92                 return r;
  93         }
  94
  95         return 0;
  96 }
  97
  98 static int remove_event_from_connection(struct connection *c, uint32_t events) {
  99         int r;
 100
 101         log_debug("Have revents=%d. Removing revents=%d.", c->events, events);
 102
 103         c->events &= ~events;
 104
 105         r = sd_event_source_set_io_events(c->w, c->events);
 106         if (r < 0) {
 107                 log_error("Error %d setting revents: %s", r, strerror(-r));
 108                 return r;
 109         }
 110
 111         if (c->events == 0) {
 112             r = sd_event_source_set_enabled(c->w, SD_EVENT_OFF);
 113             if (r < 0) {
 114                     log_error("Error %d disabling source: %s", r, strerror(-r));
 115                     return r;
 116             }
 117         }
 118
 119         return 0;
 120 }
 121
 122 static int send_buffer(struct connection *sender) {
 123         struct connection *receiver = sender->c_destination;
 124         ssize_t len;
 125         int r = 0;
 126
 127         /* We cannot assume that even a partial send() indicates that
 128          * the next send() will return EAGAIN or EWOULDBLOCK. Loop until
 129          * it does. */
 130         while (sender->buffer_filled_len > sender->buffer_sent_len) {
 131                 len = send(receiver->fd, sender->buffer + sender->buffer_sent_len, sender->buffer_filled_len - sender->buffer_sent_len, 0);
 132                 log_debug("send(%d, ...)=%ld", receiver->fd, len);
 133                 if (len < 0) {
 134                         if (errno != EWOULDBLOCK && errno != EAGAIN) {
 135                                 log_error("Error %d in send to fd=%d: %m", errno, receiver->fd);
 136                                 return -errno;
 137                         }
 138                         else {
 139                                 /* send() is in a would-block state. */
 140                                 break;
 141                         }
 142                 }
 143
 144                 /* len < 0 can't occur here. len == 0 is possible but
 145                  * undefined behavior for nonblocking send(). */
 146                 assert(len > 0);
 147                 sender->buffer_sent_len += len;
 148         }
 149
 150         log_debug("send(%d, ...) completed with %lu bytes still buffered.", receiver->fd, sender->buffer_filled_len - sender->buffer_sent_len);
 151
 152         /* Detect a would-block state or partial send. */
 153         if (sender->buffer_filled_len > sender->buffer_sent_len) {
 154
 155                 /* If the buffer is full, disable events coming for recv. */
 156                 if (sender->buffer_filled_len == BUFFER_SIZE) {
 157                     r = remove_event_from_connection(sender, EPOLLIN);
 158                     if (r < 0) {
 159                             log_error("Error %d disabling EPOLLIN for fd=%d: %s", r, sender->fd, strerror(-r));
 160                             return r;
 161                     }
 162                 }
 163
 164                 /* Watch for when the recipient can be sent data again. */
 165                 r = add_event_to_connection(receiver, EPOLLOUT);
 166                 if (r < 0) {
 167                         log_error("Error %d enabling EPOLLOUT for fd=%d: %s", r, receiver->fd, strerror(-r));
 168                         return r;
 169                 }
 170                 log_debug("Done with recv for fd=%d. Waiting on send for fd=%d.", sender->fd, receiver->fd);
 171                 return r;
 172         }
 173
 174         /* If we sent everything without any issues (would-block or
 175          * partial send), the buffer is now empty. */
 176         sender->buffer_filled_len = 0;
 177         sender->buffer_sent_len = 0;
 178
 179         /* Enable the sender's receive watcher, in case the buffer was
 180          * full and we disabled it. */
 181         r = add_event_to_connection(sender, EPOLLIN);
 182         if (r < 0) {
 183                 log_error("Error %d enabling EPOLLIN for fd=%d: %s", r, sender->fd, strerror(-r));
 184                 return r;
 185         }
 186
 187         /* Disable the other side's send watcher, as we have no data to send now. */
 188         r = remove_event_from_connection(receiver, EPOLLOUT);
 189         if (r < 0) {
 190                 log_error("Error %d disabling EPOLLOUT for fd=%d: %s", r, receiver->fd, strerror(-r));
 191                 return r;
 192         }
 193
 194         return 0;
 195 }
 196
 197 static int transfer_data_cb(sd_event_source *s, int fd, uint32_t revents, void *userdata) {
 198         struct connection *c = (struct connection *) userdata;
 199         int r = 0;
 200         ssize_t len;
 201
 202         assert(revents & (EPOLLIN | EPOLLOUT));
 203         assert(fd == c->fd);
 204         assert(s == c->w);
 205
 206         log_debug("Got event revents=%d from fd=%d (conn %p).", revents, fd, c);
 207
 208         if (revents & EPOLLIN) {
 209                 log_debug("About to recv up to %lu bytes from fd=%d (%lu/BUFFER_SIZE).", BUFFER_SIZE - c->buffer_filled_len, fd, c->buffer_filled_len);
 210
 211                 /* Receive until the buffer's full, there's no more data,
 212                  * or the client/server disconnects. */
 213                 while (c->buffer_filled_len < BUFFER_SIZE) {
 214                         len = recv(fd, c->buffer + c->buffer_filled_len, BUFFER_SIZE - c->buffer_filled_len, 0);
 215                         log_debug("recv(%d, ...)=%ld", fd, len);
 216                         if (len < 0) {
 217                                 if (errno != EWOULDBLOCK && errno != EAGAIN) {
 218                                         log_error("Error %d in recv from fd=%d: %m", errno, fd);
 219                                         return -errno;
 220                                 }
 221                                 else {
 222                                         /* recv() is in a blocking state. */
 223                                         break;
 224                                 }
 225                         }
 226                         else if (len == 0) {
 227                                 log_debug("Clean disconnection from fd=%d", fd);
 228                                 total_clients--;
 229                                 free_connection(c->c_destination);
 230                                 free_connection(c);
 231                                 return 0;
 232                         }
 233
 234                         assert(len > 0);
 235                         log_debug("Recording that the buffer got %ld more bytes full.", len);
 236                         c->buffer_filled_len += len;
 237                         log_debug("Buffer now has %ld bytes full.", c->buffer_filled_len);
 238                 }
 239
 240                 /* Try sending the data immediately. */
 241                 return send_buffer(c);
 242         }
 243         else {
 244                 return send_buffer(c->c_destination);
 245         }
 246
 247         return r;
 248 }
 249
 250 /* Once sending to the server is ready, set up the real watchers. */
 251 static int connected_to_server_cb(sd_event_source *s, int fd, uint32_t revents, void *userdata) {
 252         struct sd_event *e = NULL;
 253         struct connection *c_server_to_client = (struct connection *) userdata;
 254         struct connection *c_client_to_server = c_server_to_client->c_destination;
 255         int r;
 256
 257         assert(revents & EPOLLOUT);
 258
 259         e = sd_event_get(s);
 260
 261         /* Cancel the initial write watcher for the server. */
 262         sd_event_source_unref(s);
 263
 264         log_debug("Connected to server. Initializing watchers for receiving data.");
 265
 266         /* A recv watcher for the server. */
 267         r = sd_event_add_io(e, c_server_to_client->fd, EPOLLIN, transfer_data_cb, c_server_to_client, &c_server_to_client->w);
 268         if (r < 0) {
 269                 log_error("Error %d creating recv watcher for fd=%d: %s", r, c_server_to_client->fd, strerror(-r));
 270                 goto fail;
 271         }
 272         c_server_to_client->events = EPOLLIN;
 273
 274         /* A recv watcher for the client. */
 275         r = sd_event_add_io(e, c_client_to_server->fd, EPOLLIN, transfer_data_cb, c_client_to_server, &c_client_to_server->w);
 276         if (r < 0) {
 277                 log_error("Error %d creating recv watcher for fd=%d: %s", r, c_client_to_server->fd, strerror(-r));
 278                 goto fail;
 279         }
 280         c_client_to_server->events = EPOLLIN;
 281
 282 goto finish;
 283
 284 fail:
 285         free_connection(c_client_to_server);
 286         free_connection(c_server_to_client);
 287
 288 finish:
 289         return r;
 290 }
 291
 292 static int get_server_connection_fd(const struct proxy *proxy) {
 293         int server_fd;
 294         int r = -EBADF;
 295         int len;
 296
 297         if (proxy->remote_is_inet) {
 298                 int s;
 299                 _cleanup_freeaddrinfo_ struct addrinfo *result = NULL;
 300                 struct addrinfo hints = {.ai_family = AF_UNSPEC,
 301                                          .ai_socktype = SOCK_STREAM,
 302                                          .ai_flags = AI_PASSIVE};
 303
 304                 log_debug("Looking up address info for %s:%s", proxy->remote_host, proxy->remote_service);
 305                 s = getaddrinfo(proxy->remote_host, proxy->remote_service, &hints, &result);
 306                 if (s != 0) {
 307                         log_error("getaddrinfo error (%d): %s", s, gai_strerror(s));
 308                         return r;
 309                 }
 310
 311                 if (result == NULL) {
 312                         log_error("getaddrinfo: no result");
 313                         return r;
 314                 }
 315
 316                 /* @TODO: Try connecting to all results instead of just the first. */
 317                 server_fd = socket(result->ai_family, result->ai_socktype | SOCK_NONBLOCK, result->ai_protocol);
 318                 if (server_fd < 0) {
 319                         log_error("Error %d creating socket: %m", errno);
 320                         return r;
 321                 }
 322
 323                 r = connect(server_fd, result->ai_addr, result->ai_addrlen);
 324                 /* Ignore EINPROGRESS errors because they're expected for a nonblocking socket. */
 325                 if (r < 0 && errno != EINPROGRESS) {
 326                         log_error("Error %d while connecting to socket %s:%s: %m", errno, proxy->remote_host, proxy->remote_service);
 327                         return r;
 328                 }
 329         }
 330         else {
 331                 struct sockaddr_un remote;
 332
 333                 server_fd = socket(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK, 0);
 334                 if (server_fd < 0) {
 335                         log_error("Error %d creating socket: %m", errno);
 336                         return -EBADFD;
 337                 }
 338
 339                 remote.sun_family = AF_UNIX;
 340                 strncpy(remote.sun_path, proxy->remote_host, sizeof(remote.sun_path));
 341                 len = strlen(remote.sun_path) + sizeof(remote.sun_family);
 342                 r = connect(server_fd, (struct sockaddr *) &remote, len);
 343                 if (r < 0 && errno != EINPROGRESS) {
 344                         log_error("Error %d while connecting to Unix domain socket %s: %m", errno, proxy->remote_host);
 345                         return -EBADFD;
 346                 }
 347         }
 348
 349         log_debug("Server connection is fd=%d", server_fd);
 350         return server_fd;
 351 }
 352
 353 static int do_accept(sd_event *e, struct proxy *p, int fd) {
 354         struct connection *c_server_to_client = NULL;
 355         struct connection *c_client_to_server = NULL;
 356         int r = 0;
 357         union sockaddr_union sa;
 358         socklen_t salen = sizeof(sa);
 359         int client_fd, server_fd;
 360
 361         client_fd = accept4(fd, (struct sockaddr *) &sa, &salen, SOCK_NONBLOCK|SOCK_CLOEXEC);
 362         if (client_fd < 0) {
 363                 if (errno == EAGAIN || errno == EWOULDBLOCK)
 364                     return -errno;
 365                 log_error("Error %d accepting client connection: %m", errno);
 366                 r = -errno;
 367                 goto fail;
 368         }
 369
 370         server_fd = get_server_connection_fd(p);
 371         if (server_fd < 0) {
 372                 log_error("Error initiating server connection.");
 373                 r = server_fd;
 374                 goto fail;
 375         }
 376
 377         c_client_to_server = new0(struct connection, 1);
 378         if (c_client_to_server == NULL) {
 379                 log_oom();
 380                 goto fail;
 381         }
 382
 383         c_server_to_client = new0(struct connection, 1);
 384         if (c_server_to_client == NULL) {
 385                 log_oom();
 386                 goto fail;
 387         }
 388
 389         c_client_to_server->fd = client_fd;
 390         c_server_to_client->fd = server_fd;
 391
 392         if (sa.sa.sa_family == AF_INET || sa.sa.sa_family == AF_INET6) {
 393                 char sa_str[INET6_ADDRSTRLEN];
 394                 const char *success;
 395
 396                 success = inet_ntop(sa.sa.sa_family, &sa.in6.sin6_addr, sa_str, INET6_ADDRSTRLEN);
 397                 if (success == NULL)
 398                         log_warning("Error %d calling inet_ntop: %m", errno);
 399                 else
 400                         log_debug("Accepted client connection from %s as fd=%d", sa_str, c_client_to_server->fd);
 401         }
 402         else {
 403                 log_debug("Accepted client connection (non-IP) as fd=%d", c_client_to_server->fd);
 404         }
 405
 406         total_clients++;
 407         log_debug("Client fd=%d (conn %p) successfully connected. Total clients: %u", c_client_to_server->fd, c_client_to_server, total_clients);
 408         log_debug("Server fd=%d (conn %p) successfully initialized.", c_server_to_client->fd, c_server_to_client);
 409
 410         /* Initialize watcher for send to server; this shows connectivity. */
 411         r = sd_event_add_io(e, c_server_to_client->fd, EPOLLOUT, connected_to_server_cb, c_server_to_client, &c_server_to_client->w);
 412         if (r < 0) {
 413                 log_error("Error %d creating connectivity watcher for fd=%d: %s", r, c_server_to_client->fd, strerror(-r));
 414                 goto fail;
 415         }
 416
 417         /* Allow lookups of the opposite connection. */
 418         c_server_to_client->c_destination = c_client_to_server;
 419         c_client_to_server->c_destination = c_server_to_client;
 420
 421         goto finish;
 422
 423 fail:
 424         log_warning("Accepting a client connection or connecting to the server failed.");
 425         free_connection(c_client_to_server);
 426         free_connection(c_server_to_client);
 427
 428 finish:
 429         return r;
 430 }
 431
 432 static int accept_cb(sd_event_source *s, int fd, uint32_t revents, void *userdata) {
 433         struct proxy *p = (struct proxy *) userdata;
 434         sd_event *e = NULL;
 435         int r = 0;
 436
 437         assert(revents & EPOLLIN);
 438
 439         e = sd_event_get(s);
 440
 441         for (;;) {
 442                 r = do_accept(e, p, fd);
 443                 if (r == -EAGAIN || r == -EWOULDBLOCK)
 444                         break;
 445                 if (r < 0)
 446                         log_error("Error %d while trying to accept: %s", r, strerror(-r));
 447         }
 448
 449         /* Re-enable the watcher. */
 450         r = sd_event_source_set_enabled(s, SD_EVENT_ONESHOT);
 451         if (r < 0) {
 452                 log_error("Error %d while re-enabling listener with ONESHOT: %s", r, strerror(-r));
 453                 return r;
 454         }
 455
 456         /* Preserve the main loop even if a single accept() fails. */
 457         return 1;
 458 }
 459
 460 static int run_main_loop(struct proxy *proxy) {
 461         _cleanup_event_source_unref_ sd_event_source *w_accept = NULL;
 462         _cleanup_event_unref_ sd_event *e = NULL;
 463         int r = EXIT_SUCCESS;
 464
 465         r = sd_event_new(&e);
 466         if (r < 0) {
 467                 log_error("Failed to allocate event loop: %s", strerror(-r));
 468                 return r;
 469         }
 470
 471         r = fd_nonblock(proxy->listen_fd, true);
 472         if (r < 0) {
 473                 log_error("Failed to make listen file descriptor nonblocking: %s", strerror(-r));
 474                 return r;
 475         }
 476
 477         log_debug("Initializing main listener fd=%d", proxy->listen_fd);
 478
 479         r = sd_event_add_io(e, proxy->listen_fd, EPOLLIN, accept_cb, proxy, &w_accept);
 480         if (r < 0) {
 481                 log_error("Error %d while adding event IO source: %s", r, strerror(-r));
 482                 return r;
 483         }
 484
 485         /* Set the watcher to oneshot in case other processes are also
 486          * watching to accept(). */
 487         r = sd_event_source_set_enabled(w_accept, SD_EVENT_ONESHOT);
 488         if (r < 0) {
 489                 log_error("Error %d while setting event IO source to ONESHOT: %s", r, strerror(-r));
 490                 return r;
 491         }
 492
 493         log_debug("Initialized main listener. Entering loop.");
 494
 495         return sd_event_loop(e);
 496 }
 497
 498 static int help(void) {
 499
 500         printf("%s hostname-or-ip port-or-service\n"
 501                "%s unix-domain-socket-path\n\n"
 502                "Inherit a socket. Bidirectionally proxy.\n\n"
 503                "  -h --help       Show this help\n"
 504                "  --version       Print version and exit\n"
 505                "  --ignore-env    Ignore expected systemd environment\n",
 506                program_invocation_short_name,
 507                program_invocation_short_name);
 508
 509         return 0;
 510 }
 511
 512 static void version(void) {
 513         puts(PACKAGE_STRING " socket-proxyd");
 514 }
 515
 516 static int parse_argv(int argc, char *argv[], struct proxy *p) {
 517
 518         enum {
 519                 ARG_VERSION = 0x100,
 520                 ARG_IGNORE_ENV
 521         };
 522
 523         static const struct option options[] = {
 524                 { "help",       no_argument, NULL, 'h'           },
 525                 { "version",    no_argument, NULL, ARG_VERSION   },
 526                 { "ignore-env", no_argument, NULL, ARG_IGNORE_ENV},
 527                 { NULL,         0,           NULL, 0             }
 528         };
 529
 530         int c;
 531
 532         assert(argc >= 0);
 533         assert(argv);
 534
 535         while ((c = getopt_long(argc, argv, "h", options, NULL)) >= 0) {
 536
 537                 switch (c) {
 538
 539                 case 'h':
 540                         help();
 541                         return 0;
 542
 543                 case '?':
 544                         return -EINVAL;
 545
 546                 case ARG_VERSION:
 547                         version();
 548                         return 0;
 549
 550                 case ARG_IGNORE_ENV:
 551                         p->ignore_env = true;
 552                         continue;
 553
 554                 default:
 555                         log_error("Unknown option code %c", c);
 556                         return -EINVAL;
 557                 }
 558         }
 559
 560         if (optind + 1 != argc && optind + 2 != argc) {
 561                 log_error("Incorrect number of positional arguments.");
 562                 help();
 563                 return -EINVAL;
 564         }
 565
 566         p->remote_host = argv[optind];
 567         assert(p->remote_host);
 568
 569         p->remote_is_inet = p->remote_host[0] != '/';
 570
 571         if (optind == argc - 2) {
 572                 if (!p->remote_is_inet) {
 573                         log_error("A port or service is not allowed for Unix socket destinations.");
 574                         help();
 575                         return -EINVAL;
 576                 }
 577                 p->remote_service = argv[optind + 1];
 578                 assert(p->remote_service);
 579         } else if (p->remote_is_inet) {
 580                 log_error("A port or service is required for IP destinations.");
 581                 help();
 582                 return -EINVAL;
 583         }
 584
 585         return 1;
 586 }
 587
 588 int main(int argc, char *argv[]) {
 589         struct proxy p = {};
 590         int r;
 591
 592         log_parse_environment();
 593         log_open();
 594
 595         r = parse_argv(argc, argv, &p);
 596         if (r <= 0)
 597                 goto finish;
 598
 599         p.listen_fd = SD_LISTEN_FDS_START;
 600
 601         if (!p.ignore_env) {
 602                 int n;
 603                 n = sd_listen_fds(1);
 604                 if (n == 0) {
 605                         log_error("Found zero inheritable sockets. Are you sure this is running as a socket-activated service?");
 606                         r = EXIT_FAILURE;
 607                         goto finish;
 608                 } else if (n < 0) {
 609                         log_error("Error %d while finding inheritable sockets: %s", n, strerror(-n));
 610                         r = EXIT_FAILURE;
 611                         goto finish;
 612                 } else if (n > 1) {
 613                         log_error("Can't listen on more than one socket.");
 614                         r = EXIT_FAILURE;
 615                         goto finish;
 616                 }
 617         }
 618
 619         r = sd_is_socket(p.listen_fd, 0, SOCK_STREAM, 1);
 620         if (r < 0) {
 621                 log_error("Error %d while checking inherited socket: %s", r, strerror(-r));
 622                 goto finish;
 623         }
 624
 625         log_info("Starting the socket activation proxy with listener fd=%d.", p.listen_fd);
 626
 627         r = run_main_loop(&p);
 628
 629 finish:
 630         return r < 0 ? EXIT_FAILURE : EXIT_SUCCESS;
 631 }