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