[thirdparty/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 <stddef.h>
#include <unistd.h>

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

#include "sd-daemon.h"
#include "sd-event.h"

#include "dirent-util.h"
#include "escape.h"
#include "fd-util.h"
#include "fileio.h"
#include "io-util.h"
#include "journald-console.h"
#include "journald-kmsg.h"
#include "journald-server.h"
#include "journald-stream.h"
#include "journald-syslog.h"
#include "journald-wall.h"
#include "mkdir.h"
#include "parse-util.h"
#include "selinux-util.h"
#include "socket-util.h"
#include "string-util.h"
#include "syslog-util.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;
        char *label;
        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);
        free(s->label);
        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 fail;

        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 fail;
                }

                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 fail;
                }

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

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

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

        /* 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;
        }

        return 0;

fail:
        (void) unlink(s->state_file);

        if (temp_path)
                (void) unlink(temp_path);

        return log_error_errno(r, "Failed to save stream data %s: %m", s->state_file);
}

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;
        size_t label_len;

        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);

        label_len = s->label ? strlen(s->label) : 0;
        server_dispatch_message(s->server, iovec, n, ELEMENTSOF(iovec), &s->ucred, NULL, s->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");

        if (mac_selinux_use()) {
                r = getpeersec(fd, &stream->label);
                if (r < 0 && r != -EOPNOTSUPP)
                        (void) log_warning_errno(r, "Failed to determine peer security context: %m");
        }

        (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;
}

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) {
        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);

                (void) 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");

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