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(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->unit_id);
 344         free(s);
 345 }
 346
 347 int stdout_stream_new(Server *s) {
 348         StdoutStream *stream;
 349         int fd, r;
 350         socklen_t len;
 351         struct epoll_event ev;
 352
 353         assert(s);
 354
 355         fd = accept4(s->stdout_fd, NULL, NULL, SOCK_NONBLOCK|SOCK_CLOEXEC);
 356         if (fd < 0) {
 357                 if (errno == EAGAIN)
 358                         return 0;
 359
 360                 log_error("Failed to accept stdout connection: %m");
 361                 return -errno;
 362         }
 363
 364         if (s->n_stdout_streams >= STDOUT_STREAMS_MAX) {
 365                 log_warning("Too many stdout streams, refusing connection.");
 366                 close_nointr_nofail(fd);
 367                 return 0;
 368         }
 369
 370         stream = new0(StdoutStream, 1);
 371         if (!stream) {
 372                 close_nointr_nofail(fd);
 373                 return log_oom();
 374         }
 375
 376         stream->fd = fd;
 377
 378         len = sizeof(stream->ucred);
 379         if (getsockopt(fd, SOL_SOCKET, SO_PEERCRED, &stream->ucred, &len) < 0) {
 380                 log_error("Failed to determine peer credentials: %m");
 381                 r = -errno;
 382                 goto fail;
 383         }
 384
 385 #ifdef HAVE_SELINUX
 386         if (use_selinux()) {
 387                 if (getpeercon(fd, &stream->security_context) < 0 && errno != ENOPROTOOPT)
 388                         log_error("Failed to determine peer security context: %m");
 389         }
 390 #endif
 391
 392         if (shutdown(fd, SHUT_WR) < 0) {
 393                 log_error("Failed to shutdown writing side of socket: %m");
 394                 r = -errno;
 395                 goto fail;
 396         }
 397
 398         zero(ev);
 399         ev.data.ptr = stream;
 400         ev.events = EPOLLIN;
 401         if (epoll_ctl(s->epoll_fd, EPOLL_CTL_ADD, fd, &ev) < 0) {
 402                 log_error("Failed to add stream to event loop: %m");
 403                 r = -errno;
 404                 goto fail;
 405         }
 406
 407         stream->server = s;
 408         LIST_PREPEND(stdout_stream, s->stdout_streams, stream);
 409         s->n_stdout_streams ++;
 410
 411         return 0;
 412
 413 fail:
 414         stdout_stream_free(stream);
 415         return r;
 416 }
 417
 418 int server_open_stdout_socket(Server *s) {
 419         int r;
 420         struct epoll_event ev;
 421
 422         assert(s);
 423
 424         if (s->stdout_fd < 0) {
 425                 union sockaddr_union sa = {
 426                         .un.sun_family = AF_UNIX,
 427                         .un.sun_path = "/run/systemd/journal/stdout",
 428                 };
 429
 430                 s->stdout_fd = socket(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC|SOCK_NONBLOCK, 0);
 431                 if (s->stdout_fd < 0) {
 432                         log_error("socket() failed: %m");
 433                         return -errno;
 434                 }
 435
 436                 unlink(sa.un.sun_path);
 437
 438                 r = bind(s->stdout_fd, &sa.sa, offsetof(union sockaddr_union, un.sun_path) + strlen(sa.un.sun_path));
 439                 if (r < 0) {
 440                         log_error("bind() failed: %m");
 441                         return -errno;
 442                 }
 443
 444                 chmod(sa.un.sun_path, 0666);
 445
 446                 if (listen(s->stdout_fd, SOMAXCONN) < 0) {
 447                         log_error("listen() failed: %m");
 448                         return -errno;
 449                 }
 450         } else
 451                 fd_nonblock(s->stdout_fd, 1);
 452
 453         zero(ev);
 454         ev.events = EPOLLIN;
 455         ev.data.fd = s->stdout_fd;
 456         if (epoll_ctl(s->epoll_fd, EPOLL_CTL_ADD, s->stdout_fd, &ev) < 0) {
 457                 log_error("Failed to add stdout server fd to epoll object: %m");
 458                 return -errno;
 459         }
 460
 461         return 0;
 462 }