src/journal/journald-stream.c

   1 /* SPDX-License-Identifier: LGPL-2.1+ */
   2
   3 #include <stddef.h>
   4 #include <unistd.h>
   5
   6 #if HAVE_SELINUX
   7 #include <selinux/selinux.h>
   8 #endif
   9
  10 #include "sd-daemon.h"
  11 #include "sd-event.h"
  12
  13 #include "alloc-util.h"
  14 #include "dirent-util.h"
  15 #include "env-file.h"
  16 #include "errno-util.h"
  17 #include "escape.h"
  18 #include "fd-util.h"
  19 #include "fileio.h"
  20 #include "fs-util.h"
  21 #include "io-util.h"
  22 #include "journald-console.h"
  23 #include "journald-context.h"
  24 #include "journald-kmsg.h"
  25 #include "journald-server.h"
  26 #include "journald-stream.h"
  27 #include "journald-syslog.h"
  28 #include "journald-wall.h"
  29 #include "mkdir.h"
  30 #include "parse-util.h"
  31 #include "process-util.h"
  32 #include "selinux-util.h"
  33 #include "socket-util.h"
  34 #include "stdio-util.h"
  35 #include "string-util.h"
  36 #include "syslog-util.h"
  37 #include "tmpfile-util.h"
  38 #include "unit-name.h"
  39
  40 #define STDOUT_STREAMS_MAX 4096
  41
  42 typedef enum StdoutStreamState {
  43         STDOUT_STREAM_IDENTIFIER,
  44         STDOUT_STREAM_UNIT_ID,
  45         STDOUT_STREAM_PRIORITY,
  46         STDOUT_STREAM_LEVEL_PREFIX,
  47         STDOUT_STREAM_FORWARD_TO_SYSLOG,
  48         STDOUT_STREAM_FORWARD_TO_KMSG,
  49         STDOUT_STREAM_FORWARD_TO_CONSOLE,
  50         STDOUT_STREAM_RUNNING
  51 } StdoutStreamState;
  52
  53 /* The different types of log record terminators: a real \n was read, a NUL character was read, the maximum line length
  54  * was reached, or the end of the stream was reached */
  55
  56 typedef enum LineBreak {
  57         LINE_BREAK_NEWLINE,
  58         LINE_BREAK_NUL,
  59         LINE_BREAK_LINE_MAX,
  60         LINE_BREAK_EOF,
  61         LINE_BREAK_PID_CHANGE,
  62         _LINE_BREAK_MAX,
  63         _LINE_BREAK_INVALID = -1,
  64 } LineBreak;
  65
  66 struct StdoutStream {
  67         Server *server;
  68         StdoutStreamState state;
  69
  70         int fd;
  71
  72         struct ucred ucred;
  73         char *label;
  74         char *identifier;
  75         char *unit_id;
  76         int priority;
  77         bool level_prefix:1;
  78         bool forward_to_syslog:1;
  79         bool forward_to_kmsg:1;
  80         bool forward_to_console:1;
  81
  82         bool fdstore:1;
  83         bool in_notify_queue:1;
  84
  85         char *buffer;
  86         size_t length;
  87         size_t allocated;
  88
  89         sd_event_source *event_source;
  90
  91         char *state_file;
  92
  93         ClientContext *context;
  94
  95         LIST_FIELDS(StdoutStream, stdout_stream);
  96         LIST_FIELDS(StdoutStream, stdout_stream_notify_queue);
  97
  98         char id_field[STRLEN("_STREAM_ID=") + SD_ID128_STRING_MAX];
  99 };
 100
 101 void stdout_stream_free(StdoutStream *s) {
 102         if (!s)
 103                 return;
 104
 105         if (s->server) {
 106
 107                 if (s->context)
 108                         client_context_release(s->server, s->context);
 109
 110                 assert(s->server->n_stdout_streams > 0);
 111                 s->server->n_stdout_streams--;
 112                 LIST_REMOVE(stdout_stream, s->server->stdout_streams, s);
 113
 114                 if (s->in_notify_queue)
 115                         LIST_REMOVE(stdout_stream_notify_queue, s->server->stdout_streams_notify_queue, s);
 116
 117                 (void) server_start_or_stop_idle_timer(s->server); /* Maybe we are idle now? */
 118         }
 119
 120         if (s->event_source) {
 121                 sd_event_source_set_enabled(s->event_source, SD_EVENT_OFF);
 122                 s->event_source = sd_event_source_unref(s->event_source);
 123         }
 124
 125         safe_close(s->fd);
 126         free(s->label);
 127         free(s->identifier);
 128         free(s->unit_id);
 129         free(s->state_file);
 130         free(s->buffer);
 131
 132         free(s);
 133 }
 134
 135 DEFINE_TRIVIAL_CLEANUP_FUNC(StdoutStream*, stdout_stream_free);
 136
 137 void stdout_stream_destroy(StdoutStream *s) {
 138         if (!s)
 139                 return;
 140
 141         if (s->state_file)
 142                 (void) unlink(s->state_file);
 143
 144         stdout_stream_free(s);
 145 }
 146
 147 static int stdout_stream_save(StdoutStream *s) {
 148         _cleanup_(unlink_and_freep) char *temp_path = NULL;
 149         _cleanup_fclose_ FILE *f = NULL;
 150         int r;
 151
 152         assert(s);
 153
 154         if (s->state != STDOUT_STREAM_RUNNING)
 155                 return 0;
 156
 157         if (!s->state_file) {
 158                 struct stat st;
 159
 160                 r = fstat(s->fd, &st);
 161                 if (r < 0)
 162                         return log_warning_errno(errno, "Failed to stat connected stream: %m");
 163
 164                 /* We use device and inode numbers as identifier for the stream */
 165                 r = asprintf(&s->state_file, "%s/streams/%lu:%lu", s->server->runtime_directory, (unsigned long) st.st_dev, (unsigned long) st.st_ino);
 166                 if (r < 0)
 167                         return log_oom();
 168         }
 169
 170         (void) mkdir_parents(s->state_file, 0755);
 171
 172         r = fopen_temporary(s->state_file, &f, &temp_path);
 173         if (r < 0)
 174                 goto fail;
 175
 176         fprintf(f,
 177                 "# This is private data. Do not parse\n"
 178                 "PRIORITY=%i\n"
 179                 "LEVEL_PREFIX=%i\n"
 180                 "FORWARD_TO_SYSLOG=%i\n"
 181                 "FORWARD_TO_KMSG=%i\n"
 182                 "FORWARD_TO_CONSOLE=%i\n"
 183                 "STREAM_ID=%s\n",
 184                 s->priority,
 185                 s->level_prefix,
 186                 s->forward_to_syslog,
 187                 s->forward_to_kmsg,
 188                 s->forward_to_console,
 189                 s->id_field + STRLEN("_STREAM_ID="));
 190
 191         if (!isempty(s->identifier)) {
 192                 _cleanup_free_ char *escaped;
 193
 194                 escaped = cescape(s->identifier);
 195                 if (!escaped) {
 196                         r = -ENOMEM;
 197                         goto fail;
 198                 }
 199
 200                 fprintf(f, "IDENTIFIER=%s\n", escaped);
 201         }
 202
 203         if (!isempty(s->unit_id)) {
 204                 _cleanup_free_ char *escaped;
 205
 206                 escaped = cescape(s->unit_id);
 207                 if (!escaped) {
 208                         r = -ENOMEM;
 209                         goto fail;
 210                 }
 211
 212                 fprintf(f, "UNIT=%s\n", escaped);
 213         }
 214
 215         r = fflush_and_check(f);
 216         if (r < 0)
 217                 goto fail;
 218
 219         if (rename(temp_path, s->state_file) < 0) {
 220                 r = -errno;
 221                 goto fail;
 222         }
 223
 224         temp_path = mfree(temp_path);
 225
 226         if (!s->fdstore && !s->in_notify_queue) {
 227                 LIST_PREPEND(stdout_stream_notify_queue, s->server->stdout_streams_notify_queue, s);
 228                 s->in_notify_queue = true;
 229
 230                 if (s->server->notify_event_source) {
 231                         r = sd_event_source_set_enabled(s->server->notify_event_source, SD_EVENT_ON);
 232                         if (r < 0)
 233                                 log_warning_errno(r, "Failed to enable notify event source: %m");
 234                 }
 235         }
 236
 237         return 0;
 238
 239 fail:
 240         (void) unlink(s->state_file);
 241         return log_error_errno(r, "Failed to save stream data %s: %m", s->state_file);
 242 }
 243
 244 static int stdout_stream_log(
 245                 StdoutStream *s,
 246                 const char *p,
 247                 LineBreak line_break) {
 248
 249         struct iovec *iovec;
 250         int priority;
 251         char syslog_priority[] = "PRIORITY=\0";
 252         char syslog_facility[STRLEN("SYSLOG_FACILITY=") + DECIMAL_STR_MAX(int) + 1];
 253         _cleanup_free_ char *message = NULL, *syslog_identifier = NULL;
 254         size_t n = 0, m;
 255         int r;
 256
 257         assert(s);
 258         assert(p);
 259
 260         assert(line_break >= 0);
 261         assert(line_break < _LINE_BREAK_MAX);
 262
 263         if (s->context)
 264                 (void) client_context_maybe_refresh(s->server, s->context, NULL, NULL, 0, NULL, USEC_INFINITY);
 265         else if (pid_is_valid(s->ucred.pid)) {
 266                 r = client_context_acquire(s->server, s->ucred.pid, &s->ucred, s->label, strlen_ptr(s->label), s->unit_id, &s->context);
 267                 if (r < 0)
 268                         log_warning_errno(r, "Failed to acquire client context, ignoring: %m");
 269         }
 270
 271         priority = s->priority;
 272
 273         if (s->level_prefix)
 274                 syslog_parse_priority(&p, &priority, false);
 275
 276         if (!client_context_test_priority(s->context, priority))
 277                 return 0;
 278
 279         if (isempty(p))
 280                 return 0;
 281
 282         if (s->forward_to_syslog || s->server->forward_to_syslog)
 283                 server_forward_syslog(s->server, syslog_fixup_facility(priority), s->identifier, p, &s->ucred, NULL);
 284
 285         if (s->forward_to_kmsg || s->server->forward_to_kmsg)
 286                 server_forward_kmsg(s->server, priority, s->identifier, p, &s->ucred);
 287
 288         if (s->forward_to_console || s->server->forward_to_console)
 289                 server_forward_console(s->server, priority, s->identifier, p, &s->ucred);
 290
 291         if (s->server->forward_to_wall)
 292                 server_forward_wall(s->server, priority, s->identifier, p, &s->ucred);
 293
 294         m = N_IOVEC_META_FIELDS + 7 + client_context_extra_fields_n_iovec(s->context);
 295         iovec = newa(struct iovec, m);
 296
 297         iovec[n++] = IOVEC_MAKE_STRING("_TRANSPORT=stdout");
 298         iovec[n++] = IOVEC_MAKE_STRING(s->id_field);
 299
 300         syslog_priority[STRLEN("PRIORITY=")] = '0' + LOG_PRI(priority);
 301         iovec[n++] = IOVEC_MAKE_STRING(syslog_priority);
 302
 303         if (priority & LOG_FACMASK) {
 304                 xsprintf(syslog_facility, "SYSLOG_FACILITY=%i", LOG_FAC(priority));
 305                 iovec[n++] = IOVEC_MAKE_STRING(syslog_facility);
 306         }
 307
 308         if (s->identifier) {
 309                 syslog_identifier = strjoin("SYSLOG_IDENTIFIER=", s->identifier);
 310                 if (syslog_identifier)
 311                         iovec[n++] = IOVEC_MAKE_STRING(syslog_identifier);
 312         }
 313
 314         static const char * const line_break_field_table[_LINE_BREAK_MAX] = {
 315                 [LINE_BREAK_NEWLINE]    = NULL, /* Do not add field if traditional newline */
 316                 [LINE_BREAK_NUL]        = "_LINE_BREAK=nul",
 317                 [LINE_BREAK_LINE_MAX]   = "_LINE_BREAK=line-max",
 318                 [LINE_BREAK_EOF]        = "_LINE_BREAK=eof",
 319                 [LINE_BREAK_PID_CHANGE] = "_LINE_BREAK=pid-change",
 320         };
 321
 322         const char *c = line_break_field_table[line_break];
 323
 324         /* If this log message was generated due to an uncommon line break then mention this in the log
 325          * entry */
 326         if (c)
 327                 iovec[n++] = IOVEC_MAKE_STRING(c);
 328
 329         message = strjoin("MESSAGE=", p);
 330         if (message)
 331                 iovec[n++] = IOVEC_MAKE_STRING(message);
 332
 333         server_dispatch_message(s->server, iovec, n, m, s->context, NULL, priority, 0);
 334         return 0;
 335 }
 336
 337 static int stdout_stream_line(StdoutStream *s, char *p, LineBreak line_break) {
 338         char *orig;
 339         int r;
 340
 341         assert(s);
 342         assert(p);
 343
 344         orig = p;
 345         p = strstrip(p);
 346
 347         /* line breaks by NUL, line max length or EOF are not permissible during the negotiation part of the protocol */
 348         if (line_break != LINE_BREAK_NEWLINE && s->state != STDOUT_STREAM_RUNNING)
 349                 return log_warning_errno(SYNTHETIC_ERRNO(EINVAL),
 350                                          "Control protocol line not properly terminated.");
 351
 352         switch (s->state) {
 353
 354         case STDOUT_STREAM_IDENTIFIER:
 355                 if (!isempty(p)) {
 356                         s->identifier = strdup(p);
 357                         if (!s->identifier)
 358                                 return log_oom();
 359                 }
 360
 361                 s->state = STDOUT_STREAM_UNIT_ID;
 362                 return 0;
 363
 364         case STDOUT_STREAM_UNIT_ID:
 365                 if (s->ucred.uid == 0 &&
 366                     unit_name_is_valid(p, UNIT_NAME_PLAIN|UNIT_NAME_INSTANCE)) {
 367
 368                         s->unit_id = strdup(p);
 369                         if (!s->unit_id)
 370                                 return log_oom();
 371                 }
 372
 373                 s->state = STDOUT_STREAM_PRIORITY;
 374                 return 0;
 375
 376         case STDOUT_STREAM_PRIORITY:
 377                 r = safe_atoi(p, &s->priority);
 378                 if (r < 0 || s->priority < 0 || s->priority > 999) {
 379                         log_warning("Failed to parse log priority line.");
 380                         return -EINVAL;
 381                 }
 382
 383                 s->state = STDOUT_STREAM_LEVEL_PREFIX;
 384                 return 0;
 385
 386         case STDOUT_STREAM_LEVEL_PREFIX:
 387                 r = parse_boolean(p);
 388                 if (r < 0) {
 389                         log_warning("Failed to parse level prefix line.");
 390                         return -EINVAL;
 391                 }
 392
 393                 s->level_prefix = r;
 394                 s->state = STDOUT_STREAM_FORWARD_TO_SYSLOG;
 395                 return 0;
 396
 397         case STDOUT_STREAM_FORWARD_TO_SYSLOG:
 398                 r = parse_boolean(p);
 399                 if (r < 0) {
 400                         log_warning("Failed to parse forward to syslog line.");
 401                         return -EINVAL;
 402                 }
 403
 404                 s->forward_to_syslog = r;
 405                 s->state = STDOUT_STREAM_FORWARD_TO_KMSG;
 406                 return 0;
 407
 408         case STDOUT_STREAM_FORWARD_TO_KMSG:
 409                 r = parse_boolean(p);
 410                 if (r < 0) {
 411                         log_warning("Failed to parse copy to kmsg line.");
 412                         return -EINVAL;
 413                 }
 414
 415                 s->forward_to_kmsg = r;
 416                 s->state = STDOUT_STREAM_FORWARD_TO_CONSOLE;
 417                 return 0;
 418
 419         case STDOUT_STREAM_FORWARD_TO_CONSOLE:
 420                 r = parse_boolean(p);
 421                 if (r < 0) {
 422                         log_warning("Failed to parse copy to console line.");
 423                         return -EINVAL;
 424                 }
 425
 426                 s->forward_to_console = r;
 427                 s->state = STDOUT_STREAM_RUNNING;
 428
 429                 /* Try to save the stream, so that journald can be restarted and we can recover */
 430                 (void) stdout_stream_save(s);
 431                 return 0;
 432
 433         case STDOUT_STREAM_RUNNING:
 434                 return stdout_stream_log(s, orig, line_break);
 435         }
 436
 437         assert_not_reached("Unknown stream state");
 438 }
 439
 440 static int stdout_stream_found(
 441                 StdoutStream *s,
 442                 char *p,
 443                 size_t l,
 444                 LineBreak line_break) {
 445
 446         char saved;
 447         int r;
 448
 449         assert(s);
 450         assert(p);
 451
 452         /* Let's NUL terminate the specified buffer for this call, and revert back afterwards */
 453         saved = p[l];
 454         p[l] = 0;
 455         r = stdout_stream_line(s, p, line_break);
 456         p[l] = saved;
 457
 458         return r;
 459 }
 460
 461 static int stdout_stream_scan(
 462                 StdoutStream *s,
 463                 char *p,
 464                 size_t remaining,
 465                 LineBreak force_flush,
 466                 size_t *ret_consumed) {
 467
 468         size_t consumed = 0;
 469         int r;
 470
 471         assert(s);
 472         assert(p);
 473
 474         for (;;) {
 475                 LineBreak line_break;
 476                 size_t skip, found;
 477                 char *end1, *end2;
 478
 479                 end1 = memchr(p, '\n', remaining);
 480                 end2 = memchr(p, 0, end1 ? (size_t) (end1 - p) : remaining);
 481
 482                 if (end2) {
 483                         /* We found a NUL terminator */
 484                         found = end2 - p;
 485                         skip = found + 1;
 486                         line_break = LINE_BREAK_NUL;
 487                 } else if (end1) {
 488                         /* We found a \n terminator */
 489                         found = end1 - p;
 490                         skip = found + 1;
 491                         line_break = LINE_BREAK_NEWLINE;
 492                 } else if (remaining >= s->server->line_max) {
 493                         /* Force a line break after the maximum line length */
 494                         found = skip = s->server->line_max;
 495                         line_break = LINE_BREAK_LINE_MAX;
 496                 } else
 497                         break;
 498
 499                 r = stdout_stream_found(s, p, found, line_break);
 500                 if (r < 0)
 501                         return r;
 502
 503                 p += skip;
 504                 consumed += skip;
 505                 remaining -= skip;
 506         }
 507
 508         if (force_flush >= 0 && remaining > 0) {
 509                 r = stdout_stream_found(s, p, remaining, force_flush);
 510                 if (r < 0)
 511                         return r;
 512
 513                 consumed += remaining;
 514         }
 515
 516         if (ret_consumed)
 517                 *ret_consumed = consumed;
 518
 519         return 0;
 520 }
 521
 522 static int stdout_stream_process(sd_event_source *es, int fd, uint32_t revents, void *userdata) {
 523         CMSG_BUFFER_TYPE(CMSG_SPACE(sizeof(struct ucred))) control;
 524         StdoutStream *s = userdata;
 525         size_t limit, consumed;
 526         struct ucred *ucred;
 527         struct iovec iovec;
 528         ssize_t l;
 529         char *p;
 530         int r;
 531
 532         struct msghdr msghdr = {
 533                 .msg_iov = &iovec,
 534                 .msg_iovlen = 1,
 535                 .msg_control = &control,
 536                 .msg_controllen = sizeof(control),
 537         };
 538
 539         assert(s);
 540
 541         if ((revents|EPOLLIN|EPOLLHUP) != (EPOLLIN|EPOLLHUP)) {
 542                 log_error("Got invalid event from epoll for stdout stream: %"PRIx32, revents);
 543                 goto terminate;
 544         }
 545
 546         /* If the buffer is almost full, add room for another 1K */
 547         if (s->length + 512 >= s->allocated) {
 548                 if (!GREEDY_REALLOC(s->buffer, s->allocated, s->length + 1 + 1024)) {
 549                         log_oom();
 550                         goto terminate;
 551                 }
 552         }
 553
 554         /* Try to make use of the allocated buffer in full, but never read more than the configured line size. Also,
 555          * always leave room for a terminating NUL we might need to add. */
 556         limit = MIN(s->allocated - 1, s->server->line_max);
 557         assert(s->length <= limit);
 558         iovec = IOVEC_MAKE(s->buffer + s->length, limit - s->length);
 559
 560         l = recvmsg(s->fd, &msghdr, MSG_DONTWAIT|MSG_CMSG_CLOEXEC);
 561         if (l < 0) {
 562                 if (IN_SET(errno, EINTR, EAGAIN))
 563                         return 0;
 564
 565                 log_warning_errno(errno, "Failed to read from stream: %m");
 566                 goto terminate;
 567         }
 568         cmsg_close_all(&msghdr);
 569
 570         if (l == 0) {
 571                 (void) stdout_stream_scan(s, s->buffer, s->length, /* force_flush = */ LINE_BREAK_EOF, NULL);
 572                 goto terminate;
 573         }
 574
 575         /* Invalidate the context if the PID of the sender changed. This happens when a forked process
 576          * inherits stdout/stderr from a parent. In this case getpeercred() returns the ucred of the parent,
 577          * which can be invalid if the parent has exited in the meantime. */
 578         ucred = CMSG_FIND_DATA(&msghdr, SOL_SOCKET, SCM_CREDENTIALS, struct ucred);
 579         if (ucred && ucred->pid != s->ucred.pid) {
 580                 /* Force out any previously half-written lines from a different process, before we switch to
 581                  * the new ucred structure for everything we just added */
 582                 r = stdout_stream_scan(s, s->buffer, s->length, /* force_flush = */ LINE_BREAK_PID_CHANGE, NULL);
 583                 if (r < 0)
 584                         goto terminate;
 585
 586                 s->context = client_context_release(s->server, s->context);
 587
 588                 p = s->buffer + s->length;
 589         } else {
 590                 p = s->buffer;
 591                 l += s->length;
 592         }
 593
 594         /* Always copy in the new credentials */
 595         if (ucred)
 596                 s->ucred = *ucred;
 597
 598         r = stdout_stream_scan(s, p, l, _LINE_BREAK_INVALID, &consumed);
 599         if (r < 0)
 600                 goto terminate;
 601
 602         /* Move what wasn't consumed to the front of the buffer */
 603         assert(consumed <= (size_t) l);
 604         s->length = l - consumed;
 605         memmove(s->buffer, p + consumed, s->length);
 606
 607         return 1;
 608
 609 terminate:
 610         stdout_stream_destroy(s);
 611         return 0;
 612 }
 613
 614 int stdout_stream_install(Server *s, int fd, StdoutStream **ret) {
 615         _cleanup_(stdout_stream_freep) StdoutStream *stream = NULL;
 616         sd_id128_t id;
 617         int r;
 618
 619         assert(s);
 620         assert(fd >= 0);
 621
 622         r = sd_id128_randomize(&id);
 623         if (r < 0)
 624                 return log_error_errno(r, "Failed to generate stream ID: %m");
 625
 626         stream = new(StdoutStream, 1);
 627         if (!stream)
 628                 return log_oom();
 629
 630         *stream = (StdoutStream) {
 631                 .fd = -1,
 632                 .priority = LOG_INFO,
 633         };
 634
 635         xsprintf(stream->id_field, "_STREAM_ID=" SD_ID128_FORMAT_STR, SD_ID128_FORMAT_VAL(id));
 636
 637         r = getpeercred(fd, &stream->ucred);
 638         if (r < 0)
 639                 return log_error_errno(r, "Failed to determine peer credentials: %m");
 640
 641         r = setsockopt_int(fd, SOL_SOCKET, SO_PASSCRED, true);
 642         if (r < 0)
 643                 return log_error_errno(r, "SO_PASSCRED failed: %m");
 644
 645         if (mac_selinux_use()) {
 646                 r = getpeersec(fd, &stream->label);
 647                 if (r < 0 && r != -EOPNOTSUPP)
 648                         (void) log_warning_errno(r, "Failed to determine peer security context: %m");
 649         }
 650
 651         (void) shutdown(fd, SHUT_WR);
 652
 653         r = sd_event_add_io(s->event, &stream->event_source, fd, EPOLLIN, stdout_stream_process, stream);
 654         if (r < 0)
 655                 return log_error_errno(r, "Failed to add stream to event loop: %m");
 656
 657         r = sd_event_source_set_priority(stream->event_source, SD_EVENT_PRIORITY_NORMAL+5);
 658         if (r < 0)
 659                 return log_error_errno(r, "Failed to adjust stdout event source priority: %m");
 660
 661         stream->fd = fd;
 662
 663         stream->server = s;
 664         LIST_PREPEND(stdout_stream, s->stdout_streams, stream);
 665         s->n_stdout_streams++;
 666
 667         (void) server_start_or_stop_idle_timer(s); /* Maybe no longer idle? */
 668
 669         if (ret)
 670                 *ret = stream;
 671
 672         TAKE_PTR(stream);
 673         return 0;
 674 }
 675
 676 static int stdout_stream_new(sd_event_source *es, int listen_fd, uint32_t revents, void *userdata) {
 677         _cleanup_close_ int fd = -1;
 678         Server *s = userdata;
 679         int r;
 680
 681         assert(s);
 682
 683         if (revents != EPOLLIN)
 684                 return log_error_errno(SYNTHETIC_ERRNO(EIO),
 685                                        "Got invalid event from epoll for stdout server fd: %" PRIx32,
 686                                        revents);
 687
 688         fd = accept4(s->stdout_fd, NULL, NULL, SOCK_NONBLOCK|SOCK_CLOEXEC);
 689         if (fd < 0) {
 690                 if (ERRNO_IS_ACCEPT_AGAIN(errno))
 691                         return 0;
 692
 693                 return log_error_errno(errno, "Failed to accept stdout connection: %m");
 694         }
 695
 696         if (s->n_stdout_streams >= STDOUT_STREAMS_MAX) {
 697                 struct ucred u;
 698
 699                 r = getpeercred(fd, &u);
 700
 701                 /* By closing fd here we make sure that the client won't wait too long for journald to
 702                  * gather all the data it adds to the error message to find out that the connection has
 703                  * just been refused.
 704                  */
 705                 fd = safe_close(fd);
 706
 707                 server_driver_message(s, r < 0 ? 0 : u.pid, NULL, LOG_MESSAGE("Too many stdout streams, refusing connection."), NULL);
 708                 return 0;
 709         }
 710
 711         r = stdout_stream_install(s, fd, NULL);
 712         if (r < 0)
 713                 return r;
 714
 715         TAKE_FD(fd);
 716         return 0;
 717 }
 718
 719 static int stdout_stream_load(StdoutStream *stream, const char *fname) {
 720         _cleanup_free_ char
 721                 *priority = NULL,
 722                 *level_prefix = NULL,
 723                 *forward_to_syslog = NULL,
 724                 *forward_to_kmsg = NULL,
 725                 *forward_to_console = NULL,
 726                 *stream_id = NULL;
 727         int r;
 728
 729         assert(stream);
 730         assert(fname);
 731
 732         if (!stream->state_file) {
 733                 stream->state_file = path_join(stream->server->runtime_directory, "streams", fname);
 734                 if (!stream->state_file)
 735                         return log_oom();
 736         }
 737
 738         r = parse_env_file(NULL, stream->state_file,
 739                            "PRIORITY", &priority,
 740                            "LEVEL_PREFIX", &level_prefix,
 741                            "FORWARD_TO_SYSLOG", &forward_to_syslog,
 742                            "FORWARD_TO_KMSG", &forward_to_kmsg,
 743                            "FORWARD_TO_CONSOLE", &forward_to_console,
 744                            "IDENTIFIER", &stream->identifier,
 745                            "UNIT", &stream->unit_id,
 746                            "STREAM_ID", &stream_id);
 747         if (r < 0)
 748                 return log_error_errno(r, "Failed to read: %s", stream->state_file);
 749
 750         if (priority) {
 751                 int p;
 752
 753                 p = log_level_from_string(priority);
 754                 if (p >= 0)
 755                         stream->priority = p;
 756         }
 757
 758         if (level_prefix) {
 759                 r = parse_boolean(level_prefix);
 760                 if (r >= 0)
 761                         stream->level_prefix = r;
 762         }
 763
 764         if (forward_to_syslog) {
 765                 r = parse_boolean(forward_to_syslog);
 766                 if (r >= 0)
 767                         stream->forward_to_syslog = r;
 768         }
 769
 770         if (forward_to_kmsg) {
 771                 r = parse_boolean(forward_to_kmsg);
 772                 if (r >= 0)
 773                         stream->forward_to_kmsg = r;
 774         }
 775
 776         if (forward_to_console) {
 777                 r = parse_boolean(forward_to_console);
 778                 if (r >= 0)
 779                         stream->forward_to_console = r;
 780         }
 781
 782         if (stream_id) {
 783                 sd_id128_t id;
 784
 785                 r = sd_id128_from_string(stream_id, &id);
 786                 if (r >= 0)
 787                         xsprintf(stream->id_field, "_STREAM_ID=" SD_ID128_FORMAT_STR, SD_ID128_FORMAT_VAL(id));
 788         }
 789
 790         return 0;
 791 }
 792
 793 static int stdout_stream_restore(Server *s, const char *fname, int fd) {
 794         StdoutStream *stream;
 795         int r;
 796
 797         assert(s);
 798         assert(fname);
 799         assert(fd >= 0);
 800
 801         if (s->n_stdout_streams >= STDOUT_STREAMS_MAX) {
 802                 log_warning("Too many stdout streams, refusing restoring of stream.");
 803                 return -ENOBUFS;
 804         }
 805
 806         r = stdout_stream_install(s, fd, &stream);
 807         if (r < 0)
 808                 return r;
 809
 810         stream->state = STDOUT_STREAM_RUNNING;
 811         stream->fdstore = true;
 812
 813         /* Ignore all parsing errors */
 814         (void) stdout_stream_load(stream, fname);
 815
 816         return 0;
 817 }
 818
 819 int server_restore_streams(Server *s, FDSet *fds) {
 820         _cleanup_closedir_ DIR *d = NULL;
 821         struct dirent *de;
 822         const char *path;
 823         int r;
 824
 825         path = strjoina(s->runtime_directory, "/streams");
 826         d = opendir(path);
 827         if (!d) {
 828                 if (errno == ENOENT)
 829                         return 0;
 830
 831                 return log_warning_errno(errno, "Failed to enumerate %s: %m", path);
 832         }
 833
 834         FOREACH_DIRENT(de, d, goto fail) {
 835                 unsigned long st_dev, st_ino;
 836                 bool found = false;
 837                 Iterator i;
 838                 int fd;
 839
 840                 if (sscanf(de->d_name, "%lu:%lu", &st_dev, &st_ino) != 2)
 841                         continue;
 842
 843                 FDSET_FOREACH(fd, fds, i) {
 844                         struct stat st;
 845
 846                         if (fstat(fd, &st) < 0)
 847                                 return log_error_errno(errno, "Failed to stat %s: %m", de->d_name);
 848
 849                         if (S_ISSOCK(st.st_mode) && st.st_dev == st_dev && st.st_ino == st_ino) {
 850                                 found = true;
 851                                 break;
 852                         }
 853                 }
 854
 855                 if (!found) {
 856                         /* No file descriptor? Then let's delete the state file */
 857                         log_debug("Cannot restore stream file %s", de->d_name);
 858                         if (unlinkat(dirfd(d), de->d_name, 0) < 0)
 859                                 log_warning_errno(errno, "Failed to remove %s/%s: %m", path, de->d_name);
 860                         continue;
 861                 }
 862
 863                 fdset_remove(fds, fd);
 864
 865                 r = stdout_stream_restore(s, de->d_name, fd);
 866                 if (r < 0)
 867                         safe_close(fd);
 868         }
 869
 870         return 0;
 871
 872 fail:
 873         return log_error_errno(errno, "Failed to read streams directory: %m");
 874 }
 875
 876 int server_open_stdout_socket(Server *s, const char *stdout_socket) {
 877         int r;
 878
 879         assert(s);
 880         assert(stdout_socket);
 881
 882         if (s->stdout_fd < 0) {
 883                 union sockaddr_union sa;
 884                 socklen_t sa_len;
 885
 886                 r = sockaddr_un_set_path(&sa.un, stdout_socket);
 887                 if (r < 0)
 888                         return log_error_errno(r, "Unable to use namespace path %s for AF_UNIX socket: %m", stdout_socket);
 889                 sa_len = r;
 890
 891                 s->stdout_fd = socket(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC|SOCK_NONBLOCK, 0);
 892                 if (s->stdout_fd < 0)
 893                         return log_error_errno(errno, "socket() failed: %m");
 894
 895                 (void) sockaddr_un_unlink(&sa.un);
 896
 897                 r = bind(s->stdout_fd, &sa.sa, sa_len);
 898                 if (r < 0)
 899                         return log_error_errno(errno, "bind(%s) failed: %m", sa.un.sun_path);
 900
 901                 (void) chmod(sa.un.sun_path, 0666);
 902
 903                 if (listen(s->stdout_fd, SOMAXCONN) < 0)
 904                         return log_error_errno(errno, "listen(%s) failed: %m", sa.un.sun_path);
 905         } else
 906                 (void) fd_nonblock(s->stdout_fd, true);
 907
 908         r = sd_event_add_io(s->event, &s->stdout_event_source, s->stdout_fd, EPOLLIN, stdout_stream_new, s);
 909         if (r < 0)
 910                 return log_error_errno(r, "Failed to add stdout server fd to event source: %m");
 911
 912         r = sd_event_source_set_priority(s->stdout_event_source, SD_EVENT_PRIORITY_NORMAL+5);
 913         if (r < 0)
 914                 return log_error_errno(r, "Failed to adjust priority of stdout server event source: %m");
 915
 916         return 0;
 917 }
 918
 919 void stdout_stream_send_notify(StdoutStream *s) {
 920         struct iovec iovec = {
 921                 .iov_base = (char*) "FDSTORE=1",
 922                 .iov_len = STRLEN("FDSTORE=1"),
 923         };
 924         struct msghdr msghdr = {
 925                 .msg_iov = &iovec,
 926                 .msg_iovlen = 1,
 927         };
 928         struct cmsghdr *cmsg;
 929         ssize_t l;
 930
 931         assert(s);
 932         assert(!s->fdstore);
 933         assert(s->in_notify_queue);
 934         assert(s->server);
 935         assert(s->server->notify_fd >= 0);
 936
 937         /* Store the connection fd in PID 1, so that we get it passed
 938          * in again on next start */
 939
 940         msghdr.msg_controllen = CMSG_SPACE(sizeof(int));
 941         msghdr.msg_control = alloca0(msghdr.msg_controllen);
 942
 943         cmsg = CMSG_FIRSTHDR(&msghdr);
 944         cmsg->cmsg_level = SOL_SOCKET;
 945         cmsg->cmsg_type = SCM_RIGHTS;
 946         cmsg->cmsg_len = CMSG_LEN(sizeof(int));
 947
 948         memcpy(CMSG_DATA(cmsg), &s->fd, sizeof(int));
 949
 950         l = sendmsg(s->server->notify_fd, &msghdr, MSG_DONTWAIT|MSG_NOSIGNAL);
 951         if (l < 0) {
 952                 if (errno == EAGAIN)
 953                         return;
 954
 955                 log_error_errno(errno, "Failed to send stream file descriptor to service manager: %m");
 956         } else {
 957                 log_debug("Successfully sent stream file descriptor to service manager.");
 958                 s->fdstore = 1;
 959         }
 960
 961         LIST_REMOVE(stdout_stream_notify_queue, s->server->stdout_streams_notify_queue, s);
 962         s->in_notify_queue = false;
 963
 964 }