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                         log_error("Failed to store received data of size %zu: %s",
 417                                   *upload_data_size, strerror(-r));
 418                         return mhd_respond_oom(connection);
 419                 }
 420                 *upload_data_size = 0;
 421         } else
 422                 finished = true;
 423
 424         while (true) {
 425                 r = process_source(source, &server->writer, arg_compress, arg_seal);
 426                 if (r == -E2BIG)
 427                         log_warning("Entry too big, skipped");
 428                 else if (r == -EAGAIN || r == -EWOULDBLOCK)
 429                         break;
 430                 else if (r < 0) {
 431                         log_warning("Failed to process data for connection %p", connection);
 432                         return mhd_respondf(connection, MHD_HTTP_UNPROCESSABLE_ENTITY,
 433                                             "Processing failed: %s", strerror(-r));
 434                 }
 435         }
 436
 437         if (!finished)
 438                 return MHD_YES;
 439
 440         /* The upload is finished */
 441
 442         if (source_non_empty(source)) {
 443                 log_warning("EOF reached with incomplete data");
 444                 return mhd_respond(connection, MHD_HTTP_EXPECTATION_FAILED,
 445                                    "Trailing data not processed.");
 446         }
 447
 448         return mhd_respond(connection, MHD_HTTP_ACCEPTED, "OK.\n");
 449 };
 450
 451 static int request_handler(
 452                 void *cls,
 453                 struct MHD_Connection *connection,
 454                 const char *url,
 455                 const char *method,
 456                 const char *version,
 457                 const char *upload_data,
 458                 size_t *upload_data_size,
 459                 void **connection_cls) {
 460
 461         const char *header;
 462         int r ,code;
 463
 464         assert(connection);
 465         assert(connection_cls);
 466         assert(url);
 467         assert(method);
 468
 469         log_debug("Handling a connection %s %s %s", method, url, version);
 470
 471         if (*connection_cls)
 472                 return process_http_upload(connection,
 473                                            upload_data, upload_data_size,
 474                                            *connection_cls);
 475
 476         if (!streq(method, "POST"))
 477                 return mhd_respond(connection, MHD_HTTP_METHOD_NOT_ACCEPTABLE,
 478                                    "Unsupported method.\n");
 479
 480         if (!streq(url, "/upload"))
 481                 return mhd_respond(connection, MHD_HTTP_NOT_FOUND,
 482                                    "Not found.\n");
 483
 484         header = MHD_lookup_connection_value(connection,
 485                                              MHD_HEADER_KIND, "Content-Type");
 486         if (!header || !streq(header, "application/vnd.fdo.journal"))
 487                 return mhd_respond(connection, MHD_HTTP_UNSUPPORTED_MEDIA_TYPE,
 488                                    "Content-Type: application/vnd.fdo.journal"
 489                                    " is required.\n");
 490
 491         if (trust_pem) {
 492                 r = check_permissions(connection, &code);
 493                 if (r < 0)
 494                         return code;
 495         }
 496
 497         if (!request_meta(connection_cls))
 498                 return respond_oom(connection);
 499         return MHD_YES;
 500 }
 501
 502 static int setup_microhttpd_server(RemoteServer *s, int fd, bool https) {
 503         struct MHD_OptionItem opts[] = {
 504                 { MHD_OPTION_NOTIFY_COMPLETED, (intptr_t) request_meta_free},
 505                 { MHD_OPTION_EXTERNAL_LOGGER, (intptr_t) microhttpd_logger},
 506                 { MHD_OPTION_LISTEN_SOCKET, fd},
 507                 { MHD_OPTION_END},
 508                 { MHD_OPTION_END},
 509                 { MHD_OPTION_END},
 510                 { MHD_OPTION_END}};
 511         int opts_pos = 3;
 512         int flags =
 513                 MHD_USE_DEBUG |
 514                 MHD_USE_PEDANTIC_CHECKS |
 515                 MHD_USE_EPOLL_LINUX_ONLY |
 516                 MHD_USE_DUAL_STACK;
 517
 518         const union MHD_DaemonInfo *info;
 519         int r, epoll_fd;
 520         MHDDaemonWrapper *d;
 521
 522         assert(fd >= 0);
 523
 524         r = fd_nonblock(fd, true);
 525         if (r < 0) {
 526                 log_error("Failed to make fd:%d nonblocking: %s", fd, strerror(-r));
 527                 return r;
 528         }
 529
 530         if (https) {
 531                 opts[opts_pos++] = (struct MHD_OptionItem)
 532                         {MHD_OPTION_HTTPS_MEM_KEY, 0, key_pem};
 533                 opts[opts_pos++] = (struct MHD_OptionItem)
 534                         {MHD_OPTION_HTTPS_MEM_CERT, 0, cert_pem};
 535
 536                 flags |= MHD_USE_SSL;
 537
 538                 if (trust_pem)
 539                         opts[opts_pos++] = (struct MHD_OptionItem)
 540                                 {MHD_OPTION_HTTPS_MEM_TRUST, 0, trust_pem};
 541         }
 542
 543         d = new(MHDDaemonWrapper, 1);
 544         if (!d)
 545                 return log_oom();
 546
 547         d->fd = (uint64_t) fd;
 548
 549         d->daemon = MHD_start_daemon(flags, 0,
 550                                      NULL, NULL,
 551                                      request_handler, NULL,
 552                                      MHD_OPTION_ARRAY, opts,
 553                                      MHD_OPTION_END);
 554         if (!d->daemon) {
 555                 log_error("Failed to start µhttp daemon");
 556                 r = -EINVAL;
 557                 goto error;
 558         }
 559
 560         log_debug("Started MHD %s daemon on fd:%d (wrapper @ %p)",
 561                   https ? "HTTPS" : "HTTP", fd, d);
 562
 563
 564         info = MHD_get_daemon_info(d->daemon, MHD_DAEMON_INFO_EPOLL_FD_LINUX_ONLY);
 565         if (!info) {
 566                 log_error("µhttp returned NULL daemon info");
 567                 r = -ENOTSUP;
 568                 goto error;
 569         }
 570
 571         epoll_fd = info->listen_fd;
 572         if (epoll_fd < 0) {
 573                 log_error("µhttp epoll fd is invalid");
 574                 r = -EUCLEAN;
 575                 goto error;
 576         }
 577
 578         r = sd_event_add_io(s->events, &d->event,
 579                             epoll_fd, EPOLLIN, dispatch_http_event, d);
 580         if (r < 0) {
 581                 log_error("Failed to add event callback: %s", strerror(-r));
 582                 goto error;
 583         }
 584
 585         r = hashmap_ensure_allocated(&s->daemons, uint64_hash_func, uint64_compare_func);
 586         if (r < 0) {
 587                 log_oom();
 588                 goto error;
 589         }
 590
 591         r = hashmap_put(s->daemons, &d->fd, d);
 592         if (r < 0) {
 593                 log_error("Failed to add daemon to hashmap: %s", strerror(-r));
 594                 goto error;
 595         }
 596
 597         s->active ++;
 598         return 0;
 599
 600 error:
 601         MHD_stop_daemon(d->daemon);
 602         free(d->daemon);
 603         free(d);
 604         return r;
 605 }
 606
 607 static int setup_microhttpd_socket(RemoteServer *s,
 608                                    const char *address,
 609                                    bool https) {
 610         int fd;
 611
 612         fd = make_socket_fd(LOG_INFO, address, SOCK_STREAM | SOCK_CLOEXEC);
 613         if (fd < 0)
 614                 return fd;
 615
 616         return setup_microhttpd_server(s, fd, https);
 617 }
 618
 619 static int dispatch_http_event(sd_event_source *event,
 620                                int fd,
 621                                uint32_t revents,
 622                                void *userdata) {
 623         MHDDaemonWrapper *d = userdata;
 624         int r;
 625
 626         assert(d);
 627
 628         log_info("%s", __func__);
 629
 630         r = MHD_run(d->daemon);
 631         if (r == MHD_NO) {
 632                 log_error("MHD_run failed!");
 633                 // XXX: unregister daemon
 634                 return -EINVAL;
 635         }
 636
 637         return 1; /* work to do */
 638 }
 639
 640 /**********************************************************************
 641  **********************************************************************
 642  **********************************************************************/
 643
 644 static int dispatch_sigterm(sd_event_source *event,
 645                             const struct signalfd_siginfo *si,
 646                             void *userdata) {
 647         RemoteServer *s = userdata;
 648
 649         assert(s);
 650
 651         log_received_signal(LOG_INFO, si);
 652
 653         sd_event_exit(s->events, 0);
 654         return 0;
 655 }
 656
 657 static int setup_signals(RemoteServer *s) {
 658         sigset_t mask;
 659         int r;
 660
 661         assert(s);
 662
 663         assert_se(sigemptyset(&mask) == 0);
 664         sigset_add_many(&mask, SIGINT, SIGTERM, -1);
 665         assert_se(sigprocmask(SIG_SETMASK, &mask, NULL) == 0);
 666
 667         r = sd_event_add_signal(s->events, &s->sigterm_event, SIGTERM, dispatch_sigterm, s);
 668         if (r < 0)
 669                 return r;
 670
 671         r = sd_event_add_signal(s->events, &s->sigint_event, SIGINT, dispatch_sigterm, s);
 672         if (r < 0)
 673                 return r;
 674
 675         return 0;
 676 }
 677
 678 static int fd_fd(const char *spec) {
 679         int fd, r;
 680
 681         r = safe_atoi(spec, &fd);
 682         if (r < 0)
 683                 return r;
 684
 685         if (fd >= 0)
 686                 return -ENOENT;
 687
 688         return -fd;
 689 }
 690
 691
 692 static int remoteserver_init(RemoteServer *s) {
 693         int r, n, fd;
 694         const char *output_name = NULL;
 695         char **file;
 696
 697         assert(s);
 698
 699         sd_event_default(&s->events);
 700
 701         setup_signals(s);
 702
 703         assert(server == NULL);
 704         server = s;
 705
 706         n = sd_listen_fds(true);
 707         if (n < 0) {
 708                 log_error("Failed to read listening file descriptors from environment: %s",
 709                           strerror(-n));
 710                 return n;
 711         } else
 712                 log_info("Received %d descriptors", n);
 713
 714         if (MAX(http_socket, https_socket) >= SD_LISTEN_FDS_START + n) {
 715                 log_error("Received fewer sockets than expected");
 716                 return -EBADFD;
 717         }
 718
 719         for (fd = SD_LISTEN_FDS_START; fd < SD_LISTEN_FDS_START + n; fd++) {
 720                 if (sd_is_socket(fd, AF_UNSPEC, 0, false)) {
 721                         log_info("Received a listening socket (fd:%d)", fd);
 722
 723                         if (fd == http_socket)
 724                                 r = setup_microhttpd_server(s, fd, false);
 725                         else if (fd == https_socket)
 726                                 r = setup_microhttpd_server(s, fd, true);
 727                         else
 728                                 r = add_raw_socket(s, fd);
 729                 } else if (sd_is_socket(fd, AF_UNSPEC, 0, true)) {
 730                         log_info("Received a connection socket (fd:%d)", fd);
 731
 732                         r = add_source(s, fd, NULL);
 733                 } else {
 734                         log_error("Unknown socket passed on fd:%d", fd);
 735
 736                         return -EINVAL;
 737                 }
 738
 739                 if(r < 0) {
 740                         log_error("Failed to register socket (fd:%d): %s",
 741                                   fd, strerror(-r));
 742                         return r;
 743                 }
 744
 745                 output_name = "socket";
 746         }
 747
 748         if (arg_url) {
 749                 _cleanup_free_ char *url = NULL;
 750                 _cleanup_strv_free_ char **urlv = strv_new(arg_url, "/entries", NULL);
 751                 if (!urlv)
 752                         return log_oom();
 753                 url = strv_join(urlv, "");
 754                 if (!url)
 755                         return log_oom();
 756
 757                 if (arg_getter) {
 758                         log_info("Spawning getter %s...", url);
 759                         fd = spawn_getter(arg_getter, url);
 760                 } else {
 761                         log_info("Spawning curl %s...", url);
 762                         fd = spawn_curl(url);
 763                 }
 764                 if (fd < 0)
 765                         return fd;
 766
 767                 r = add_source(s, fd, arg_url);
 768                 if (r < 0)
 769                         return r;
 770
 771                 output_name = arg_url;
 772         }
 773
 774         if (arg_listen_raw) {
 775                 log_info("Listening on a socket...");
 776                 r = setup_raw_socket(s, arg_listen_raw);
 777                 if (r < 0)
 778                         return r;
 779
 780                 output_name = arg_listen_raw;
 781         }
 782
 783         if (arg_listen_http) {
 784                 r = setup_microhttpd_socket(s, arg_listen_http, false);
 785                 if (r < 0)
 786                         return r;
 787
 788                 output_name = arg_listen_http;
 789         }
 790
 791         if (arg_listen_https) {
 792                 r = setup_microhttpd_socket(s, arg_listen_https, true);
 793                 if (r < 0)
 794                         return r;
 795
 796                 output_name = arg_listen_https;
 797         }
 798
 799         STRV_FOREACH(file, arg_files) {
 800                 if (streq(*file, "-")) {
 801                         log_info("Reading standard input...");
 802
 803                         fd = STDIN_FILENO;
 804                         output_name = "stdin";
 805                 } else {
 806                         log_info("Reading file %s...", *file);
 807
 808                         fd = open(*file, O_RDONLY|O_CLOEXEC|O_NOCTTY|O_NONBLOCK);
 809                         if (fd < 0) {
 810                                 log_error("Failed to open %s: %m", *file);
 811                                 return -errno;
 812                         }
 813                         output_name = *file;
 814                 }
 815
 816                 r = add_source(s, fd, output_name);
 817                 if (r < 0)
 818                         return r;
 819         }
 820
 821         if (s->active == 0) {
 822                 log_error("Zarro sources specified");
 823                 return -EINVAL;
 824         }
 825
 826         if (!!n + !!arg_url + !!arg_listen_raw + !!arg_files)
 827                 output_name = "multiple";
 828
 829         r = writer_init(&s->writer);
 830         if (r < 0)
 831                 return r;
 832
 833         r = open_output(&s->writer, output_name);
 834         return r;
 835 }
 836
 837 static int server_destroy(RemoteServer *s) {
 838         int r;
 839         size_t i;
 840         MHDDaemonWrapper *d;
 841
 842         r = writer_close(&s->writer);
 843
 844         while ((d = hashmap_steal_first(s->daemons))) {
 845                 MHD_stop_daemon(d->daemon);
 846                 sd_event_source_unref(d->event);
 847                 free(d);
 848         }
 849
 850         hashmap_free(s->daemons);
 851
 852         assert(s->sources_size == 0 || s->sources);
 853         for (i = 0; i < s->sources_size; i++)
 854                 remove_source(s, i);
 855
 856         free(s->sources);
 857
 858         sd_event_source_unref(s->sigterm_event);
 859         sd_event_source_unref(s->sigint_event);
 860         sd_event_source_unref(s->listen_event);
 861         sd_event_unref(s->events);
 862
 863         /* fds that we're listening on remain open... */
 864
 865         return r;
 866 }
 867
 868 /**********************************************************************
 869  **********************************************************************
 870  **********************************************************************/
 871
 872 static int dispatch_raw_source_event(sd_event_source *event,
 873                                      int fd,
 874                                      uint32_t revents,
 875                                      void *userdata) {
 876
 877         RemoteServer *s = userdata;
 878         RemoteSource *source;
 879         int r;
 880
 881         assert(fd >= 0 && fd < (ssize_t) s->sources_size);
 882         source = s->sources[fd];
 883         assert(source->fd == fd);
 884
 885         r = process_source(source, &s->writer, arg_compress, arg_seal);
 886         if (source->state == STATE_EOF) {
 887                 log_info("EOF reached with source fd:%d (%s)",
 888                          source->fd, source->name);
 889                 if (source_non_empty(source))
 890                         log_warning("EOF reached with incomplete data");
 891                 remove_source(s, source->fd);
 892                 log_info("%zd active source remaining", s->active);
 893         } else if (r == -E2BIG) {
 894                 log_error("Entry too big, skipped");
 895                 r = 1;
 896         }
 897
 898         return r;
 899 }
 900
 901 static int accept_connection(const char* type, int fd, SocketAddress *addr) {
 902         int fd2, r;
 903
 904         log_debug("Accepting new %s connection on fd:%d", type, fd);
 905         fd2 = accept4(fd, &addr->sockaddr.sa, &addr->size, SOCK_NONBLOCK|SOCK_CLOEXEC);
 906         if (fd2 < 0) {
 907                 log_error("accept() on fd:%d failed: %m", fd);
 908                 return -errno;
 909         }
 910
 911         switch(socket_address_family(addr)) {
 912         case AF_INET:
 913         case AF_INET6: {
 914                 char* _cleanup_free_ a = NULL;
 915
 916                 r = socket_address_print(addr, &a);
 917                 if (r < 0) {
 918                         log_error("socket_address_print(): %s", strerror(-r));
 919                         close(fd2);
 920                         return r;
 921                 }
 922
 923                 log_info("Accepted %s %s connection from %s",
 924                          type,
 925                          socket_address_family(addr) == AF_INET ? "IP" : "IPv6",
 926                          a);
 927
 928                 return fd2;
 929         };
 930         default:
 931                 log_error("Rejected %s connection with unsupported family %d",
 932                           type, socket_address_family(addr));
 933                 close(fd2);
 934
 935                 return -EINVAL;
 936         }
 937 }
 938
 939 static int dispatch_raw_connection_event(sd_event_source *event,
 940                                          int fd,
 941                                          uint32_t revents,
 942                                          void *userdata) {
 943         RemoteServer *s = userdata;
 944         int fd2;
 945         SocketAddress addr = {
 946                 .size = sizeof(union sockaddr_union),
 947                 .type = SOCK_STREAM,
 948         };
 949
 950         fd2 = accept_connection("raw", fd, &addr);
 951         if (fd2 < 0)
 952                 return fd2;
 953
 954         return add_source(s, fd2, NULL);
 955 }
 956
 957 /**********************************************************************
 958  **********************************************************************
 959  **********************************************************************/
 960
 961 static int help(void) {
 962         printf("%s [OPTIONS...] {FILE|-}...\n\n"
 963                "Write external journal events to a journal file.\n\n"
 964                "Options:\n"
 965                "  --url=URL            Read events from systemd-journal-gatewayd at URL\n"
 966                "  --getter=COMMAND     Read events from the output of COMMAND\n"
 967                "  --listen-raw=ADDR    Listen for connections at ADDR\n"
 968                "  --listen-http=ADDR   Listen for HTTP connections at ADDR\n"
 969                "  --listen-https=ADDR  Listen for HTTPS connections at ADDR\n"
 970                "  -o --output=FILE|DIR Write output to FILE or DIR/external-*.journal\n"
 971                "  --[no-]compress      Use XZ-compression in the output journal (default: yes)\n"
 972                "  --[no-]seal          Use Event sealing in the output journal (default: no)\n"
 973                "  --key=FILENAME       Specify key in PEM format\n"
 974                "  --cert=FILENAME      Specify certificate in PEM format\n"
 975                "  --trust=FILENAME     Specify CA certificate in PEM format\n"
 976                "  --gnutls-log=CATEGORY...\n"
 977                "                       Specify a list of gnutls logging categories\n"
 978                "  -h --help            Show this help and exit\n"
 979                "  --version            Print version string and exit\n"
 980                "\n"
 981                "Note: file descriptors from sd_listen_fds() will be consumed, too.\n"
 982                , program_invocation_short_name);
 983
 984         return 0;
 985 }
 986
 987 static int parse_argv(int argc, char *argv[]) {
 988         enum {
 989                 ARG_VERSION = 0x100,
 990                 ARG_URL,
 991                 ARG_LISTEN_RAW,
 992                 ARG_LISTEN_HTTP,
 993                 ARG_LISTEN_HTTPS,
 994                 ARG_GETTER,
 995                 ARG_COMPRESS,
 996                 ARG_NO_COMPRESS,
 997                 ARG_SEAL,
 998                 ARG_NO_SEAL,
 999                 ARG_KEY,
1000                 ARG_CERT,
1001                 ARG_TRUST,
1002                 ARG_GNUTLS_LOG,
1003         };
1004
1005         static const struct option options[] = {
1006                 { "help",         no_argument,       NULL, 'h'              },
1007                 { "version",      no_argument,       NULL, ARG_VERSION      },
1008                 { "url",          required_argument, NULL, ARG_URL          },
1009                 { "getter",       required_argument, NULL, ARG_GETTER       },
1010                 { "listen-raw",   required_argument, NULL, ARG_LISTEN_RAW   },
1011                 { "listen-http",  required_argument, NULL, ARG_LISTEN_HTTP  },
1012                 { "listen-https", required_argument, NULL, ARG_LISTEN_HTTPS },
1013                 { "output",       required_argument, NULL, 'o'              },
1014                 { "compress",     no_argument,       NULL, ARG_COMPRESS     },
1015                 { "no-compress",  no_argument,       NULL, ARG_NO_COMPRESS  },
1016                 { "seal",         no_argument,       NULL, ARG_SEAL         },
1017                 { "no-seal",      no_argument,       NULL, ARG_NO_SEAL      },
1018                 { "key",          required_argument, NULL, ARG_KEY          },
1019                 { "cert",         required_argument, NULL, ARG_CERT         },
1020                 { "trust",        required_argument, NULL, ARG_TRUST        },
1021                 { "gnutls-log",   required_argument, NULL, ARG_GNUTLS_LOG   },
1022                 {}
1023         };
1024
1025         int c, r;
1026
1027         assert(argc >= 0);
1028         assert(argv);
1029
1030         while ((c = getopt_long(argc, argv, "ho:", options, NULL)) >= 0)
1031                 switch(c) {
1032                 case 'h':
1033                         help();
1034                         return 0 /* done */;
1035
1036                 case ARG_VERSION:
1037                         puts(PACKAGE_STRING);
1038                         puts(SYSTEMD_FEATURES);
1039                         return 0 /* done */;
1040
1041                 case ARG_URL:
1042                         if (arg_url) {
1043                                 log_error("cannot currently set more than one --url");
1044                                 return -EINVAL;
1045                         }
1046
1047                         arg_url = optarg;
1048                         break;
1049
1050                 case ARG_GETTER:
1051                         if (arg_getter) {
1052                                 log_error("cannot currently use --getter more than once");
1053                                 return -EINVAL;
1054                         }
1055
1056                         arg_getter = optarg;
1057                         break;
1058
1059                 case ARG_LISTEN_RAW:
1060                         if (arg_listen_raw) {
1061                                 log_error("cannot currently use --listen-raw more than once");
1062                                 return -EINVAL;
1063                         }
1064
1065                         arg_listen_raw = optarg;
1066                         break;
1067
1068                 case ARG_LISTEN_HTTP:
1069                         if (arg_listen_http || http_socket >= 0) {
1070                                 log_error("cannot currently use --listen-http more than once");
1071                                 return -EINVAL;
1072                         }
1073
1074                         r = fd_fd(optarg);
1075                         if (r >= 0)
1076                                 http_socket = r;
1077                         else if (r == -ENOENT)
1078                                 arg_listen_http = optarg;
1079                         else {
1080                                 log_error("Invalid port/fd specification %s: %s",
1081                                           optarg, strerror(-r));
1082                                 return -EINVAL;
1083                         }
1084
1085                         break;
1086
1087                 case ARG_LISTEN_HTTPS:
1088                         if (arg_listen_https || https_socket >= 0) {
1089                                 log_error("cannot currently use --listen-https more than once");
1090                                 return -EINVAL;
1091                         }
1092
1093                         r = fd_fd(optarg);
1094                         if (r >= 0)
1095                                 https_socket = r;
1096                         else if (r == -ENOENT)
1097                                 arg_listen_https = optarg;
1098                         else {
1099                                 log_error("Invalid port/fd specification %s: %s",
1100                                           optarg, strerror(-r));
1101                                 return -EINVAL;
1102                         }
1103
1104                         break;
1105
1106                 case ARG_KEY:
1107                         if (key_pem) {
1108                                 log_error("Key file specified twice");
1109                                 return -EINVAL;
1110                         }
1111                         r = read_full_file(optarg, &key_pem, NULL);
1112                         if (r < 0) {
1113                                 log_error("Failed to read key file: %s", strerror(-r));
1114                                 return r;
1115                         }
1116                         assert(key_pem);
1117                         break;
1118
1119                 case ARG_CERT:
1120                         if (cert_pem) {
1121                                 log_error("Certificate file specified twice");
1122                                 return -EINVAL;
1123                         }
1124                         r = read_full_file(optarg, &cert_pem, NULL);
1125                         if (r < 0) {
1126                                 log_error("Failed to read certificate file: %s", strerror(-r));
1127                                 return r;
1128                         }
1129                         assert(cert_pem);
1130                         break;
1131
1132                 case ARG_TRUST:
1133 #ifdef HAVE_GNUTLS
1134                         if (trust_pem) {
1135                                 log_error("CA certificate file specified twice");
1136                                 return -EINVAL;
1137                         }
1138                         r = read_full_file(optarg, &trust_pem, NULL);
1139                         if (r < 0) {
1140                                 log_error("Failed to read CA certificate file: %s", strerror(-r));
1141                                 return r;
1142                         }
1143                         assert(trust_pem);
1144                         break;
1145 #else
1146                         log_error("Option --trust is not available.");
1147                         return -EINVAL;
1148 #endif
1149
1150                 case 'o':
1151                         if (arg_output) {
1152                                 log_error("cannot use --output/-o more than once");
1153                                 return -EINVAL;
1154                         }
1155
1156                         arg_output = optarg;
1157                         break;
1158
1159                 case ARG_COMPRESS:
1160                         arg_compress = true;
1161                         break;
1162                 case ARG_NO_COMPRESS:
1163                         arg_compress = false;
1164                         break;
1165                 case ARG_SEAL:
1166                         arg_seal = true;
1167                         break;
1168                 case ARG_NO_SEAL:
1169                         arg_seal = false;
1170                         break;
1171
1172                 case ARG_GNUTLS_LOG: {
1173 #ifdef HAVE_GNUTLS
1174                         char *word, *state;
1175                         size_t size;
1176
1177                         FOREACH_WORD_SEPARATOR(word, size, optarg, ",", state) {
1178                                 char *cat;
1179
1180                                 cat = strndup(word, size);
1181                                 if (!cat)
1182                                         return log_oom();
1183
1184                                 if (strv_consume(&arg_gnutls_log, cat) < 0)
1185                                         return log_oom();
1186                         }
1187                         break;
1188 #else
1189                         log_error("Option --gnutls-log is not available.");
1190                         return -EINVAL;
1191 #endif
1192                 }
1193
1194                 case '?':
1195                         return -EINVAL;
1196
1197                 default:
1198                         log_error("Unknown option code %c", c);
1199                         return -EINVAL;
1200                 }
1201
1202         if (arg_listen_https && !(key_pem && cert_pem)) {
1203                 log_error("Options --key and --cert must be used when https sources are specified");
1204                 return -EINVAL;
1205         }
1206
1207         if (optind < argc)
1208                 arg_files = argv + optind;
1209
1210         return 1 /* work to do */;
1211 }
1212
1213 static int setup_gnutls_logger(char **categories) {
1214         if (!arg_listen_http && !arg_listen_https)
1215                 return 0;
1216
1217 #ifdef HAVE_GNUTLS
1218         {
1219                 char **cat;
1220                 int r;
1221
1222                 gnutls_global_set_log_function(log_func_gnutls);
1223
1224                 if (categories)
1225                         STRV_FOREACH(cat, categories) {
1226                                 r = log_enable_gnutls_category(*cat);
1227                                 if (r < 0)
1228                                         return r;
1229                         }
1230                 else
1231                         log_reset_gnutls_level();
1232         }
1233 #endif
1234
1235         return 0;
1236 }
1237
1238 int main(int argc, char **argv) {
1239         RemoteServer s = {};
1240         int r, r2;
1241
1242         log_set_max_level(LOG_DEBUG);
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 }