[systemd.git] / src / journal / journald-stream.c

/*-*- Mode: C; c-basic-offset: 8; indent-tabs-mode: nil -*-*/

/***
  This file is part of systemd.

  Copyright 2011 Lennart Poettering

  systemd is free software; you can redistribute it and/or modify it
  under the terms of the GNU Lesser General Public License as published by
  the Free Software Foundation; either version 2.1 of the License, or
  (at your option) any later version.

  systemd is distributed in the hope that it will be useful, but
  WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  Lesser General Public License for more details.

  You should have received a copy of the GNU Lesser General Public License
  along with systemd; If not, see <http://www.gnu.org/licenses/>.
***/

#include <fcntl.h>
#include <unistd.h>
#include <stddef.h>

#ifdef HAVE_SELINUX
#include <selinux/selinux.h>
#endif

#include "sd-event.h"
#include "sd-daemon.h"
#include "socket-util.h"
#include "selinux-util.h"
#include "mkdir.h"
#include "fileio.h"
#include "journald-server.h"
#include "journald-stream.h"
#include "journald-syslog.h"
#include "journald-kmsg.h"
#include "journald-console.h"
#include "journald-wall.h"

#define STDOUT_STREAMS_MAX 4096

typedef enum StdoutStreamState {
        STDOUT_STREAM_IDENTIFIER,
        STDOUT_STREAM_UNIT_ID,
        STDOUT_STREAM_PRIORITY,
        STDOUT_STREAM_LEVEL_PREFIX,
        STDOUT_STREAM_FORWARD_TO_SYSLOG,
        STDOUT_STREAM_FORWARD_TO_KMSG,
        STDOUT_STREAM_FORWARD_TO_CONSOLE,
        STDOUT_STREAM_RUNNING
} StdoutStreamState;

struct StdoutStream {
        Server *server;
        StdoutStreamState state;

        int fd;

        struct ucred ucred;
#ifdef HAVE_SELINUX
        security_context_t security_context;
#endif

        char *identifier;
        char *unit_id;
        int priority;
        bool level_prefix:1;
        bool forward_to_syslog:1;
        bool forward_to_kmsg:1;
        bool forward_to_console:1;

        bool fdstore:1;

        char buffer[LINE_MAX+1];
        size_t length;

        sd_event_source *event_source;

        char *state_file;

        LIST_FIELDS(StdoutStream, stdout_stream);
};

void stdout_stream_free(StdoutStream *s) {
        if (!s)
                return;

        if (s->server) {
                assert(s->server->n_stdout_streams > 0);
                s->server->n_stdout_streams --;
                LIST_REMOVE(stdout_stream, s->server->stdout_streams, s);
        }

        if (s->event_source) {
                sd_event_source_set_enabled(s->event_source, SD_EVENT_OFF);
                s->event_source = sd_event_source_unref(s->event_source);
        }

        safe_close(s->fd);

#ifdef HAVE_SELINUX
        if (s->security_context)
                freecon(s->security_context);
#endif

        free(s->identifier);
        free(s->unit_id);
        free(s->state_file);

        free(s);
}

DEFINE_TRIVIAL_CLEANUP_FUNC(StdoutStream*, stdout_stream_free);

static void stdout_stream_destroy(StdoutStream *s) {
        if (!s)
                return;

        if (s->state_file)
                unlink(s->state_file);

        stdout_stream_free(s);
}

static int stdout_stream_save(StdoutStream *s) {
        _cleanup_free_ char *temp_path = NULL;
        _cleanup_fclose_ FILE *f = NULL;
        int r;

        assert(s);

        if (s->state != STDOUT_STREAM_RUNNING)
                return 0;

        if (!s->state_file) {
                struct stat st;

                r = fstat(s->fd, &st);
                if (r < 0)
                        return log_warning_errno(errno, "Failed to stat connected stream: %m");

                /* We use device and inode numbers as identifier for the stream */
                if (asprintf(&s->state_file, "/run/systemd/journal/streams/%lu:%lu", (unsigned long) st.st_dev, (unsigned long) st.st_ino) < 0)
                        return log_oom();
        }

        mkdir_p("/run/systemd/journal/streams", 0755);

        r = fopen_temporary(s->state_file, &f, &temp_path);
        if (r < 0)
                goto finish;

        fprintf(f,
                "# This is private data. Do not parse\n"
                "PRIORITY=%i\n"
                "LEVEL_PREFIX=%i\n"
                "FORWARD_TO_SYSLOG=%i\n"
                "FORWARD_TO_KMSG=%i\n"
                "FORWARD_TO_CONSOLE=%i\n",
                s->priority,
                s->level_prefix,
                s->forward_to_syslog,
                s->forward_to_kmsg,
                s->forward_to_console);

        if (!isempty(s->identifier)) {
                _cleanup_free_ char *escaped;

                escaped = cescape(s->identifier);
                if (!escaped) {
                        r = -ENOMEM;
                        goto finish;
                }

                fprintf(f, "IDENTIFIER=%s\n", escaped);
        }

        if (!isempty(s->unit_id)) {
                _cleanup_free_ char *escaped;

                escaped = cescape(s->unit_id);
                if (!escaped) {
                        r = -ENOMEM;
                        goto finish;
                }

                fprintf(f, "UNIT=%s\n", escaped);
        }

        r = fflush_and_check(f);
        if (r < 0)
                goto finish;

        if (rename(temp_path, s->state_file) < 0) {
                r = -errno;
                goto finish;
        }

        free(temp_path);
        temp_path = NULL;

        /* Store the connection fd in PID 1, so that we get it passed
         * in again on next start */
        if (!s->fdstore) {
                sd_pid_notify_with_fds(0, false, "FDSTORE=1", &s->fd, 1);
                s->fdstore = true;
        }

finish:
        if (temp_path)
                unlink(temp_path);

        if (r < 0)
                log_error_errno(r, "Failed to save stream data %s: %m", s->state_file);

        return r;
}

static int stdout_stream_log(StdoutStream *s, const char *p) {
        struct iovec iovec[N_IOVEC_META_FIELDS + 5];
        int priority;
        char syslog_priority[] = "PRIORITY=\0";
        char syslog_facility[sizeof("SYSLOG_FACILITY=")-1 + DECIMAL_STR_MAX(int) + 1];
        _cleanup_free_ char *message = NULL, *syslog_identifier = NULL;
        unsigned n = 0;
        char *label = NULL;
        size_t label_len = 0;

        assert(s);
        assert(p);

        if (isempty(p))
                return 0;

        priority = s->priority;

        if (s->level_prefix)
                syslog_parse_priority(&p, &priority, false);

        if (s->forward_to_syslog || s->server->forward_to_syslog)
                server_forward_syslog(s->server, syslog_fixup_facility(priority), s->identifier, p, &s->ucred, NULL);

        if (s->forward_to_kmsg || s->server->forward_to_kmsg)
                server_forward_kmsg(s->server, priority, s->identifier, p, &s->ucred);

        if (s->forward_to_console || s->server->forward_to_console)
                server_forward_console(s->server, priority, s->identifier, p, &s->ucred);

        if (s->server->forward_to_wall)
                server_forward_wall(s->server, priority, s->identifier, p, &s->ucred);

        IOVEC_SET_STRING(iovec[n++], "_TRANSPORT=stdout");

        syslog_priority[strlen("PRIORITY=")] = '0' + LOG_PRI(priority);
        IOVEC_SET_STRING(iovec[n++], syslog_priority);

        if (priority & LOG_FACMASK) {
                xsprintf(syslog_facility, "SYSLOG_FACILITY=%i", LOG_FAC(priority));
                IOVEC_SET_STRING(iovec[n++], syslog_facility);
        }

        if (s->identifier) {
                syslog_identifier = strappend("SYSLOG_IDENTIFIER=", s->identifier);
                if (syslog_identifier)
                        IOVEC_SET_STRING(iovec[n++], syslog_identifier);
        }

        message = strappend("MESSAGE=", p);
        if (message)
                IOVEC_SET_STRING(iovec[n++], message);

#ifdef HAVE_SELINUX
        if (s->security_context) {
                label = (char*) s->security_context;
                label_len = strlen((char*) s->security_context);
        }
#endif

        server_dispatch_message(s->server, iovec, n, ELEMENTSOF(iovec), &s->ucred, NULL, label, label_len, s->unit_id, priority, 0);
        return 0;
}

static int stdout_stream_line(StdoutStream *s, char *p) {
        int r;

        assert(s);
        assert(p);

        p = strstrip(p);

        switch (s->state) {

        case STDOUT_STREAM_IDENTIFIER:
                if (isempty(p))
                        s->identifier = NULL;
                else  {
                        s->identifier = strdup(p);
                        if (!s->identifier)
                                return log_oom();
                }

                s->state = STDOUT_STREAM_UNIT_ID;
                return 0;

        case STDOUT_STREAM_UNIT_ID:
                if (s->ucred.uid == 0) {
                        if (isempty(p))
                                s->unit_id = NULL;
                        else  {
                                s->unit_id = strdup(p);
                                if (!s->unit_id)
                                        return log_oom();
                        }
                }

                s->state = STDOUT_STREAM_PRIORITY;
                return 0;

        case STDOUT_STREAM_PRIORITY:
                r = safe_atoi(p, &s->priority);
                if (r < 0 || s->priority < 0 || s->priority > 999) {
                        log_warning("Failed to parse log priority line.");
                        return -EINVAL;
                }

                s->state = STDOUT_STREAM_LEVEL_PREFIX;
                return 0;

        case STDOUT_STREAM_LEVEL_PREFIX:
                r = parse_boolean(p);
                if (r < 0) {
                        log_warning("Failed to parse level prefix line.");
                        return -EINVAL;
                }

                s->level_prefix = !!r;
                s->state = STDOUT_STREAM_FORWARD_TO_SYSLOG;
                return 0;

        case STDOUT_STREAM_FORWARD_TO_SYSLOG:
                r = parse_boolean(p);
                if (r < 0) {
                        log_warning("Failed to parse forward to syslog line.");
                        return -EINVAL;
                }

                s->forward_to_syslog = !!r;
                s->state = STDOUT_STREAM_FORWARD_TO_KMSG;
                return 0;

        case STDOUT_STREAM_FORWARD_TO_KMSG:
                r = parse_boolean(p);
                if (r < 0) {
                        log_warning("Failed to parse copy to kmsg line.");
                        return -EINVAL;
                }

                s->forward_to_kmsg = !!r;
                s->state = STDOUT_STREAM_FORWARD_TO_CONSOLE;
                return 0;

        case STDOUT_STREAM_FORWARD_TO_CONSOLE:
                r = parse_boolean(p);
                if (r < 0) {
                        log_warning("Failed to parse copy to console line.");
                        return -EINVAL;
                }

                s->forward_to_console = !!r;
                s->state = STDOUT_STREAM_RUNNING;

                /* Try to save the stream, so that journald can be restarted and we can recover */
                (void) stdout_stream_save(s);
                return 0;

        case STDOUT_STREAM_RUNNING:
                return stdout_stream_log(s, p);
        }

        assert_not_reached("Unknown stream state");
}

static int stdout_stream_scan(StdoutStream *s, bool force_flush) {
        char *p;
        size_t remaining;
        int r;

        assert(s);

        p = s->buffer;
        remaining = s->length;
        for (;;) {
                char *end;
                size_t skip;

                end = memchr(p, '\n', remaining);
                if (end)
                        skip = end - p + 1;
                else if (remaining >= sizeof(s->buffer) - 1) {
                        end = p + sizeof(s->buffer) - 1;
                        skip = remaining;
                } else
                        break;

                *end = 0;

                r = stdout_stream_line(s, p);
                if (r < 0)
                        return r;

                remaining -= skip;
                p += skip;
        }

        if (force_flush && remaining > 0) {
                p[remaining] = 0;
                r = stdout_stream_line(s, p);
                if (r < 0)
                        return r;

                p += remaining;
                remaining = 0;
        }

        if (p > s->buffer) {
                memmove(s->buffer, p, remaining);
                s->length = remaining;
        }

        return 0;
}

static int stdout_stream_process(sd_event_source *es, int fd, uint32_t revents, void *userdata) {
        StdoutStream *s = userdata;
        ssize_t l;
        int r;

        assert(s);

        if ((revents|EPOLLIN|EPOLLHUP) != (EPOLLIN|EPOLLHUP)) {
                log_error("Got invalid event from epoll for stdout stream: %"PRIx32, revents);
                goto terminate;
        }

        l = read(s->fd, s->buffer+s->length, sizeof(s->buffer)-1-s->length);
        if (l < 0) {

                if (errno == EAGAIN)
                        return 0;

                log_warning_errno(errno, "Failed to read from stream: %m");
                goto terminate;
        }

        if (l == 0) {
                stdout_stream_scan(s, true);
                goto terminate;
        }

        s->length += l;
        r = stdout_stream_scan(s, false);
        if (r < 0)
                goto terminate;

        return 1;

terminate:
        stdout_stream_destroy(s);
        return 0;
}

static int stdout_stream_install(Server *s, int fd, StdoutStream **ret) {
        _cleanup_(stdout_stream_freep) StdoutStream *stream = NULL;
        int r;

        assert(s);
        assert(fd >= 0);

        stream = new0(StdoutStream, 1);
        if (!stream)
                return log_oom();

        stream->fd = -1;
        stream->priority = LOG_INFO;

        r = getpeercred(fd, &stream->ucred);
        if (r < 0)
                return log_error_errno(r, "Failed to determine peer credentials: %m");

#ifdef HAVE_SELINUX
        if (mac_selinux_use()) {
                if (getpeercon(fd, &stream->security_context) < 0 && errno != ENOPROTOOPT)
                        log_error_errno(errno, "Failed to determine peer security context: %m");
        }
#endif

        (void) shutdown(fd, SHUT_WR);

        r = sd_event_add_io(s->event, &stream->event_source, fd, EPOLLIN, stdout_stream_process, stream);
        if (r < 0)
                return log_error_errno(r, "Failed to add stream to event loop: %m");

        r = sd_event_source_set_priority(stream->event_source, SD_EVENT_PRIORITY_NORMAL+5);
        if (r < 0)
                return log_error_errno(r, "Failed to adjust stdout event source priority: %m");

        stream->fd = fd;

        stream->server = s;
        LIST_PREPEND(stdout_stream, s->stdout_streams, stream);
        s->n_stdout_streams ++;

        if (ret)
                *ret = stream;

        stream = NULL;

        return 0;
}

static int stdout_stream_new(sd_event_source *es, int listen_fd, uint32_t revents, void *userdata) {
        _cleanup_close_ int fd = -1;
        Server *s = userdata;
        int r;

        assert(s);

        if (revents != EPOLLIN) {
                log_error("Got invalid event from epoll for stdout server fd: %"PRIx32, revents);
                return -EIO;
        }

        fd = accept4(s->stdout_fd, NULL, NULL, SOCK_NONBLOCK|SOCK_CLOEXEC);
        if (fd < 0) {
                if (errno == EAGAIN)
                        return 0;

                log_error_errno(errno, "Failed to accept stdout connection: %m");
                return -errno;
        }

        if (s->n_stdout_streams >= STDOUT_STREAMS_MAX) {
                log_warning("Too many stdout streams, refusing connection.");
                return 0;
        }

        r = stdout_stream_install(s, fd, NULL);
        if (r < 0)
                return r;

        fd = -1;
        return 0;
}

static int stdout_stream_load(StdoutStream *stream, const char *fname) {
        _cleanup_free_ char
                *priority = NULL,
                *level_prefix = NULL,
                *forward_to_syslog = NULL,
                *forward_to_kmsg = NULL,
                *forward_to_console = NULL;
        int r;

        assert(stream);
        assert(fname);

        if (!stream->state_file) {
                stream->state_file = strappend("/run/systemd/journal/streams/", fname);
                if (!stream->state_file)
                        return log_oom();
        }

        r = parse_env_file(stream->state_file, NEWLINE,
                           "PRIORITY", &priority,
                           "LEVEL_PREFIX", &level_prefix,
                           "FORWARD_TO_SYSLOG", &forward_to_syslog,
                           "FORWARD_TO_KMSG", &forward_to_kmsg,
                           "FORWARD_TO_CONSOLE", &forward_to_console,
                           "IDENTIFIER", &stream->identifier,
                           "UNIT", &stream->unit_id,
                           NULL);
        if (r < 0)
                return log_error_errno(r, "Failed to read: %s", stream->state_file);

        if (priority) {
                int p;

                p = log_level_from_string(priority);
                if (p >= 0)
                        stream->priority = p;
        }

        if (level_prefix) {
                r = parse_boolean(level_prefix);
                if (r >= 0)
                        stream->level_prefix = r;
        }

        if (forward_to_syslog) {
                r = parse_boolean(forward_to_syslog);
                if (r >= 0)
                        stream->forward_to_syslog = r;
        }

        if (forward_to_kmsg) {
                r = parse_boolean(forward_to_kmsg);
                if (r >= 0)
                        stream->forward_to_kmsg = r;
        }

        if (forward_to_console) {
                r = parse_boolean(forward_to_console);
                if (r >= 0)
                        stream->forward_to_console = r;
        }

        return 0;
}

static int stdout_stream_restore(Server *s, const char *fname, int fd) {
        StdoutStream *stream;
        int r;

        assert(s);
        assert(fname);
        assert(fd >= 0);

        if (s->n_stdout_streams >= STDOUT_STREAMS_MAX) {
                log_warning("Too many stdout streams, refusing restoring of stream.");
                return -ENOBUFS;
        }

        r = stdout_stream_install(s, fd, &stream);
        if (r < 0)
                return r;

        stream->state = STDOUT_STREAM_RUNNING;
        stream->fdstore = true;

        /* Ignore all parsing errors */
        (void) stdout_stream_load(stream, fname);

        return 0;
}

static int server_restore_streams(Server *s, FDSet *fds) {
        _cleanup_closedir_ DIR *d = NULL;
        struct dirent *de;
        int r;

        d = opendir("/run/systemd/journal/streams");
        if (!d) {
                if (errno == ENOENT)
                        return 0;

                return log_warning_errno(errno, "Failed to enumerate /run/systemd/journal/streams: %m");
        }

        FOREACH_DIRENT(de, d, goto fail) {
                unsigned long st_dev, st_ino;
                bool found = false;
                Iterator i;
                int fd;

                if (sscanf(de->d_name, "%lu:%lu", &st_dev, &st_ino) != 2)
                        continue;

                FDSET_FOREACH(fd, fds, i) {
                        struct stat st;

                        if (fstat(fd, &st) < 0)
                                return log_error_errno(errno, "Failed to stat %s: %m", de->d_name);

                        if (S_ISSOCK(st.st_mode) && st.st_dev == st_dev && st.st_ino == st_ino) {
                                found = true;
                                break;
                        }
                }

                if (!found) {
                        /* No file descriptor? Then let's delete the state file */
                        log_debug("Cannot restore stream file %s", de->d_name);
                        unlinkat(dirfd(d), de->d_name, 0);
                        continue;
                }

                fdset_remove(fds, fd);

                r = stdout_stream_restore(s, de->d_name, fd);
                if (r < 0)
                        safe_close(fd);
        }

        return 0;

fail:
        return log_error_errno(errno, "Failed to read streams directory: %m");
}

int server_open_stdout_socket(Server *s, FDSet *fds) {
        int r;

        assert(s);

        if (s->stdout_fd < 0) {
                union sockaddr_union sa = {
                        .un.sun_family = AF_UNIX,
                        .un.sun_path = "/run/systemd/journal/stdout",
                };

                s->stdout_fd = socket(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC|SOCK_NONBLOCK, 0);
                if (s->stdout_fd < 0)
                        return log_error_errno(errno, "socket() failed: %m");

                unlink(sa.un.sun_path);

                r = bind(s->stdout_fd, &sa.sa, offsetof(union sockaddr_union, un.sun_path) + strlen(sa.un.sun_path));
                if (r < 0)
                        return log_error_errno(errno, "bind(%s) failed: %m", sa.un.sun_path);

                chmod(sa.un.sun_path, 0666);

                if (listen(s->stdout_fd, SOMAXCONN) < 0)
                        return log_error_errno(errno, "listen(%s) failed: %m", sa.un.sun_path);
        } else
                fd_nonblock(s->stdout_fd, 1);

        r = sd_event_add_io(s->event, &s->stdout_event_source, s->stdout_fd, EPOLLIN, stdout_stream_new, s);
        if (r < 0)
                return log_error_errno(r, "Failed to add stdout server fd to event source: %m");

        r = sd_event_source_set_priority(s->stdout_event_source, SD_EVENT_PRIORITY_NORMAL+10);
        if (r < 0)
                return log_error_errno(r, "Failed to adjust priority of stdout server event source: %m");

        /* Try to restore streams, but don't bother if this fails */
        (void) server_restore_streams(s, fds);

        return 0;
}
Commit	Line	Data
663996b3 MS	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>
663996b3 MS	25
	26	#ifdef HAVE_SELINUX
	27	#include <selinux/selinux.h>
	28	#endif
	29
60f067b4	30	#include "sd-event.h"
e735f4d4	31	#include "sd-daemon.h"
663996b3	32	#include "socket-util.h"
60f067b4	33	#include "selinux-util.h"
e735f4d4 MP	34	#include "mkdir.h"
e735f4d4 MP	35	#include "fileio.h"
663996b3 MS	36	#include "journald-server.h"
	37	#include "journald-stream.h"
	38	#include "journald-syslog.h"
	39	#include "journald-kmsg.h"
	40	#include "journald-console.h"
60f067b4	41	#include "journald-wall.h"
663996b3 MS	42
	43	#define STDOUT_STREAMS_MAX 4096
	44
	45	typedef enum StdoutStreamState {
	46	STDOUT_STREAM_IDENTIFIER,
	47	STDOUT_STREAM_UNIT_ID,
	48	STDOUT_STREAM_PRIORITY,
	49	STDOUT_STREAM_LEVEL_PREFIX,
	50	STDOUT_STREAM_FORWARD_TO_SYSLOG,
	51	STDOUT_STREAM_FORWARD_TO_KMSG,
	52	STDOUT_STREAM_FORWARD_TO_CONSOLE,
	53	STDOUT_STREAM_RUNNING
	54	} StdoutStreamState;
	55
	56	struct StdoutStream {
	57	Server *server;
	58	StdoutStreamState state;
	59
	60	int fd;
	61
	62	struct ucred ucred;
	63	#ifdef HAVE_SELINUX
	64	security_context_t security_context;
	65	#endif
	66
	67	char *identifier;
	68	char *unit_id;
	69	int priority;
	70	bool level_prefix:1;
	71	bool forward_to_syslog:1;
	72	bool forward_to_kmsg:1;
	73	bool forward_to_console:1;
	74
e735f4d4 MP	75	bool fdstore:1;
e735f4d4 MP	76
663996b3 MS	77	char buffer[LINE_MAX+1];
	78	size_t length;
	79
60f067b4 JS	80	sd_event_source *event_source;
60f067b4 JS	81
e735f4d4 MP	82	char *state_file;
e735f4d4 MP	83
663996b3 MS	84	LIST_FIELDS(StdoutStream, stdout_stream);
	85	};
	86
e735f4d4 MP	87	void stdout_stream_free(StdoutStream *s) {
	88	if (!s)
	89	return;
	90
	91	if (s->server) {
	92	assert(s->server->n_stdout_streams > 0);
	93	s->server->n_stdout_streams --;
	94	LIST_REMOVE(stdout_stream, s->server->stdout_streams, s);
	95	}
	96
	97	if (s->event_source) {
	98	sd_event_source_set_enabled(s->event_source, SD_EVENT_OFF);
	99	s->event_source = sd_event_source_unref(s->event_source);
	100	}
	101
	102	safe_close(s->fd);
	103
	104	#ifdef HAVE_SELINUX
	105	if (s->security_context)
	106	freecon(s->security_context);
	107	#endif
	108
	109	free(s->identifier);
	110	free(s->unit_id);
	111	free(s->state_file);
	112
	113	free(s);
	114	}
	115
	116	DEFINE_TRIVIAL_CLEANUP_FUNC(StdoutStream*, stdout_stream_free);
	117
	118	static void stdout_stream_destroy(StdoutStream *s) {
	119	if (!s)
	120	return;
	121
	122	if (s->state_file)
	123	unlink(s->state_file);
	124
	125	stdout_stream_free(s);
	126	}
	127
	128	static int stdout_stream_save(StdoutStream *s) {
	129	_cleanup_free_ char *temp_path = NULL;
	130	_cleanup_fclose_ FILE *f = NULL;
	131	int r;
	132
	133	assert(s);
	134
	135	if (s->state != STDOUT_STREAM_RUNNING)
	136	return 0;
	137
	138	if (!s->state_file) {
	139	struct stat st;
	140
	141	r = fstat(s->fd, &st);
	142	if (r < 0)
	143	return log_warning_errno(errno, "Failed to stat connected stream: %m");
	144
	145	/* We use device and inode numbers as identifier for the stream */
	146	if (asprintf(&s->state_file, "/run/systemd/journal/streams/%lu:%lu", (unsigned long) st.st_dev, (unsigned long) st.st_ino) < 0)
	147	return log_oom();
	148	}
	149
	150	mkdir_p("/run/systemd/journal/streams", 0755);
151
152	r = fopen_temporary(s->state_file, &f, &temp_path);
153	if (r < 0)
154	goto finish;
155
156	fprintf(f,
157	"# This is private data. Do not parse\n"
158	"PRIORITY=%i\n"
159	"LEVEL_PREFIX=%i\n"
160	"FORWARD_TO_SYSLOG=%i\n"
161	"FORWARD_TO_KMSG=%i\n"
162	"FORWARD_TO_CONSOLE=%i\n",
163	s->priority,
164	s->level_prefix,
165	s->forward_to_syslog,
166	s->forward_to_kmsg,
167	s->forward_to_console);
168
169	if (!isempty(s->identifier)) {
170	_cleanup_free_ char *escaped;
171
172	escaped = cescape(s->identifier);
173	if (!escaped) {
174	r = -ENOMEM;
175	goto finish;
176	}
177
178	fprintf(f, "IDENTIFIER=%s\n", escaped);
179	}
180
181	if (!isempty(s->unit_id)) {
182	_cleanup_free_ char *escaped;
183
184	escaped = cescape(s->unit_id);
185	if (!escaped) {
186	r = -ENOMEM;
187	goto finish;
188	}
189
190	fprintf(f, "UNIT=%s\n", escaped);
191	}
192
193	r = fflush_and_check(f);
194	if (r < 0)
195	goto finish;
196
197	if (rename(temp_path, s->state_file) < 0) {
198	r = -errno;
199	goto finish;
200	}
201
202	free(temp_path);
203	temp_path = NULL;
204
205	/* Store the connection fd in PID 1, so that we get it passed
206	* in again on next start */
207	if (!s->fdstore) {
208	sd_pid_notify_with_fds(0, false, "FDSTORE=1", &s->fd, 1);
209	s->fdstore = true;
210	}
211
212	finish:
213	if (temp_path)
214	unlink(temp_path);
215
216	if (r < 0)
217	log_error_errno(r, "Failed to save stream data %s: %m", s->state_file);
218
219	return r;
220	}
221
663996b3 MS	222	static int stdout_stream_log(StdoutStream s, const char p) {
663996b3 MS	223	struct iovec iovec[N_IOVEC_META_FIELDS + 5];
663996b3	224	int priority;
60f067b4	225	char syslog_priority[] = "PRIORITY=\0";
e735f4d4	226	char syslog_facility[sizeof("SYSLOG_FACILITY=")-1 + DECIMAL_STR_MAX(int) + 1];
60f067b4 JS	227	_cleanup_free_ char message = NULL, syslog_identifier = NULL;
60f067b4 JS	228	unsigned n = 0;
663996b3 MS	229	char *label = NULL;
	230	size_t label_len = 0;
	231
	232	assert(s);
	233	assert(p);
	234
	235	if (isempty(p))
	236	return 0;
	237
	238	priority = s->priority;
	239
	240	if (s->level_prefix)
60f067b4	241	syslog_parse_priority(&p, &priority, false);
663996b3 MS	242
	243	if (s->forward_to_syslog \|\| s->server->forward_to_syslog)
	244	server_forward_syslog(s->server, syslog_fixup_facility(priority), s->identifier, p, &s->ucred, NULL);
	245
	246	if (s->forward_to_kmsg \|\| s->server->forward_to_kmsg)
	247	server_forward_kmsg(s->server, priority, s->identifier, p, &s->ucred);
	248
	249	if (s->forward_to_console \|\| s->server->forward_to_console)
	250	server_forward_console(s->server, priority, s->identifier, p, &s->ucred);
	251
60f067b4 JS	252	if (s->server->forward_to_wall)
	253	server_forward_wall(s->server, priority, s->identifier, p, &s->ucred);
	254
663996b3 MS	255	IOVEC_SET_STRING(iovec[n++], "_TRANSPORT=stdout");
663996b3 MS	256
60f067b4 JS	257	syslog_priority[strlen("PRIORITY=")] = '0' + LOG_PRI(priority);
60f067b4 JS	258	IOVEC_SET_STRING(iovec[n++], syslog_priority);
663996b3	259
60f067b4	260	if (priority & LOG_FACMASK) {
e735f4d4	261	xsprintf(syslog_facility, "SYSLOG_FACILITY=%i", LOG_FAC(priority));
60f067b4 JS	262	IOVEC_SET_STRING(iovec[n++], syslog_facility);
60f067b4 JS	263	}
663996b3 MS	264
	265	if (s->identifier) {
	266	syslog_identifier = strappend("SYSLOG_IDENTIFIER=", s->identifier);
	267	if (syslog_identifier)
	268	IOVEC_SET_STRING(iovec[n++], syslog_identifier);
	269	}
	270
	271	message = strappend("MESSAGE=", p);
	272	if (message)
	273	IOVEC_SET_STRING(iovec[n++], message);
	274
	275	#ifdef HAVE_SELINUX
	276	if (s->security_context) {
	277	label = (char*) s->security_context;
	278	label_len = strlen((char*) s->security_context);
	279	}
	280	#endif
	281
14228c0d	282	server_dispatch_message(s->server, iovec, n, ELEMENTSOF(iovec), &s->ucred, NULL, label, label_len, s->unit_id, priority, 0);
663996b3 MS	283	return 0;
	284	}
	285
	286	static int stdout_stream_line(StdoutStream s, char p) {
	287	int r;
	288
	289	assert(s);
	290	assert(p);
	291
	292	p = strstrip(p);
	293
	294	switch (s->state) {
	295
	296	case STDOUT_STREAM_IDENTIFIER:
	297	if (isempty(p))
	298	s->identifier = NULL;
	299	else {
	300	s->identifier = strdup(p);
	301	if (!s->identifier)
	302	return log_oom();
	303	}
	304
	305	s->state = STDOUT_STREAM_UNIT_ID;
	306	return 0;
	307
	308	case STDOUT_STREAM_UNIT_ID:
	309	if (s->ucred.uid == 0) {
	310	if (isempty(p))
	311	s->unit_id = NULL;
	312	else {
	313	s->unit_id = strdup(p);
	314	if (!s->unit_id)
	315	return log_oom();
	316	}
	317	}
	318
	319	s->state = STDOUT_STREAM_PRIORITY;
	320	return 0;
	321
	322	case STDOUT_STREAM_PRIORITY:
	323	r = safe_atoi(p, &s->priority);
	324	if (r < 0 \|\| s->priority < 0 \|\| s->priority > 999) {
	325	log_warning("Failed to parse log priority line.");
	326	return -EINVAL;
	327	}
	328
	329	s->state = STDOUT_STREAM_LEVEL_PREFIX;
	330	return 0;
	331
	332	case STDOUT_STREAM_LEVEL_PREFIX:
	333	r = parse_boolean(p);
	334	if (r < 0) {
	335	log_warning("Failed to parse level prefix line.");
	336	return -EINVAL;
	337	}
	338
	339	s->level_prefix = !!r;
	340	s->state = STDOUT_STREAM_FORWARD_TO_SYSLOG;
	341	return 0;
	342
	343	case STDOUT_STREAM_FORWARD_TO_SYSLOG:
	344	r = parse_boolean(p);
	345	if (r < 0) {
	346	log_warning("Failed to parse forward to syslog line.");
347	return -EINVAL;
348	}
349
350	s->forward_to_syslog = !!r;
351	s->state = STDOUT_STREAM_FORWARD_TO_KMSG;
352	return 0;
353
354	case STDOUT_STREAM_FORWARD_TO_KMSG:
355	r = parse_boolean(p);
356	if (r < 0) {
357	log_warning("Failed to parse copy to kmsg line.");
358	return -EINVAL;
359	}
360
361	s->forward_to_kmsg = !!r;
362	s->state = STDOUT_STREAM_FORWARD_TO_CONSOLE;
363	return 0;
364
365	case STDOUT_STREAM_FORWARD_TO_CONSOLE:
366	r = parse_boolean(p);
367	if (r < 0) {
368	log_warning("Failed to parse copy to console line.");
369	return -EINVAL;
370	}
371
372	s->forward_to_console = !!r;
373	s->state = STDOUT_STREAM_RUNNING;
e735f4d4 MP	374
	375	/* Try to save the stream, so that journald can be restarted and we can recover */
	376	(void) stdout_stream_save(s);
663996b3 MS	377	return 0;
	378
	379	case STDOUT_STREAM_RUNNING:
	380	return stdout_stream_log(s, p);
	381	}
	382
	383	assert_not_reached("Unknown stream state");
	384	}
	385
	386	static int stdout_stream_scan(StdoutStream *s, bool force_flush) {
	387	char *p;
	388	size_t remaining;
	389	int r;
	390
	391	assert(s);
	392
	393	p = s->buffer;
	394	remaining = s->length;
	395	for (;;) {
	396	char *end;
	397	size_t skip;
	398
	399	end = memchr(p, '\n', remaining);
	400	if (end)
	401	skip = end - p + 1;
	402	else if (remaining >= sizeof(s->buffer) - 1) {
	403	end = p + sizeof(s->buffer) - 1;
	404	skip = remaining;
	405	} else
	406	break;
	407
	408	*end = 0;
	409
	410	r = stdout_stream_line(s, p);
	411	if (r < 0)
	412	return r;
	413
	414	remaining -= skip;
	415	p += skip;
	416	}
	417
	418	if (force_flush && remaining > 0) {
	419	p[remaining] = 0;
	420	r = stdout_stream_line(s, p);
	421	if (r < 0)
	422	return r;
	423
	424	p += remaining;
	425	remaining = 0;
	426	}
	427
	428	if (p > s->buffer) {
	429	memmove(s->buffer, p, remaining);
	430	s->length = remaining;
	431	}
	432
	433	return 0;
	434	}
	435
60f067b4 JS	436	static int stdout_stream_process(sd_event_source es, int fd, uint32_t revents, void userdata) {
60f067b4 JS	437	StdoutStream *s = userdata;
663996b3 MS	438	ssize_t l;
	439	int r;
	440
	441	assert(s);
	442
60f067b4 JS	443	if ((revents\|EPOLLIN\|EPOLLHUP) != (EPOLLIN\|EPOLLHUP)) {
	444	log_error("Got invalid event from epoll for stdout stream: %"PRIx32, revents);
	445	goto terminate;
	446	}
	447
663996b3 MS	448	l = read(s->fd, s->buffer+s->length, sizeof(s->buffer)-1-s->length);
	449	if (l < 0) {
	450
	451	if (errno == EAGAIN)
	452	return 0;
	453
f47781d8	454	log_warning_errno(errno, "Failed to read from stream: %m");
60f067b4	455	goto terminate;
663996b3 MS	456	}
	457
	458	if (l == 0) {
60f067b4 JS	459	stdout_stream_scan(s, true);
60f067b4 JS	460	goto terminate;
663996b3 MS	461	}
	462
	463	s->length += l;
	464	r = stdout_stream_scan(s, false);
	465	if (r < 0)
60f067b4	466	goto terminate;
663996b3 MS	467
	468	return 1;
	469
60f067b4	470	terminate:
e735f4d4	471	stdout_stream_destroy(s);
60f067b4	472	return 0;
663996b3 MS	473	}
663996b3 MS	474
e735f4d4 MP	475	static int stdout_stream_install(Server s, int fd, StdoutStream *ret) {
	476	_cleanup_(stdout_stream_freep) StdoutStream *stream = NULL;
	477	int r;
	478
663996b3	479	assert(s);
e735f4d4	480	assert(fd >= 0);
663996b3	481
e735f4d4 MP	482	stream = new0(StdoutStream, 1);
	483	if (!stream)
	484	return log_oom();
663996b3	485
e735f4d4 MP	486	stream->fd = -1;
e735f4d4 MP	487	stream->priority = LOG_INFO;
663996b3	488
e735f4d4 MP	489	r = getpeercred(fd, &stream->ucred);
	490	if (r < 0)
	491	return log_error_errno(r, "Failed to determine peer credentials: %m");
60f067b4	492
663996b3	493	#ifdef HAVE_SELINUX
e735f4d4 MP	494	if (mac_selinux_use()) {
	495	if (getpeercon(fd, &stream->security_context) < 0 && errno != ENOPROTOOPT)
	496	log_error_errno(errno, "Failed to determine peer security context: %m");
	497	}
663996b3 MS	498	#endif
663996b3 MS	499
e735f4d4 MP	500	(void) shutdown(fd, SHUT_WR);
	501
	502	r = sd_event_add_io(s->event, &stream->event_source, fd, EPOLLIN, stdout_stream_process, stream);
	503	if (r < 0)
	504	return log_error_errno(r, "Failed to add stream to event loop: %m");
	505
	506	r = sd_event_source_set_priority(stream->event_source, SD_EVENT_PRIORITY_NORMAL+5);
	507	if (r < 0)
	508	return log_error_errno(r, "Failed to adjust stdout event source priority: %m");
	509
	510	stream->fd = fd;
	511
	512	stream->server = s;
	513	LIST_PREPEND(stdout_stream, s->stdout_streams, stream);
	514	s->n_stdout_streams ++;
	515
	516	if (ret)
	517	*ret = stream;
	518
	519	stream = NULL;
	520
	521	return 0;
663996b3 MS	522	}
663996b3 MS	523
60f067b4	524	static int stdout_stream_new(sd_event_source es, int listen_fd, uint32_t revents, void userdata) {
e735f4d4	525	_cleanup_close_ int fd = -1;
60f067b4	526	Server *s = userdata;
e735f4d4	527	int r;
663996b3 MS	528
	529	assert(s);
	530
60f067b4 JS	531	if (revents != EPOLLIN) {
	532	log_error("Got invalid event from epoll for stdout server fd: %"PRIx32, revents);
	533	return -EIO;
	534	}
	535
663996b3 MS	536	fd = accept4(s->stdout_fd, NULL, NULL, SOCK_NONBLOCK\|SOCK_CLOEXEC);
	537	if (fd < 0) {
	538	if (errno == EAGAIN)
	539	return 0;
	540
f47781d8	541	log_error_errno(errno, "Failed to accept stdout connection: %m");
663996b3 MS	542	return -errno;
	543	}
	544
	545	if (s->n_stdout_streams >= STDOUT_STREAMS_MAX) {
	546	log_warning("Too many stdout streams, refusing connection.");
663996b3 MS	547	return 0;
	548	}
	549
e735f4d4 MP	550	r = stdout_stream_install(s, fd, NULL);
	551	if (r < 0)
	552	return r;
	553
	554	fd = -1;
	555	return 0;
	556	}
	557
	558	static int stdout_stream_load(StdoutStream stream, const char fname) {
	559	_cleanup_free_ char
	560	*priority = NULL,
	561	*level_prefix = NULL,
	562	*forward_to_syslog = NULL,
	563	*forward_to_kmsg = NULL,
	564	*forward_to_console = NULL;
	565	int r;
	566
	567	assert(stream);
	568	assert(fname);
	569
	570	if (!stream->state_file) {
	571	stream->state_file = strappend("/run/systemd/journal/streams/", fname);
	572	if (!stream->state_file)
	573	return log_oom();
663996b3 MS	574	}
663996b3 MS	575
e735f4d4 MP	576	r = parse_env_file(stream->state_file, NEWLINE,
	577	"PRIORITY", &priority,
	578	"LEVEL_PREFIX", &level_prefix,
	579	"FORWARD_TO_SYSLOG", &forward_to_syslog,
	580	"FORWARD_TO_KMSG", &forward_to_kmsg,
	581	"FORWARD_TO_CONSOLE", &forward_to_console,
	582	"IDENTIFIER", &stream->identifier,
	583	"UNIT", &stream->unit_id,
	584	NULL);
	585	if (r < 0)
	586	return log_error_errno(r, "Failed to read: %s", stream->state_file);
663996b3	587
e735f4d4 MP	588	if (priority) {
	589	int p;
	590
	591	p = log_level_from_string(priority);
	592	if (p >= 0)
	593	stream->priority = p;
663996b3 MS	594	}
663996b3 MS	595
e735f4d4 MP	596	if (level_prefix) {
	597	r = parse_boolean(level_prefix);
	598	if (r >= 0)
	599	stream->level_prefix = r;
60f067b4	600	}
663996b3	601
e735f4d4 MP	602	if (forward_to_syslog) {
	603	r = parse_boolean(forward_to_syslog);
	604	if (r >= 0)
	605	stream->forward_to_syslog = r;
663996b3 MS	606	}
663996b3 MS	607
e735f4d4 MP	608	if (forward_to_kmsg) {
	609	r = parse_boolean(forward_to_kmsg);
	610	if (r >= 0)
	611	stream->forward_to_kmsg = r;
60f067b4 JS	612	}
60f067b4 JS	613
e735f4d4 MP	614	if (forward_to_console) {
	615	r = parse_boolean(forward_to_console);
	616	if (r >= 0)
	617	stream->forward_to_console = r;
663996b3 MS	618	}
663996b3 MS	619
e735f4d4 MP	620	return 0;
	621	}
	622
	623	static int stdout_stream_restore(Server s, const char fname, int fd) {
	624	StdoutStream *stream;
	625	int r;
	626
	627	assert(s);
	628	assert(fname);
	629	assert(fd >= 0);
	630
	631	if (s->n_stdout_streams >= STDOUT_STREAMS_MAX) {
	632	log_warning("Too many stdout streams, refusing restoring of stream.");
	633	return -ENOBUFS;
	634	}
	635
	636	r = stdout_stream_install(s, fd, &stream);
	637	if (r < 0)
	638	return r;
	639
	640	stream->state = STDOUT_STREAM_RUNNING;
	641	stream->fdstore = true;
	642
	643	/* Ignore all parsing errors */
	644	(void) stdout_stream_load(stream, fname);
663996b3 MS	645
663996b3 MS	646	return 0;
e735f4d4 MP	647	}
	648
	649	static int server_restore_streams(Server s, FDSet fds) {
	650	_cleanup_closedir_ DIR *d = NULL;
	651	struct dirent *de;
	652	int r;
	653
	654	d = opendir("/run/systemd/journal/streams");
	655	if (!d) {
	656	if (errno == ENOENT)
	657	return 0;
	658
	659	return log_warning_errno(errno, "Failed to enumerate /run/systemd/journal/streams: %m");
	660	}
	661
	662	FOREACH_DIRENT(de, d, goto fail) {
	663	unsigned long st_dev, st_ino;
	664	bool found = false;
	665	Iterator i;
	666	int fd;
	667
	668	if (sscanf(de->d_name, "%lu:%lu", &st_dev, &st_ino) != 2)
	669	continue;
	670
	671	FDSET_FOREACH(fd, fds, i) {
	672	struct stat st;
	673
	674	if (fstat(fd, &st) < 0)
	675	return log_error_errno(errno, "Failed to stat %s: %m", de->d_name);
	676
	677	if (S_ISSOCK(st.st_mode) && st.st_dev == st_dev && st.st_ino == st_ino) {
	678	found = true;
	679	break;
	680	}
	681	}
	682
	683	if (!found) {
	684	/* No file descriptor? Then let's delete the state file */
	685	log_debug("Cannot restore stream file %s", de->d_name);
	686	unlinkat(dirfd(d), de->d_name, 0);
	687	continue;
	688	}
	689
	690	fdset_remove(fds, fd);
	691
	692	r = stdout_stream_restore(s, de->d_name, fd);
	693	if (r < 0)
	694	safe_close(fd);
	695	}
663996b3	696
60f067b4	697	return 0;
e735f4d4 MP	698
	699	fail:
	700	return log_error_errno(errno, "Failed to read streams directory: %m");
663996b3 MS	701	}
663996b3 MS	702
e735f4d4	703	int server_open_stdout_socket(Server s, FDSet fds) {
663996b3	704	int r;
663996b3 MS	705
	706	assert(s);
	707
	708	if (s->stdout_fd < 0) {
	709	union sockaddr_union sa = {
	710	.un.sun_family = AF_UNIX,
	711	.un.sun_path = "/run/systemd/journal/stdout",
	712	};
	713
	714	s->stdout_fd = socket(AF_UNIX, SOCK_STREAM\|SOCK_CLOEXEC\|SOCK_NONBLOCK, 0);
f47781d8 MP	715	if (s->stdout_fd < 0)
f47781d8 MP	716	return log_error_errno(errno, "socket() failed: %m");
663996b3 MS	717
	718	unlink(sa.un.sun_path);
	719
	720	r = bind(s->stdout_fd, &sa.sa, offsetof(union sockaddr_union, un.sun_path) + strlen(sa.un.sun_path));
f47781d8 MP	721	if (r < 0)
f47781d8 MP	722	return log_error_errno(errno, "bind(%s) failed: %m", sa.un.sun_path);
663996b3 MS	723
	724	chmod(sa.un.sun_path, 0666);
	725
f47781d8 MP	726	if (listen(s->stdout_fd, SOMAXCONN) < 0)
f47781d8 MP	727	return log_error_errno(errno, "listen(%s) failed: %m", sa.un.sun_path);
663996b3 MS	728	} else
	729	fd_nonblock(s->stdout_fd, 1);
	730
60f067b4	731	r = sd_event_add_io(s->event, &s->stdout_event_source, s->stdout_fd, EPOLLIN, stdout_stream_new, s);
f47781d8 MP	732	if (r < 0)
f47781d8 MP	733	return log_error_errno(r, "Failed to add stdout server fd to event source: %m");
60f067b4 JS	734
60f067b4 JS	735	r = sd_event_source_set_priority(s->stdout_event_source, SD_EVENT_PRIORITY_NORMAL+10);
f47781d8 MP	736	if (r < 0)
f47781d8 MP	737	return log_error_errno(r, "Failed to adjust priority of stdout server event source: %m");
663996b3	738
e735f4d4 MP	739	/* Try to restore streams, but don't bother if this fails */
	740	(void) server_restore_streams(s, fds);
	741
663996b3 MS	742	return 0;
663996b3 MS	743	}