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