Send outgoing messages to the relay server on startup.

Also attempt to retry messages that could not be relayed periodically.

MANIFEST

@@ -348,9 +348,10 @@ logsrvd/logsrv_util.c
 logsrvd/logsrv_util.h
 logsrvd/logsrvd.c
 logsrvd/logsrvd.h
-logsrvd/logsrvd_local.c
 logsrvd/logsrvd_conf.c
 logsrvd/logsrvd_journal.c
+logsrvd/logsrvd_local.c
+logsrvd/logsrvd_queue.c
 logsrvd/logsrvd_relay.c
 logsrvd/regress/corpus/seed/logsrvd_conf/logsrvd.conf.1
 logsrvd/regress/corpus/seed/logsrvd_conf/logsrvd.conf.2

logsrvd/Makefile.in

@@ -121,7 +121,7 @@ PROGS = sudo_logsrvd sudo_sendlog
 
 LOGSRVD_OBJS = logsrv_util.o iolog_writer.o logsrvd.o logsrvd_conf.o \
 	       logsrvd_journal.o logsrvd_local.o logsrvd_relay.o \
-	       tls_client.o tls_init.o
+	       logsrvd_queue.o tls_client.o tls_init.o
 
 SENDLOG_OBJS = logsrv_util.o sendlog.o tls_client.o tls_init.o
 
@@ -430,6 +430,28 @@ logsrvd_local.i: $(srcdir)/logsrvd_local.c $(incdir)/compat/stdbool.h \
 	$(CC) -E -o $@ $(CPPFLAGS) $<
 logsrvd_local.plog: logsrvd_local.i
 	rm -f $@; pvs-studio --cfg $(PVS_CFG) --sourcetree-root $(top_srcdir) --skip-cl-exe yes --source-file $(srcdir)/logsrvd_local.c --i-file $< --output-file $@
+logsrvd_queue.o: $(srcdir)/logsrvd_queue.c $(incdir)/compat/stdbool.h \
+                 $(incdir)/log_server.pb-c.h $(incdir)/protobuf-c/protobuf-c.h \
+                 $(incdir)/sudo_compat.h $(incdir)/sudo_conf.h \
+                 $(incdir)/sudo_debug.h $(incdir)/sudo_event.h \
+                 $(incdir)/sudo_eventlog.h $(incdir)/sudo_gettext.h \
+                 $(incdir)/sudo_iolog.h $(incdir)/sudo_queue.h \
+                 $(incdir)/sudo_util.h $(srcdir)/logsrv_util.h \
+                 $(srcdir)/logsrvd.h $(srcdir)/tls_common.h \
+                 $(top_builddir)/config.h
+	$(CC) -c $(CPPFLAGS) $(CFLAGS) $(ASAN_CFLAGS) $(PIE_CFLAGS) $(SSP_CFLAGS) $(srcdir)/logsrvd_queue.c
+logsrvd_queue.i: $(srcdir)/logsrvd_queue.c $(incdir)/compat/stdbool.h \
+                 $(incdir)/log_server.pb-c.h $(incdir)/protobuf-c/protobuf-c.h \
+                 $(incdir)/sudo_compat.h $(incdir)/sudo_conf.h \
+                 $(incdir)/sudo_debug.h $(incdir)/sudo_event.h \
+                 $(incdir)/sudo_eventlog.h $(incdir)/sudo_gettext.h \
+                 $(incdir)/sudo_iolog.h $(incdir)/sudo_queue.h \
+                 $(incdir)/sudo_util.h $(srcdir)/logsrv_util.h \
+                 $(srcdir)/logsrvd.h $(srcdir)/tls_common.h \
+                 $(top_builddir)/config.h
+	$(CC) -E -o $@ $(CPPFLAGS) $<
+logsrvd_queue.plog: logsrvd_queue.i
+	rm -f $@; pvs-studio --cfg $(PVS_CFG) --sourcetree-root $(top_srcdir) --skip-cl-exe yes --source-file $(srcdir)/logsrvd_queue.c --i-file $< --output-file $@
 logsrvd_relay.o: $(srcdir)/logsrvd_relay.c $(incdir)/compat/stdbool.h \
                  $(incdir)/log_server.pb-c.h $(incdir)/protobuf-c/protobuf-c.h \
                  $(incdir)/sudo_compat.h $(incdir)/sudo_debug.h \

logsrvd/logsrvd.c

@@ -110,6 +110,11 @@ connection_closure_free(struct connection_closure *closure)
 	struct connection_buffer *buf;
 
 	TAILQ_REMOVE(&connections, closure, entries);
+
+	if (closure->state == CONNECTING && closure->journal != NULL) {
+	    /* Failed to relay journal file, retry later. */
+	    logsrvd_queue_insert(closure);
+	}
 	if (closure->relay_closure != NULL)
 	    relay_closure_free(closure->relay_closure);
 #if defined(HAVE_OPENSSL)
@@ -159,7 +164,7 @@ connection_closure_free(struct connection_closure *closure)
 /*
  * Allocate a new connection closure.
  */
-static struct connection_closure *
+struct connection_closure *
 connection_closure_alloc(int fd, bool tls, bool relay_only,
     struct sudo_event_base *base)
 {
@@ -269,6 +274,9 @@ connection_close(struct connection_closure *closure)
 	sudo_debug_printf(SUDO_DEBUG_INFO|SUDO_DEBUG_LINENO,
 	    "removing journal file %s", closure->journal_path);
 	unlink(closure->journal_path);
+
+	/* Process the next outgoing file (if any). */
+	logsrvd_queue_enable(0, closure->evbase);
     }
     connection_closure_free(closure);
 
@@ -1669,14 +1677,20 @@ server_dump_stats(void)
 	    if (closure->sock == -1) {
 		sudo_debug_printf(SUDO_DEBUG_INFO, "  %2d: journal %s", n,
 		    closure->journal_path ? closure->journal_path : "none");
+		sudo_debug_printf(SUDO_DEBUG_INFO, "  %2d: fd %d", n,
+		    closure->journal ? fileno(closure->journal) : -1);
 	    } else {
 		sudo_debug_printf(SUDO_DEBUG_INFO, "  %2d: addr %s%s", n,
 		    closure->ipaddr, closure->tls ? " (TLS)" : "");
+		sudo_debug_printf(SUDO_DEBUG_INFO, "  %2d: sock %d", n,
+		    closure->sock);
 	    }
 	    if (relay_closure != NULL) {
 		sudo_debug_printf(SUDO_DEBUG_INFO, "      relay: %s (%s)",
 		    relay_closure->relay_name.name,
 		    relay_closure->relay_name.ipaddr);
+		sudo_debug_printf(SUDO_DEBUG_INFO, "      relay sock: %d",
+		    relay_closure->sock);
 	    }
 	    sudo_debug_printf(SUDO_DEBUG_INFO, "      state: %d", closure->state);
 	    if (closure->errstr != NULL) {
@@ -1696,6 +1710,7 @@ server_dump_stats(void)
 	}
 	sudo_debug_printf(SUDO_DEBUG_INFO, "%d client connection(s)\n", n);
     }
+    logsrvd_queue_dump();
 
     debug_return;
 }
@@ -1946,6 +1961,7 @@ main(int argc, char *argv[])
     daemonize(nofork);
     signal(SIGPIPE, SIG_IGN);
 
+    logsrvd_queue_scan(evbase);
     sudo_ev_dispatch(evbase);
     if (!nofork && logsrvd_conf_pid_file() != NULL)
 	unlink(logsrvd_conf_pid_file());

logsrvd/logsrvd.h

@@ -41,6 +41,9 @@
 /* Shutdown timeout (in seconds) in case client connections time out. */
 #define SHUTDOWN_TIMEO	10
 
+/* Template for mkstemp(3) when creating temporary files. */
+#define RELAY_TEMPLATE	"relay.XXXXXXXX"
+
 /*
  * Connection status.
  * In the RUNNING state we expect I/O log buffers.
@@ -168,6 +171,15 @@ struct listener {
 };
 TAILQ_HEAD(listener_list, listener);
 
+/*
+ * Queue of finished journal files to be relayed.
+ */
+struct outgoing_journal {
+    TAILQ_ENTRY(outgoing_journal) entries;
+    char *journal_path;
+};
+TAILQ_HEAD(outgoing_journal_queue, outgoing_journal);
+
 /* iolog_writer.c */
 struct eventlog *evlog_new(TimeSpec *submit_time, InfoMessage **info_msgs, size_t infolen, struct connection_closure *closure);
 bool iolog_init(AcceptMessage *msg, struct connection_closure *closure);
@@ -184,6 +196,7 @@ bool schedule_commit_point(TimeSpec *commit_point, struct connection_closure *cl
 bool fmt_log_id_message(const char *id, struct connection_closure *closure);
 bool schedule_error_message(const char *errstr, struct connection_closure *closure);
 struct connection_buffer *get_free_buf(size_t, struct connection_closure *closure);
+struct connection_closure *connection_closure_alloc(int fd, bool tls, bool relay_only, struct sudo_event_base *base);
 
 /* logsrvd_conf.c */
 bool logsrvd_conf_read(const char *path);
@@ -225,6 +238,12 @@ bool store_iobuf_local(int iofd, IoBuffer *iobuf, uint8_t *buf, size_t len, stru
 bool store_winsize_local(ChangeWindowSize *msg, uint8_t *buf, size_t len, struct connection_closure *closure);
 bool store_suspend_local(CommandSuspend *msg, uint8_t *buf, size_t len, struct connection_closure *closure);
 
+/* logsrvd_queue.c */
+bool logsrvd_queue_enable(int timeout, struct sudo_event_base *evbase);
+bool logsrvd_queue_insert(struct connection_closure *closure);
+bool logsrvd_queue_scan(struct sudo_event_base *evbase);
+void logsrvd_queue_dump(void);
+
 /* logsrvd_relay.c */
 extern struct client_message_switch cms_relay;
 void relay_closure_free(struct relay_closure *relay_closure);

logsrvd/logsrvd_journal.c

@@ -88,12 +88,12 @@ journal_mkstemp(const char *parent_dir, char *pathbuf, int pathlen)
     int fd, len;
     debug_decl(journal_mkstemp, SUDO_DEBUG_UTIL);
 
-    len = snprintf(pathbuf, pathlen, "%s/%s/relay.XXXXXXXX",
+    len = snprintf(pathbuf, pathlen, "%s/%s/%s",
-	logsrvd_conf_relay_dir(), parent_dir);
+	logsrvd_conf_relay_dir(), parent_dir, RELAY_TEMPLATE);
     if (len >= pathlen) {
 	errno = ENAMETOOLONG;
 	sudo_debug_printf(SUDO_DEBUG_ERROR|SUDO_DEBUG_LINENO|SUDO_DEBUG_ERRNO,
-	    "%s/%s/relay.XXXXXXXX", parent_dir, logsrvd_conf_relay_dir());
+	    "%s/%s/%s", logsrvd_conf_relay_dir(), parent_dir, RELAY_TEMPLATE);
 	debug_return_int(-1);
     }
     if (!sudo_mkdir_parents(pathbuf, ROOT_UID, ROOT_GID,
@@ -126,9 +126,18 @@ journal_create(struct connection_closure *closure)
 	closure->errstr = _("unable to create journal file");
 	debug_return_bool(false);
     }
+    if (!sudo_lock_file(fd, SUDO_TLOCK)) {
+	sudo_debug_printf(SUDO_DEBUG_ERROR|SUDO_DEBUG_LINENO|SUDO_DEBUG_ERRNO,
+	    "unable to lock journal file %s", journal_path);
+	unlink(journal_path);
+	close(fd);
+	closure->errstr = _("unable to lock journal file");
+	debug_return_bool(false);
+    }
     if (!journal_fdopen(fd, journal_path, closure)) {
 	sudo_debug_printf(SUDO_DEBUG_ERROR|SUDO_DEBUG_LINENO|SUDO_DEBUG_ERRNO,
 	    "unable to fdopen journal file %s", journal_path);
+	unlink(journal_path);
 	close(fd);
 	closure->errstr = _("unable to allocate memory");
 	debug_return_bool(false);

logsrvd/logsrvd_queue.c

@@ -0,0 +1,288 @@
+/*
+ * SPDX-License-Identifier: ISC
+ *
+ * Copyright (c) 2021 Todd C. Miller <Todd.Miller@sudo.ws>
+ *
+ * Permission to use, copy, modify, and distribute this software for any
+ * purpose with or without fee is hereby granted, provided that the above
+ * copyright notice and this permission notice appear in all copies.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+ * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+ * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+ * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+ * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
+ * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+ * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+ */
+
+#include "config.h"
+
+#include <sys/stat.h>
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <netinet/tcp.h>
+#include <arpa/inet.h>
+
+#include <errno.h>
+#include <fcntl.h>
+#include <limits.h>
+#ifdef HAVE_STDBOOL_H
+# include <stdbool.h>
+#else
+# include "compat/stdbool.h"
+#endif /* HAVE_STDBOOL_H */
+#if defined(HAVE_STDINT_H)
+# include <stdint.h>
+#elif defined(HAVE_INTTYPES_H)
+# include <inttypes.h>
+#endif
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <dirent.h>
+#include <time.h>
+#include <unistd.h>
+
+#include "sudo_compat.h"
+#include "sudo_conf.h"
+#include "sudo_debug.h"
+#include "sudo_event.h"
+#include "sudo_eventlog.h"
+#include "sudo_gettext.h"
+#include "sudo_iolog.h"
+#include "sudo_queue.h"
+#include "sudo_util.h"
+
+#include "log_server.pb-c.h"
+#include "logsrvd.h"
+
+#if defined(HAVE_STRUCT_DIRENT_D_NAMLEN) && HAVE_STRUCT_DIRENT_D_NAMLEN
+# define NAMLEN(dirent) (dirent)->d_namlen
+#else
+# define NAMLEN(dirent) strlen((dirent)->d_name)
+#endif
+
+static struct outgoing_journal_queue outgoing_journal_queue =
+    TAILQ_HEAD_INITIALIZER(outgoing_journal_queue);
+
+static struct sudo_event *outgoing_queue_event;
+
+/*
+ * Callback that runs when the outgoing queue retry timer fires.
+ * Tries to relay the first entry in the outgoing queue.
+ */
+static void
+outgoing_queue_cb(int unused, int what, void *v)
+{
+    struct connection_closure *closure;
+    struct outgoing_journal *oj, *next;
+    struct sudo_event_base *evbase = v;
+    bool success = false;
+    debug_decl(outgoing_queue_cb, SUDO_DEBUG_UTIL);
+
+    /* Must have at least one relay server. */
+    if (TAILQ_EMPTY(logsrvd_conf_relay_address()))
+	debug_return;
+
+    /* Process first journal. */
+    TAILQ_FOREACH_SAFE(oj, &outgoing_journal_queue, entries, next) {
+	FILE *fp;
+	int fd;
+
+	fd = open(oj->journal_path, O_RDWR);
+	if (fd == -1) {
+	    if (errno == ENOENT) {
+		TAILQ_REMOVE(&outgoing_journal_queue, oj, entries);
+		free(oj->journal_path);
+		free(oj);
+	    }
+	    continue;
+	}
+	if (!sudo_lock_file(fd, SUDO_TLOCK)) {
+	    sudo_debug_printf(SUDO_DEBUG_ERROR|SUDO_DEBUG_LINENO|SUDO_DEBUG_ERRNO,
+		"unable to lock fd %d (%s)", fd, oj->journal_path);
+	    close(fd);
+	    continue;
+	}
+	fp = fdopen(fd, "r");
+	if (fp == NULL) {
+	    close(fd);
+	    sudo_debug_printf(SUDO_DEBUG_ERROR|SUDO_DEBUG_LINENO|SUDO_DEBUG_ERRNO,
+		"unable to fdopen %s", oj->journal_path);
+	    break;
+	}
+
+	/* Allocate a connection closure and fill in journal vars. */
+	closure = connection_closure_alloc(fd, false, true, evbase);
+	if (closure == NULL) {
+	    fclose(fp);
+	    break;
+	}
+	closure->journal = fp;
+	closure->journal_path = oj->journal_path;
+
+	/* Done with oj now, closure owns journal_path. */
+	TAILQ_REMOVE(&outgoing_journal_queue, oj, entries);
+	free(oj);
+
+	success = connect_relay(closure);
+	if (!success) {
+	    sudo_debug_printf(SUDO_DEBUG_ERROR|SUDO_DEBUG_LINENO,
+		"unable to connect to relay");
+	    connection_close(closure);
+	}
+	break;
+    }
+}
+
+/*
+ * Schedule the outgoing_queue_event, creating it as necessary.
+ * The event will fire after the specified timeout elapses.
+ */
+bool
+logsrvd_queue_enable(int timeout, struct sudo_event_base *evbase)
+{
+    debug_decl(logsrvd_queue_enable, SUDO_DEBUG_UTIL);
+
+    if (!TAILQ_EMPTY(&outgoing_journal_queue)) {
+	struct timespec tv = { timeout, 0 };
+
+	if (outgoing_queue_event == NULL) {
+	    outgoing_queue_event = sudo_ev_alloc(-1, SUDO_EV_TIMEOUT,
+		outgoing_queue_cb, evbase);
+	    if (outgoing_queue_event == NULL) {
+		sudo_debug_printf(SUDO_DEBUG_ERROR|SUDO_DEBUG_LINENO,
+		    "unable to allocate memory");
+		debug_return_bool(false);
+	    }
+	}
+	if (sudo_ev_add(evbase, outgoing_queue_event, &tv, false) == -1) {
+	    sudo_debug_printf(SUDO_DEBUG_ERROR|SUDO_DEBUG_LINENO,
+		"unable to add server write event");
+	    debug_return_bool(false);
+	}
+    }
+    debug_return_bool(true);
+}
+
+/*
+ * Allocate a queue item based on the connection and push it on
+ * the outgoing queue.
+ * Consumes journal_path from the closure.
+ */
+bool
+logsrvd_queue_insert(struct connection_closure *closure)
+{
+    struct outgoing_journal *oj;
+    debug_decl(logsrvd_queue_insert, SUDO_DEBUG_UTIL);
+
+    if (closure->journal_path == NULL) {
+	sudo_debug_printf(SUDO_DEBUG_ERROR|SUDO_DEBUG_LINENO,
+	    "missing journal_path for closure %p", closure);
+	debug_return_bool(false);
+    }
+
+    if ((oj = malloc(sizeof(*oj))) == NULL) {
+	sudo_debug_printf(SUDO_DEBUG_ERROR|SUDO_DEBUG_LINENO,
+	    "unable to allocate memory");
+	debug_return_bool(false);
+    }
+    oj->journal_path = closure->journal_path;
+    closure->journal_path = NULL;
+    TAILQ_INSERT_TAIL(&outgoing_journal_queue, oj, entries);
+
+    if (!logsrvd_queue_enable(30, closure->evbase))
+	debug_return_bool(false);
+
+    debug_return_bool(true);
+}
+
+/*
+ * Scan the outgoing queue at startup and populate the
+ * outgoing_journal_queue.
+ */
+bool
+logsrvd_queue_scan(struct sudo_event_base *evbase)
+{
+    char path[PATH_MAX];
+    struct dirent *dent;
+    size_t prefix_len;
+    int dirlen;
+    DIR *dirp;
+    debug_decl(logsrvd_queue_scan, SUDO_DEBUG_UTIL);
+
+    /* Must have at least one relay server. */
+    if (TAILQ_EMPTY(logsrvd_conf_relay_address()))
+	debug_return_bool(true);
+
+    dirlen = snprintf(path, sizeof(path), "%s/outgoing/%s",
+	logsrvd_conf_relay_dir(), RELAY_TEMPLATE);
+    if (dirlen >= ssizeof(path)) {
+	errno = ENAMETOOLONG;
+	sudo_debug_printf(SUDO_DEBUG_ERROR|SUDO_DEBUG_LINENO|SUDO_DEBUG_ERRNO,
+	    "%s/outgoing/%s", logsrvd_conf_relay_dir(), RELAY_TEMPLATE);
+	debug_return_bool(false);
+    }
+    dirlen -= sizeof(RELAY_TEMPLATE) - 1;
+    path[dirlen] = '\0';
+
+    dirp = opendir(path);
+    if (dirp == NULL) {
+	sudo_debug_printf(SUDO_DEBUG_ERROR|SUDO_DEBUG_LINENO|SUDO_DEBUG_ERRNO,
+	    "unable to opendir %s", path);
+	debug_return_bool(false);
+    }
+    prefix_len = strcspn(RELAY_TEMPLATE, "X");
+    while ((dent = readdir(dirp)) != NULL) {
+	struct outgoing_journal *oj;
+
+	/* Skip anything that is not a relay temp file. */
+	if (NAMLEN(dent) != sizeof(RELAY_TEMPLATE) - 1)
+	    continue;
+	if (strncmp(dent->d_name, RELAY_TEMPLATE, prefix_len) != 0)
+	    continue;
+
+	/* Add to queue. */
+	path[dirlen] = '\0';
+	if (strlcat(path, dent->d_name, sizeof(path)) >= sizeof(path))
+	    continue;
+	if ((oj = malloc(sizeof(*oj))) == NULL) {
+	    sudo_debug_printf(SUDO_DEBUG_ERROR|SUDO_DEBUG_LINENO,
+		"unable to allocate memory");
+	    debug_return_bool(false);
+	}
+	if ((oj->journal_path = strdup(path)) == NULL) {
+	    free(oj);
+	    sudo_debug_printf(SUDO_DEBUG_ERROR|SUDO_DEBUG_LINENO,
+		"unable to allocate memory");
+	    debug_return_bool(false);
+	}
+	TAILQ_INSERT_TAIL(&outgoing_journal_queue, oj, entries);
+    }
+
+    /* Process the queue immediately. */
+    if (!logsrvd_queue_enable(0, evbase))
+	debug_return_bool(false);
+
+    debug_return_bool(true);
+}
+
+/*
+ * Dump outgoing queue in response to SIGUSR1.
+ */
+void
+logsrvd_queue_dump(void)
+{
+    struct outgoing_journal *oj;
+    debug_decl(logsrvd_queue_dump, SUDO_DEBUG_UTIL);
+
+    if (TAILQ_EMPTY(&outgoing_journal_queue))
+	debug_return;
+
+    sudo_debug_printf(SUDO_DEBUG_INFO, "outgoing journal queue:");
+    TAILQ_FOREACH(oj, &outgoing_journal_queue, entries) {
+	sudo_debug_printf(SUDO_DEBUG_INFO, "  %s", oj->journal_path);
+    }
+}

logsrvd/logsrvd_relay.c

@@ -365,6 +365,8 @@ connect_relay_next(struct connection_closure *closure)
     relay_closure->relay_name.name = sudo_rcstr_addref(relay->sa_host);
 
     if (ret == 0) {
+	if (relay_closure->sock != -1)
+	    close(relay_closure->sock);
 	relay_closure->sock = sock;
 #if defined(HAVE_OPENSSL)
 	/* Relay connection succeeded, start TLS handshake. */
@@ -390,6 +392,8 @@ connect_relay_next(struct connection_closure *closure)
 		"unable to add server connect event");
 	    goto bad;
 	}
+	if (relay_closure->sock != -1)
+	    close(relay_closure->sock);
 	relay_closure->sock = sock;
 	closure->state = CONNECTING;
     }