routing: add new methods to check the freshness of an edge/node

In this commit, we add a set of new methods to check the freshness of
an edge/node. This will allow callers to skip expensive validation in
the case that the router already knows of an item, or knows of a
fresher version of that time.

A set of tests have been added to ensure basic correctness of these new
methods.

routing/router.go

@@ -33,8 +33,8 @@ const (
 	DefaultFinalCLTVDelta = 9
 )
 
-// ChannelGraphSource represents the source of information about the topology of
+// ChannelGraphSource represents the source of information about the topology
-// the lightning network. It's responsible for the addition of nodes, edges,
+// of the lightning network. It's responsible for the addition of nodes, edges,
 // applying edge updates, and returning the current block height with which the
 // topology is synchronized.
 type ChannelGraphSource interface {
@@ -56,6 +56,22 @@ type ChannelGraphSource interface {
 	// edge considered as not fully constructed.
 	UpdateEdge(policy *channeldb.ChannelEdgePolicy) error
 
+	// IsStaleNode returns true if the graph source has a node announcement
+	// for the target node with a more recent timestamp. This method will
+	// also return true if we don't have an active channel announcement for
+	// the target node.
+	IsStaleNode(node Vertex, timestamp time.Time) bool
+
+	// IsKnownEdge returns true if the graph source already knows of the
+	// passed channel ID.
+	IsKnownEdge(chanID lnwire.ShortChannelID) bool
+
+	// IsStaleEdgePolicy returns true if the graph source has a channel
+	// edge for the passed channel ID (and flags) that have a more recent
+	// timestamp.
+	IsStaleEdgePolicy(chanID lnwire.ShortChannelID, timestamp time.Time,
+		flags lnwire.ChanUpdateFlag) bool
+
 	// ForAllOutgoingChannels is used to iterate over all channels
 	// emanating from the "source" node which is the center of the
 	// star-graph.
@@ -819,6 +835,42 @@ func (r *ChannelRouter) networkHandler() {
 	}
 }
 
+// assertNodeAnnFreshness returns a non-nil error if we have an announcement in
+// the database for the passed node with a timestamp newer than the passed
+// timestamp. ErrIgnored will be returned if we already have the node, and
+// ErrOutdated will be returned if we have a timestamp that's after the new
+// timestamp.
+func (r *ChannelRouter) assertNodeAnnFreshness(node Vertex,
+	msgTimestamp time.Time) error {
+
+	// If we are not already aware of this node, it means that we don't
+	// know about any channel using this node. To avoid a DoS attack by
+	// node announcements, we will ignore such nodes. If we do know about
+	// this node, check that this update brings info newer than what we
+	// already have.
+	lastUpdate, exists, err := r.cfg.Graph.HasLightningNode(node)
+	if err != nil {
+		return errors.Errorf("unable to query for the "+
+			"existence of node: %v", err)
+	}
+	if !exists {
+		return newErrf(ErrIgnored, "Ignoring node announcement"+
+			" for node not found in channel graph (%x)",
+			node[:])
+	}
+
+	// If we've reached this point then we're aware of the vertex being
+	// advertised. So we now check if the new message has a new time stamp,
+	// if not then we won't accept the new data as it would override newer
+	// data.
+	if !lastUpdate.Before(msgTimestamp) {
+		return newErrf(ErrOutdated, "Ignoring outdated "+
+			"announcement for %x", node[:])
+	}
+
+	return nil
+}
+
 // processUpdate processes a new relate authenticated channel/edge, node or
 // channel/edge update network update. If the update didn't affect the internal
 // state of the draft due to either being out of date, invalid, or redundant,
@@ -829,31 +881,12 @@ func (r *ChannelRouter) processUpdate(msg interface{}) error {
 
 	switch msg := msg.(type) {
 	case *channeldb.LightningNode:
-		// If we are not already aware of this node, it means that we
+		// Before we add the node to the database, we'll check to see
-		// don't know about any channel using this node. To avoid a DoS
+		// if the announcement is "fresh" or not. If it isn't, then
-		// attack by node announcements, we will ignore such nodes. If
+		// we'll return an error.
-		// we do know about this node, check that this update brings
+		err := r.assertNodeAnnFreshness(msg.PubKeyBytes, msg.LastUpdate)
-		// info newer than what we already have.
-		lastUpdate, exists, err := r.cfg.Graph.HasLightningNode(msg.PubKeyBytes)
 		if err != nil {
-			return errors.Errorf("unable to query for the "+
+			return err
-				"existence of node: %v", err)
-		}
-		if !exists {
-			return newErrf(ErrIgnored, "Ignoring node announcement"+
-				" for node not found in channel graph (%x)",
-				msg.PubKeyBytes)
-		}
-
-		// If we've reached this point then we're aware of the vertex
-		// being advertised. So we now check if the new message has a
-		// new time stamp, if not then we won't accept the new data as
-		// it would override newer data.
-		if exists && lastUpdate.After(msg.LastUpdate) ||
-			lastUpdate.Equal(msg.LastUpdate) {
-
-			return newErrf(ErrOutdated, "Ignoring outdated "+
-				"announcement for %x", msg.PubKeyBytes)
 		}
 
 		if err := r.cfg.Graph.AddLightningNode(msg); err != nil {
@@ -1070,8 +1103,7 @@ func (r *ChannelRouter) processUpdate(msg interface{}) error {
 		}
 
 		invalidateCache = true
-		log.Infof("New channel update applied: %v",
+		log.Debugf("New channel update applied: %v", spew.Sdump(msg))
-			spew.Sdump(msg))
 
 	default:
 		return errors.Errorf("wrong routing update message type")
@@ -1907,3 +1939,63 @@ func (r *ChannelRouter) AddProof(chanID lnwire.ShortChannelID,
 	info.AuthProof = proof
 	return r.cfg.Graph.UpdateChannelEdge(info)
 }
+
+// IsStaleNode returns true if the graph source has a node announcement for the
+// target node with a more recent timestamp.
+//
+// NOTE: This method is part of the ChannelGraphSource interface.
+func (r *ChannelRouter) IsStaleNode(node Vertex, timestamp time.Time) bool {
+	// If our attempt to assert that the node announcement is fresh fails,
+	// then we know that this is actually a stale announcement.
+	return r.assertNodeAnnFreshness(node, timestamp) != nil
+}
+
+// IsKnownEdge returns true if the graph source already knows of the passed
+// channel ID.
+//
+// NOTE: This method is part of the ChannelGraphSource interface.
+func (r *ChannelRouter) IsKnownEdge(chanID lnwire.ShortChannelID) bool {
+	_, _, exists, _ := r.cfg.Graph.HasChannelEdge(chanID.ToUint64())
+	return exists
+}
+
+// IsStaleEdgePolicy returns true if the graph soruce has a channel edge for
+// the passed channel ID (and flags) that have a more recent timestamp.
+//
+// NOTE: This method is part of the ChannelGraphSource interface.
+func (r *ChannelRouter) IsStaleEdgePolicy(chanID lnwire.ShortChannelID,
+	timestamp time.Time, flags lnwire.ChanUpdateFlag) bool {
+
+	edge1Timestamp, edge2Timestamp, exists, err := r.cfg.Graph.HasChannelEdge(
+		chanID.ToUint64(),
+	)
+	if err != nil {
+		return false
+
+	}
+
+	// If we don't know of the edge, then it means it's fresh (thus not
+	// stale).
+	if !exists {
+		return false
+	}
+
+	// As edges are directional edge node has a unique policy for the
+	// direction of the edge they control. Therefore we first check if we
+	// already have the most up to date information for that edge. If so,
+	// then we can exit early.
+	switch {
+
+	// A flag set of 0 indicates this is an announcement for the "first"
+	// node in the channel.
+	case flags&lnwire.ChanUpdateDirection == 0:
+		return !edge1Timestamp.Before(timestamp)
+
+	// Similarly, a flag set of 1 indicates this is an announcement for the
+	// "second" node in the channel.
+	case flags&lnwire.ChanUpdateDirection == 1:
+		return !edge2Timestamp.Before(timestamp)
+	}
+
+	return false
+}

routing/router_test.go

@@ -1391,3 +1391,242 @@ func TestFindPathFeeWeighting(t *testing.T) {
 		t.Fatalf("wrong node: %v", path[0].Node.Alias)
 	}
 }
+
+// TestIsStaleNode tests that the IsStaleNode method properly detects stale
+// node announcements.
+func TestIsStaleNode(t *testing.T) {
+	t.Parallel()
+
+	const startingBlockHeight = 101
+	ctx, cleanUp, err := createTestCtx(startingBlockHeight)
+	defer cleanUp()
+	if err != nil {
+		t.Fatalf("unable to create router: %v", err)
+	}
+
+	// Before we can insert a node in to the database, we need to create a
+	// channel that it's linked to.
+	var (
+		pub1 [33]byte
+		pub2 [33]byte
+	)
+	copy(pub1[:], priv1.PubKey().SerializeCompressed())
+	copy(pub2[:], priv2.PubKey().SerializeCompressed())
+
+	fundingTx, _, chanID, err := createChannelEdge(ctx,
+		bitcoinKey1.SerializeCompressed(),
+		bitcoinKey2.SerializeCompressed(),
+		10000, 500)
+	if err != nil {
+		t.Fatalf("unable to create channel edge: %v", err)
+	}
+	fundingBlock := &wire.MsgBlock{
+		Transactions: []*wire.MsgTx{fundingTx},
+	}
+	ctx.chain.addBlock(fundingBlock, chanID.BlockHeight, chanID.BlockHeight)
+
+	edge := &channeldb.ChannelEdgeInfo{
+		ChannelID:        chanID.ToUint64(),
+		NodeKey1Bytes:    pub1,
+		NodeKey2Bytes:    pub2,
+		BitcoinKey1Bytes: pub1,
+		BitcoinKey2Bytes: pub2,
+		AuthProof:        nil,
+	}
+	if err := ctx.router.AddEdge(edge); err != nil {
+		t.Fatalf("unable to add edge: %v", err)
+	}
+
+	// Before we add the node, if we query for staleness, we should get
+	// false, as we haven't added the full node.
+	updateTimeStamp := time.Unix(123, 0)
+	if ctx.router.IsStaleNode(pub1, updateTimeStamp) {
+		t.Fatalf("incorrectly detected node as stale")
+	}
+
+	// With the node stub in the database, we'll add the fully node
+	// announcement to the database.
+	n1 := &channeldb.LightningNode{
+		HaveNodeAnnouncement: true,
+		LastUpdate:           updateTimeStamp,
+		Addresses:            testAddrs,
+		Color:                color.RGBA{1, 2, 3, 0},
+		Alias:                "node11",
+		AuthSigBytes:         testSig.Serialize(),
+		Features:             testFeatures,
+	}
+	copy(n1.PubKeyBytes[:], priv1.PubKey().SerializeCompressed())
+	if err := ctx.router.AddNode(n1); err != nil {
+		t.Fatalf("could not add node: %v", err)
+	}
+
+	// If we use the same timestamp and query for staleness, we should get
+	// true.
+	if !ctx.router.IsStaleNode(pub1, updateTimeStamp) {
+		t.Fatalf("failure to detect stale node update")
+	}
+
+	// If we update the timestamp and once again query for staleness, it
+	// should report false.
+	newTimeStamp := time.Unix(1234, 0)
+	if ctx.router.IsStaleNode(pub1, newTimeStamp) {
+		t.Fatalf("incorrectly detected node as stale")
+	}
+}
+
+// TestIsKnownEdge tests that the IsKnownEdge method properly detects stale
+// channel announcements.
+func TestIsKnownEdge(t *testing.T) {
+	t.Parallel()
+
+	const startingBlockHeight = 101
+	ctx, cleanUp, err := createTestCtx(startingBlockHeight)
+	defer cleanUp()
+	if err != nil {
+		t.Fatalf("unable to create router: %v", err)
+	}
+
+	// First, we'll create a new channel edge (just the info) and insert it
+	// into the database.
+	var (
+		pub1 [33]byte
+		pub2 [33]byte
+	)
+	copy(pub1[:], priv1.PubKey().SerializeCompressed())
+	copy(pub2[:], priv2.PubKey().SerializeCompressed())
+
+	fundingTx, _, chanID, err := createChannelEdge(ctx,
+		bitcoinKey1.SerializeCompressed(),
+		bitcoinKey2.SerializeCompressed(),
+		10000, 500)
+	if err != nil {
+		t.Fatalf("unable to create channel edge: %v", err)
+	}
+	fundingBlock := &wire.MsgBlock{
+		Transactions: []*wire.MsgTx{fundingTx},
+	}
+	ctx.chain.addBlock(fundingBlock, chanID.BlockHeight, chanID.BlockHeight)
+
+	edge := &channeldb.ChannelEdgeInfo{
+		ChannelID:        chanID.ToUint64(),
+		NodeKey1Bytes:    pub1,
+		NodeKey2Bytes:    pub2,
+		BitcoinKey1Bytes: pub1,
+		BitcoinKey2Bytes: pub2,
+		AuthProof:        nil,
+	}
+	if err := ctx.router.AddEdge(edge); err != nil {
+		t.Fatalf("unable to add edge: %v", err)
+	}
+
+	// Now that the edge has been inserted, query is the router already
+	// knows of the edge should return true.
+	if !ctx.router.IsKnownEdge(*chanID) {
+		t.Fatalf("router should detect edge as known")
+	}
+}
+
+// TestIsStaleEdgePolicy tests that the IsStaleEdgePolicy properly detects
+// stale channel edge update announcements.
+func TestIsStaleEdgePolicy(t *testing.T) {
+	t.Parallel()
+
+	const startingBlockHeight = 101
+	ctx, cleanUp, err := createTestCtx(startingBlockHeight,
+		basicGraphFilePath)
+	defer cleanUp()
+	if err != nil {
+		t.Fatalf("unable to create router: %v", err)
+	}
+
+	// First, we'll create a new channel edge (just the info) and insert it
+	// into the database.
+	var (
+		pub1 [33]byte
+		pub2 [33]byte
+	)
+	copy(pub1[:], priv1.PubKey().SerializeCompressed())
+	copy(pub2[:], priv2.PubKey().SerializeCompressed())
+
+	fundingTx, _, chanID, err := createChannelEdge(ctx,
+		bitcoinKey1.SerializeCompressed(),
+		bitcoinKey2.SerializeCompressed(),
+		10000, 500)
+	if err != nil {
+		t.Fatalf("unable to create channel edge: %v", err)
+	}
+	fundingBlock := &wire.MsgBlock{
+		Transactions: []*wire.MsgTx{fundingTx},
+	}
+	ctx.chain.addBlock(fundingBlock, chanID.BlockHeight, chanID.BlockHeight)
+
+	// If we query for staleness before adding the edge, we should get
+	// false.
+	updateTimeStamp := time.Unix(123, 0)
+	if ctx.router.IsStaleEdgePolicy(*chanID, updateTimeStamp, 0) {
+		t.Fatalf("router failed to detect fresh edge policy")
+	}
+	if ctx.router.IsStaleEdgePolicy(*chanID, updateTimeStamp, 1) {
+		t.Fatalf("router failed to detect fresh edge policy")
+	}
+
+	edge := &channeldb.ChannelEdgeInfo{
+		ChannelID:        chanID.ToUint64(),
+		NodeKey1Bytes:    pub1,
+		NodeKey2Bytes:    pub2,
+		BitcoinKey1Bytes: pub1,
+		BitcoinKey2Bytes: pub2,
+		AuthProof:        nil,
+	}
+	if err := ctx.router.AddEdge(edge); err != nil {
+		t.Fatalf("unable to add edge: %v", err)
+	}
+
+	// We'll also add two edge policies, one for each direction.
+	edgePolicy := &channeldb.ChannelEdgePolicy{
+		SigBytes:                  testSig.Serialize(),
+		ChannelID:                 edge.ChannelID,
+		LastUpdate:                updateTimeStamp,
+		TimeLockDelta:             10,
+		MinHTLC:                   1,
+		FeeBaseMSat:               10,
+		FeeProportionalMillionths: 10000,
+	}
+	edgePolicy.Flags = 0
+	if err := ctx.router.UpdateEdge(edgePolicy); err != nil {
+		t.Fatalf("unable to update edge policy: %v", err)
+	}
+
+	edgePolicy = &channeldb.ChannelEdgePolicy{
+		SigBytes:                  testSig.Serialize(),
+		ChannelID:                 edge.ChannelID,
+		LastUpdate:                updateTimeStamp,
+		TimeLockDelta:             10,
+		MinHTLC:                   1,
+		FeeBaseMSat:               10,
+		FeeProportionalMillionths: 10000,
+	}
+	edgePolicy.Flags = 1
+	if err := ctx.router.UpdateEdge(edgePolicy); err != nil {
+		t.Fatalf("unable to update edge policy: %v", err)
+	}
+
+	// Now that the edges have been added, an identical (chanID, flag,
+	// timestamp) tuple for each edge should be detected as a stale edge.
+	if !ctx.router.IsStaleEdgePolicy(*chanID, updateTimeStamp, 0) {
+		t.Fatalf("router failed to detect stale edge policy")
+	}
+	if !ctx.router.IsStaleEdgePolicy(*chanID, updateTimeStamp, 1) {
+		t.Fatalf("router failed to detect stale edge policy")
+	}
+
+	// If we now update the timestamp for both edges, the router should
+	// detect that this tuple represents a fresh edge.
+	updateTimeStamp = time.Unix(9999, 0)
+	if ctx.router.IsStaleEdgePolicy(*chanID, updateTimeStamp, 0) {
+		t.Fatalf("router failed to detect fresh edge policy")
+	}
+	if ctx.router.IsStaleEdgePolicy(*chanID, updateTimeStamp, 1) {
+		t.Fatalf("router failed to detect fresh edge policy")
+	}
+}