Add --exec-{send,receive}

adsbus/Makefile

@@ -10,7 +10,7 @@ VALGRIND ?= valgrind
 VALGRIND_FLAGS ?= --error-exitcode=1 --track-fds=yes --show-leak-kinds=all --leak-check=full
 ADSBUS_TEST_FLAGS ?= --stdin --stdout=airspy_adsb --stdout=beast --stdout=json --stdout=proto --stdout=raw --stdout=stats
 
-OBJ_NETWORK = incoming.o outgoing.o receive.o send.o
+OBJ_TRANSPORT = exec.o incoming.o outgoing.o receive.o send.o
 OBJ_PROTOCOL = airspy_adsb.o beast.o json.o proto.o raw.o stats.o
 OBJ_UTIL = buf.o hex.o list.o opts.o packet.o peer.o rand.o resolve.o server.o socket.o uuid.o wakeup.o
 OBJ_PROTO = adsb.pb-c.o
@@ -26,8 +26,8 @@ clean:
 adsb.pb-c.c: ../proto/adsb.proto
 	protoc-c --c_out=./ --proto_path=$(dir $<) $<
 
-adsbus: adsbus.o $(OBJ_NETWORK) $(OBJ_PROTOCOL) $(OBJ_UTIL) $(OBJ_PROTO)
+adsbus: adsbus.o $(OBJ_TRANSPORT) $(OBJ_PROTOCOL) $(OBJ_UTIL) $(OBJ_PROTO)
-	$(COMP) $(LDFLAGS) -o adsbus adsbus.o $(OBJ_NETWORK) $(OBJ_PROTOCOL) $(OBJ_UTIL) $(OBJ_PROTO) $(LIBS)
+	$(COMP) $(LDFLAGS) -o adsbus adsbus.o $(OBJ_TRANSPORT) $(OBJ_PROTOCOL) $(OBJ_UTIL) $(OBJ_PROTO) $(LIBS)
 
 fuzz:
 	rm -rf findings

adsbus/adsbus.c

@@ -6,6 +6,7 @@
 #include <unistd.h>
 
 #include "beast.h"
+#include "exec.h"
 #include "hex.h"
 #include "incoming.h"
 #include "json.h"
@@ -35,6 +36,8 @@ static void print_usage(const char *name) {
 			"\t--file-read=PATH\n"
 			"\t--file-write=FORMAT=PATH\n"
 			"\t--file-append=FORMAT=PATH\n"
+			"\t--exec-receive=COMMAND\n"
+			"\t--exec-send=FORMAT=COMMAND\n"
 			"\t--stdin\n"
 			"\t--stdout=FORMAT\n"
 			, name);
@@ -51,6 +54,8 @@ static bool parse_opts(int argc, char *argv[]) {
 		{"file-read",       required_argument, 0, 'r'},
 		{"file-write",      required_argument, 0, 'w'},
 		{"file-append",     required_argument, 0, 'a'},
+		{"exec-receive",    required_argument, 0, 'e'},
+		{"exec-send",       required_argument, 0, 'f'},
 		{"stdin",           no_argument,       0, 'i'},
 		{"stdout",          required_argument, 0, 'o'},
 		{"help",            no_argument,       0, 'h'},
@@ -89,6 +94,14 @@ static bool parse_opts(int argc, char *argv[]) {
 				handler = opts_add_file_append;
 				break;
 
+			case 'e':
+				handler = opts_add_exec_receive;
+				break;
+
+			case 'f':
+				handler = opts_add_exec_send;
+				break;
+
 		  case 'i':
 				handler = opts_add_stdin;
 				break;
@@ -151,6 +164,7 @@ int main(int argc, char *argv[]) {
 	send_cleanup();
 	incoming_cleanup();
 	outgoing_cleanup();
+	exec_cleanup();
 
 	json_cleanup();
 	proto_cleanup();

adsbus/exec.c

@@ -0,0 +1,150 @@
+#include <assert.h>
+#include <signal.h>
+#include <stdlib.h>
+#include <stdio.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <sys/types.h>
+#include <sys/wait.h>
+#include <unistd.h>
+
+#include "buf.h"
+#include "list.h"
+#include "peer.h"
+#include "uuid.h"
+#include "wakeup.h"
+
+#include "exec.h"
+
+struct exec {
+	struct peer peer;
+	uint8_t id[UUID_LEN];
+	char *command;
+	exec_connection_handler handler;
+	exec_get_hello hello;
+	void *passthrough;
+	uint32_t *count;
+	pid_t child;
+	struct list_head exec_list;
+};
+
+static struct list_head exec_head = LIST_HEAD_INIT(exec_head);
+
+static void exec_spawn_wrapper(struct peer *);
+
+static void exec_del(struct exec *exec) {
+	(*exec->count)--;
+	if (exec->child > 0) {
+		fprintf(stderr, "E %s: Sending SIGTERM to child process %d\n", exec->id, exec->child);
+		// Racy with the process terminating, so don't assert on it
+		kill(exec->child, SIGTERM);
+		assert(waitpid(exec->child, NULL, 0) == exec->child);
+	}
+	free(exec->command);
+	free(exec);
+}
+
+static void exec_close_handler(struct peer *peer) {
+	struct exec *exec = (struct exec *) peer;
+	int status;
+	assert(waitpid(exec->child, &status, WNOHANG) == exec->child);
+	exec->child = -1;
+	if (WIFEXITED(status)) {
+		fprintf(stderr, "E %s: Client exited with status %d\n", exec->id, WEXITSTATUS(status));
+	} else {
+		assert(WIFSIGNALED(status));
+		fprintf(stderr, "E %s: Client exited with signal %d\n", exec->id, WTERMSIG(status));
+	}
+	uint32_t delay = wakeup_get_retry_delay_ms(1);
+	fprintf(stderr, "E %s: Will retry in %ds\n", exec->id, delay / 1000);
+	exec->peer.event_handler = exec_spawn_wrapper;
+	wakeup_add((struct peer *) exec, delay);
+}
+
+static bool exec_hello(int fd, struct exec *exec) {
+	if (!exec->hello) {
+		return true;
+	}
+	struct buf buf = BUF_INIT, *buf_ptr = &buf;
+	exec->hello(&buf_ptr, exec->passthrough);
+	if (!buf_ptr->length) {
+		return true;
+	}
+	return (write(fd, buf_at(buf_ptr, 0), buf_ptr->length) == (ssize_t) buf_ptr->length);
+}
+
+static void exec_parent(struct exec *exec, pid_t child, int fd) {
+	exec->child = child;
+	fprintf(stderr, "E %s: Child started as process %d\n", exec->id, exec->child);
+
+	if (!exec_hello(fd, exec)) {
+		assert(!close(fd));
+		exec_close_handler((struct peer *) exec);
+		return;
+	}
+
+	exec->peer.event_handler = exec_close_handler;
+	exec->handler(fd, exec->passthrough, (struct peer *) exec);
+}
+
+static void __attribute__ ((noreturn)) exec_child(const struct exec *exec, int fd) {
+	assert(setsid() != -1);
+	// We leave stderr open from child to parent
+	// Other than that, fds should have CLOEXEC set
+	if (fd != 0) {
+		assert(dup2(fd, 0) == 0);
+	}
+	if (fd != 1) {
+		assert(dup2(fd, 1) == 1);
+	}
+	if (fd != 0 && fd != 1) {
+		assert(!close(fd));
+	}
+	assert(!execl("/bin/sh", "sh", "-c", exec->command, NULL));
+	abort();
+}
+
+static void exec_spawn(struct exec *exec) {
+	fprintf(stderr, "E %s: Executing: %s\n", exec->id, exec->command);
+	int fds[2];
+	assert(!socketpair(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK, 0, fds));
+	
+	int res = fork();
+	assert(res >= 0);
+	if (res) {
+		assert(!close(fds[1]));
+		exec_parent(exec, res, fds[0]);
+	} else {
+		assert(!close(fds[0]));
+		exec_child(exec, fds[1]);
+	}
+}
+
+static void exec_spawn_wrapper(struct peer *peer) {
+	struct exec *exec = (struct exec *) peer;
+	exec_spawn(exec);
+}
+
+void exec_cleanup() {
+	struct exec *iter, *next;
+	list_for_each_entry_safe(iter, next, &exec_head, exec_list) {
+		exec_del(iter);
+	}
+}
+
+void exec_new(char *command, exec_connection_handler handler, exec_get_hello hello, void *passthrough, uint32_t *count) {
+	(*count)++;
+
+	struct exec *exec = malloc(sizeof(*exec));
+	exec->peer.fd = -1;
+	uuid_gen(exec->id);
+	exec->command = strdup(command);
+	exec->handler = handler;
+	exec->hello = hello;
+	exec->passthrough = passthrough;
+	exec->count = count;
+
+	list_add(&exec->exec_list, &exec_head);
+
+	exec_spawn(exec);
+}

adsbus/exec.h

@@ -0,0 +1,11 @@
+#pragma once
+
+#include <stdint.h>
+
+struct buf;
+struct peer;
+
+void exec_cleanup(void);
+typedef void (*exec_connection_handler)(int fd, void *, struct peer *);
+typedef void (*exec_get_hello)(struct buf **, void *);
+void exec_new(char *, exec_connection_handler, exec_get_hello, void *, uint32_t *);

adsbus/opts.c

@@ -7,6 +7,7 @@
 #include <unistd.h>
 
 #include "buf.h"
+#include "exec.h"
 #include "incoming.h"
 #include "outgoing.h"
 #include "peer.h"
@@ -127,6 +128,21 @@ bool opts_add_file_append(char *arg) {
 	return send_new_hello(fd, serializer, NULL);
 }
 
+bool opts_add_exec_receive(char *arg) {
+	exec_new(arg, receive_new, NULL, NULL, &peer_count_in);
+	return true;
+}
+
+bool opts_add_exec_send(char *arg) {
+	struct serializer *serializer = opts_get_serializer(&arg);
+	if (!serializer) {
+		return NULL;
+	}
+
+	exec_new(arg, send_new_wrapper, send_hello, serializer, &peer_count_out);
+	return true;
+}
+
 bool opts_add_stdin(char __attribute__((unused)) *arg) {
 	int fd = dup(0);
 	assert(!fcntl(fd, F_SETFD, FD_CLOEXEC));

adsbus/opts.h

@@ -9,5 +9,7 @@ bool opts_add_listen_send(char *);
 bool opts_add_file_read(char *);
 bool opts_add_file_write(char *);
 bool opts_add_file_append(char *);
+bool opts_add_exec_receive(char *);
+bool opts_add_exec_send(char *);
 bool opts_add_stdout(char *);
 bool opts_add_stdin(char *);

adsbus/outgoing.c

@@ -41,7 +41,7 @@ static void outgoing_resolve(struct outgoing *);
 static void outgoing_resolve_wrapper(struct peer *);
 
 static void outgoing_retry(struct outgoing *outgoing) {
-	uint32_t delay = wakeup_get_retry_delay_ms(outgoing->attempt++);
+	uint32_t delay = wakeup_get_retry_delay_ms(++outgoing->attempt);
 	fprintf(stderr, "O %s: Will retry in %ds\n", outgoing->id, delay / 1000);
 	outgoing->peer.event_handler = outgoing_resolve_wrapper;
 	wakeup_add((struct peer *) outgoing, delay);

adsbus/rand.c

@@ -16,7 +16,7 @@ static struct buf rand_buf = BUF_INIT;
 static int rand_fd;
 
 void rand_init() {
-	rand_fd = open("/dev/urandom", O_RDONLY);
+	rand_fd = open("/dev/urandom", O_RDONLY | O_CLOEXEC);
 	assert(rand_fd >= 0);
 	assert(read(rand_fd, buf_at(&rand_buf, 0), BUF_LEN_MAX) == BUF_LEN_MAX);
 	rand_buf.length = BUF_LEN_MAX;