Skip to content

Commit

Permalink
connectd/peer_exchange_initmsg: handle peer comms ourselves.
Browse files Browse the repository at this point in the history 
connectd is the only user of the cryptomsg async APIs; better to
open-code it here.  We need to expose a little from cryptomsg(),
but we remove the 'struct peer' entirely from connectd.

One trick is that we still need to defer telling lightningd when a
peer reconnects (until it tells us the old one is disconnected).  So
now we generate the message for lightningd and send it once we're woken.

Signed-off-by: Rusty Russell <[email protected]>
  • Loading branch information
@rustyrussell
rustyrussell committed Sep 28, 2018
1 parent cfd54d6 commit a1bdaa8
Show file tree
Hide file tree
Showing 7 changed files with 289 additions and 145 deletions.
2 changes: 1 addition & 1 deletion common/cryptomsg.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -315,7 +315,7 @@ u8 *cryptomsg_encrypt_msg(const tal_t *ctx,
tal_hexstr(trc, msg, mlen),
tal_hexstr(trc, npub, sizeof(npub)),
tal_hexstr(trc, &cs->sk, sizeof(cs->sk)),
tal_hexstr(trc, out + 18, clen));
tal_hexstr(trc, out + CRYPTOMSG_HDR_SIZE, clen));
#endif

maybe_rotate_key(&cs->sn, &cs->sk, &cs->s_ck);
Expand Down
16 changes: 16 additions & 0 deletions common/cryptomsg.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,22 @@ struct io_plan *peer_write_message(struct io_conn *conn,
const u8 *msg,
struct io_plan *(*next)(struct io_conn *,
struct peer *));
/* BOLT #8:
*
* ### Receiving and Decrypting Messages
*
* In order to decrypt the _next_ message in the network stream, the
* following steps are completed:
*
* 1. Read _exactly_ 18 bytes from the network buffer.
*/
#define CRYPTOMSG_HDR_SIZE 18

/* BOLT #8:
*
* 4. Read _exactly_ `l+16` bytes from the network buffer
*/
#define CRYPTOMSG_BODY_OVERHEAD 16

/* Low-level functions for sync comms: doesn't discard unknowns! */
u8 *cryptomsg_encrypt_msg(const tal_t *ctx,
Expand Down
1 change: 1 addition & 0 deletions connectd/Makefile
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@ LIGHTNINGD_CONNECT_CONTROL_OBJS := $(LIGHTNINGD_CONNECT_CONTROL_SRC:.c=.o)
LIGHTNINGD_CONNECT_HEADERS := connectd/gen_connect_wire.h \
connectd/gen_connect_gossip_wire.h \
connectd/connectd.h \
connectd/peer_exchange_initmsg.h \
connectd/handshake.h \
connectd/netaddress.h \
connectd/tor_autoservice.h \
Expand Down
208 changes: 64 additions & 144 deletions connectd/connectd.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
#include <connectd/gen_connect_wire.h>
#include <connectd/handshake.h>
#include <connectd/netaddress.h>
#include <connectd/peer_exchange_initmsg.h>
#include <connectd/tor.h>
#include <connectd/tor_autoservice.h>
#include <errno.h>
Expand Down Expand Up @@ -140,27 +141,6 @@ struct reaching {
u32 seconds_waited;
};

/* This is a transitory structure: we hand off to the master daemon as soon
* as we've completed INIT read/write. */
struct peer {
struct daemon *daemon;

/* The ID of the peer */
struct pubkey id;

/* Where it's connected to. */
struct wireaddr_internal addr;

/* Feature bitmaps. */
u8 *gfeatures, *lfeatures;

/* Cryptostate */
struct peer_crypto_state pcs;

/* Our connection (and owner) */
struct io_conn *conn;
};

/* Mutual recursion */
static void try_reach_one_addr(struct reaching *reach);

Expand Down Expand Up @@ -194,24 +174,6 @@ static bool broken_resolver(struct daemon *daemon)
return daemon->broken_resolver_response != NULL;
}

static struct peer *new_peer(struct io_conn *conn,
struct daemon *daemon,
const struct pubkey *their_id,
const struct wireaddr_internal *addr,
const struct crypto_state *cs)
{
struct peer *peer = tal(conn, struct peer);

peer->conn = conn;
peer->id = *their_id;
peer->addr = *addr;
peer->daemon = daemon;
init_peer_crypto_state(peer, &peer->pcs);
peer->pcs.cs = *cs;

return peer;
}

static void destroy_reaching(struct reaching *reach)
{
list_del_from(&reach->daemon->reaching, &reach->list);
Expand All @@ -228,10 +190,12 @@ static struct reaching *find_reaching(struct daemon *daemon,
return NULL;
}

static void reached_peer(struct peer *peer, struct io_conn *conn)
static void reached_peer(struct io_conn *conn,
struct daemon *daemon,
const struct pubkey *id)
{
/* OK, we've reached the peer successfully, tell everyone. */
struct reaching *r = find_reaching(peer->daemon, &peer->id);
struct reaching *r = find_reaching(daemon, id);

if (!r)
return;
Expand All @@ -240,166 +204,122 @@ static void reached_peer(struct peer *peer, struct io_conn *conn)
io_set_finish(conn, NULL, NULL);

/* Don't free conn with reach */
tal_steal(peer->daemon, conn);
tal_steal(daemon, conn);
tal_free(r);
}

static int get_gossipfd(struct peer *peer)
static int get_gossipfd(struct daemon *daemon,
const struct pubkey *id,
const u8 *lfeatures)
{
bool gossip_queries_feature, initial_routing_sync, success;
u8 *msg;

gossip_queries_feature
= feature_offered(peer->lfeatures, LOCAL_GOSSIP_QUERIES)
&& feature_offered(peer->daemon->localfeatures,
= feature_offered(lfeatures, LOCAL_GOSSIP_QUERIES)
&& feature_offered(daemon->localfeatures,
LOCAL_GOSSIP_QUERIES);
initial_routing_sync
= feature_offered(peer->lfeatures, LOCAL_INITIAL_ROUTING_SYNC);
= feature_offered(lfeatures, LOCAL_INITIAL_ROUTING_SYNC);

/* We do this communication sync. */
msg = towire_gossip_new_peer(NULL, &peer->id, gossip_queries_feature,
msg = towire_gossip_new_peer(NULL, id, gossip_queries_feature,
initial_routing_sync);
if (!wire_sync_write(GOSSIPCTL_FD, take(msg)))
status_failed(STATUS_FAIL_INTERNAL_ERROR,
"Failed writing to gossipctl: %s",
strerror(errno));

msg = wire_sync_read(peer, GOSSIPCTL_FD);
msg = wire_sync_read(tmpctx, GOSSIPCTL_FD);
if (!fromwire_gossip_new_peer_reply(msg, &success))
status_failed(STATUS_FAIL_INTERNAL_ERROR,
"Failed parsing msg gossipctl: %s",
tal_hex(tmpctx, msg));
if (!success) {
status_broken("Gossipd did not give us an fd: losing peer %s",
type_to_string(tmpctx, struct pubkey, &peer->id));
type_to_string(tmpctx, struct pubkey, id));
return -1;
}
return fdpass_recv(GOSSIPCTL_FD);
}

static struct io_plan *peer_close_after_error(struct io_conn *conn,
struct peer *peer)
{
status_trace("%s: we sent them a fatal error, closing",
type_to_string(tmpctx, struct pubkey, &peer->id));
return io_close(conn);
}
struct peer_reconnected {
struct daemon *daemon;
struct pubkey id;
const u8 *peer_connected_msg;
const u8 *lfeatures;
};

/* Mutual recursion */
static struct io_plan *peer_connected(struct io_conn *conn, struct peer *peer);
static struct io_plan *retry_peer_connected(struct io_conn *conn,
struct peer *peer)
struct peer_reconnected *pr)
{
struct io_plan *plan;

status_trace("peer %s: processing now old peer gone",
type_to_string(tmpctx, struct pubkey, &peer->id));
type_to_string(tmpctx, struct pubkey, &pr->id));

return peer_connected(conn, peer);
plan = peer_connected(conn, pr->daemon, &pr->id,
take(pr->peer_connected_msg),
take(pr->lfeatures));
tal_free(pr);
return plan;
}

static struct io_plan *peer_connected(struct io_conn *conn, struct peer *peer)
struct io_plan *peer_connected(struct io_conn *conn,
struct daemon *daemon,
const struct pubkey *id TAKES,
const u8 *peer_connected_msg TAKES,
const u8 *lfeatures TAKES)
{
struct daemon *daemon = peer->daemon;
u8 *msg;
int gossip_fd;
struct pubkey *key;
int gossip_fd;

/* FIXME: We could do this before exchanging init msgs. */
key = pubkey_set_get(&daemon->peers, &peer->id);
key = pubkey_set_get(&daemon->peers, id);
if (key) {
u8 *msg;
struct peer_reconnected *r;

status_trace("peer %s: reconnect",
type_to_string(tmpctx, struct pubkey, &peer->id));
type_to_string(tmpctx, struct pubkey, id));

/* Tell master to kill it: will send peer_disconnect */
msg = towire_connect_reconnected(NULL, &peer->id);
msg = towire_connect_reconnected(NULL, id);
daemon_conn_send(&daemon->master, take(msg));
return io_wait(conn, key, retry_peer_connected, peer);

/* Save arguments for next time. */
r = tal(daemon, struct peer_reconnected);
r->daemon = daemon;
r->id = *id;
r->peer_connected_msg
= tal_dup_arr(r, u8, peer_connected_msg,
tal_count(peer_connected_msg), 0);
r->lfeatures
= tal_dup_arr(r, u8, lfeatures, tal_count(lfeatures), 0);
return io_wait(conn, key, retry_peer_connected, r);
}

reached_peer(peer, conn);
reached_peer(conn, daemon, id);

gossip_fd = get_gossipfd(daemon, id, lfeatures);

/* We promised we'd take it. */
if (taken(lfeatures))
tal_free(lfeatures);

gossip_fd = get_gossipfd(peer);
if (gossip_fd < 0)
return io_close(conn);

msg = towire_connect_peer_connected(tmpctx, &peer->id, &peer->addr,
&peer->pcs.cs,
peer->gfeatures, peer->lfeatures);
daemon_conn_send(&daemon->master, msg);
daemon_conn_send(&daemon->master, peer_connected_msg);
daemon_conn_send_fd(&daemon->master, io_conn_fd(conn));
daemon_conn_send_fd(&daemon->master, gossip_fd);

pubkey_set_add(&daemon->peers,
tal_dup(daemon, struct pubkey, &peer->id));
pubkey_set_add(&daemon->peers, tal_dup(daemon, struct pubkey, id));

/* This frees the peer. */
return io_close_taken_fd(conn);
}

static struct io_plan *peer_init_received(struct io_conn *conn,
struct peer *peer,
u8 *msg)
{
if (!fromwire_init(peer, msg, &peer->gfeatures, &peer->lfeatures)) {
status_trace("peer %s bad fromwire_init '%s', closing",
type_to_string(tmpctx, struct pubkey, &peer->id),
tal_hex(tmpctx, msg));
return io_close(conn);
}

if (!features_supported(peer->gfeatures, peer->lfeatures)) {
const u8 *global_features = get_offered_global_features(msg);
const u8 *local_features = get_offered_local_features(msg);
msg = towire_errorfmt(NULL, NULL, "Unsupported features %s/%s:"
" we only offer globalfeatures %s"
" and localfeatures %s",
tal_hex(msg, peer->gfeatures),
tal_hex(msg, peer->lfeatures),
tal_hexstr(msg,
global_features,
tal_count(global_features)),
tal_hexstr(msg,
local_features,
tal_count(local_features)));
return peer_write_message(conn, &peer->pcs, take(msg),
peer_close_after_error);
}

return peer_connected(conn, peer);
}

static struct io_plan *read_init(struct io_conn *conn, struct peer *peer)
{
/* BOLT #1:
*
* The receiving node:
* - MUST wait to receive `init` before sending any other messages.
*/
return peer_read_message(conn, &peer->pcs, peer_init_received);
}

/* This creates a temporary peer which is not in the list and is owner
* by the connection; it's placed in the list and owned by daemon once
* we have the features. */
static struct io_plan *init_new_peer(struct io_conn *conn,
const struct pubkey *their_id,
const struct wireaddr_internal *addr,
const struct crypto_state *cs,
struct daemon *daemon)
{
struct peer *peer = new_peer(conn, daemon, their_id, addr, cs);
u8 *initmsg;

/* BOLT #1:
*
* The sending node:
* - MUST send `init` as the first Lightning message for any
* connection.
*/
initmsg = towire_init(NULL,
daemon->globalfeatures, daemon->localfeatures);
return peer_write_message(conn, &peer->pcs, take(initmsg), read_init);
}

struct listen_fd {
int fd;
/* If we bind() IPv6 then IPv4 to same port, we *may* fail to listen()
Expand Down Expand Up @@ -456,7 +376,7 @@ static struct io_plan *handshake_in_success(struct io_conn *conn,
{
status_trace("Connect IN from %s",
type_to_string(tmpctx, struct pubkey, id));
return init_new_peer(conn, id, addr, cs, daemon);
return peer_exchange_initmsg(conn, daemon, cs, id, addr);
}

static struct io_plan *connection_in(struct io_conn *conn, struct daemon *daemon)
Expand Down Expand Up @@ -830,7 +750,7 @@ static struct io_plan *handshake_out_success(struct io_conn *conn,
reach->connstate = "Exchanging init messages";
status_trace("Connect OUT to %s",
type_to_string(tmpctx, struct pubkey, id));
return init_new_peer(conn, id, addr, cs, reach->daemon);
return peer_exchange_initmsg(conn, reach->daemon, cs, id, addr);
}

struct io_plan *connection_out(struct io_conn *conn, struct reaching *reach)
Expand Down
12 changes: 12 additions & 0 deletions connectd/connectd.h
View file
Original file line number Diff line number Diff line change
@@ -1,10 +1,22 @@
#ifndef LIGHTNING_CONNECTD_CONNECTD_H
#define LIGHTNING_CONNECTD_CONNECTD_H
#include "config.h"
#include <bitcoin/pubkey.h>
#include <common/crypto_state.h>

struct io_conn;
struct peer;
struct reaching;
struct daemon;

/* Called by io_tor_connect once it has a connection out. */
struct io_plan *connection_out(struct io_conn *conn, struct reaching *reach);

/* Called by peer_exchange_initmsg if successful. */
struct io_plan *peer_connected(struct io_conn *conn,
struct daemon *daemon,
const struct pubkey *id TAKES,
const u8 *peer_connected_msg TAKES,
const u8 *lfeatures TAKES);

#endif /* LIGHTNING_CONNECTD_CONNECTD_H */
Loading

0 comments on commit a1bdaa8

Please sign in to comment.