Merge pull request #775 from cfromknecht/sphinx-replay

Switch Persistence [1/4]: Infra for Sphinx Batched Decoding and Replay Protection

glide.lock

@@ -1,5 +1,5 @@
-hash: e0cbbf733425814948ced513c5921d8a4920f179d059f39eeafbf337c32b4ead
+hash: 607d774162a30647479b037b9291e372c92d1ab6b60808837b22c205ac58981f
-updated: 2018-03-08T20:06:04.267833-05:00
+updated: 2018-03-08T21:10:50.953704-05:00
 imports:
 - name: git.schwanenlied.me/yawning/bsaes.git
   version: e06297f34865a50b8e473105e52cb64ad1b55da8
@@ -84,7 +84,7 @@ imports:
   - filterdb
   - headerfs
 - name: github.com/lightningnetwork/lightning-onion
-  version: dbb6dc0eaf32a043c9bc3cfed0fd5fd8db08e3b9
+  version: 9e4b184daf3e32e0c2e523b92ec1b2d6c51ad77f
 - name: github.com/ltcsuite/ltcd
   version: 5f654d5faab99ee2b3488fabba98e5f7a5257ee3
   subpackages:

glide.yaml

@@ -59,7 +59,7 @@ import:
 - package: google.golang.org/grpc
   version: b3ddf786825de56a4178401b7e174ee332173b66
 - package: github.com/lightningnetwork/lightning-onion
-  version: dbb6dc0eaf32a043c9bc3cfed0fd5fd8db08e3b9
+  version: 9e4b184daf3e32e0c2e523b92ec1b2d6c51ad77f
 - package: github.com/grpc-ecosystem/grpc-gateway
   version: f2862b476edcef83412c7af8687c9cd8e4097c0f
 - package: github.com/go-errors/errors

htlcswitch/failure.go

@@ -3,6 +3,7 @@ package htlcswitch
 import (
 	"bytes"
 	"fmt"
+	"io"
 
 	"github.com/lightningnetwork/lightning-onion"
 	"github.com/lightningnetwork/lnd/lnwire"
@@ -45,6 +46,37 @@ type ErrorDecrypter interface {
 	DecryptError(lnwire.OpaqueReason) (*ForwardingError, error)
 }
 
+// EncrypterType establishes an enum used in serialization to indicate how to
+// decode a concrete instance of the ErrorEncrypter interface.
+type EncrypterType byte
+
+const (
+	// EncrypterTypeNone signals that no error encyrpter is present, this
+	// can happen if the htlc is originates in the switch.
+	EncrypterTypeNone EncrypterType = 0
+
+	// EncrypterTypeSphinx is used to identify a sphinx onion error
+	// encrypter instance.
+	EncrypterTypeSphinx = 1
+
+	// EncrypterTypeMock is used to identify a mock obfuscator instance.
+	EncrypterTypeMock = 2
+)
+
+// UnknownEncrypterType is an error message used to signal that an unexpected
+// EncrypterType was encountered during decoding.
+type UnknownEncrypterType EncrypterType
+
+// Error returns a formatted error indicating the invalid EncrypterType.
+func (e UnknownEncrypterType) Error() string {
+	return fmt.Sprintf("unknown error encrypter type: %d", e)
+}
+
+// ErrorEncrypterExtracter defines a function signature that extracts an
+// ErrorEncrypter from an sphinx OnionPacket.
+type ErrorEncrypterExtracter func(*sphinx.OnionPacket) (ErrorEncrypter,
+	lnwire.FailCode)
+
 // ErrorEncrypter is an interface that is used to encrypt HTLC related errors
 // at the source of the error, and also at each intermediate hop all the way
 // back to the source of the payment.
@@ -59,6 +91,16 @@ type ErrorEncrypter interface {
 	// in an additional layer of onion encryption. This process repeats
 	// until the error arrives at the source of the payment.
 	IntermediateEncrypt(lnwire.OpaqueReason) lnwire.OpaqueReason
+
+	// Type returns an enum indicating the underlying concrete instance
+	// backing this interface.
+	Type() EncrypterType
+
+	// Encode serializes the encrypter to the given io.Writer.
+	Encode(io.Writer) error
+
+	// Decode deserializes the encrypter from the given io.Reader.
+	Decode(io.Reader) error
 }
 
 // SphinxErrorEncrypter is a concrete implementation of both the ErrorEncrypter
@@ -67,6 +109,16 @@ type ErrorEncrypter interface {
 // encryption and must be treated as such accordingly.
 type SphinxErrorEncrypter struct {
 	*sphinx.OnionErrorEncrypter
+
+	ogPacket *sphinx.OnionPacket
+}
+
+// NewSphinxErrorEncrypter initializes a new sphinx error encrypter as well as
+// the embedded onion error encrypter.
+func NewSphinxErrorEncrypter() *SphinxErrorEncrypter {
+	return &SphinxErrorEncrypter{
+		OnionErrorEncrypter: &sphinx.OnionErrorEncrypter{},
+	}
 }
 
 // EncryptFirstHop transforms a concrete failure message into an encrypted
@@ -97,6 +149,24 @@ func (s *SphinxErrorEncrypter) IntermediateEncrypt(reason lnwire.OpaqueReason) l
 	return s.EncryptError(false, reason)
 }
 
+// Type returns the identifier for a sphinx error encrypter.
+func (s *SphinxErrorEncrypter) Type() EncrypterType {
+	return EncrypterTypeSphinx
+}
+
+// Encode serializes the error encrypter to the provided io.Writer.
+func (s *SphinxErrorEncrypter) Encode(w io.Writer) error {
+	return s.OnionErrorEncrypter.Encode(w)
+}
+
+// Decode reconstructs the error encrypter from the provided io.Reader.
+func (s *SphinxErrorEncrypter) Decode(r io.Reader) error {
+	if s.OnionErrorEncrypter == nil {
+		s.OnionErrorEncrypter = &sphinx.OnionErrorEncrypter{}
+	}
+	return s.OnionErrorEncrypter.Decode(r)
+}
+
 // A compile time check to ensure SphinxErrorEncrypter implements the
 // ErrorEncrypter interface.
 var _ ErrorEncrypter = (*SphinxErrorEncrypter)(nil)

htlcswitch/iterator.go

@@ -40,6 +40,10 @@ var (
 	// exitHop is a special "hop" which denotes that an incoming HTLC is
 	// meant to pay finally to the receiving node.
 	exitHop lnwire.ShortChannelID
+
+	// sourceHop is a sentinel value denoting that an incoming HTLC is
+	// initiated by our own switch.
+	sourceHop lnwire.ShortChannelID
 )
 
 // ForwardingInfo contains all the information that is necessary to forward and
@@ -85,14 +89,20 @@ type HopIterator interface {
 	// EncodeNextHop encodes the onion packet destined for the next hop
 	// into the passed io.Writer.
 	EncodeNextHop(w io.Writer) error
+
+	// ExtractErrorEncrypter returns the ErrorEncrypter needed for this hop,
+	// along with a failure code to signal if the decoding was successful.
+	ExtractErrorEncrypter(ErrorEncrypterExtracter) (ErrorEncrypter,
+		lnwire.FailCode)
 }
 
 // sphinxHopIterator is the Sphinx implementation of hop iterator which uses
 // onion routing to encode the payment route  in such a way so that node might
 // see only the next hop in the route..
 type sphinxHopIterator struct {
-	// nextPacket is the decoded onion packet for the _next_ hop.
+	// ogPacket is the original packet from which the processed packet is
-	nextPacket *sphinx.OnionPacket
+	// derived.
+	ogPacket *sphinx.OnionPacket
 
 	// processedPacket is the outcome of processing an onion packet. It
 	// includes the information required to properly forward the packet to
@@ -100,6 +110,17 @@ type sphinxHopIterator struct {
 	processedPacket *sphinx.ProcessedPacket
 }
 
+// makeSphinxHopIterator converts a processed packet returned from a sphinx
+// router and converts it into an hop iterator for usage in the link.
+func makeSphinxHopIterator(ogPacket *sphinx.OnionPacket,
+	packet *sphinx.ProcessedPacket) *sphinxHopIterator {
+
+	return &sphinxHopIterator{
+		ogPacket:        ogPacket,
+		processedPacket: packet,
+	}
+}
+
 // A compile time check to ensure sphinxHopIterator implements the HopIterator
 // interface.
 var _ HopIterator = (*sphinxHopIterator)(nil)
@@ -108,7 +129,7 @@ var _ HopIterator = (*sphinxHopIterator)(nil)
 //
 // NOTE: Part of the HopIterator interface.
 func (r *sphinxHopIterator) EncodeNextHop(w io.Writer) error {
-	return r.nextPacket.Encode(w)
+	return r.processedPacket.NextPacket.Encode(w)
 }
 
 // ForwardingInstructions returns the set of fields that detail exactly _how_
@@ -137,6 +158,18 @@ func (r *sphinxHopIterator) ForwardingInstructions() ForwardingInfo {
 	}
 }
 
+// ExtractErrorEncrypter decodes and returns the ErrorEncrypter for this hop,
+// along with a failure code to signal if the decoding was successful. The
+// ErrorEncrypter is used to encrypt errors back to the sender in the event that
+// a payment fails.
+//
+// NOTE: Part of the HopIterator interface.
+func (r *sphinxHopIterator) ExtractErrorEncrypter(
+	extracter ErrorEncrypterExtracter) (ErrorEncrypter, lnwire.FailCode) {
+
+	return extracter(r.ogPacket)
+}
+
 // OnionProcessor is responsible for keeping all sphinx dependent parts inside
 // and expose only decoding function. With such approach we give freedom for
 // subsystems which wants to decode sphinx path to not be dependable from
@@ -155,11 +188,22 @@ func NewOnionProcessor(router *sphinx.Router) *OnionProcessor {
 	return &OnionProcessor{router}
 }
 
+// Start spins up the onion processor's sphinx router.
+func (p *OnionProcessor) Start() error {
+	return p.router.Start()
+}
+
+// Stop shutsdown the onion processor's sphinx router.
+func (p *OnionProcessor) Stop() error {
+	p.router.Stop()
+	return nil
+}
+
 // DecodeHopIterator attempts to decode a valid sphinx packet from the passed io.Reader
 // instance using the rHash as the associated data when checking the relevant
 // MACs during the decoding process.
-func (p *OnionProcessor) DecodeHopIterator(r io.Reader, rHash []byte) (HopIterator,
+func (p *OnionProcessor) DecodeHopIterator(r io.Reader, rHash []byte,
-	lnwire.FailCode) {
+	incomingCltv uint32) (HopIterator, lnwire.FailCode) {
 
 	onionPkt := &sphinx.OnionPacket{}
 	if err := onionPkt.Decode(r); err != nil {
@@ -179,7 +223,9 @@ func (p *OnionProcessor) DecodeHopIterator(r io.Reader, rHash []byte) (HopIterat
 	// associated data in order to thwart attempts a replay attacks. In the
 	// case of a replay, an attacker is *forced* to use the same payment
 	// hash twice, thereby losing their money entirely.
-	sphinxPacket, err := p.router.ProcessOnionPacket(onionPkt, rHash)
+	sphinxPacket, err := p.router.ProcessOnionPacket(
+		onionPkt, rHash, incomingCltv,
+	)
 	if err != nil {
 		switch err {
 		case sphinx.ErrInvalidOnionVersion:
@@ -194,10 +240,160 @@ func (p *OnionProcessor) DecodeHopIterator(r io.Reader, rHash []byte) (HopIterat
 		}
 	}
 
-	return &sphinxHopIterator{
+	return makeSphinxHopIterator(onionPkt, sphinxPacket), lnwire.CodeNone
-		nextPacket:      sphinxPacket.NextPacket,
+}
-		processedPacket: sphinxPacket,
+
-	}, lnwire.CodeNone
+// DecodeHopIteratorRequest encapsulates all date necessary to process an onion
+// packet, perform sphinx replay detection, and schedule the entry for garbage
+// collection.
+type DecodeHopIteratorRequest struct {
+	OnionReader  io.Reader
+	RHash        []byte
+	IncomingCltv uint32
+}
+
+// DecodeHopIteratorResponse encapsulates the outcome of a batched sphinx onion
+// processing.
+type DecodeHopIteratorResponse struct {
+	HopIterator HopIterator
+	FailCode    lnwire.FailCode
+}
+
+// Result returns the (HopIterator, lnwire.FailCode) tuple, which should
+// correspond to the index of a particular DecodeHopIteratorRequest.
+//
+// NOTE: The HopIterator should be considered invalid if the fail code is
+// anything but lnwire.CodeNone.
+func (r *DecodeHopIteratorResponse) Result() (HopIterator, lnwire.FailCode) {
+	return r.HopIterator, r.FailCode
+}
+
+// DecodeHopIterators performs batched decoding and validation of incoming
+// sphinx packets. For the same `id`, this method will return the same iterators
+// and failcodes upon subsequent invocations.
+//
+// NOTE: In order for the responses to be valid, the caller must guarantee that
+// the presented readers and rhashes *NEVER* deviate across invocations for the
+// same id.
+func (p *OnionProcessor) DecodeHopIterators(id []byte,
+	reqs []DecodeHopIteratorRequest) ([]DecodeHopIteratorResponse, error) {
+
+	var (
+		batchSize = len(reqs)
+		onionPkts = make([]sphinx.OnionPacket, batchSize)
+		resps     = make([]DecodeHopIteratorResponse, batchSize)
+	)
+
+	tx := p.router.BeginTxn(id, batchSize)
+
+	for i, req := range reqs {
+		onionPkt := &onionPkts[i]
+		resp := &resps[i]
+
+		err := onionPkt.Decode(req.OnionReader)
+		switch err {
+		case nil:
+			// success
+
+		case sphinx.ErrInvalidOnionVersion:
+			resp.FailCode = lnwire.CodeInvalidOnionVersion
+			continue
+
+		case sphinx.ErrInvalidOnionKey:
+			resp.FailCode = lnwire.CodeInvalidOnionKey
+			continue
+
+		default:
+			log.Errorf("unable to decode onion packet: %v", err)
+			resp.FailCode = lnwire.CodeInvalidOnionKey
+			continue
+		}
+
+		err = tx.ProcessOnionPacket(
+			uint16(i), onionPkt, req.RHash, req.IncomingCltv,
+		)
+		switch err {
+		case nil:
+			// success
+
+		case sphinx.ErrInvalidOnionVersion:
+			resp.FailCode = lnwire.CodeInvalidOnionVersion
+			continue
+
+		case sphinx.ErrInvalidOnionHMAC:
+			resp.FailCode = lnwire.CodeInvalidOnionHmac
+			continue
+
+		case sphinx.ErrInvalidOnionKey:
+			resp.FailCode = lnwire.CodeInvalidOnionKey
+			continue
+
+		default:
+			log.Errorf("unable to process onion packet: %v", err)
+			resp.FailCode = lnwire.CodeInvalidOnionKey
+			continue
+		}
+	}
+
+	// With that batch created, we will now attempt to write the shared
+	// secrets to disk. This operation will returns the set of indices that
+	// were detected as replays, and the computed sphinx packets for all
+	// indices that did not fail the above loop. Only indices that are not
+	// in the replay set should be considered valid, as they are
+	// opportunistically computed.
+	packets, replays, err := tx.Commit()
+	if err != nil {
+		log.Errorf("unable to process onion packet batch %x: %v",
+			id, err)
+
+		// If we failed to commit the batch to the secret share log, we
+		// will mark all not-yet-failed channels with a temporary
+		// channel failure and exit since we cannot proceed.
+		for i := range resps {
+			resp := &resps[i]
+
+			// Skip any indexes that already failed onion decoding.
+			if resp.FailCode != lnwire.CodeNone {
+				continue
+			}
+
+			log.Errorf("unable to process onion packet %x-%v",
+				id, i)
+			resp.FailCode = lnwire.CodeTemporaryChannelFailure
+		}
+
+		// TODO(conner): return real errors to caller so link can fail?
+		return resps, err
+	}
+
+	// Otherwise, the commit was successful. Now we will post process any
+	// remaining packets, additionally failing any that were included in the
+	// replay set.
+	for i := range resps {
+		resp := &resps[i]
+
+		// Skip any indexes that already failed onion decoding.
+		if resp.FailCode != lnwire.CodeNone {
+			continue
+		}
+
+		// If this index is contained in the replay set, mark it with a
+		// temporary channel failure error code. We infer that the
+		// offending error was due to a replayed packet because this
+		// index was found in the replay set.
+		if replays.Contains(uint16(i)) {
+			log.Errorf("unable to process onion packet: %v",
+				sphinx.ErrReplayedPacket)
+			resp.FailCode = lnwire.CodeTemporaryChannelFailure
+			continue
+		}
+
+		// Finally, construct a hop iterator from our processed sphinx
+		// packet, simultaneously caching the original onion packet.
+		resp.HopIterator = makeSphinxHopIterator(&onionPkts[i], &packets[i])
+	}
+
+	return resps, nil
 }
 
 // ExtractErrorEncrypter takes an io.Reader which should contain the onion
@@ -205,20 +401,8 @@ func (p *OnionProcessor) DecodeHopIterator(r io.Reader, rHash []byte) (HopIterat
 // ErrorEncrypter instance using the derived shared secret. In the case that en
 // error occurs, a lnwire failure code detailing the parsing failure will be
 // returned.
-func (p *OnionProcessor) ExtractErrorEncrypter(r io.Reader) (ErrorEncrypter, lnwire.FailCode) {
+func (p *OnionProcessor) ExtractErrorEncrypter(onionPkt *sphinx.OnionPacket) (
-
+	ErrorEncrypter, lnwire.FailCode) {
-	onionPkt := &sphinx.OnionPacket{}
-	if err := onionPkt.Decode(r); err != nil {
-		switch err {
-		case sphinx.ErrInvalidOnionVersion:
-			return nil, lnwire.CodeInvalidOnionVersion
-		case sphinx.ErrInvalidOnionKey:
-			return nil, lnwire.CodeInvalidOnionKey
-		default:
-			log.Errorf("unable to decode onion packet: %v", err)
-			return nil, lnwire.CodeInvalidOnionKey
-		}
-	}
 
 	onionObfuscator, err := sphinx.NewOnionErrorEncrypter(p.router,
 		onionPkt.EphemeralKey)
@@ -238,5 +422,6 @@ func (p *OnionProcessor) ExtractErrorEncrypter(r io.Reader) (ErrorEncrypter, lnw
 
 	return &SphinxErrorEncrypter{
 		OnionErrorEncrypter: onionObfuscator,
+		ogPacket:            onionPkt,
 	}, lnwire.CodeNone
 }

htlcswitch/link.go

@@ -3,12 +3,11 @@ package htlcswitch
 import (
 	"bytes"
 	"fmt"
+	"io"
 	"sync"
 	"sync/atomic"
 	"time"
 
-	"io"
-
 	"crypto/sha256"
 
 	"github.com/go-errors/errors"
@@ -130,11 +129,19 @@ type ChannelLinkConfig struct {
 	// DecodeHopIterator function is responsible for decoding HTLC Sphinx
 	// onion blob, and creating hop iterator which will give us next
 	// destination of HTLC.
-	DecodeHopIterator func(r io.Reader, rHash []byte) (HopIterator, lnwire.FailCode)
+	DecodeHopIterator func(r io.Reader, rHash []byte,
+		cltv uint32) (HopIterator, lnwire.FailCode)
+
+	// DecodeHopIterators facilitates batched decoding of HTLC Sphinx onion
+	// blobs, which are then used to inform how to forward an HTLC.
+	// NOTE: This function assumes the same set of readers and preimages are
+	// always presented for the same identifier.
+	DecodeHopIterators func([]byte, []DecodeHopIteratorRequest) (
+		[]DecodeHopIteratorResponse, error)
 
 	// DecodeOnionObfuscator function is responsible for decoding HTLC
 	// Sphinx onion blob, and creating onion failure obfuscator.
-	DecodeOnionObfuscator func(r io.Reader) (ErrorEncrypter, lnwire.FailCode)
+	DecodeOnionObfuscator ErrorEncrypterExtracter
 
 	// GetLastChannelUpdate retrieves the latest routing policy for this
 	// particular channel. This will be used to provide payment senders our
@@ -1448,26 +1455,6 @@ func (l *channelLink) processLockedInHtlcs(
 			var onionBlob [lnwire.OnionPacketSize]byte
 			copy(onionBlob[:], pd.OnionBlob)
 
-			// Retrieve onion obfuscator from onion blob in order
-			// to produce initial obfuscation of the onion
-			// failureCode.
-			onionReader := bytes.NewReader(onionBlob[:])
-			obfuscator, failureCode := l.cfg.DecodeOnionObfuscator(
-				onionReader,
-			)
-			if failureCode != lnwire.CodeNone {
-				// If we're unable to process the onion blob
-				// than we should send the malformed htlc error
-				// to payment sender.
-				l.sendMalformedHTLCError(pd.HtlcIndex, failureCode,
-					onionBlob[:])
-				needUpdate = true
-
-				log.Errorf("unable to decode onion "+
-					"obfuscator: %v", failureCode)
-				continue
-			}
-
 			// Before adding the new htlc to the state machine,
 			// parse the onion object in order to obtain the
 			// routing information with DecodeHopIterator function
@@ -1479,9 +1466,9 @@ func (l *channelLink) processLockedInHtlcs(
 			// attacks. In the case of a replay, an attacker is
 			// *forced* to use the same payment hash twice, thereby
 			// losing their money entirely.
-			onionReader = bytes.NewReader(onionBlob[:])
+			onionReader := bytes.NewReader(onionBlob[:])
 			chanIterator, failureCode := l.cfg.DecodeHopIterator(
-				onionReader, pd.RHash[:],
+				onionReader, pd.RHash[:], pd.Timeout,
 			)
 			if failureCode != lnwire.CodeNone {
 				// If we're unable to process the onion blob
@@ -1496,6 +1483,25 @@ func (l *channelLink) processLockedInHtlcs(
 				continue
 			}
 
+			// Retrieve onion obfuscator from onion blob in order
+			// to produce initial obfuscation of the onion
+			// failureCode.
+			obfuscator, failureCode := chanIterator.ExtractErrorEncrypter(
+				l.cfg.DecodeOnionObfuscator,
+			)
+			if failureCode != lnwire.CodeNone {
+				// If we're unable to process the onion blob
+				// than we should send the malformed htlc error
+				// to payment sender.
+				l.sendMalformedHTLCError(pd.HtlcIndex, failureCode,
+					onionBlob[:])
+				needUpdate = true
+
+				log.Errorf("unable to decode onion "+
+					"obfuscator: %v", failureCode)
+				continue
+			}
+
 			heightNow := l.bestHeight
 
 			fwdInfo := chanIterator.ForwardingInstructions()

htlcswitch/link_test.go

@@ -9,12 +9,11 @@ import (
 	"testing"
 	"time"
 
-	"io"
-
 	"math"
 
 	"github.com/davecgh/go-spew/spew"
 	"github.com/go-errors/errors"
+	"github.com/lightningnetwork/lightning-onion"
 	"github.com/lightningnetwork/lnd/chainntnfs"
 	"github.com/lightningnetwork/lnd/channeldb"
 	"github.com/lightningnetwork/lnd/contractcourt"
@@ -1119,7 +1118,7 @@ func TestChannelLinkMultiHopDecodeError(t *testing.T) {
 
 	// Replace decode function with another which throws an error.
 	n.carolChannelLink.cfg.DecodeOnionObfuscator = func(
-		r io.Reader) (ErrorEncrypter, lnwire.FailCode) {
+		*sphinx.OnionPacket) (ErrorEncrypter, lnwire.FailCode) {
 		return nil, lnwire.CodeInvalidOnionVersion
 	}
 
@@ -1450,7 +1449,7 @@ func newSingleLinkTestHarness(chanAmt, chanReserve btcutil.Amount) (
 		Peer:              alicePeer,
 		Switch:            New(Config{}),
 		DecodeHopIterator: decoder.DecodeHopIterator,
-		DecodeOnionObfuscator: func(io.Reader) (ErrorEncrypter, lnwire.FailCode) {
+		DecodeOnionObfuscator: func(*sphinx.OnionPacket) (ErrorEncrypter, lnwire.FailCode) {
 			return obfuscator, lnwire.CodeNone
 		},
 		GetLastChannelUpdate: mockGetChanUpdateMessage,

htlcswitch/mock.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/btcsuite/fastsha256"
 	"github.com/go-errors/errors"
+	"github.com/lightningnetwork/lightning-onion"
 	"github.com/lightningnetwork/lnd/chainntnfs"
 	"github.com/lightningnetwork/lnd/channeldb"
 	"github.com/lightningnetwork/lnd/contractcourt"
@@ -195,6 +196,10 @@ type mockHopIterator struct {
 	hops []ForwardingInfo
 }
 
+func (r *mockHopIterator) OnionPacket() *sphinx.OnionPacket {
+	return nil
+}
+
 func newMockHopIterator(hops ...ForwardingInfo) HopIterator {
 	return &mockHopIterator{hops: hops}
 }
@@ -205,6 +210,12 @@ func (r *mockHopIterator) ForwardingInstructions() ForwardingInfo {
 	return h
 }
 
+func (r *mockHopIterator) ExtractErrorEncrypter(
+	extracter ErrorEncrypterExtracter) (ErrorEncrypter, lnwire.FailCode) {
+
+	return extracter(nil)
+}
+
 func (r *mockHopIterator) EncodeNextHop(w io.Writer) error {
 	var hopLength [4]byte
 	binary.BigEndian.PutUint32(hopLength[:], uint32(len(r.hops)))
@@ -246,12 +257,30 @@ var _ HopIterator = (*mockHopIterator)(nil)
 
 // mockObfuscator mock implementation of the failure obfuscator which only
 // encodes the failure and do not makes any onion obfuscation.
-type mockObfuscator struct{}
+type mockObfuscator struct {
+	ogPacket *sphinx.OnionPacket
+}
 
 func newMockObfuscator() ErrorEncrypter {
 	return &mockObfuscator{}
 }
 
+func (o *mockObfuscator) OnionPacket() *sphinx.OnionPacket {
+	return o.ogPacket
+}
+
+func (o *mockObfuscator) Type() EncrypterType {
+	return EncrypterTypeMock
+}
+
+func (o *mockObfuscator) Encode(w io.Writer) error {
+	return nil
+}
+
+func (o *mockObfuscator) Decode(r io.Reader) error {
+	return nil
+}
+
 func (o *mockObfuscator) EncryptFirstHop(failure lnwire.FailureMessage) (
 	lnwire.OpaqueReason, error) {
 
@@ -292,10 +321,20 @@ var _ ErrorDecrypter = (*mockDeobfuscator)(nil)
 
 // mockIteratorDecoder test version of hop iterator decoder which decodes the
 // encoded array of hops.
-type mockIteratorDecoder struct{}
+type mockIteratorDecoder struct {
+	mu sync.RWMutex
 
-func (p *mockIteratorDecoder) DecodeHopIterator(r io.Reader, meta []byte) (
+	responses map[[32]byte][]DecodeHopIteratorResponse
-	HopIterator, lnwire.FailCode) {
+}
+
+func newMockIteratorDecoder() *mockIteratorDecoder {
+	return &mockIteratorDecoder{
+		responses: make(map[[32]byte][]DecodeHopIteratorResponse),
+	}
+}
+
+func (p *mockIteratorDecoder) DecodeHopIterator(r io.Reader, rHash []byte,
+	cltv uint32) (HopIterator, lnwire.FailCode) {
 
 	var b [4]byte
 	_, err := r.Read(b[:])
@@ -317,6 +356,40 @@ func (p *mockIteratorDecoder) DecodeHopIterator(r io.Reader, meta []byte) (
 	return newMockHopIterator(hops...), lnwire.CodeNone
 }
 
+func (p *mockIteratorDecoder) DecodeHopIterators(id []byte,
+	reqs []DecodeHopIteratorRequest) ([]DecodeHopIteratorResponse, error) {
+
+	idHash := sha256.Sum256(id)
+
+	p.mu.RLock()
+	if resps, ok := p.responses[idHash]; ok {
+		p.mu.RUnlock()
+		return resps, nil
+	}
+	p.mu.RUnlock()
+
+	batchSize := len(reqs)
+
+	resps := make([]DecodeHopIteratorResponse, 0, batchSize)
+	for _, req := range reqs {
+		iterator, failcode := p.DecodeHopIterator(
+			req.OnionReader, req.RHash, req.IncomingCltv,
+		)
+
+		resp := DecodeHopIteratorResponse{
+			HopIterator: iterator,
+			FailCode:    failcode,
+		}
+		resps = append(resps, resp)
+	}
+
+	p.mu.Lock()
+	p.responses[idHash] = resps
+	p.mu.Unlock()
+
+	return resps, nil
+}
+
 func (f *ForwardingInfo) decode(r io.Reader) error {
 	var net [1]byte
 	if _, err := r.Read(net[:]); err != nil {

htlcswitch/test_utils.go

@@ -11,14 +11,13 @@ import (
 	"io/ioutil"
 	"os"
 
-	"io"
-
 	"math/big"
 
 	"net"
 
 	"github.com/btcsuite/fastsha256"
 	"github.com/go-errors/errors"
+	"github.com/lightningnetwork/lightning-onion"
 	"github.com/lightningnetwork/lnd/chainntnfs"
 	"github.com/lightningnetwork/lnd/channeldb"
 	"github.com/lightningnetwork/lnd/contractcourt"
@@ -800,7 +799,7 @@ func newThreeHopNetwork(t testing.TB, aliceChannel, firstBobChannel,
 			Peer:              bobServer,
 			Switch:            aliceServer.htlcSwitch,
 			DecodeHopIterator: decoder.DecodeHopIterator,
-			DecodeOnionObfuscator: func(io.Reader) (ErrorEncrypter,
+			DecodeOnionObfuscator: func(*sphinx.OnionPacket) (ErrorEncrypter,
 				lnwire.FailCode) {
 				return obfuscator, lnwire.CodeNone
 			},
@@ -846,7 +845,7 @@ func newThreeHopNetwork(t testing.TB, aliceChannel, firstBobChannel,
 			Peer:              aliceServer,
 			Switch:            bobServer.htlcSwitch,
 			DecodeHopIterator: decoder.DecodeHopIterator,
-			DecodeOnionObfuscator: func(io.Reader) (ErrorEncrypter,
+			DecodeOnionObfuscator: func(*sphinx.OnionPacket) (ErrorEncrypter,
 				lnwire.FailCode) {
 				return obfuscator, lnwire.CodeNone
 			},
@@ -892,7 +891,7 @@ func newThreeHopNetwork(t testing.TB, aliceChannel, firstBobChannel,
 			Peer:              carolServer,
 			Switch:            bobServer.htlcSwitch,
 			DecodeHopIterator: decoder.DecodeHopIterator,
-			DecodeOnionObfuscator: func(io.Reader) (ErrorEncrypter,
+			DecodeOnionObfuscator: func(*sphinx.OnionPacket) (ErrorEncrypter,
 				lnwire.FailCode) {
 				return obfuscator, lnwire.CodeNone
 			},
@@ -938,7 +937,7 @@ func newThreeHopNetwork(t testing.TB, aliceChannel, firstBobChannel,
 			Peer:              bobServer,
 			Switch:            carolServer.htlcSwitch,
 			DecodeHopIterator: decoder.DecodeHopIterator,
-			DecodeOnionObfuscator: func(io.Reader) (ErrorEncrypter,
+			DecodeOnionObfuscator: func(*sphinx.OnionPacket) (ErrorEncrypter,
 				lnwire.FailCode) {
 				return obfuscator, lnwire.CodeNone
 			},

lnd_test.go

@@ -5006,7 +5006,7 @@ func testBidirectionalAsyncPayments(net *lntest.NetworkHarness, t *harnessTest)
 
 	// Wait for Alice and Bob receive their payments, and throw and error
 	// if something goes wrong.
-	maxTime := 20 * time.Second
+	maxTime := 60 * time.Second
 	for i := 0; i < 2; i++ {
 		select {
 		case err := <-errChan:

server.go

@@ -9,6 +9,7 @@ import (
 	"image/color"
 	"math/big"
 	"net"
+	"path/filepath"
 	"strconv"
 	"sync"
 	"sync/atomic"
@@ -151,6 +152,15 @@ func newServer(listenAddrs []string, chanDB *channeldb.DB, cc *chainControl,
 	globalFeatures := lnwire.NewRawFeatureVector()
 
 	serializedPubKey := privKey.PubKey().SerializeCompressed()
+
+	// Initialize the sphinx router, placing it's persistent replay log in
+	// the same directory as the channel graph database.
+	graphDir := filepath.Dir(chanDB.Path())
+	sharedSecretPath := filepath.Join(graphDir, "sphinxreplay.db")
+	sphinxRouter := sphinx.NewRouter(
+		sharedSecretPath, privKey, activeNetParams.Params, cc.chainNotifier,
+	)
+
 	s := &server{
 		chanDB: chanDB,
 		cc:     cc,
@@ -162,8 +172,7 @@ func newServer(listenAddrs []string, chanDB *channeldb.DB, cc *chainControl,
 
 		// TODO(roasbeef): derive proper onion key based on rotation
 		// schedule
-		sphinx: htlcswitch.NewOnionProcessor(
+		sphinx:      htlcswitch.NewOnionProcessor(sphinxRouter),
-			sphinx.NewRouter(privKey, activeNetParams.Params)),
 		lightningID: sha256.Sum256(serializedPubKey),
 
 		persistentPeers:        make(map[string]struct{}),
@@ -491,7 +500,9 @@ func (s *server) Start() error {
 	if err := s.cc.chainNotifier.Start(); err != nil {
 		return err
 	}
-
+	if err := s.sphinx.Start(); err != nil {
+		return err
+	}
 	if err := s.htlcSwitch.Start(); err != nil {
 		return err
 	}
@@ -554,6 +565,7 @@ func (s *server) Stop() error {
 	s.cc.chainNotifier.Stop()
 	s.chanRouter.Stop()
 	s.htlcSwitch.Stop()
+	s.sphinx.Stop()
 	s.utxoNursery.Stop()
 	s.breachArbiter.Stop()
 	s.authGossiper.Stop()