src/journal/journal-remote.c

   1 /*-*- Mode: C; c-basic-offset: 8; indent-tabs-mode: nil -*-*/
   2
   3 /***
   4   This file is part of systemd.
   5
   6   Copyright 2012 Zbigniew Jędrzejewski-Szmek
   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 <errno.h>
  23 #include <fcntl.h>
  24 #include <inttypes.h>
  25 #include <stdio.h>
  26 #include <stdlib.h>
  27 #include <string.h>
  28 #include <sys/prctl.h>
  29 #include <sys/socket.h>
  30 #include <sys/stat.h>
  31 #include <sys/types.h>
  32 #include <unistd.h>
  33 #include <getopt.h>
  34
  35 #include "sd-daemon.h"
  36 #include "sd-event.h"
  37 #include "journal-file.h"
  38 #include "journald-native.h"
  39 #include "socket-util.h"
  40 #include "mkdir.h"
  41 #include "build.h"
  42 #include "macro.h"
  43 #include "strv.h"
  44 #include "fileio.h"
  45 #include "socket-util.h"
  46 #include "microhttpd-util.h"
  47
  48 #ifdef HAVE_GNUTLS
  49 #include <gnutls/gnutls.h>
  50 #endif
  51
  52 #include "journal-remote-parse.h"
  53 #include "journal-remote-write.h"
  54
  55 #define REMOTE_JOURNAL_PATH "/var/log/journal/" SD_ID128_FORMAT_STR "/remote-%s.journal"
  56
  57 static char* arg_output = NULL;
  58 static char* arg_url = NULL;
  59 static char* arg_getter = NULL;
  60 static bool arg_stdin = false;
  61 static char* arg_listen_raw = NULL;
  62 static char* arg_listen_http = NULL;
  63 static char* arg_listen_https = NULL;
  64 static int arg_compress = true;
  65 static int arg_seal = false;
  66
  67 static char *key_pem = NULL;
  68 static char *cert_pem = NULL;
  69 static char *trust_pem = NULL;
  70
  71 /**********************************************************************
  72  **********************************************************************
  73  **********************************************************************/
  74
  75 static int spawn_child(const char* child, char** argv) {
  76         int fd[2];
  77         pid_t parent_pid, child_pid;
  78         int r;
  79
  80         if (pipe(fd) < 0) {
  81                 log_error("Failed to create pager pipe: %m");
  82                 return -errno;
  83         }
  84
  85         parent_pid = getpid();
  86
  87         child_pid = fork();
  88         if (child_pid < 0) {
  89                 r = -errno;
  90                 log_error("Failed to fork: %m");
  91                 close_pipe(fd);
  92                 return r;
  93         }
  94
  95         /* In the child */
  96         if (child_pid == 0) {
  97                 r = dup2(fd[1], STDOUT_FILENO);
  98                 if (r < 0) {
  99                         log_error("Failed to dup pipe to stdout: %m");
 100                         _exit(EXIT_FAILURE);
 101                 }
 102
 103                 r = close_pipe(fd);
 104                 if (r < 0)
 105                         log_warning("Failed to close pipe fds: %m");
 106
 107                 /* Make sure the child goes away when the parent dies */
 108                 if (prctl(PR_SET_PDEATHSIG, SIGTERM) < 0)
 109                         _exit(EXIT_FAILURE);
 110
 111                 /* Check whether our parent died before we were able
 112                  * to set the death signal */
 113                 if (getppid() != parent_pid)
 114                         _exit(EXIT_SUCCESS);
 115
 116                 execvp(child, argv);
 117                 log_error("Failed to exec child %s: %m", child);
 118                 _exit(EXIT_FAILURE);
 119         }
 120
 121         r = close(fd[1]);
 122         if (r < 0)
 123                 log_warning("Failed to close write end of pipe: %m");
 124
 125         return fd[0];
 126 }
 127
 128 static int spawn_curl(char* url) {
 129         char **argv = STRV_MAKE("curl",
 130                                 "-HAccept: application/vnd.fdo.journal",
 131                                 "--silent",
 132                                 "--show-error",
 133                                 url);
 134         int r;
 135
 136         r = spawn_child("curl", argv);
 137         if (r < 0)
 138                 log_error("Failed to spawn curl: %m");
 139         return r;
 140 }
 141
 142 static int spawn_getter(char *getter, char *url) {
 143         int r;
 144         char _cleanup_strv_free_ **words = NULL, **words2 = NULL;
 145
 146         assert(getter);
 147         words = strv_split_quoted(getter);
 148         if (!words)
 149                 return log_oom();
 150
 151         r = spawn_child(words[0], words);
 152         if (r < 0)
 153                 log_error("Failed to spawn getter %s: %m", getter);
 154
 155         return r;
 156 }
 157
 158 static int open_output(Writer *s, const char* url) {
 159         char _cleanup_free_ *name, *output = NULL;
 160         char *c;
 161         int r;
 162
 163         assert(url);
 164         name = strdup(url);
 165         if (!name)
 166                 return log_oom();
 167
 168         for(c = name; *c; c++) {
 169                 if (*c == '/' || *c == ':' || *c == ' ')
 170                         *c = '~';
 171                 else if (*c == '?') {
 172                         *c = '\0';
 173                         break;
 174                 }
 175         }
 176
 177         if (!arg_output) {
 178                 sd_id128_t machine;
 179                 r = sd_id128_get_machine(&machine);
 180                 if (r < 0) {
 181                         log_error("failed to determine machine ID128: %s", strerror(-r));
 182                         return r;
 183                 }
 184
 185                 r = asprintf(&output, REMOTE_JOURNAL_PATH,
 186                              SD_ID128_FORMAT_VAL(machine), name);
 187                 if (r < 0)
 188                         return log_oom();
 189         } else {
 190                 r = is_dir(arg_output);
 191                 if (r > 0) {
 192                         r = asprintf(&output,
 193                                      "%s/remote-%s.journal", arg_output, name);
 194                         if (r < 0)
 195                                 return log_oom();
 196                 } else {
 197                         output = strdup(arg_output);
 198                         if (!output)
 199                                 return log_oom();
 200                 }
 201         }
 202
 203         r = journal_file_open_reliably(output,
 204                                        O_RDWR|O_CREAT, 0640,
 205                                        arg_compress, arg_seal,
 206                                        &s->metrics,
 207                                        s->mmap,
 208                                        NULL, &s->journal);
 209         if (r < 0)
 210                 log_error("Failed to open output journal %s: %s",
 211                           arg_output, strerror(-r));
 212         else
 213                 log_info("Opened output file %s", s->journal->path);
 214         return r;
 215 }
 216
 217 /**********************************************************************
 218  **********************************************************************
 219  **********************************************************************/
 220
 221 typedef struct MHDDaemonWrapper {
 222         uint64_t fd;
 223         struct MHD_Daemon *daemon;
 224
 225         sd_event_source *event;
 226 } MHDDaemonWrapper;
 227
 228 typedef struct RemoteServer {
 229         RemoteSource **sources;
 230         ssize_t sources_size;
 231         ssize_t active;
 232
 233         sd_event *events;
 234         sd_event_source *sigterm_event, *sigint_event, *listen_event;
 235
 236         Writer writer;
 237
 238         Hashmap *daemons;
 239 } RemoteServer;
 240
 241 /* This should go away as soon as µhttpd allows state to be passed around. */
 242 static RemoteServer *server;
 243
 244 static int dispatch_raw_source_event(sd_event_source *event,
 245                                      int fd,
 246                                      uint32_t revents,
 247                                      void *userdata);
 248 static int dispatch_raw_connection_event(sd_event_source *event,
 249                                          int fd,
 250                                          uint32_t revents,
 251                                          void *userdata);
 252 static int dispatch_http_event(sd_event_source *event,
 253                                int fd,
 254                                uint32_t revents,
 255                                void *userdata);
 256
 257 static int get_source_for_fd(RemoteServer *s, int fd, RemoteSource **source) {
 258         assert(fd >= 0);
 259         assert(source);
 260
 261         if (!GREEDY_REALLOC0_T(s->sources, s->sources_size, fd + 1))
 262                 return log_oom();
 263
 264         if (s->sources[fd] == NULL) {
 265                 s->sources[fd] = new0(RemoteSource, 1);
 266                 if (!s->sources[fd])
 267                         return log_oom();
 268                 s->sources[fd]->fd = -1;
 269                 s->active++;
 270         }
 271
 272         *source = s->sources[fd];
 273         return 0;
 274 }
 275
 276 static int remove_source(RemoteServer *s, int fd) {
 277         RemoteSource *source;
 278
 279         assert(s);
 280         assert(fd >= 0);
 281         assert(fd < s->sources_size);
 282
 283         source = s->sources[fd];
 284         if (source) {
 285                 source_free(source);
 286                 s->sources[fd] = NULL;
 287                 s->active--;
 288         }
 289
 290         close(fd);
 291
 292         return 0;
 293 }
 294
 295 static int add_source(RemoteServer *s, int fd, const char* name) {
 296         RemoteSource *source = NULL;
 297         char *realname;
 298         int r;
 299
 300         assert(s);
 301         assert(fd >= 0);
 302
 303         if (name) {
 304                 realname = strdup(name);
 305                 if (!realname)
 306                         return log_oom();
 307         } else {
 308                 r = asprintf(&realname, "fd:%d", fd);
 309                 if (r < 0)
 310                         return log_oom();
 311         }
 312
 313         log_debug("Creating source for fd:%d (%s)", fd, name);
 314
 315         r = get_source_for_fd(s, fd, &source);
 316         if (r < 0) {
 317                 log_error("Failed to create source for fd:%d (%s)", fd, name);
 318                 return r;
 319         }
 320         assert(source);
 321         assert(source->fd < 0);
 322         source->fd = fd;
 323
 324         r = sd_event_add_io(s->events, &source->event,
 325                             fd, EPOLLIN, dispatch_raw_source_event, s);
 326         if (r < 0) {
 327                 log_error("Failed to register event source for fd:%d: %s",
 328                           fd, strerror(-r));
 329                 goto error;
 330         }
 331
 332         return 1; /* work to do */
 333
 334  error:
 335         remove_source(s, fd);
 336         return r;
 337 }
 338
 339 static int setup_raw_socket(RemoteServer *s, const char *address) {
 340         int fd, r;
 341
 342         fd = make_socket_fd(LOG_INFO, address, SOCK_STREAM | SOCK_CLOEXEC);
 343         if (fd < 0)
 344                 return fd;
 345
 346         r = sd_event_add_io(s->events, &s->listen_event, fd, EPOLLIN,
 347                             dispatch_raw_connection_event, s);
 348         if (r < 0) {
 349                 close(fd);
 350                 return r;
 351         }
 352
 353         s->active ++;
 354         return 0;
 355 }
 356
 357 /**********************************************************************
 358  **********************************************************************
 359  **********************************************************************/
 360
 361 static RemoteSource *request_meta(void **connection_cls) {
 362         RemoteSource *source;
 363
 364         assert(connection_cls);
 365         if (*connection_cls)
 366                 return *connection_cls;
 367
 368         source = new0(RemoteSource, 1);
 369         if (!source)
 370                 return NULL;
 371         source->fd = -1;
 372
 373         log_debug("Added RemoteSource as connection metadata %p", source);
 374
 375         *connection_cls = source;
 376         return source;
 377 }
 378
 379 static void request_meta_free(void *cls,
 380                               struct MHD_Connection *connection,
 381                               void **connection_cls,
 382                               enum MHD_RequestTerminationCode toe) {
 383         RemoteSource *s;
 384
 385         assert(connection_cls);
 386         s = *connection_cls;
 387
 388         log_debug("Cleaning up connection metadata %p", s);
 389         source_free(s);
 390         *connection_cls = NULL;
 391 }
 392
 393 static int process_http_upload(
 394                 struct MHD_Connection *connection,
 395                 const char *upload_data,
 396                 size_t *upload_data_size,
 397                 RemoteSource *source) {
 398
 399         bool finished = false;
 400         int r;
 401
 402         assert(source);
 403
 404         log_debug("request_handler_upload: connection %p, %zu bytes",
 405                   connection, *upload_data_size);
 406
 407         if (*upload_data_size) {
 408                 log_info("Received %zu bytes", *upload_data_size);
 409
 410                 r = push_data(source, upload_data, *upload_data_size);
 411                 if (r < 0) {
 412                         log_error("Failed to store received data of size %zu: %s",
 413                                   *upload_data_size, strerror(-r));
 414                         return respond_oom_internal(connection);
 415                 }
 416                 *upload_data_size = 0;
 417         } else
 418                 finished = true;
 419
 420         while (true) {
 421                 r = process_source(source, &server->writer, arg_compress, arg_seal);
 422                 if (r == -E2BIG)
 423                         log_warning("Entry too big, skipped");
 424                 else if (r == -EAGAIN || r == -EWOULDBLOCK)
 425                         break;
 426                 else if (r < 0) {
 427                         log_warning("Failed to process data for connection %p", connection);
 428                         return respond_error(connection, MHD_HTTP_UNPROCESSABLE_ENTITY,
 429                                              "Processing failed: %s", strerror(-r));
 430                 }
 431         }
 432
 433         if (!finished)
 434                 return MHD_YES;
 435
 436         /* The upload is finished */
 437
 438         if (source_non_empty(source)) {
 439                 log_warning("EOF reached with incomplete data");
 440                 return respond_error(connection, MHD_HTTP_EXPECTATION_FAILED,
 441                                      "Trailing data not processed.");
 442         }
 443
 444         return respond_error(connection, MHD_HTTP_ACCEPTED, "OK.\n");
 445 };
 446
 447 static int request_handler(
 448                 void *cls,
 449                 struct MHD_Connection *connection,
 450                 const char *url,
 451                 const char *method,
 452                 const char *version,
 453                 const char *upload_data,
 454                 size_t *upload_data_size,
 455                 void **connection_cls) {
 456
 457         const char *header;
 458         int r ,code;
 459
 460         assert(connection);
 461         assert(connection_cls);
 462         assert(url);
 463         assert(method);
 464
 465         log_debug("Handling a connection %s %s %s", method, url, version);
 466
 467         if (*connection_cls)
 468                 return process_http_upload(connection,
 469                                            upload_data, upload_data_size,
 470                                            *connection_cls);
 471
 472         if (!streq(method, "POST"))
 473                 return respond_error(connection, MHD_HTTP_METHOD_NOT_ACCEPTABLE,
 474                                      "Unsupported method.\n");
 475
 476         if (!streq(url, "/upload"))
 477                 return respond_error(connection, MHD_HTTP_NOT_FOUND,
 478                                      "Not found.\n");
 479
 480         header = MHD_lookup_connection_value(connection,
 481                                              MHD_HEADER_KIND, "Content-Type");
 482         if (!header || !streq(header, "application/vnd.fdo.journal"))
 483                 return respond_error(connection, MHD_HTTP_UNSUPPORTED_MEDIA_TYPE,
 484                                      "Content-Type: application/vnd.fdo.journal"
 485                                      " is required.\n");
 486
 487         if (trust_pem) {
 488                 r = check_permissions(connection, &code);
 489                 if (r < 0)
 490                         return code;
 491         }
 492
 493         if (!request_meta(connection_cls))
 494                 return respond_oom(connection);
 495         return MHD_YES;
 496 }
 497
 498 static int setup_microhttpd_server(RemoteServer *s, int fd, bool https) {
 499         struct MHD_OptionItem opts[] = {
 500                 { MHD_OPTION_NOTIFY_COMPLETED, (intptr_t) request_meta_free},
 501                 { MHD_OPTION_EXTERNAL_LOGGER, (intptr_t) microhttpd_logger},
 502                 { MHD_OPTION_LISTEN_SOCKET, fd},
 503                 { MHD_OPTION_END},
 504                 { MHD_OPTION_END},
 505                 { MHD_OPTION_END},
 506                 { MHD_OPTION_END}};
 507         int opts_pos = 3;
 508         int flags =
 509                 MHD_USE_DEBUG |
 510                 MHD_USE_PEDANTIC_CHECKS |
 511                 MHD_USE_EPOLL_LINUX_ONLY |
 512                 MHD_USE_DUAL_STACK;
 513
 514         const union MHD_DaemonInfo *info;
 515         int r, epoll_fd;
 516         MHDDaemonWrapper *d;
 517
 518         assert(fd >= 0);
 519
 520         r = fd_nonblock(fd, true);
 521         if (r < 0) {
 522                 log_error("Failed to make fd:%d nonblocking: %s", fd, strerror(-r));
 523                 return r;
 524         }
 525
 526         if (https) {
 527                 opts[opts_pos++] = (struct MHD_OptionItem)
 528                         {MHD_OPTION_HTTPS_MEM_KEY, 0, key_pem};
 529                 opts[opts_pos++] = (struct MHD_OptionItem)
 530                         {MHD_OPTION_HTTPS_MEM_CERT, 0, cert_pem};
 531
 532                 flags |= MHD_USE_SSL;
 533
 534                 if (trust_pem)
 535                         opts[opts_pos++] = (struct MHD_OptionItem)
 536                                 {MHD_OPTION_HTTPS_MEM_TRUST, 0, trust_pem};
 537         }
 538
 539         d = new(MHDDaemonWrapper, 1);
 540         if (!d)
 541                 return log_oom();
 542
 543         d->fd = (uint64_t) fd;
 544
 545         d->daemon = MHD_start_daemon(flags, 0,
 546                                      NULL, NULL,
 547                                      request_handler, NULL,
 548                                      MHD_OPTION_ARRAY, opts,
 549                                      MHD_OPTION_END);
 550         if (!d->daemon) {
 551                 log_error("Failed to start µhttp daemon");
 552                 r = -EINVAL;
 553                 goto error;
 554         }
 555
 556         log_debug("Started MHD %s daemon on fd:%d (wrapper @ %p)",
 557                   https ? "HTTPS" : "HTTP", fd, d);
 558
 559
 560         info = MHD_get_daemon_info(d->daemon, MHD_DAEMON_INFO_EPOLL_FD_LINUX_ONLY);
 561         if (!info) {
 562                 log_error("µhttp returned NULL daemon info");
 563                 r = -ENOTSUP;
 564                 goto error;
 565         }
 566
 567         epoll_fd = info->listen_fd;
 568         if (epoll_fd < 0) {
 569                 log_error("µhttp epoll fd is invalid");
 570                 r = -EUCLEAN;
 571                 goto error;
 572         }
 573
 574         r = sd_event_add_io(s->events, &d->event,
 575                             epoll_fd, EPOLLIN, dispatch_http_event, d);
 576         if (r < 0) {
 577                 log_error("Failed to add event callback: %s", strerror(-r));
 578                 goto error;
 579         }
 580
 581         r = hashmap_ensure_allocated(&s->daemons, uint64_hash_func, uint64_compare_func);
 582         if (r < 0) {
 583                 log_oom();
 584                 goto error;
 585         }
 586
 587         r = hashmap_put(s->daemons, &d->fd, d);
 588         if (r < 0) {
 589                 log_error("Failed to add daemon to hashmap: %s", strerror(-r));
 590                 goto error;
 591         }
 592
 593         s->active ++;
 594         return 0;
 595
 596 error:
 597         MHD_stop_daemon(d->daemon);
 598         free(d->daemon);
 599         free(d);
 600         return r;
 601 }
 602
 603 static int setup_microhttpd_socket(RemoteServer *s,
 604                                    const char *address,
 605                                    bool https) {
 606         int fd;
 607
 608         fd = make_socket_fd(LOG_INFO, address, SOCK_STREAM | SOCK_CLOEXEC);
 609         if (fd < 0)
 610                 return fd;
 611
 612         return setup_microhttpd_server(s, fd, https);
 613 }
 614
 615 static int dispatch_http_event(sd_event_source *event,
 616                                int fd,
 617                                uint32_t revents,
 618                                void *userdata) {
 619         MHDDaemonWrapper *d = userdata;
 620         int r;
 621
 622         assert(d);
 623
 624         log_info("%s", __func__);
 625
 626         r = MHD_run(d->daemon);
 627         if (r == MHD_NO) {
 628                 log_error("MHD_run failed!");
 629                 // XXX: unregister daemon
 630                 return -EINVAL;
 631         }
 632
 633         return 1; /* work to do */
 634 }
 635
 636 /**********************************************************************
 637  **********************************************************************
 638  **********************************************************************/
 639
 640 static int dispatch_sigterm(sd_event_source *event,
 641                             const struct signalfd_siginfo *si,
 642                             void *userdata) {
 643         RemoteServer *s = userdata;
 644
 645         assert(s);
 646
 647         log_received_signal(LOG_INFO, si);
 648
 649         sd_event_exit(s->events, 0);
 650         return 0;
 651 }
 652
 653 static int setup_signals(RemoteServer *s) {
 654         sigset_t mask;
 655         int r;
 656
 657         assert(s);
 658
 659         assert_se(sigemptyset(&mask) == 0);
 660         sigset_add_many(&mask, SIGINT, SIGTERM, -1);
 661         assert_se(sigprocmask(SIG_SETMASK, &mask, NULL) == 0);
 662
 663         r = sd_event_add_signal(s->events, &s->sigterm_event, SIGTERM, dispatch_sigterm, s);
 664         if (r < 0)
 665                 return r;
 666
 667         r = sd_event_add_signal(s->events, &s->sigint_event, SIGINT, dispatch_sigterm, s);
 668         if (r < 0)
 669                 return r;
 670
 671         return 0;
 672 }
 673
 674 static int remoteserver_init(RemoteServer *s) {
 675         int r, n, fd;
 676         const char *output_name = NULL;
 677
 678         assert(s);
 679
 680         sd_event_default(&s->events);
 681
 682         setup_signals(s);
 683
 684         assert(server == NULL);
 685         server = s;
 686
 687         n = sd_listen_fds(true);
 688         if (n < 0) {
 689                 log_error("Failed to read listening file descriptors from environment: %s",
 690                           strerror(-n));
 691                 return n;
 692         } else
 693                 log_info("Received %d descriptors", n);
 694
 695         for (fd = SD_LISTEN_FDS_START; fd < SD_LISTEN_FDS_START + n; fd++) {
 696                 if (sd_is_socket(fd, AF_UNSPEC, 0, false)) {
 697                         assert_not_reached("not implemented");
 698                 } else if (sd_is_socket(fd, AF_UNSPEC, 0, true)) {
 699                         log_info("Received a connection socket (fd:%d)", fd);
 700
 701                         r = add_source(s, fd, NULL);
 702                         output_name = "socket";
 703                 } else {
 704                         log_error("Unknown socket passed on fd:%d", fd);
 705                         return -EINVAL;
 706                 }
 707         }
 708
 709         if (arg_url) {
 710                 char _cleanup_free_ *url = NULL;
 711                 char _cleanup_strv_free_ **urlv = strv_new(arg_url, "/entries", NULL);
 712                 if (!urlv)
 713                         return log_oom();
 714                 url = strv_join(urlv, "");
 715                 if (!url)
 716                         return log_oom();
 717
 718                 if (arg_getter) {
 719                         log_info("Spawning getter %s...", url);
 720                         fd = spawn_getter(arg_getter, url);
 721                 } else {
 722                         log_info("Spawning curl %s...", url);
 723                         fd = spawn_curl(url);
 724                 }
 725                 if (fd < 0)
 726                         return fd;
 727
 728                 r = add_source(s, fd, arg_url);
 729                 if (r < 0)
 730                         return r;
 731
 732                 output_name = arg_url;
 733         }
 734
 735         if (arg_listen_raw) {
 736                 log_info("Listening on a socket...");
 737                 r = setup_raw_socket(s, arg_listen_raw);
 738                 if (r < 0)
 739                         return r;
 740
 741                 output_name = arg_listen_raw;
 742         }
 743
 744         if (arg_listen_http) {
 745                 r = setup_microhttpd_socket(s, arg_listen_http, false);
 746                 if (r < 0)
 747                         return r;
 748
 749                 output_name = arg_listen_http;
 750         }
 751
 752         if (arg_listen_https) {
 753                 r = setup_microhttpd_socket(s, arg_listen_https, true);
 754                 if (r < 0)
 755                         return r;
 756
 757                 output_name = arg_listen_https;
 758         }
 759
 760         if (arg_stdin) {
 761                 log_info("Reading standard input...");
 762                 r = add_source(s, STDIN_FILENO, "stdin");
 763                 if (r < 0)
 764                         return r;
 765
 766                 output_name = "stdin";
 767         }
 768
 769         if (s->active == 0) {
 770                 log_error("Zarro sources specified");
 771                 return -EINVAL;
 772         }
 773
 774         if (!!n + !!arg_url + !!arg_listen_raw + !!arg_stdin > 1)
 775                 output_name = "multiple";
 776
 777         r = writer_init(&s->writer);
 778         if (r < 0)
 779                 return r;
 780
 781         r = open_output(&s->writer, output_name);
 782         return r;
 783 }
 784
 785 static int server_destroy(RemoteServer *s) {
 786         int r;
 787         ssize_t i;
 788         MHDDaemonWrapper *d;
 789
 790         r = writer_close(&s->writer);
 791
 792         while ((d = hashmap_steal_first(s->daemons))) {
 793                 MHD_stop_daemon(d->daemon);
 794                 sd_event_source_unref(d->event);
 795                 free(d);
 796         }
 797
 798         hashmap_free(s->daemons);
 799
 800         assert(s->sources_size == 0 || s->sources);
 801         for (i = 0; i < s->sources_size; i++)
 802                 remove_source(s, i);
 803
 804         free(s->sources);
 805
 806         sd_event_source_unref(s->sigterm_event);
 807         sd_event_source_unref(s->sigint_event);
 808         sd_event_source_unref(s->listen_event);
 809         sd_event_unref(s->events);
 810
 811         /* fds that we're listening on remain open... */
 812
 813         return r;
 814 }
 815
 816 /**********************************************************************
 817  **********************************************************************
 818  **********************************************************************/
 819
 820 static int dispatch_raw_source_event(sd_event_source *event,
 821                                      int fd,
 822                                      uint32_t revents,
 823                                      void *userdata) {
 824
 825         RemoteServer *s = userdata;
 826         RemoteSource *source;
 827         int r;
 828
 829         assert(fd < s->sources_size);
 830         source = s->sources[fd];
 831         assert(source->fd == fd);
 832
 833         r = process_source(source, &s->writer, arg_compress, arg_seal);
 834         if (source->state == STATE_EOF) {
 835                 log_info("EOF reached with source fd:%d (%s)",
 836                          source->fd, source->name);
 837                 if (source_non_empty(source))
 838                         log_warning("EOF reached with incomplete data");
 839                 remove_source(s, source->fd);
 840                 log_info("%zd active source remaining", s->active);
 841         } else if (r == -E2BIG) {
 842                 log_error("Entry too big, skipped");
 843                 r = 1;
 844         }
 845
 846         return r;
 847 }
 848
 849 static int accept_connection(const char* type, int fd, SocketAddress *addr) {
 850         int fd2, r;
 851
 852         log_debug("Accepting new %s connection on fd:%d", type, fd);
 853         fd2 = accept4(fd, &addr->sockaddr.sa, &addr->size, SOCK_NONBLOCK|SOCK_CLOEXEC);
 854         if (fd2 < 0) {
 855                 log_error("accept() on fd:%d failed: %m", fd);
 856                 return -errno;
 857         }
 858
 859         switch(socket_address_family(addr)) {
 860         case AF_INET:
 861         case AF_INET6: {
 862                 char* _cleanup_free_ a = NULL;
 863
 864                 r = socket_address_print(addr, &a);
 865                 if (r < 0) {
 866                         log_error("socket_address_print(): %s", strerror(-r));
 867                         close(fd2);
 868                         return r;
 869                 }
 870
 871                 log_info("Accepted %s %s connection from %s",
 872                          type,
 873                          socket_address_family(addr) == AF_INET ? "IP" : "IPv6",
 874                          a);
 875
 876                 return fd2;
 877         };
 878         default:
 879                 log_error("Rejected %s connection with unsupported family %d",
 880                           type, socket_address_family(addr));
 881                 close(fd2);
 882
 883                 return -EINVAL;
 884         }
 885 }
 886
 887 static int dispatch_raw_connection_event(sd_event_source *event,
 888                                          int fd,
 889                                          uint32_t revents,
 890                                          void *userdata) {
 891         RemoteServer *s = userdata;
 892         int fd2;
 893         SocketAddress addr = {
 894                 .size = sizeof(union sockaddr_union),
 895                 .type = SOCK_STREAM,
 896         };
 897
 898         fd2 = accept_connection("raw", fd, &addr);
 899         if (fd2 < 0)
 900                 return fd2;
 901
 902         return add_source(s, fd2, NULL);
 903 }
 904
 905 /**********************************************************************
 906  **********************************************************************
 907  **********************************************************************/
 908
 909 static int help(void) {
 910         printf("%s [OPTIONS...]\n\n"
 911                "Write external journal events to a journal file.\n\n"
 912                "Options:\n"
 913                "  --url=URL            Read events from systemd-journal-gatewayd at URL\n"
 914                "  --getter=COMMAND     Read events from the output of COMMAND\n"
 915                "  --listen-raw=ADDR    Listen for connections at ADDR\n"
 916                "  --listen-http=ADDR   Listen for HTTP connections at ADDR\n"
 917                "  --listen-https=ADDR  Listen for HTTPS connections at ADDR\n"
 918                "  --stdin              Read events from standard input\n"
 919                "  -o --output=FILE|DIR Write output to FILE or DIR/external-*.journal\n"
 920                "  --[no-]compress      Use XZ-compression in the output journal (default: yes)\n"
 921                "  --[no-]seal          Use Event sealing in the output journal (default: no)\n"
 922                "  -h --help            Show this help and exit\n"
 923                "  --version            Print version string and exit\n"
 924                "\n"
 925                "Note: file descriptors from sd_listen_fds() will be consumed, too.\n"
 926                , program_invocation_short_name);
 927
 928         return 0;
 929 }
 930
 931 static int parse_argv(int argc, char *argv[]) {
 932         enum {
 933                 ARG_VERSION = 0x100,
 934                 ARG_URL,
 935                 ARG_LISTEN_RAW,
 936                 ARG_LISTEN_HTTP,
 937                 ARG_LISTEN_HTTPS,
 938                 ARG_STDIN,
 939                 ARG_GETTER,
 940                 ARG_COMPRESS,
 941                 ARG_NO_COMPRESS,
 942                 ARG_SEAL,
 943                 ARG_NO_SEAL,
 944                 ARG_KEY,
 945                 ARG_CERT,
 946                 ARG_TRUST,
 947         };
 948
 949         static const struct option options[] = {
 950                 { "help",         no_argument,       NULL, 'h'              },
 951                 { "version",      no_argument,       NULL, ARG_VERSION      },
 952                 { "url",          required_argument, NULL, ARG_URL          },
 953                 { "getter",       required_argument, NULL, ARG_GETTER       },
 954                 { "listen-raw",   required_argument, NULL, ARG_LISTEN_RAW   },
 955                 { "listen-http",  required_argument, NULL, ARG_LISTEN_HTTP  },
 956                 { "listen-https", required_argument, NULL, ARG_LISTEN_HTTPS },
 957                 { "stdin",        no_argument,       NULL, ARG_STDIN        },
 958                 { "output",       required_argument, NULL, 'o'              },
 959                 { "compress",     no_argument,       NULL, ARG_COMPRESS     },
 960                 { "no-compress",  no_argument,       NULL, ARG_NO_COMPRESS  },
 961                 { "seal",         no_argument,       NULL, ARG_SEAL         },
 962                 { "no-seal",      no_argument,       NULL, ARG_NO_SEAL      },
 963                 { "key",          required_argument, NULL, ARG_KEY          },
 964                 { "cert",         required_argument, NULL, ARG_CERT         },
 965                 { "trust",        required_argument, NULL, ARG_TRUST        },
 966                 {}
 967         };
 968
 969         int c, r;
 970
 971         assert(argc >= 0);
 972         assert(argv);
 973
 974         while ((c = getopt_long(argc, argv, "ho:", options, NULL)) >= 0)
 975                 switch(c) {
 976                 case 'h':
 977                         help();
 978                         return 0 /* done */;
 979
 980                 case ARG_VERSION:
 981                         puts(PACKAGE_STRING);
 982                         puts(SYSTEMD_FEATURES);
 983                         return 0 /* done */;
 984
 985                 case ARG_URL:
 986                         if (arg_url) {
 987                                 log_error("cannot currently set more than one --url");
 988                                 return -EINVAL;
 989                         }
 990
 991                         arg_url = optarg;
 992                         break;
 993
 994                 case ARG_GETTER:
 995                         if (arg_getter) {
 996                                 log_error("cannot currently use --getter more than once");
 997                                 return -EINVAL;
 998                         }
 999
1000                         arg_getter = optarg;
1001                         break;
1002
1003                 case ARG_LISTEN_RAW:
1004                         if (arg_listen_raw) {
1005                                 log_error("cannot currently use --listen-raw more than once");
1006                                 return -EINVAL;
1007                         }
1008
1009                         arg_listen_raw = optarg;
1010                         break;
1011
1012                 case ARG_LISTEN_HTTP:
1013                         if (arg_listen_http) {
1014                                 log_error("cannot currently use --listen-http more than once");
1015                                 return -EINVAL;
1016                         }
1017
1018                         arg_listen_http = optarg;
1019                         break;
1020
1021                 case ARG_LISTEN_HTTPS:
1022                         if (arg_listen_https) {
1023                                 log_error("cannot currently use --listen-https more than once");
1024                                 return -EINVAL;
1025                         }
1026
1027                         arg_listen_https = optarg;
1028                         break;
1029
1030                 case ARG_KEY:
1031                         if (key_pem) {
1032                                 log_error("Key file specified twice");
1033                                 return -EINVAL;
1034                         }
1035                         r = read_full_file(optarg, &key_pem, NULL);
1036                         if (r < 0) {
1037                                 log_error("Failed to read key file: %s", strerror(-r));
1038                                 return r;
1039                         }
1040                         assert(key_pem);
1041                         break;
1042
1043                 case ARG_CERT:
1044                         if (cert_pem) {
1045                                 log_error("Certificate file specified twice");
1046                                 return -EINVAL;
1047                         }
1048                         r = read_full_file(optarg, &cert_pem, NULL);
1049                         if (r < 0) {
1050                                 log_error("Failed to read certificate file: %s", strerror(-r));
1051                                 return r;
1052                         }
1053                         assert(cert_pem);
1054                         break;
1055
1056                 case ARG_TRUST:
1057 #ifdef HAVE_GNUTLS
1058                         if (trust_pem) {
1059                                 log_error("CA certificate file specified twice");
1060                                 return -EINVAL;
1061                         }
1062                         r = read_full_file(optarg, &trust_pem, NULL);
1063                         if (r < 0) {
1064                                 log_error("Failed to read CA certificate file: %s", strerror(-r));
1065                                 return r;
1066                         }
1067                         assert(trust_pem);
1068                         break;
1069 #else
1070                         log_error("Option --trust is not available.");
1071 #endif
1072
1073                 case ARG_STDIN:
1074                         arg_stdin = true;
1075                         break;
1076
1077                 case 'o':
1078                         if (arg_output) {
1079                                 log_error("cannot use --output/-o more than once");
1080                                 return -EINVAL;
1081                         }
1082
1083                         arg_output = optarg;
1084                         break;
1085
1086                 case ARG_COMPRESS:
1087                         arg_compress = true;
1088                         break;
1089                 case ARG_NO_COMPRESS:
1090                         arg_compress = false;
1091                         break;
1092                 case ARG_SEAL:
1093                         arg_seal = true;
1094                         break;
1095                 case ARG_NO_SEAL:
1096                         arg_seal = false;
1097                         break;
1098
1099                 case '?':
1100                         return -EINVAL;
1101
1102                 default:
1103                         log_error("Unknown option code %c", c);
1104                         return -EINVAL;
1105                 }
1106
1107         if (arg_listen_https && !(key_pem && cert_pem)) {
1108                 log_error("Options --key and --cert must be used when https sources are specified");
1109                 return -EINVAL;
1110         }
1111
1112         if (optind < argc) {
1113                 log_error("This program takes no positional arguments");
1114                 return -EINVAL;
1115         }
1116
1117         return 1 /* work to do */;
1118 }
1119
1120 static int setup_gnutls_logger(void) {
1121         if (!arg_listen_http && !arg_listen_https)
1122                 return 0;
1123
1124 #ifdef HAVE_GNUTLS
1125         gnutls_global_set_log_function(log_func_gnutls);
1126         gnutls_global_set_log_level(GNUTLS_LOG_LEVEL);
1127 #endif
1128
1129         return 0;
1130 }
1131
1132 int main(int argc, char **argv) {
1133         RemoteServer s = {};
1134         int r, r2;
1135
1136         log_set_max_level(LOG_DEBUG);
1137         log_show_color(true);
1138         log_parse_environment();
1139
1140         r = parse_argv(argc, argv);
1141         if (r <= 0)
1142                 return r == 0 ? EXIT_SUCCESS : EXIT_FAILURE;
1143
1144         r = setup_gnutls_logger();
1145         if (r < 0)
1146                 return EXIT_FAILURE;
1147
1148         if (remoteserver_init(&s) < 0)
1149                 return EXIT_FAILURE;
1150
1151         log_debug("%s running as pid %lu",
1152                   program_invocation_short_name, (unsigned long) getpid());
1153         sd_notify(false,
1154                   "READY=1\n"
1155                   "STATUS=Processing requests...");
1156
1157         while (s.active) {
1158                 r = sd_event_get_state(s.events);
1159                 if (r < 0)
1160                         break;
1161                 if (r == SD_EVENT_FINISHED)
1162                         break;
1163
1164                 r = sd_event_run(s.events, -1);
1165                 if (r < 0) {
1166                         log_error("Failed to run event loop: %s", strerror(-r));
1167                         break;
1168                 }
1169         }
1170
1171         log_info("Finishing after writing %" PRIu64 " entries", s.writer.seqnum);
1172         r2 = server_destroy(&s);
1173
1174         sd_notify(false, "STATUS=Shutting down...");
1175
1176         return r >= 0 && r2 >= 0 ? EXIT_SUCCESS : EXIT_FAILURE;
1177 }