connectd: allow forking a proxy which handles websockets.

If set, we spawn it (lightning_websocketd) on any invalid incoming
act1 handshake.  That means websocketd is a per-peer daemon, but it means
every other daemon uses the connection normally (it's just actually talking
to websocketd instead of the client directly).

Signed-off-by: Rusty Russell <[email protected]>

connectd/connectd.c

@@ -163,6 +163,9 @@ struct daemon {
 
 	/* Our features, as lightningd told us */
 	struct feature_set *our_features;
+
+	/* Subdaemon to proxy websocket requests, if any. */
+	char *websocket_helper;
 };
 
 /* Peers we're trying to reach: we iterate through addrs until we succeed
@@ -611,6 +614,7 @@ static struct io_plan *connection_in(struct io_conn *conn, struct daemon *daemon
 	 * code from thinking `conn` (which we don't keep a pointer to) is
 	 * leaked */
 	return responder_handshake(notleak(conn), &daemon->mykey, &addr,
+				   daemon->websocket_helper,
 				   handshake_in_success, daemon);
 }
 
@@ -1316,7 +1320,8 @@ static struct io_plan *connect_init(struct io_conn *conn,
 		&daemon->dev_allow_localhost, &daemon->use_dns,
 		&tor_password,
 		&daemon->use_v3_autotor,
-		    &daemon->timeout_secs)) {
+		&daemon->timeout_secs,
+		&daemon->websocket_helper)) {
 		/* This is a helper which prints the type expected and the actual
 		 * message, then exits (it should never be called!). */
 		master_badmsg(WIRE_CONNECTD_INIT, msg);

connectd/connectd_wire.csv

@@ -18,6 +18,7 @@ msgdata,connectd_init,use_dns,bool,
 msgdata,connectd_init,tor_password,wirestring,
 msgdata,connectd_init,use_v3_autotor,bool,
 msgdata,connectd_init,timeout_secs,u32,
+msgdata,connectd_init,websocket_helper,?wirestring,
 
 # Connectd->master, here are the addresses I bound, can announce.
 msgtype,connectd_init_reply,2100

connectd/handshake.c

@@ -7,19 +7,24 @@
 #include <ccan/endian/endian.h>
 #include <ccan/io/io.h>
 #include <ccan/mem/mem.h>
+#include <ccan/noerr/noerr.h>
 #include <common/crypto_state.h>
 #include <common/ecdh.h>
+#include <common/memleak.h>
 #include <common/status.h>
 #include <common/type_to_string.h>
 #include <common/utils.h>
 #include <common/wireaddr.h>
 #include <connectd/handshake.h>
 #include <errno.h>
+#include <fcntl.h>
 #include <secp256k1.h>
 #include <secp256k1_ecdh.h>
 #include <sodium/crypto_aead_chacha20poly1305.h>
 #include <sodium/randombytes.h>
 #include <stdio.h>
+#include <sys/types.h>
+#include <sys/wait.h>
 #include <unistd.h>
 #include <wire/wire.h>
 
@@ -179,6 +184,9 @@ struct handshake {
 	/* Are we initiator or responder. */
 	enum bolt8_side side;
 
+	/* Subdaemon to proxy via if we incorrect handshake */
+	const char *websocket_helper;
+
 	/* Function to call once handshake complete. */
 	struct io_plan *(*cb)(struct io_conn *conn,
 			      const struct pubkey *their_id,
@@ -863,6 +871,107 @@ static struct io_plan *act_two_responder(struct io_conn *conn,
 	return io_write(conn, &h->act2, ACT_TWO_SIZE, act_three_responder, h);
 }
 
+/* For helper, we don't populate ->websocket_helper so it doesn't loop! */
+static struct io_plan *helper_init(struct io_conn *conn, struct handshake *old_h)
+{
+	return responder_handshake(conn, &old_h->my_id, &old_h->addr, NULL,
+				   old_h->cb, old_h->cbarg);
+}
+
+static struct io_plan *try_websocket_server(struct io_conn *conn,
+					    struct handshake *h)
+{
+	int childmsg[2], execfail[2];
+	pid_t childpid;
+	int err;
+
+	if (!h->websocket_helper) {
+		status_debug("No websocket helper, failing for '%s'",
+			     tal_hexstr(tmpctx, &h->act1, sizeof(h->act1)));
+		return handshake_failed(conn, h);
+	}
+
+	status_debug("Websocket helper: trying for '%s'",
+		     tal_hexstr(tmpctx, &h->act1, sizeof(h->act1)));
+
+	if (socketpair(AF_LOCAL, SOCK_STREAM, 0, childmsg) != 0)
+		goto fail;
+
+	if (pipe(execfail) != 0)
+		goto close_msgfd_fail;
+
+	if (fcntl(execfail[1], F_SETFD, fcntl(execfail[1], F_GETFD)
+		  | FD_CLOEXEC) < 0)
+		goto close_execfail_fail;
+
+	childpid = fork();
+	if (childpid < 0)
+		goto close_execfail_fail;
+
+	if (childpid == 0) {
+		size_t max;
+		close(childmsg[0]);
+		close(execfail[0]);
+
+		/* Attach remote socket to stdin. */
+		if (dup2(io_conn_fd(conn), STDIN_FILENO) == -1)
+			goto child_errno_fail;
+
+		/* Attach our socket to stdout. */
+		if (dup2(childmsg[1], STDOUT_FILENO) == -1)
+			goto child_errno_fail;
+
+		/* Make (fairly!) sure all other fds are closed. */
+		max = sysconf(_SC_OPEN_MAX);
+		for (size_t i = STDERR_FILENO + 1; i < max; i++)
+			close(i);
+
+		/* Tell websocket helper what we read so far. */
+		execlp(h->websocket_helper, h->websocket_helper,
+		       tal_hexstr(tmpctx, &h->act1, sizeof(h->act1)),
+		       NULL);
+
+	child_errno_fail:
+		err = errno;
+		/* Gcc's warn-unused-result fail. */
+		if (write(execfail[1], &err, sizeof(err))) {
+			;
+		}
+		exit(127);
+	}
+
+	close(childmsg[1]);
+	close(execfail[1]);
+
+	/* Child will close this without writing on successful exec. */
+	if (read(execfail[0], &err, sizeof(err)) == sizeof(err)) {
+		close(execfail[0]);
+		waitpid(childpid, NULL, 0);
+		status_broken("Exec of helper %s failed: %s",
+			      h->websocket_helper, strerror(err));
+		errno = err;
+		return io_close(conn);
+	}
+
+	close(execfail[0]);
+
+	/* Now, we try handshake again, but this time via the helper proxy */
+	notleak(io_new_conn(tal_parent(conn), childmsg[0], helper_init, h));
+
+	/* Abandon original (doesn't close since child has dup'd fd) */
+	return io_close(conn);
+
+close_execfail_fail:
+	close_noerr(execfail[0]);
+	close_noerr(execfail[1]);
+close_msgfd_fail:
+	close_noerr(childmsg[0]);
+	close_noerr(childmsg[1]);
+fail:
+	status_broken("Preparation of helper failed: %s",
+		      strerror(errno));
+	return io_close(conn);
+}
 
 static struct io_plan *act_one_responder2(struct io_conn *conn,
 					 struct handshake *h)
@@ -873,7 +982,7 @@ static struct io_plan *act_one_responder2(struct io_conn *conn,
 	 *     MUST abort the connection attempt.
 	 */
 	if (h->act1.v != 0)
-		return handshake_failed(conn, h);
+		return try_websocket_server(conn, h);
 
 	/* BOLT #8:
 	 *
@@ -963,6 +1072,7 @@ static struct io_plan *act_one_responder(struct io_conn *conn,
 struct io_plan *responder_handshake_(struct io_conn *conn,
 				     const struct pubkey *my_id,
 				     const struct wireaddr_internal *addr,
+				     const char *websocket_helper,
 				     struct io_plan *(*cb)(struct io_conn *,
 							   const struct pubkey *,
 							   const struct wireaddr_internal *,
@@ -973,6 +1083,7 @@ struct io_plan *responder_handshake_(struct io_conn *conn,
 	struct handshake *h = new_handshake(conn, my_id);
 
 	h->side = RESPONDER;
+	h->websocket_helper = websocket_helper;
 	h->my_id = *my_id;
 	h->addr = *addr;
 	h->cbarg = cbarg;

connectd/handshake.h

@@ -32,8 +32,8 @@ struct io_plan *initiator_handshake_(struct io_conn *conn,
 				     void *cbarg);
 
 
-#define responder_handshake(conn, my_id, addr, cb, cbarg)		\
-	responder_handshake_((conn), (my_id), (addr),			\
+#define responder_handshake(conn, my_id, addr, websocket_helper, cb, cbarg) \
+	responder_handshake_((conn), (my_id), (addr), (websocket_helper), \
 			     typesafe_cb_preargs(struct io_plan *, void *, \
 						 (cb), (cbarg),		\
 						 struct io_conn *,	\
@@ -45,6 +45,7 @@ struct io_plan *initiator_handshake_(struct io_conn *conn,
 struct io_plan *responder_handshake_(struct io_conn *conn,
 				     const struct pubkey *my_id,
 				     const struct wireaddr_internal *addr,
+				     const char *websocket_helper,
 				     struct io_plan *(*cb)(struct io_conn *,
 							   const struct pubkey *,
 							   const struct wireaddr_internal *,

connectd/test/run-initiator-success.c

@@ -83,6 +83,9 @@ u8 fromwire_u8(const u8 **cursor UNNEEDED, size_t *max UNNEEDED)
 /* Generated stub for fromwire_u8_array */
 void fromwire_u8_array(const u8 **cursor UNNEEDED, size_t *max UNNEEDED, u8 *arr UNNEEDED, size_t num UNNEEDED)
 { fprintf(stderr, "fromwire_u8_array called!\n"); abort(); }
+/* Generated stub for notleak_ */
+void *notleak_(const void *ptr UNNEEDED, bool plus_children UNNEEDED)
+{ fprintf(stderr, "notleak_ called!\n"); abort(); }
 /* Generated stub for towire */
 void towire(u8 **pptr UNNEEDED, const void *data UNNEEDED, size_t len UNNEEDED)
 { fprintf(stderr, "towire called!\n"); abort(); }

connectd/test/run-responder-success.c

@@ -83,6 +83,9 @@ u8 fromwire_u8(const u8 **cursor UNNEEDED, size_t *max UNNEEDED)
 /* Generated stub for fromwire_u8_array */
 void fromwire_u8_array(const u8 **cursor UNNEEDED, size_t *max UNNEEDED, u8 *arr UNNEEDED, size_t num UNNEEDED)
 { fprintf(stderr, "fromwire_u8_array called!\n"); abort(); }
+/* Generated stub for notleak_ */
+void *notleak_(const void *ptr UNNEEDED, bool plus_children UNNEEDED)
+{ fprintf(stderr, "notleak_ called!\n"); abort(); }
 /* Generated stub for towire */
 void towire(u8 **pptr UNNEEDED, const void *data UNNEEDED, size_t len UNNEEDED)
 { fprintf(stderr, "towire called!\n"); abort(); }
@@ -320,7 +323,7 @@ int main(int argc, char *argv[])
 
 	dummy.itype = ADDR_INTERNAL_WIREADDR;
 	dummy.u.wireaddr.addrlen = 0;
-	responder_handshake((void *)tmpctx, &ls_pub, &dummy, success, NULL);
+	responder_handshake((void *)tmpctx, &ls_pub, &dummy, NULL, success, NULL);
 	/* Should not exit! */
 	abort();
 }

lightningd/connect_control.c

@@ -363,6 +363,7 @@ int connectd_init(struct lightningd *ld)
 	int hsmfd;
 	struct wireaddr_internal *wireaddrs = ld->proposed_wireaddr;
 	enum addr_listen_announce *listen_announce = ld->proposed_listen_announce;
+	const char *websocket_helper_path = NULL;
 
 	if (socketpair(AF_LOCAL, SOCK_STREAM, 0, fds) != 0)
 		fatal("Could not socketpair for connectd<->gossipd");
@@ -394,7 +395,8 @@ int connectd_init(struct lightningd *ld)
 	    IFDEV(ld->dev_allow_localhost, false), ld->config.use_dns,
 	    ld->tor_service_password ? ld->tor_service_password : "",
 	    ld->config.use_v3_autotor,
-	    ld->config.connection_timeout_secs);
+	    ld->config.connection_timeout_secs,
+	    websocket_helper_path);
 
 	subd_req(ld->connectd, ld->connectd, take(msg), -1, 0,
 		 connect_init_done, NULL);