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