sphinx: adds batched processing of onion pkts via Tx

Author: cfromknecht

sphinx.go

@@ -11,7 +11,6 @@ import (
 	"math"
 	"math/big"
 
-	"github.com/Crypt-iQ/lightning-onion/persistlog"
 	"github.com/aead/chacha20"
 	"github.com/lightningnetwork/lnd/chainntnfs"
 	"github.com/roasbeef/btcd/btcec"
@@ -70,6 +69,10 @@ const (
 	baseVersion = 0
 )
 
+// Hash256 is a statically sized, 32-byte array, typically containing
+// the output of a SHA256 hash.
+type Hash256 [sha256.Size]byte
+
 var (
 	// paddingBytes are the padding bytes used to fill out the remainder of the
 	// unused portion of the per-hop payload.
@@ -207,14 +210,14 @@ func (hd *HopData) Decode(r io.Reader) error {
 // generateSharedSecrets by the given nodes pubkeys, generates the shared
 // secrets.
 func generateSharedSecrets(paymentPath []*btcec.PublicKey,
-	sessionKey *btcec.PrivateKey) [][sha256.Size]byte {
+	sessionKey *btcec.PrivateKey) []Hash256 {
 	// Each hop performs ECDH with our ephemeral key pair to arrive at a
 	// shared secret. Additionally, each hop randomizes the group element
 	// for the next hop by multiplying it by the blinding factor. This way
 	// we only need to transmit a single group element, and hops can't link
 	// a session back to us if they have several nodes in the path.
 	numHops := len(paymentPath)
-	hopSharedSecrets := make([][sha256.Size]byte, numHops)
+	hopSharedSecrets := make([]Hash256, numHops)
 
 	// Compute the triplet for the first hop outside of the main loop.
 	// Within the loop each new triplet will be computed recursively based
@@ -301,8 +304,8 @@ func NewOnionPacket(paymentPath []*btcec.PublicKey, sessionKey *btcec.PrivateKey
 		// We'll derive the two keys we need for each hop in order to:
 		// generate our stream cipher bytes for the mixHeader, and
 		// calculate the MAC over the entire constructed packet.
-		rhoKey := generateKey("rho", hopSharedSecrets[i])
-		muKey := generateKey("mu", hopSharedSecrets[i])
+		rhoKey := generateKey("rho", &hopSharedSecrets[i])
+		muKey := generateKey("mu", &hopSharedSecrets[i])
 
 		// The HMAC for the final hop is simply zeroes. This allows the
 		// last hop to recognize that it is the destination for a
@@ -379,13 +382,13 @@ func rightShift(slice []byte, num int) {
 // "filler" bytes produced by this function at the last hop. Using this
 // methodology, the size of the field stays constant at each hop.
 func generateHeaderPadding(key string, numHops int, hopSize int,
-	sharedSecrets [][sharedSecretSize]byte) []byte {
+	sharedSecrets []Hash256) []byte {
 
 	filler := make([]byte, (numHops-1)*hopSize)
 	for i := 1; i < numHops; i++ {
 		totalFillerSize := ((NumMaxHops - i) + 1) * hopSize
 
-		streamKey := generateKey(key, sharedSecrets[i-1])
+		streamKey := generateKey(key, &sharedSecrets[i-1])
 		streamBytes := generateCipherStream(streamKey, numStreamBytes)
 
 		xor(filler, filler, streamBytes[totalFillerSize:totalFillerSize+i*hopSize])
@@ -486,7 +489,7 @@ func xor(dst, a, b []byte) int {
 // construction/processing based off of the denoted keyType. Within Sphinx
 // various keys are used within the same onion packet for padding generation,
 // MAC generation, and encryption/decryption.
-func generateKey(keyType string, sharedKey [sharedSecretSize]byte) [keyLen]byte {
+func generateKey(keyType string, sharedKey *Hash256) [keyLen]byte {
 	mac := hmac.New(sha256.New, []byte(keyType))
 	mac.Write(sharedKey[:])
 	h := mac.Sum(nil)
@@ -517,12 +520,14 @@ func generateCipherStream(key [keyLen]byte, numBytes uint) []byte {
 // computeBlindingFactor for the next hop given the ephemeral pubKey and
 // sharedSecret for this hop. The blinding factor is computed as the
 // sha-256(pubkey || sharedSecret).
-func computeBlindingFactor(hopPubKey *btcec.PublicKey, hopSharedSecret []byte) [sha256.Size]byte {
+func computeBlindingFactor(hopPubKey *btcec.PublicKey,
+	hopSharedSecret []byte) Hash256 {
+
 	sha := sha256.New()
 	sha.Write(hopPubKey.SerializeCompressed())
 	sha.Write(hopSharedSecret)
 
-	var hash [sha256.Size]byte
+	var hash Hash256
 	copy(hash[:], sha.Sum(nil))
 	return hash
 }
@@ -547,7 +552,7 @@ func blindBaseElement(blindingFactor []byte) *btcec.PublicKey {
 // key. We then take the _entire_ point generated by the ECDH operation,
 // serialize that using a compressed format, then feed the raw bytes through a
 // single SHA256 invocation.  The resulting value is the shared secret.
-func generateSharedSecret(pub *btcec.PublicKey, priv *btcec.PrivateKey) [32]byte {
+func generateSharedSecret(pub *btcec.PublicKey, priv *btcec.PrivateKey) Hash256 {
 	s := &btcec.PublicKey{}
 	x, y := btcec.S256().ScalarMult(pub.X, pub.Y, priv.D.Bytes())
 	s.X = x
@@ -622,23 +627,19 @@ type Router struct {
 
 	onionKey *btcec.PrivateKey
 
-	d *persistlog.DecayedLog
+	log ReplayLog
 }
 
 // NewRouter creates a new instance of a Sphinx onion Router given the node's
 // currently advertised onion private key, and the target Bitcoin network.
-func NewRouter(nodeKey *btcec.PrivateKey, net *chaincfg.Params,
+func NewRouter(dbPath string, nodeKey *btcec.PrivateKey, net *chaincfg.Params,
 	chainNotifier chainntnfs.ChainNotifier) *Router {
 	var nodeID [addressSize]byte
 	copy(nodeID[:], btcutil.Hash160(nodeKey.PubKey().SerializeCompressed()))
 
 	// Safe to ignore the error here, nodeID is 20 bytes.
 	nodeAddr, _ := btcutil.NewAddressPubKeyHash(nodeID[:], net)
 
-	d := &persistlog.DecayedLog{
-		Notifier: chainNotifier,
-	}
-
 	return &Router{
 		nodeID:   nodeID,
 		nodeAddr: nodeAddr,
@@ -652,20 +653,20 @@ func NewRouter(nodeKey *btcec.PrivateKey, net *chaincfg.Params,
 		},
 		// TODO(roasbeef): replace instead with bloom filter?
 		// * https://moderncrypto.org/mail-archive/messaging/2015/001911.html
-		d: d,
+		log: NewDecayedLog(dbPath, chainNotifier),
 	}
 }
 
 // Start starts / opens the DecayedLog's channeldb and its accompanying
 // garbage collector goroutine.
 func (r *Router) Start() error {
-	return r.d.Start("")
+	return r.log.Start()
 }
 
 // Stop stops / closes the DecayedLog's channeldb and its accompanying
 // garbage collector goroutine.
 func (r *Router) Stop() {
-	r.d.Stop()
+	r.log.Stop()
 }
 
 // ProcessOnionPacket processes an incoming onion packet which has been forward
@@ -677,28 +678,46 @@ func (r *Router) Stop() {
 // In the case of a successful packet processing, and ProcessedPacket struct is
 // returned which houses the newly parsed packet, along with instructions on
 // what to do next.
-func (r *Router) ProcessOnionPacket(onionPkt *OnionPacket, assocData []byte) (*ProcessedPacket, error) {
-	dhKey := onionPkt.EphemeralKey
-	routeInfo := onionPkt.RoutingInfo
-	headerMac := onionPkt.HeaderMAC
+func (r *Router) ProcessOnionPacket(onionPkt *OnionPacket,
+	assocData []byte, incomingCltv uint32) (*ProcessedPacket, error) {
 
+	// Compute the shared secret for this onion packet.
 	sharedSecret, err := r.generateSharedSecret(onionPkt.EphemeralKey)
 	if err != nil {
 		return nil, err
 	}
 
-	// In order to mitigate replay attacks, if we've seen this particular
-	// shared secret before, cease processing and just drop this forwarding
-	// message.
-	hashedSecret := persistlog.HashSharedSecret(sharedSecret)
-	cltv, err := r.d.Get(hashedSecret[:])
+	// Additionally, compute the hash prefix of the shared secret, which
+	// will serve as an identifier for detecting replayed packets.
+	hashPrefix := hashSharedSecret(&sharedSecret)
+
+	// Continue to optimistically process this packet, deferring replay
+	// protection until the end to reduce the penalty of multiple IO
+	// operations.
+	packet, err := processOnionPacket(onionPkt, &sharedSecret, assocData)
 	if err != nil {
 		return nil, err
 	}
-	if cltv != math.MaxUint32 {
-		return nil, ErrReplayedPacket
+
+	// Atomically compare this hash prefix with the contents of the on-disk
+	// log, persisting it only if this entry was not detected as a replay.
+	if err := r.log.Put(&hashPrefix, incomingCltv); err != nil {
+		return nil, err
 	}
 
+	return packet, nil
+}
+
+// processOnionPacket performs the primary key derivation and handling of onion
+// packets. The processed packets returned from this method should only be used
+// if the packet was not flagged as a replayed packet.
+func processOnionPacket(onionPkt *OnionPacket,
+	sharedSecret *Hash256, assocData []byte) (*ProcessedPacket, error) {
+
+	dhKey := onionPkt.EphemeralKey
+	routeInfo := onionPkt.RoutingInfo
+	headerMac := onionPkt.HeaderMAC
+
 	// Using the derived shared secret, ensure the integrity of the routing
 	// information by checking the attached MAC without leaking timing
 	// information.
@@ -711,9 +730,14 @@ func (r *Router) ProcessOnionPacket(onionPkt *OnionPacket, assocData []byte) (*P
 	// Attach the padding zeroes in order to properly strip an encryption
 	// layer off the routing info revealing the routing information for the
 	// next hop.
+	streamBytes := generateCipherStream(
+		generateKey("rho", sharedSecret),
+		numStreamBytes,
+	)
+	headerWithPadding := append(routeInfo[:],
+		bytes.Repeat([]byte{0}, hopDataSize)...)
+
 	var hopInfo [numStreamBytes]byte
-	streamBytes := generateCipherStream(generateKey("rho", sharedSecret), numStreamBytes)
-	headerWithPadding := append(routeInfo[:], bytes.Repeat([]byte{0}, hopDataSize)...)
 	xor(hopInfo[:], headerWithPadding, streamBytes)
 
 	// Randomize the DH group element for the next hop using the
@@ -729,23 +753,6 @@ func (r *Router) ProcessOnionPacket(onionPkt *OnionPacket, assocData []byte) (*P
 		return nil, err
 	}
 
-	// The MAC checks out, mark this current shared secret as processed in
-	// order to mitigate future replay attacks. We need to check to see if
-	// we already know the secret again since a replay might have happened
-	// while we were checking the MAC and decoding the HopData.
-	cltv, err = r.d.Get(hashedSecret[:])
-	if err != nil {
-		return nil, err
-	}
-	if cltv != math.MaxUint32 {
-		return nil, ErrReplayedPacket
-	}
-
-	err = r.d.Put(hashedSecret[:], hopData.OutgoingCltv)
-	if err != nil {
-		return nil, err
-	}
-
 	// With the necessary items extracted, we'll copy of the onion packet
 	// for the next node, snipping off our per-hop data.
 	var nextMixHeader [routingInfoSize]byte
@@ -761,7 +768,7 @@ func (r *Router) ProcessOnionPacket(onionPkt *OnionPacket, assocData []byte) (*P
 	// However if the uncovered 'nextMac' is all zeroes, then this
 	// indicates that we're the final hop in the route.
 	var action ProcessCode = MoreHops
-	if bytes.Compare(bytes.Repeat([]byte{0x00}, hmacSize), hopData.HMAC[:]) == 0 {
+	if bytes.Compare(zeroHMAC[:], hopData.HMAC[:]) == 0 {
 		action = ExitNode
 	}
 
@@ -773,9 +780,9 @@ func (r *Router) ProcessOnionPacket(onionPkt *OnionPacket, assocData []byte) (*P
 }
 
 // generateSharedSecret generates the shared secret by given ephemeral key.
-func (r *Router) generateSharedSecret(dhKey *btcec.PublicKey) ([sha256.Size]byte,
+func (r *Router) generateSharedSecret(dhKey *btcec.PublicKey) (Hash256,
 	error) {
-	var sharedSecret [sha256.Size]byte
+	var sharedSecret Hash256
 
 	// Ensure that the public key is on our curve.
 	if !btcec.S256().IsOnCurve(dhKey.X, dhKey.Y) {
@@ -786,3 +793,96 @@ func (r *Router) generateSharedSecret(dhKey *btcec.PublicKey) ([sha256.Size]byte
 	sharedSecret = generateSharedSecret(dhKey, r.onionKey)
 	return sharedSecret, nil
 }
+
+// Tx is a transaction consisting of a number of sphinx packets to be atomically
+// written to the replay log. This structure helps to coordinate construction of
+// the underlying Batch object, and to ensure that the result of the processing
+// is idempotent.
+type Tx struct {
+	// batch is the set of packets to be incrementally processed and
+	// ultimately committed in this transaction
+	batch *Batch
+
+	// router is a reference to the sphinx router that created this
+	// transaction. Committing this transaction will utilize this router's
+	// replay log.
+	router *Router
+
+	// packets contains a potentially sparse list of optimistically processed
+	// packets for this batch. The contents of a particular index should
+	// only be accessed if the index is *not* included in the replay set, or
+	// otherwise failed any other stage of the processing.
+	packets []ProcessedPacket
+}
+
+// BeginTxn creates a new transaction that can later be committed back to the
+// sphinx router's replay log.
+//
+// NOTE: The nels parameter should represent the maximum number of that could be
+// added to the batch, using sequence numbers that match or exceed this value
+// could result in an out-of-bounds panic.
+func (r *Router) BeginTxn(id []byte, nels int) *Tx {
+	return &Tx{
+		batch:   NewBatch(id),
+		router:  r,
+		packets: make([]ProcessedPacket, nels),
+	}
+}
+
+// ProcessOnionPacket processes an incoming onion packet which has been forward
+// to the target Sphinx router. If the encoded ephemeral key isn't on the
+// target Elliptic Curve, then the packet is rejected. Similarly, if the
+// derived shared secret has been seen before the packet is rejected.  Finally
+// if the MAC doesn't check the packet is again rejected.
+//
+// In the case of a successful packet processing, and ProcessedPacket struct is
+// returned which houses the newly parsed packet, along with instructions on
+// what to do next.
+func (t *Tx) ProcessOnionPacket(seqNum uint16, onionPkt *OnionPacket,
+	assocData []byte, incomingCltv uint32) error {
+
+	// Compute the shared secret for this onion packet.
+	sharedSecret, err := t.router.generateSharedSecret(
+		onionPkt.EphemeralKey)
+	if err != nil {
+		return err
+	}
+
+	// Additionally, compute the hash prefix of the shared secret, which
+	// will serve as an identifier for detecting replayed packets.
+	hashPrefix := hashSharedSecret(&sharedSecret)
+
+	// Continue to optimistically process this packet, deferring replay
+	// protection until the end to reduce the penalty of multiple IO
+	// operations.
+	packet, err := processOnionPacket(onionPkt, &sharedSecret, assocData)
+	if err != nil {
+		return err
+	}
+
+	// Add the hash prefix to pending batch of shared secrets that will be
+	// written later via Commit().
+	err = t.batch.Put(seqNum, &hashPrefix, incomingCltv)
+	if err != nil {
+		return err
+	}
+
+	// If we successfully added this packet to the batch, cache the processed
+	// packet within the Tx which can be accessed after committing if this
+	// sequence number does not appear in the replay set.
+	t.packets[seqNum] = *packet
+
+	return nil
+}
+
+// Commit writes this transaction's batch of sphinx packets to the replay log,
+// performing a final check against the log for replays.
+func (t *Tx) Commit() ([]ProcessedPacket, *ReplaySet, error) {
+	if t.batch.isCommitted {
+		return t.packets, t.batch.replaySet, nil
+	}
+
+	rs, err := t.router.log.PutBatch(t.batch)
+
+	return t.packets, rs, err
+}