sphinx: replay protection and garbage collector

This commit introduces persistent replay protection against prior
shared secrets from prior HTLC's. A new data structure DecayedLog
was created that stores all shared secrets in boltdb and it
persists on startup. Additionally, it comes with a set of tests
that assert its persistence guarantees. DecayedLog adheres to the
newly created PersistLog interface. DecayedLog also comes with a
garbage collector that removes expired shared secrets from the
back-end boltdb.

Author: nsa

bench_test.go

@@ -5,6 +5,7 @@ import (
 	"testing"
 
 	"github.com/roasbeef/btcd/btcec"
+	"github.com/lightningnetwork/lightning-onion/persistlog"
 )
 
 var (
@@ -55,7 +56,14 @@ func BenchmarkPathPacketConstruction(b *testing.B) {
 
 func BenchmarkProcessPacket(b *testing.B) {
 	b.StopTimer()
-	path, _, sphinxPacket, err := newTestRoute(1)
+
+	// Create the DecayedLog object
+	d := persistlog.DecayedLog{}
+	if err := d.Start(); err != nil {
+		b.Fatalf("unable to start channeldb")
+	}
+
+	path, _, sphinxPacket, err := newTestRoute(1, d)
 	if err != nil {
 		b.Fatalf("unable to create test route: %v", err)
 	}
@@ -65,15 +73,10 @@ func BenchmarkProcessPacket(b *testing.B) {
 	var (
 		pkt *ProcessedPacket
 	)
-	for i := 0; i < b.N; i++ {
-		pkt, err = path[0].ProcessOnionPacket(sphinxPacket, nil)
-		if err != nil {
-			b.Fatalf("unable to process packet: %v", err)
-		}
 
-		b.StopTimer()
-		path[0].seenSecrets = make(map[[sharedSecretSize]byte]struct{})
-		b.StartTimer()
+	pkt, err = path[0].ProcessOnionPacket(sphinxPacket, nil)
+	if err != nil {
+		b.Fatalf("unable to process packet: %v", err)
 	}
 
 	p = pkt

cmd/main.go

@@ -76,7 +76,7 @@ func main() {
 		}
 
 		privkey, _ := btcec.PrivKeyFromBytes(btcec.S256(), binKey)
-		s := sphinx.NewRouter(privkey, &chaincfg.TestNet3Params)
+		s := sphinx.NewRouter(privkey, &chaincfg.TestNet3Params, nil)
 
 		var packet sphinx.OnionPacket
 		err = packet.Decode(bytes.NewBuffer(binMsg))

glide.lock

@@ -1,11 +1,18 @@
 package: github.com/lightningnetwork/lightning-onion
 import:
+- package: github.com/boltdb/bolt
+  version: ^1.2.1
 - package: github.com/aead/chacha20
   version: d31a916ded42d1640b9d89a26f8abd53cc96790c
+- package: github.com/lightningnetwork/lnd
+  subpackages:
+  - chainntnfs
+  - channeldb
 - package: github.com/roasbeef/btcd
   subpackages:
   - btcec
   - chaincfg
+  - wire
 - package: github.com/roasbeef/btcutil
 - package: golang.org/x/crypto
   subpackages:

glide.yaml

@@ -0,0 +1,236 @@
+package persistlog
+
+import (
+	"crypto/sha256"
+	"encoding/binary"
+	"fmt"
+	"github.com/boltdb/bolt"
+	"github.com/lightningnetwork/lnd/chainntnfs"
+	"github.com/lightningnetwork/lnd/channeldb"
+	"sync"
+)
+
+const (
+	// defaultDbDirectory is the default directory where our decayed log
+	// will store our (sharedHash, CLTV expiry height) key-value pairs.
+	defaultDbDirectory = "sharedsecret"
+
+	// sharedHashSize is the size in bytes of the keys we will be storing
+	// in the DecayedLog. It represents the first 20 bytes of a truncated
+	// sha-256 hash of a secret generated by ECDH.
+	sharedHashSize = 20
+
+	// sharedSecretSize is the size in bytes of the shared secrets.
+	sharedSecretSize = 32
+)
+
+var (
+	// sharedHashBucket is a bucket which houses all the first sharedHashSize
+	// bytes of a received HTLC's hashed shared secret and the HTLC's
+	// expiry block height.
+	sharedHashBucket = []byte("shared-hash")
+)
+
+// DecayedLog implements the PersistLog interface. It stores the first
+// sharedHashSize bytes of a sha256-hashed shared secret along with a node's
+// CLTV value. It is a decaying log meaning there will be a garbage collector
+// to collect entries which are expired according to their stored CLTV value
+// and the current block height. DecayedLog wraps channeldb for simplicity, but
+// must batch writes to the database to decrease write contention.
+type DecayedLog struct {
+	db       *channeldb.DB
+	wg       sync.WaitGroup
+	quit     chan (struct{})
+	Notifier chainntnfs.ChainNotifier
+}
+
+// garbageCollector deletes entries from sharedHashBucket whose expiry height
+// has already past. This function MUST be run as a goroutine.
+func (d *DecayedLog) garbageCollector() error {
+	defer d.wg.Done()
+
+	epochClient, err := d.Notifier.RegisterBlockEpochNtfn()
+	if err != nil {
+		return fmt.Errorf("Unable to register for epoch "+
+			"notification: %v", err)
+	}
+	defer epochClient.Cancel()
+
+outer:
+	for {
+		select {
+		case epoch, ok := <-epochClient.Epochs:
+			if !ok {
+				return fmt.Errorf("Epoch client shutting " +
+					"down")
+			}
+
+			var expiredCltv [][]byte
+			err := d.db.View(func(tx *bolt.Tx) error {
+				// Grab the shared hash bucket
+				sharedHashes := tx.Bucket(sharedHashBucket)
+				if sharedHashes == nil {
+					return fmt.Errorf("sharedHashBucket " +
+						"is nil")
+				}
+
+				sharedHashes.ForEach(func(k, v []byte) error {
+					cltv := uint32(binary.BigEndian.Uint32(v))
+					if uint32(epoch.Height) > cltv {
+						// Store expired hash in array
+						expiredCltv = append(expiredCltv, k)
+					}
+					return nil
+				})
+
+				return nil
+			})
+			if err != nil {
+				return fmt.Errorf("Error viewing channeldb: "+
+					"%v", err)
+			}
+
+			// Delete every item in array
+			for _, hash := range expiredCltv {
+				err = d.Delete(hash)
+				if err != nil {
+					return fmt.Errorf("Unable to delete"+
+						"expired secret: %v", err)
+				}
+			}
+
+		case <-d.quit:
+			break outer
+		}
+	}
+
+	return nil
+}
+
+// A compile time check to see if DecayedLog adheres to the PersistLog
+// interface.
+var _ PersistLog = (*DecayedLog)(nil)
+
+// HashSharedSecret Sha-256 hashes the shared secret and returns the first
+// sharedHashSize bytes of the hash.
+func HashSharedSecret(sharedSecret [sharedSecretSize]byte) [sharedHashSize]byte {
+	// Sha256 hash of sharedSecret
+	h := sha256.New()
+	h.Write(sharedSecret[:])
+
+	var sharedHash [sharedHashSize]byte
+
+	// Copy bytes to sharedHash
+	copy(sharedHash[:], h.Sum(nil)[:sharedHashSize])
+	return sharedHash
+}
+
+// Delete removes a <shared secret hash, CLTV value> key-pair from the
+// sharedHashBucket.
+func (d *DecayedLog) Delete(hash []byte) error {
+	return d.db.Update(func(tx *bolt.Tx) error {
+		sharedHashes, err := tx.CreateBucketIfNotExists(sharedHashBucket)
+		if err != nil {
+			return fmt.Errorf("Unable to created sharedHashes bucket:"+
+				" %v", err)
+		}
+
+		return sharedHashes.Delete(hash)
+	})
+}
+
+// Get retrieves the CLTV value of a processed HTLC given the first 20 bytes
+// of the Sha-256 hash of the shared secret used during sphinx processing.
+func (d *DecayedLog) Get(hash []byte) (
+	uint32, error) {
+	var value uint32
+
+	err := d.db.View(func(tx *bolt.Tx) error {
+		// Grab the shared hash bucket which stores the mapping from
+		// truncated sha-256 hashes of shared secrets to CLTV values.
+		sharedHashes := tx.Bucket(sharedHashBucket)
+		if sharedHashes == nil {
+			return fmt.Errorf("sharedHashes is nil, could " +
+				"not retrieve CLTV value")
+		}
+
+		// If the sharedHash is found, we use it to find the associated
+		// CLTV in the sharedHashBucket.
+		valueBytes := sharedHashes.Get(hash)
+		if valueBytes == nil {
+			return nil
+		}
+
+		value = uint32(binary.BigEndian.Uint32(valueBytes))
+
+		return nil
+	})
+	if err != nil {
+		return value, err
+	}
+
+	return value, nil
+}
+
+// Put stores a <shared secret hash, CLTV value> key-pair into the
+// sharedHashBucket.
+func (d *DecayedLog) Put(hash []byte,
+	value uint32) error {
+	return d.db.Batch(func(tx *bolt.Tx) error {
+		var scratch [4]byte
+
+		sharedHashes, err := tx.CreateBucketIfNotExists(sharedHashBucket)
+		if err != nil {
+			return fmt.Errorf("Unable to create bucket sharedHashes:"+
+				" %v", err)
+		}
+
+		// Store value into scratch
+		binary.BigEndian.PutUint32(scratch[:], value)
+
+		return sharedHashes.Put(hash, scratch[:])
+	})
+}
+
+// Start opens the database we will be using to store hashed shared secrets.
+// It also starts the garbage collector in a goroutine to remove stale
+// database entries.
+func (d *DecayedLog) Start() error {
+	// Create the quit channel
+	d.quit = make(chan struct{})
+
+	// Open the channeldb for use.
+	var err error
+	if d.db, err = channeldb.Open(defaultDbDirectory); err != nil {
+		return fmt.Errorf("Could not open channeldb: %v", err)
+	}
+
+	err = d.db.Update(func(tx *bolt.Tx) error {
+		_, err := tx.CreateBucketIfNotExists(sharedHashBucket)
+		if err != nil {
+			return fmt.Errorf("Unable to create bucket sharedHashes:"+
+				" %v", err)
+		}
+		return nil
+	})
+	if err != nil {
+		return fmt.Errorf("Could not create sharedHashes")
+	}
+
+	// Start garbage collector.
+	if d.Notifier != nil {
+		d.wg.Add(1)
+		go d.garbageCollector()
+	}
+
+	return nil
+}
+
+// Stop halts the garbage collector and closes channeldb.
+func (d *DecayedLog) Stop() {
+	// Stop garbage collector.
+	close(d.quit)
+
+	// Close channeldb.
+	d.db.Close()
+}