lightningd: addgossip API to inject gossip messages.

Importantly, this is synchronous, so pay will be able to use it
reliably.

Signed-off-by: Rusty Russell <rusty@rustcorp.com.au>

contrib/pyln-client/pyln/client/lightning.py

@@ -484,6 +484,15 @@ class LightningRpc(UnixDomainSocketRpc):
         if patch_json:
             monkey_patch_json(patch=True)
 
+    def addgossip(self, message):
+        """
+        Inject this (hex-encoded) gossip message.
+        """
+        payload = {
+            "message": message,
+        }
+        return self.call("addgossip", payload)
+
     def autocleaninvoice(self, cycle_seconds=None, expired_by=None):
         """
         Sets up automatic cleaning of expired invoices. {cycle_seconds} sets

gossipd/gossipd.c

@@ -230,7 +230,8 @@ static bool get_node_announcement_by_id(const tal_t *ctx,
  * message, and puts the announcemnt on an internal 'pending'
  * queue.  We'll send a request to lightningd to look it up, and continue
  * processing in `handle_txout_reply`. */
-static const u8 *handle_channel_announcement_msg(struct peer *peer,
+static const u8 *handle_channel_announcement_msg(struct daemon *daemon,
+						 struct peer *peer,
 						 const u8 *msg)
 {
 	const struct short_channel_id *scid;
@@ -239,20 +240,20 @@ static const u8 *handle_channel_announcement_msg(struct peer *peer,
 	/* If it's OK, tells us the short_channel_id to lookup; it notes
 	 * if this is the unknown channel the peer was looking for (in
 	 * which case, it frees and NULLs that ptr) */
-	err = handle_channel_announcement(peer->daemon->rstate, msg,
-					  peer->daemon->current_blockheight,
+	err = handle_channel_announcement(daemon->rstate, msg,
+					  daemon->current_blockheight,
 					  &scid, peer);
 	if (err)
 		return err;
 	else if (scid) {
 		/* We give them some grace period, in case we don't know about
 		 * block yet. */
-		if (peer->daemon->current_blockheight == 0
+		if (daemon->current_blockheight == 0
 		    || !is_scid_depth_announceable(scid,
-						   peer->daemon->current_blockheight)) {
-			tal_arr_expand(&peer->daemon->deferred_txouts, *scid);
+						   daemon->current_blockheight)) {
+			tal_arr_expand(&daemon->deferred_txouts, *scid);
 		} else {
-			daemon_conn_send(peer->daemon->master,
+			daemon_conn_send(daemon->master,
 					 take(towire_gossipd_get_txout(NULL,
 								      scid)));
 		}
@@ -420,7 +421,7 @@ static bool handle_local_channel_announcement(struct daemon *daemon,
 		return false;
 	}
 
-	err = handle_channel_announcement_msg(peer, cannouncement);
+	err = handle_channel_announcement_msg(daemon, peer, cannouncement);
 	if (err) {
 		status_broken("peer %s invalid local_channel_announcement %s (%s)",
 			      type_to_string(tmpctx, struct node_id, &peer->id),
@@ -705,7 +706,7 @@ static struct io_plan *peer_msg_in(struct io_conn *conn,
 	/* These are messages relayed from peer */
 	switch ((enum peer_wire)fromwire_peektype(msg)) {
 	case WIRE_CHANNEL_ANNOUNCEMENT:
-		err = handle_channel_announcement_msg(peer, msg);
+		err = handle_channel_announcement_msg(peer->daemon, peer, msg);
 		goto handled_relay;
 	case WIRE_CHANNEL_UPDATE:
 		err = handle_channel_update_msg(peer, msg);
@@ -1809,6 +1810,53 @@ static struct io_plan *handle_payment_failure(struct io_conn *conn,
 	return daemon_conn_read_next(conn, daemon->master);
 }
 
+/*~ lightningd tells us when about a gossip message directly, when told to by
+ * the addgossip RPC call.  That's usually used when a plugin gets an update
+ * returned in an payment error. */
+static struct io_plan *inject_gossip(struct io_conn *conn,
+				     struct daemon *daemon,
+				     const u8 *msg)
+{
+	u8 *goss;
+	const u8 *errmsg;
+	const char *err;
+
+	if (!fromwire_gossipd_addgossip(msg, msg, &goss))
+		master_badmsg(WIRE_GOSSIPD_ADDGOSSIP, msg);
+
+	switch (fromwire_peektype(goss)) {
+	case WIRE_CHANNEL_ANNOUNCEMENT:
+		errmsg = handle_channel_announcement_msg(daemon, NULL, goss);
+		break;
+	case WIRE_NODE_ANNOUNCEMENT:
+		errmsg = handle_node_announcement(daemon->rstate, goss,
+						  NULL, NULL);
+		break;
+	case WIRE_CHANNEL_UPDATE:
+		errmsg = handle_channel_update(daemon->rstate, goss,
+					       NULL, NULL, true);
+		break;
+	default:
+		err = tal_fmt(tmpctx, "unknown gossip type %i",
+			      fromwire_peektype(goss));
+		goto err_extracted;
+	}
+
+	/* The APIs above are designed to send error messages back to peers:
+	 * we extract the raw string instead. */
+	if (errmsg) {
+		err = sanitize_error(tmpctx, errmsg, NULL);
+		tal_free(errmsg);
+	} else
+		/* Send empty string if no error. */
+		err = "";
+
+err_extracted:
+	daemon_conn_send(daemon->master,
+			 take(towire_gossipd_addgossip_reply(NULL, err)));
+	return daemon_conn_read_next(conn, daemon->master);
+}
+
 /*~ This is where lightningd tells us that a channel's funding transaction has
  * been spent. */
 static struct io_plan *handle_outpoint_spent(struct io_conn *conn,
@@ -1908,6 +1956,9 @@ static struct io_plan *recv_req(struct io_conn *conn,
 	case WIRE_GOSSIPD_NEW_BLOCKHEIGHT:
 		return new_blockheight(conn, daemon, msg);
 
+	case WIRE_GOSSIPD_ADDGOSSIP:
+		return inject_gossip(conn, daemon, msg);
+
 #if DEVELOPER
 	case WIRE_GOSSIPD_DEV_SET_MAX_SCIDS_ENCODE_SIZE:
 		return dev_set_max_scids_encode_size(conn, daemon, msg);
@@ -1942,6 +1993,7 @@ static struct io_plan *recv_req(struct io_conn *conn,
 	case WIRE_GOSSIPD_DEV_COMPACT_STORE_REPLY:
 	case WIRE_GOSSIPD_GOT_ONIONMSG_TO_US:
 	case WIRE_GOSSIPD_GOT_ONIONMSG_FORWARD:
+	case WIRE_GOSSIPD_ADDGOSSIP_REPLY:
 		break;
 	}
 

gossipd/gossipd_wire.csv

@@ -161,3 +161,13 @@ msgdata,gossipd_send_onionmsg,id,node_id,
 msgdata,gossipd_send_onionmsg,onion_len,u16,
 msgdata,gossipd_send_onionmsg,onion,u8,onion_len
 msgdata,gossipd_send_onionmsg,blinding,?pubkey,
+
+# Lightningd tells us to inject a gossip message (for addgossip RPC)
+msgtype,gossipd_addgossip,3044
+msgdata,gossipd_addgossip,len,u16,
+msgdata,gossipd_addgossip,msg,u8,len
+
+# Empty string means no problem.
+msgtype,gossipd_addgossip_reply,3144
+msgdata,gossipd_addgossip_reply,err,wirestring,
+

gossipd/gossipd_wiregen.c

@@ -49,6 +49,8 @@ const char *gossipd_wire_name(int e)
 	case WIRE_GOSSIPD_GOT_ONIONMSG_TO_US: return "WIRE_GOSSIPD_GOT_ONIONMSG_TO_US";
 	case WIRE_GOSSIPD_GOT_ONIONMSG_FORWARD: return "WIRE_GOSSIPD_GOT_ONIONMSG_FORWARD";
 	case WIRE_GOSSIPD_SEND_ONIONMSG: return "WIRE_GOSSIPD_SEND_ONIONMSG";
+	case WIRE_GOSSIPD_ADDGOSSIP: return "WIRE_GOSSIPD_ADDGOSSIP";
+	case WIRE_GOSSIPD_ADDGOSSIP_REPLY: return "WIRE_GOSSIPD_ADDGOSSIP_REPLY";
 	}
 
 	snprintf(invalidbuf, sizeof(invalidbuf), "INVALID %i", e);
@@ -87,6 +89,8 @@ bool gossipd_wire_is_defined(u16 type)
 	case WIRE_GOSSIPD_GOT_ONIONMSG_TO_US:;
 	case WIRE_GOSSIPD_GOT_ONIONMSG_FORWARD:;
 	case WIRE_GOSSIPD_SEND_ONIONMSG:;
+	case WIRE_GOSSIPD_ADDGOSSIP:;
+	case WIRE_GOSSIPD_ADDGOSSIP_REPLY:;
 	      return true;
 	}
 	return false;
@@ -1033,4 +1037,55 @@ bool fromwire_gossipd_send_onionmsg(const tal_t *ctx, const void *p, struct node
 	}
 	return cursor != NULL;
 }
-// SHA256STAMP:1da012a28ad84883f18920e51c39a0af77f85e309e981f9ea8d158d0698f6a59
+
+/* WIRE: GOSSIPD_ADDGOSSIP */
+/* Lightningd tells us to inject a gossip message (for addgossip RPC) */
+u8 *towire_gossipd_addgossip(const tal_t *ctx, const u8 *msg)
+{
+	u16 len = tal_count(msg);
+	u8 *p = tal_arr(ctx, u8, 0);
+
+	towire_u16(&p, WIRE_GOSSIPD_ADDGOSSIP);
+	towire_u16(&p, len);
+	towire_u8_array(&p, msg, len);
+
+	return memcheck(p, tal_count(p));
+}
+bool fromwire_gossipd_addgossip(const tal_t *ctx, const void *p, u8 **msg)
+{
+	u16 len;
+
+	const u8 *cursor = p;
+	size_t plen = tal_count(p);
+
+	if (fromwire_u16(&cursor, &plen) != WIRE_GOSSIPD_ADDGOSSIP)
+		return false;
+ 	len = fromwire_u16(&cursor, &plen);
+ 	// 2nd case msg
+	*msg = len ? tal_arr(ctx, u8, len) : NULL;
+	fromwire_u8_array(&cursor, &plen, *msg, len);
+	return cursor != NULL;
+}
+
+/* WIRE: GOSSIPD_ADDGOSSIP_REPLY */
+/* Empty string means no problem. */
+u8 *towire_gossipd_addgossip_reply(const tal_t *ctx, const wirestring *err)
+{
+	u8 *p = tal_arr(ctx, u8, 0);
+
+	towire_u16(&p, WIRE_GOSSIPD_ADDGOSSIP_REPLY);
+	towire_wirestring(&p, err);
+
+	return memcheck(p, tal_count(p));
+}
+bool fromwire_gossipd_addgossip_reply(const tal_t *ctx, const void *p, wirestring **err)
+{
+	const u8 *cursor = p;
+	size_t plen = tal_count(p);
+
+	if (fromwire_u16(&cursor, &plen) != WIRE_GOSSIPD_ADDGOSSIP_REPLY)
+		return false;
+ 	*err = fromwire_wirestring(ctx, &cursor, &plen);
+	return cursor != NULL;
+}
+// SHA256STAMP:e82edc5625085e21b02b27a2293d9d757556f3090a8a20b142dcb73411307a0c

gossipd/gossipd_wiregen.h

@@ -65,6 +65,10 @@ enum gossipd_wire {
         WIRE_GOSSIPD_GOT_ONIONMSG_FORWARD = 3143,
         /*  Lightningd tells us to send a onion message. */
         WIRE_GOSSIPD_SEND_ONIONMSG = 3040,
+        /*  Lightningd tells us to inject a gossip message (for addgossip RPC) */
+        WIRE_GOSSIPD_ADDGOSSIP = 3044,
+        /*  Empty string means no problem. */
+        WIRE_GOSSIPD_ADDGOSSIP_REPLY = 3144,
 };
 
 const char *gossipd_wire_name(int e);
@@ -216,6 +220,16 @@ bool fromwire_gossipd_got_onionmsg_forward(const tal_t *ctx, const void *p, stru
 u8 *towire_gossipd_send_onionmsg(const tal_t *ctx, const struct node_id *id, const u8 *onion, const struct pubkey *blinding);
 bool fromwire_gossipd_send_onionmsg(const tal_t *ctx, const void *p, struct node_id *id, u8 **onion, struct pubkey **blinding);
 
+/* WIRE: GOSSIPD_ADDGOSSIP */
+/*  Lightningd tells us to inject a gossip message (for addgossip RPC) */
+u8 *towire_gossipd_addgossip(const tal_t *ctx, const u8 *msg);
+bool fromwire_gossipd_addgossip(const tal_t *ctx, const void *p, u8 **msg);
+
+/* WIRE: GOSSIPD_ADDGOSSIP_REPLY */
+/*  Empty string means no problem. */
+u8 *towire_gossipd_addgossip_reply(const tal_t *ctx, const wirestring *err);
+bool fromwire_gossipd_addgossip_reply(const tal_t *ctx, const void *p, wirestring **err);
+
 
 #endif /* LIGHTNING_GOSSIPD_GOSSIPD_WIREGEN_H */
-// SHA256STAMP:1da012a28ad84883f18920e51c39a0af77f85e309e981f9ea8d158d0698f6a59
+// SHA256STAMP:e82edc5625085e21b02b27a2293d9d757556f3090a8a20b142dcb73411307a0c

gossipd/routing.c

@@ -1394,13 +1394,13 @@ static u8 *check_channel_update(const tal_t *ctx,
 		return towire_errorfmt(ctx, NULL,
 				       "Bad signature for %s hash %s"
 				       " on channel_update %s",
-				       type_to_string(ctx,
+				       type_to_string(tmpctx,
 						      secp256k1_ecdsa_signature,
 						      node_sig),
-				       type_to_string(ctx,
+				       type_to_string(tmpctx,
 						      struct sha256_double,
 						      &hash),
-				       tal_hex(ctx, update));
+				       tal_hex(tmpctx, update));
 	return NULL;
 }
 
@@ -1422,49 +1422,49 @@ static u8 *check_channel_announcement(const tal_t *ctx,
 		return towire_errorfmt(ctx, NULL,
 				       "Bad node_signature_1 %s hash %s"
 				       " on channel_announcement %s",
-				       type_to_string(ctx,
+				       type_to_string(tmpctx,
 						      secp256k1_ecdsa_signature,
 						      node1_sig),
-				       type_to_string(ctx,
+				       type_to_string(tmpctx,
 						      struct sha256_double,
 						      &hash),
-				       tal_hex(ctx, announcement));
+				       tal_hex(tmpctx, announcement));
 	}
 	if (!check_signed_hash_nodeid(&hash, node2_sig, node2_id)) {
 		return towire_errorfmt(ctx, NULL,
 				       "Bad node_signature_2 %s hash %s"
 				       " on channel_announcement %s",
-				       type_to_string(ctx,
+				       type_to_string(tmpctx,
 						      secp256k1_ecdsa_signature,
 						      node2_sig),
-				       type_to_string(ctx,
+				       type_to_string(tmpctx,
 						      struct sha256_double,
 						      &hash),
-				       tal_hex(ctx, announcement));
+				       tal_hex(tmpctx, announcement));
 	}
 	if (!check_signed_hash(&hash, bitcoin1_sig, bitcoin1_key)) {
 		return towire_errorfmt(ctx, NULL,
 				       "Bad bitcoin_signature_1 %s hash %s"
 				       " on channel_announcement %s",
-				       type_to_string(ctx,
+				       type_to_string(tmpctx,
 						      secp256k1_ecdsa_signature,
 						      bitcoin1_sig),
-				       type_to_string(ctx,
+				       type_to_string(tmpctx,
 						      struct sha256_double,
 						      &hash),
-				       tal_hex(ctx, announcement));
+				       tal_hex(tmpctx, announcement));
 	}
 	if (!check_signed_hash(&hash, bitcoin2_sig, bitcoin2_key)) {
 		return towire_errorfmt(ctx, NULL,
 				       "Bad bitcoin_signature_2 %s hash %s"
 				       " on channel_announcement %s",
-				       type_to_string(ctx,
+				       type_to_string(tmpctx,
 						      secp256k1_ecdsa_signature,
 						      bitcoin2_sig),
-				       type_to_string(ctx,
+				       type_to_string(tmpctx,
 						      struct sha256_double,
 						      &hash),
-				       tal_hex(ctx, announcement));
+				       tal_hex(tmpctx, announcement));
 	}
 	return NULL;
 }

lightningd/gossip_control.c

@@ -151,6 +151,7 @@ static unsigned gossip_msg(struct subd *gossip, const u8 *msg, const int *fds)
 	case WIRE_GOSSIPD_DEV_SET_TIME:
 	case WIRE_GOSSIPD_NEW_BLOCKHEIGHT:
 	case WIRE_GOSSIPD_SEND_ONIONMSG:
+	case WIRE_GOSSIPD_ADDGOSSIP:
 	/* This is a reply, so never gets through to here. */
 	case WIRE_GOSSIPD_GETNODES_REPLY:
 	case WIRE_GOSSIPD_GETROUTE_REPLY:
@@ -159,6 +160,7 @@ static unsigned gossip_msg(struct subd *gossip, const u8 *msg, const int *fds)
 	case WIRE_GOSSIPD_DEV_MEMLEAK_REPLY:
 	case WIRE_GOSSIPD_DEV_COMPACT_STORE_REPLY:
 	case WIRE_GOSSIPD_GET_STRIPPED_CUPDATE_REPLY:
+	case WIRE_GOSSIPD_ADDGOSSIP_REPLY:
 		break;
 
 	case WIRE_GOSSIPD_GOT_ONIONMSG_TO_US:
@@ -576,6 +578,55 @@ static const struct json_command listchannels_command = {
 };
 AUTODATA(json_command, &listchannels_command);
 
+/* Called upon receiving a addgossip_reply from `gossipd` */
+static void json_addgossip_reply(struct subd *gossip UNUSED, const u8 *reply,
+				 const int *fds UNUSED,
+				 struct command *cmd)
+{
+	char *err;
+
+	if (!fromwire_gossipd_addgossip_reply(reply, reply, &err)) {
+		/* Shouldn't happen: just end json stream. */
+		log_broken(cmd->ld->log,
+			   "Invalid addgossip_reply from gossipd: %s",
+			   tal_hex(tmpctx, reply));
+		was_pending(command_fail(cmd, LIGHTNINGD,
+					 "Invalid reply from gossipd"));
+		return;
+	}
+
+	if (strlen(err))
+		was_pending(command_fail(cmd, LIGHTNINGD, "%s", err));
+	else
+		was_pending(command_success(cmd, json_stream_success(cmd)));
+}
+
+static struct command_result *json_addgossip(struct command *cmd,
+					     const char *buffer,
+					     const jsmntok_t *obj UNNEEDED,
+					     const jsmntok_t *params)
+{
+	u8 *req, *gossip_msg;
+	if (!param(cmd, buffer, params,
+		   p_req("message", param_bin_from_hex, &gossip_msg),
+		   NULL))
+		return command_param_failed();
+
+	req = towire_gossipd_addgossip(cmd, gossip_msg);
+	subd_req(cmd->ld->gossip, cmd->ld->gossip,
+		 req, -1, 0, json_addgossip_reply, cmd);
+
+	return command_still_pending(cmd);
+}
+
+static const struct json_command addgossip_command = {
+	"addgossip",
+	"utility",
+	json_addgossip,
+	"Inject gossip {message} into gossipd"
+};
+AUTODATA(json_command, &addgossip_command);
+
 #if DEVELOPER
 static struct command_result *
 json_dev_set_max_scids_encode_size(struct command *cmd,

tests/test_gossip.py

@@ -1799,3 +1799,47 @@ def test_routetool(node_factory):
                     l1.info['id'],
                     l2.info['id']],
                    check=True, timeout=TIMEOUT)
+
+
+def test_addgossip(node_factory):
+    l1, l2 = node_factory.line_graph(2, fundchannel=True, wait_for_announce=True,
+                                     opts={'log-level': 'io'})
+
+    # We should get two node_announcements, one channel_announcement, and two
+    # channel_update.
+    l3 = node_factory.get_node()
+
+    # 0x0100 = channel_announcement
+    # 0x0102 = channel_update
+    # 0x0101 = node_announcement
+    ann = l1.daemon.is_in_log(r"\[OUT\] 0100.*")
+    if ann is None:
+        ann = l2.daemon.is_in_log(r"\[OUT\] 0100.*")
+
+    upd1 = l1.daemon.is_in_log(r"\[OUT\] 0102.*")
+    upd2 = l2.daemon.is_in_log(r"\[OUT\] 0102.*")
+
+    nann1 = l1.daemon.is_in_log(r"\[OUT\] 0101.*")
+    nann2 = l2.daemon.is_in_log(r"\[OUT\] 0101.*")
+
+    # Feed them to l3 (Each one starts with TIMESTAMP chanid-xxx: [OUT] ...)
+    l3.rpc.addgossip(ann.split()[3])
+
+    l3.rpc.addgossip(upd1.split()[3])
+    l3.rpc.addgossip(upd2.split()[3])
+    l3.rpc.addgossip(nann1.split()[3])
+    l3.rpc.addgossip(nann2.split()[3])
+
+    # In this case, it can actually have to wait, since it does scid lookup.
+    wait_for(lambda: len(l3.rpc.listchannels()['channels']) == 2)
+    wait_for(lambda: len(l3.rpc.listnodes()['nodes']) == 2)
+
+    # Now corrupt an update
+    badupdate = upd1.split()[3]
+    if badupdate.endswith('f'):
+        badupdate = badupdate[:-1] + 'e'
+    else:
+        badupdate = badupdate[:-1] + 'f'
+
+    with pytest.raises(RpcError, match='Bad signature'):
+        l3.rpc.addgossip(badupdate)