src/journal/journald-stream.c

   1 /*-*- Mode: C; c-basic-offset: 8; indent-tabs-mode: nil -*-*/
   2
   3 /***
   4   This file is part of systemd.
   5
   6   Copyright 2011 Lennart Poettering
   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 <fcntl.h>
  23 #include <unistd.h>
  24 #include <sys/epoll.h>
  25
  26 #ifdef HAVE_SELINUX
  27 #include <selinux/selinux.h>
  28 #endif
  29
  30 #include "socket-util.h"
  31 #include "journald.h"
  32 #include "journald-stream.h"
  33 #include "journald-syslog.h"
  34 #include "journald-kmsg.h"
  35
  36 #define STDOUT_STREAMS_MAX 4096
  37
  38 typedef enum StdoutStreamState {
  39         STDOUT_STREAM_IDENTIFIER,
  40         STDOUT_STREAM_UNIT_ID,
  41         STDOUT_STREAM_PRIORITY,
  42         STDOUT_STREAM_LEVEL_PREFIX,
  43         STDOUT_STREAM_FORWARD_TO_SYSLOG,
  44         STDOUT_STREAM_FORWARD_TO_KMSG,
  45         STDOUT_STREAM_FORWARD_TO_CONSOLE,
  46         STDOUT_STREAM_RUNNING
  47 } StdoutStreamState;
  48
  49 struct StdoutStream {
  50         Server *server;
  51         StdoutStreamState state;
  52
  53         int fd;
  54
  55         struct ucred ucred;
  56 #ifdef HAVE_SELINUX
  57         security_context_t security_context;
  58 #endif
  59
  60         char *identifier;
  61         char *unit_id;
  62         int priority;
  63         bool level_prefix:1;
  64         bool forward_to_syslog:1;
  65         bool forward_to_kmsg:1;
  66         bool forward_to_console:1;
  67
  68         char buffer[LINE_MAX+1];
  69         size_t length;
  70
  71         LIST_FIELDS(StdoutStream, stdout_stream);
  72 };
  73
  74 static int stdout_stream_log(StdoutStream *s, const char *p) {
  75         struct iovec iovec[N_IOVEC_META_FIELDS + 5];
  76         char *message = NULL, *syslog_priority = NULL, *syslog_facility = NULL, *syslog_identifier = NULL;
  77         unsigned n = 0;
  78         int priority;
  79         char *label = NULL;
  80         size_t label_len = 0;
  81
  82         assert(s);
  83         assert(p);
  84
  85         if (isempty(p))
  86                 return 0;
  87
  88         priority = s->priority;
  89
  90         if (s->level_prefix)
  91                 syslog_parse_priority((char**) &p, &priority);
  92
  93         if (s->forward_to_syslog || s->server->forward_to_syslog)
  94                 server_forward_syslog(s->server, syslog_fixup_facility(priority), s->identifier, p, &s->ucred, NULL);
  95
  96         if (s->forward_to_kmsg || s->server->forward_to_kmsg)
  97                 server_forward_kmsg(s->server, priority, s->identifier, p, &s->ucred);
  98
  99         if (s->forward_to_console || s->server->forward_to_console)
 100                 server_forward_console(s->server, priority, s->identifier, p, &s->ucred);
 101
 102         IOVEC_SET_STRING(iovec[n++], "_TRANSPORT=stdout");
 103
 104         if (asprintf(&syslog_priority, "PRIORITY=%i", priority & LOG_PRIMASK) >= 0)
 105                 IOVEC_SET_STRING(iovec[n++], syslog_priority);
 106
 107         if (priority & LOG_FACMASK)
 108                 if (asprintf(&syslog_facility, "SYSLOG_FACILITY=%i", LOG_FAC(priority)) >= 0)
 109                         IOVEC_SET_STRING(iovec[n++], syslog_facility);
 110
 111         if (s->identifier) {
 112                 syslog_identifier = strappend("SYSLOG_IDENTIFIER=", s->identifier);
 113                 if (syslog_identifier)
 114                         IOVEC_SET_STRING(iovec[n++], syslog_identifier);
 115         }
 116
 117         message = strappend("MESSAGE=", p);
 118         if (message)
 119                 IOVEC_SET_STRING(iovec[n++], message);
 120
 121 #ifdef HAVE_SELINUX
 122         if (s->security_context) {
 123                 label = (char*) s->security_context;
 124                 label_len = strlen((char*) s->security_context);
 125         }
 126 #endif
 127
 128         server_dispatch_message(s->server, iovec, n, ELEMENTSOF(iovec), &s->ucred, NULL, label, label_len, s->unit_id, priority);
 129
 130         free(message);
 131         free(syslog_priority);
 132         free(syslog_facility);
 133         free(syslog_identifier);
 134
 135         return 0;
 136 }
 137
 138 static int stdout_stream_line(StdoutStream *s, char *p) {
 139         int r;
 140
 141         assert(s);
 142         assert(p);
 143
 144         p = strstrip(p);
 145
 146         switch (s->state) {
 147
 148         case STDOUT_STREAM_IDENTIFIER:
 149                 if (isempty(p))
 150                         s->identifier = NULL;
 151                 else  {
 152                         s->identifier = strdup(p);
 153                         if (!s->identifier)
 154                                 return log_oom();
 155                 }
 156
 157                 s->state = STDOUT_STREAM_UNIT_ID;
 158                 return 0;
 159
 160         case STDOUT_STREAM_UNIT_ID:
 161                 if (s->ucred.uid == 0) {
 162                         if (isempty(p))
 163                                 s->unit_id = NULL;
 164                         else  {
 165                                 s->unit_id = strdup(p);
 166                                 if (!s->unit_id)
 167                                         return log_oom();
 168                         }
 169                 }
 170
 171                 s->state = STDOUT_STREAM_PRIORITY;
 172                 return 0;
 173
 174         case STDOUT_STREAM_PRIORITY:
 175                 r = safe_atoi(p, &s->priority);
 176                 if (r < 0 || s->priority <= 0 || s->priority >= 999) {
 177                         log_warning("Failed to parse log priority line.");
 178                         return -EINVAL;
 179                 }
 180
 181                 s->state = STDOUT_STREAM_LEVEL_PREFIX;
 182                 return 0;
 183
 184         case STDOUT_STREAM_LEVEL_PREFIX:
 185                 r = parse_boolean(p);
 186                 if (r < 0) {
 187                         log_warning("Failed to parse level prefix line.");
 188                         return -EINVAL;
 189                 }
 190
 191                 s->level_prefix = !!r;
 192                 s->state = STDOUT_STREAM_FORWARD_TO_SYSLOG;
 193                 return 0;
 194
 195         case STDOUT_STREAM_FORWARD_TO_SYSLOG:
 196                 r = parse_boolean(p);
 197                 if (r < 0) {
 198                         log_warning("Failed to parse forward to syslog line.");
 199                         return -EINVAL;
 200                 }
 201
 202                 s->forward_to_syslog = !!r;
 203                 s->state = STDOUT_STREAM_FORWARD_TO_KMSG;
 204                 return 0;
 205
 206         case STDOUT_STREAM_FORWARD_TO_KMSG:
 207                 r = parse_boolean(p);
 208                 if (r < 0) {
 209                         log_warning("Failed to parse copy to kmsg line.");
 210                         return -EINVAL;
 211                 }
 212
 213                 s->forward_to_kmsg = !!r;
 214                 s->state = STDOUT_STREAM_FORWARD_TO_CONSOLE;
 215                 return 0;
 216
 217         case STDOUT_STREAM_FORWARD_TO_CONSOLE:
 218                 r = parse_boolean(p);
 219                 if (r < 0) {
 220                         log_warning("Failed to parse copy to console line.");
 221                         return -EINVAL;
 222                 }
 223
 224                 s->forward_to_console = !!r;
 225                 s->state = STDOUT_STREAM_RUNNING;
 226                 return 0;
 227
 228         case STDOUT_STREAM_RUNNING:
 229                 return stdout_stream_log(s, p);
 230         }
 231
 232         assert_not_reached("Unknown stream state");
 233 }
 234
 235 static int stdout_stream_scan(StdoutStream *s, bool force_flush) {
 236         char *p;
 237         size_t remaining;
 238         int r;
 239
 240         assert(s);
 241
 242         p = s->buffer;
 243         remaining = s->length;
 244         for (;;) {
 245                 char *end;
 246                 size_t skip;
 247
 248                 end = memchr(p, '\n', remaining);
 249                 if (end)
 250                         skip = end - p + 1;
 251                 else if (remaining >= sizeof(s->buffer) - 1) {
 252                         end = p + sizeof(s->buffer) - 1;
 253                         skip = remaining;
 254                 } else
 255                         break;
 256
 257                 *end = 0;
 258
 259                 r = stdout_stream_line(s, p);
 260                 if (r < 0)
 261                         return r;
 262
 263                 remaining -= skip;
 264                 p += skip;
 265         }
 266
 267         if (force_flush && remaining > 0) {
 268                 p[remaining] = 0;
 269                 r = stdout_stream_line(s, p);
 270                 if (r < 0)
 271                         return r;
 272
 273                 p += remaining;
 274                 remaining = 0;
 275         }
 276
 277         if (p > s->buffer) {
 278                 memmove(s->buffer, p, remaining);
 279                 s->length = remaining;
 280         }
 281
 282         return 0;
 283 }
 284
 285 int stdout_stream_process(StdoutStream *s) {
 286         ssize_t l;
 287         int r;
 288
 289         assert(s);
 290
 291         l = read(s->fd, s->buffer+s->length, sizeof(s->buffer)-1-s->length);
 292         if (l < 0) {
 293
 294                 if (errno == EAGAIN)
 295                         return 0;
 296
 297                 log_warning("Failed to read from stream: %m");
 298                 return -errno;
 299         }
 300
 301         if (l == 0) {
 302                 r = stdout_stream_scan(s, true);
 303                 if (r < 0)
 304                         return r;
 305
 306                 return 0;
 307         }
 308
 309         s->length += l;
 310         r = stdout_stream_scan(s, false);
 311         if (r < 0)
 312                 return r;
 313
 314         return 1;
 315
 316 }
 317
 318 void stdout_stream_free(StdoutStream *s) {
 319         assert(s);
 320
 321         if (s->server) {
 322                 assert(s->server->n_stdout_streams > 0);
 323                 s->server->n_stdout_streams --;
 324                 LIST_REMOVE(StdoutStream, stdout_stream, s->server->stdout_streams, s);
 325         }
 326
 327         if (s->fd >= 0) {
 328                 if (s->server)
 329                         epoll_ctl(s->server->epoll_fd, EPOLL_CTL_DEL, s->fd, NULL);
 330
 331                 close_nointr_nofail(s->fd);
 332         }
 333
 334 #ifdef HAVE_SELINUX
 335         if (s->security_context)
 336                 freecon(s->security_context);
 337 #endif
 338
 339         free(s->identifier);
 340         free(s);
 341 }
 342
 343 int stdout_stream_new(Server *s) {
 344         StdoutStream *stream;
 345         int fd, r;
 346         socklen_t len;
 347         struct epoll_event ev;
 348
 349         assert(s);
 350
 351         fd = accept4(s->stdout_fd, NULL, NULL, SOCK_NONBLOCK|SOCK_CLOEXEC);
 352         if (fd < 0) {
 353                 if (errno == EAGAIN)
 354                         return 0;
 355
 356                 log_error("Failed to accept stdout connection: %m");
 357                 return -errno;
 358         }
 359
 360         if (s->n_stdout_streams >= STDOUT_STREAMS_MAX) {
 361                 log_warning("Too many stdout streams, refusing connection.");
 362                 close_nointr_nofail(fd);
 363                 return 0;
 364         }
 365
 366         stream = new0(StdoutStream, 1);
 367         if (!stream) {
 368                 close_nointr_nofail(fd);
 369                 return log_oom();
 370         }
 371
 372         stream->fd = fd;
 373
 374         len = sizeof(stream->ucred);
 375         if (getsockopt(fd, SOL_SOCKET, SO_PEERCRED, &stream->ucred, &len) < 0) {
 376                 log_error("Failed to determine peer credentials: %m");
 377                 r = -errno;
 378                 goto fail;
 379         }
 380
 381 #ifdef HAVE_SELINUX
 382         if (getpeercon(fd, &stream->security_context) < 0 && errno != ENOPROTOOPT)
 383                 log_error("Failed to determine peer security context: %m");
 384 #endif
 385
 386         if (shutdown(fd, SHUT_WR) < 0) {
 387                 log_error("Failed to shutdown writing side of socket: %m");
 388                 r = -errno;
 389                 goto fail;
 390         }
 391
 392         zero(ev);
 393         ev.data.ptr = stream;
 394         ev.events = EPOLLIN;
 395         if (epoll_ctl(s->epoll_fd, EPOLL_CTL_ADD, fd, &ev) < 0) {
 396                 log_error("Failed to add stream to event loop: %m");
 397                 r = -errno;
 398                 goto fail;
 399         }
 400
 401         stream->server = s;
 402         LIST_PREPEND(StdoutStream, stdout_stream, s->stdout_streams, stream);
 403         s->n_stdout_streams ++;
 404
 405         return 0;
 406
 407 fail:
 408         stdout_stream_free(stream);
 409         return r;
 410 }
 411
 412 int server_open_stdout_socket(Server *s) {
 413         union sockaddr_union sa;
 414         int r;
 415         struct epoll_event ev;
 416
 417         assert(s);
 418
 419         if (s->stdout_fd < 0) {
 420
 421                 s->stdout_fd = socket(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC|SOCK_NONBLOCK, 0);
 422                 if (s->stdout_fd < 0) {
 423                         log_error("socket() failed: %m");
 424                         return -errno;
 425                 }
 426
 427                 zero(sa);
 428                 sa.un.sun_family = AF_UNIX;
 429                 strncpy(sa.un.sun_path, "/run/systemd/journal/stdout", sizeof(sa.un.sun_path));
 430
 431                 unlink(sa.un.sun_path);
 432
 433                 r = bind(s->stdout_fd, &sa.sa, offsetof(union sockaddr_union, un.sun_path) + strlen(sa.un.sun_path));
 434                 if (r < 0) {
 435                         log_error("bind() failed: %m");
 436                         return -errno;
 437                 }
 438
 439                 chmod(sa.un.sun_path, 0666);
 440
 441                 if (listen(s->stdout_fd, SOMAXCONN) < 0) {
 442                         log_error("liste() failed: %m");
 443                         return -errno;
 444                 }
 445         } else
 446                 fd_nonblock(s->stdout_fd, 1);
 447
 448         zero(ev);
 449         ev.events = EPOLLIN;
 450         ev.data.fd = s->stdout_fd;
 451         if (epoll_ctl(s->epoll_fd, EPOLL_CTL_ADD, s->stdout_fd, &ev) < 0) {
 452                 log_error("Failed to add stdout server fd to epoll object: %m");
 453                 return -errno;
 454         }
 455
 456         return 0;
 457 }