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 #include "build.h"
  41
  42 #define BUFFER_SIZE 16384
  43 #define _cleanup_freeaddrinfo_ _cleanup_(freeaddrinfop)
  44
  45 unsigned int total_clients = 0;
  46
  47 DEFINE_TRIVIAL_CLEANUP_FUNC(struct addrinfo *, freeaddrinfo);
  48
  49 struct proxy {
  50         int listen_fd;
  51         bool ignore_env;
  52         bool remote_is_inet;
  53         const char *remote_host;
  54         const char *remote_service;
  55 };
  56
  57 struct connection {
  58         int fd;
  59         uint32_t events;
  60         sd_event_source *w;
  61         struct connection *c_destination;
  62         size_t buffer_filled_len;
  63         size_t buffer_sent_len;
  64         char buffer[BUFFER_SIZE];
  65 };
  66
  67 static void free_connection(struct connection *c) {
  68         if (c != NULL) {
  69                 log_debug("Freeing fd=%d (conn %p).", c->fd, c);
  70                 sd_event_source_unref(c->w);
  71                 if (c->fd > 0)
  72                         close_nointr_nofail(c->fd);
  73                 free(c);
  74         }
  75 }
  76
  77 static int add_event_to_connection(struct connection *c, uint32_t events) {
  78         int r;
  79
  80         log_debug("Have revents=%d. Adding revents=%d.", c->events, events);
  81
  82         c->events |= events;
  83
  84         r = sd_event_source_set_io_events(c->w, c->events);
  85         if (r < 0) {
  86                 log_error("Error %d setting revents: %s", r, strerror(-r));
  87                 return r;
  88         }
  89
  90         r = sd_event_source_set_enabled(c->w, SD_EVENT_ON);
  91         if (r < 0) {
  92                 log_error("Error %d enabling source: %s", r, strerror(-r));
  93                 return r;
  94         }
  95
  96         return 0;
  97 }
  98
  99 static int remove_event_from_connection(struct connection *c, uint32_t events) {
 100         int r;
 101
 102         log_debug("Have revents=%d. Removing revents=%d.", c->events, events);
 103
 104         c->events &= ~events;
 105
 106         r = sd_event_source_set_io_events(c->w, c->events);
 107         if (r < 0) {
 108                 log_error("Error %d setting revents: %s", r, strerror(-r));
 109                 return r;
 110         }
 111
 112         if (c->events == 0) {
 113                 r = sd_event_source_set_enabled(c->w, SD_EVENT_OFF);
 114                 if (r < 0) {
 115                         log_error("Error %d disabling source: %s", r, strerror(-r));
 116                         return r;
 117                 }
 118         }
 119
 120         return 0;
 121 }
 122
 123 static int send_buffer(struct connection *sender) {
 124         struct connection *receiver = sender->c_destination;
 125         ssize_t len;
 126         int r = 0;
 127
 128         /* We cannot assume that even a partial send() indicates that
 129          * the next send() will return EAGAIN or EWOULDBLOCK. Loop until
 130          * it does. */
 131         while (sender->buffer_filled_len > sender->buffer_sent_len) {
 132                 len = send(receiver->fd, sender->buffer + sender->buffer_sent_len, sender->buffer_filled_len - sender->buffer_sent_len, 0);
 133                 log_debug("send(%d, ...)=%zd", receiver->fd, len);
 134                 if (len < 0) {
 135                         if (errno != EWOULDBLOCK && errno != EAGAIN) {
 136                                 log_error("Error %d in send to fd=%d: %m", errno, receiver->fd);
 137                                 return -errno;
 138                         }
 139                         else {
 140                                 /* send() is in a would-block state. */
 141                                 break;
 142                         }
 143                 }
 144
 145                 /* len < 0 can't occur here. len == 0 is possible but
 146                  * undefined behavior for nonblocking send(). */
 147                 assert(len > 0);
 148                 sender->buffer_sent_len += len;
 149         }
 150
 151         log_debug("send(%d, ...) completed with %zu bytes still buffered.", receiver->fd, sender->buffer_filled_len - sender->buffer_sent_len);
 152
 153         /* Detect a would-block state or partial send. */
 154         if (sender->buffer_filled_len > sender->buffer_sent_len) {
 155
 156                 /* If the buffer is full, disable events coming for recv. */
 157                 if (sender->buffer_filled_len == BUFFER_SIZE) {
 158                         r = remove_event_from_connection(sender, EPOLLIN);
 159                         if (r < 0) {
 160                                 log_error("Error %d disabling EPOLLIN for fd=%d: %s", r, sender->fd, strerror(-r));
 161                                 return r;
 162                         }
 163                 }
 164
 165                 /* Watch for when the recipient can be sent data again. */
 166                 r = add_event_to_connection(receiver, EPOLLOUT);
 167                 if (r < 0) {
 168                         log_error("Error %d enabling EPOLLOUT for fd=%d: %s", r, receiver->fd, strerror(-r));
 169                         return r;
 170                 }
 171                 log_debug("Done with recv for fd=%d. Waiting on send for fd=%d.", sender->fd, receiver->fd);
 172                 return r;
 173         }
 174
 175         /* If we sent everything without any issues (would-block or
 176          * partial send), the buffer is now empty. */
 177         sender->buffer_filled_len = 0;
 178         sender->buffer_sent_len = 0;
 179
 180         /* Enable the sender's receive watcher, in case the buffer was
 181          * full and we disabled it. */
 182         r = add_event_to_connection(sender, EPOLLIN);
 183         if (r < 0) {
 184                 log_error("Error %d enabling EPOLLIN for fd=%d: %s", r, sender->fd, strerror(-r));
 185                 return r;
 186         }
 187
 188         /* Disable the other side's send watcher, as we have no data to send now. */
 189         r = remove_event_from_connection(receiver, EPOLLOUT);
 190         if (r < 0) {
 191                 log_error("Error %d disabling EPOLLOUT for fd=%d: %s", r, receiver->fd, strerror(-r));
 192                 return r;
 193         }
 194
 195         return 0;
 196 }
 197
 198 static int transfer_data_cb(sd_event_source *s, int fd, uint32_t revents, void *userdata) {
 199         struct connection *c = (struct connection *) userdata;
 200         int r = 0;
 201         ssize_t len;
 202
 203         assert(revents & (EPOLLIN | EPOLLOUT));
 204         assert(fd == c->fd);
 205         assert(s == c->w);
 206
 207         log_debug("Got event revents=%d from fd=%d (conn %p).", revents, fd, c);
 208
 209         if (revents & EPOLLIN) {
 210                 log_debug("About to recv up to %zu bytes from fd=%d (%zu/BUFFER_SIZE).", BUFFER_SIZE - c->buffer_filled_len, fd, c->buffer_filled_len);
 211
 212                 /* Receive until the buffer's full, there's no more data,
 213                  * or the client/server disconnects. */
 214                 while (c->buffer_filled_len < BUFFER_SIZE) {
 215                         len = recv(fd, c->buffer + c->buffer_filled_len, BUFFER_SIZE - c->buffer_filled_len, 0);
 216                         log_debug("recv(%d, ...)=%zd", fd, len);
 217                         if (len < 0) {
 218                                 if (errno != EWOULDBLOCK && errno != EAGAIN) {
 219                                         log_error("Error %d in recv from fd=%d: %m", errno, fd);
 220                                         return -errno;
 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 %zd more bytes full.", len);
 236                         c->buffer_filled_len += len;
 237                         log_debug("Buffer now has %zu 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                         break;
 448                 }
 449         }
 450
 451         /* Re-enable the watcher. */
 452         r = sd_event_source_set_enabled(s, SD_EVENT_ONESHOT);
 453         if (r < 0) {
 454                 log_error("Error %d while re-enabling listener with ONESHOT: %s", r, strerror(-r));
 455                 return r;
 456         }
 457
 458         /* Preserve the main loop even if a single accept() fails. */
 459         return 1;
 460 }
 461
 462 static int run_main_loop(struct proxy *proxy) {
 463         _cleanup_event_source_unref_ sd_event_source *w_accept = NULL;
 464         _cleanup_event_unref_ sd_event *e = NULL;
 465         int r = EXIT_SUCCESS;
 466
 467         r = sd_event_new(&e);
 468         if (r < 0) {
 469                 log_error("Failed to allocate event loop: %s", strerror(-r));
 470                 return r;
 471         }
 472
 473         r = fd_nonblock(proxy->listen_fd, true);
 474         if (r < 0) {
 475                 log_error("Failed to make listen file descriptor nonblocking: %s", strerror(-r));
 476                 return r;
 477         }
 478
 479         log_debug("Initializing main listener fd=%d", proxy->listen_fd);
 480
 481         r = sd_event_add_io(e, proxy->listen_fd, EPOLLIN, accept_cb, proxy, &w_accept);
 482         if (r < 0) {
 483                 log_error("Error %d while adding event IO source: %s", r, strerror(-r));
 484                 return r;
 485         }
 486
 487         /* Set the watcher to oneshot in case other processes are also
 488          * watching to accept(). */
 489         r = sd_event_source_set_enabled(w_accept, SD_EVENT_ONESHOT);
 490         if (r < 0) {
 491                 log_error("Error %d while setting event IO source to ONESHOT: %s", r, strerror(-r));
 492                 return r;
 493         }
 494
 495         log_debug("Initialized main listener. Entering loop.");
 496
 497         return sd_event_loop(e);
 498 }
 499
 500 static int help(void) {
 501
 502         printf("%s hostname-or-ip port-or-service\n"
 503                "%s unix-domain-socket-path\n\n"
 504                "Inherit a socket. Bidirectionally proxy.\n\n"
 505                "  -h --help       Show this help\n"
 506                "  --version       Print version and exit\n"
 507                "  --ignore-env    Ignore expected systemd environment\n",
 508                program_invocation_short_name,
 509                program_invocation_short_name);
 510
 511         return 0;
 512 }
 513
 514 static int parse_argv(int argc, char *argv[], struct proxy *p) {
 515
 516         enum {
 517                 ARG_VERSION = 0x100,
 518                 ARG_IGNORE_ENV
 519         };
 520
 521         static const struct option options[] = {
 522                 { "help",       no_argument, NULL, 'h'           },
 523                 { "version",    no_argument, NULL, ARG_VERSION   },
 524                 { "ignore-env", no_argument, NULL, ARG_IGNORE_ENV},
 525                 {}
 526         };
 527
 528         int c;
 529
 530         assert(argc >= 0);
 531         assert(argv);
 532
 533         while ((c = getopt_long(argc, argv, "h", options, NULL)) >= 0) {
 534
 535                 switch (c) {
 536
 537                 case 'h':
 538                         return help();
 539
 540                 case ARG_VERSION:
 541                         puts(PACKAGE_STRING);
 542                         puts(SYSTEMD_FEATURES);
 543                         return 0;
 544
 545                 case ARG_IGNORE_ENV:
 546                         p->ignore_env = true;
 547                         continue;
 548
 549                 case '?':
 550                         return -EINVAL;
 551
 552                 default:
 553                         assert_not_reached("Unhandled option");
 554                 }
 555         }
 556
 557         if (optind + 1 != argc && optind + 2 != argc) {
 558                 log_error("Incorrect number of positional arguments.");
 559                 help();
 560                 return -EINVAL;
 561         }
 562
 563         p->remote_host = argv[optind];
 564         assert(p->remote_host);
 565
 566         p->remote_is_inet = p->remote_host[0] != '/';
 567
 568         if (optind == argc - 2) {
 569                 if (!p->remote_is_inet) {
 570                         log_error("A port or service is not allowed for Unix socket destinations.");
 571                         help();
 572                         return -EINVAL;
 573                 }
 574                 p->remote_service = argv[optind + 1];
 575                 assert(p->remote_service);
 576         } else if (p->remote_is_inet) {
 577                 log_error("A port or service is required for IP destinations.");
 578                 help();
 579                 return -EINVAL;
 580         }
 581
 582         return 1;
 583 }
 584
 585 int main(int argc, char *argv[]) {
 586         struct proxy p = {};
 587         int r;
 588
 589         log_parse_environment();
 590         log_open();
 591
 592         r = parse_argv(argc, argv, &p);
 593         if (r <= 0)
 594                 goto finish;
 595
 596         p.listen_fd = SD_LISTEN_FDS_START;
 597
 598         if (!p.ignore_env) {
 599                 int n;
 600                 n = sd_listen_fds(1);
 601                 if (n == 0) {
 602                         log_error("Found zero inheritable sockets. Are you sure this is running as a socket-activated service?");
 603                         r = EXIT_FAILURE;
 604                         goto finish;
 605                 } else if (n < 0) {
 606                         log_error("Error %d while finding inheritable sockets: %s", n, strerror(-n));
 607                         r = EXIT_FAILURE;
 608                         goto finish;
 609                 } else if (n > 1) {
 610                         log_error("Can't listen on more than one socket.");
 611                         r = EXIT_FAILURE;
 612                         goto finish;
 613                 }
 614         }
 615
 616         r = sd_is_socket(p.listen_fd, 0, SOCK_STREAM, 1);
 617         if (r < 0) {
 618                 log_error("Error %d while checking inherited socket: %s", r, strerror(-r));
 619                 goto finish;
 620         }
 621
 622         log_info("Starting the socket activation proxy with listener fd=%d.", p.listen_fd);
 623
 624         r = run_main_loop(&p);
 625
 626 finish:
 627         return r < 0 ? EXIT_FAILURE : EXIT_SUCCESS;
 628 }