Add replay detection to handshakes.

This is done by maintaining a map keyed off the SipHash-2-4 digest of
the MAC_C component of the handshake.  Collisions, while possible are
unlikely in the extreme and are thus treated as replays.

In concept this is fairly similar to the ScrambleSuit `replay.py` code,
with a few modifications:

 * There is a upper bound on how large the replay filter can grow.
   Currently this is set to 102400 entries, though it is unlikely that
   this limit will be hit.

 * A doubly linked list is also maintained parallel to the map, so the
   filter compaction process does not need to iterate over the entire
   filter.

5 files changed, 256 insertions, 8 deletions

diff --git a/handshake_ntor.go b/handshake_ntor.go
index bff500c..3a8b36e 100644
--- a/handshake_ntor.go
+++ b/handshake_ntor.go
@@ -73,6 +73,11 @@ var ErrMarkNotFoundYet = errors.New("handshake: M_[C,S] not found yet")
 // connection MUST be dropped.
 var ErrInvalidHandshake = errors.New("handshake: Failed to find M_[C,S]")
 
+// ErrReplayedHandshake is the error returned when the obfs4 handshake fails
+// due it being replayed.  This error is fatal and the connection MUST be
+// dropped.
+var ErrReplayedHandshake = errors.New("handshake: Replay detected")
+
 // ErrNtorFailed is the error returned when the ntor handshake fails.  This
 // error is fatal and the connection MUST be dropped.
 var ErrNtorFailed = errors.New("handshake: ntor handshake failure")
@@ -242,7 +247,7 @@ func newServerHandshake(nodeID *ntor.NodeID, serverIdentity *ntor.Keypair) *serv
 	return hs
 }
 
-func (hs *serverHandshake) parseClientHandshake(resp []byte) ([]byte, error) {
+func (hs *serverHandshake) parseClientHandshake(filter *replayFilter, resp []byte) ([]byte, error) {
 	// No point in examining the data unless the miminum plausible response has
 	// been received.
 	if clientMinHandshakeLength > len(resp) {
@@ -281,14 +286,19 @@ func (hs *serverHandshake) parseClientHandshake(resp []byte) ([]byte, error) {
 		macCmp := hs.mac.Sum(nil)[:macLength]
 		macRx := resp[pos+markLength : pos+markLength+macLength]
 		if hmac.Equal(macCmp, macRx) {
+			// Ensure that this handshake has not been seen previously.
+			if filter.testAndSet(time.Now().Unix(), macRx) {
+				// The client either happened to generate exactly the same
+				// session key and padding, or someone is replaying a previous
+				// handshake.  In either case, fuck them.
+				return nil, ErrReplayedHandshake
+			}
+
 			macFound = true
 			hs.epochHour = epochHour
 
-			// In theory, we should always evaluate all 3 MACs, but at this
-			// point we are reasonably confident that the client knows the
-			// correct NodeID/Public key, and if this fails, we just ignore the
-			// client for a random interval and drop the connection anyway.
-			break
+			// We could break out here, but in the name of reducing timing
+			// variation, evaluate all 3 MACs.
 		}
 	}
 	if !macFound {
diff --git a/handshake_ntor_test.go b/handshake_ntor_test.go
index 41780e9..73b43bf 100644
--- a/handshake_ntor_test.go
+++ b/handshake_ntor_test.go
@@ -45,6 +45,7 @@ func TestHandshakeNtor(t *testing.T) {
 		t.Fatal("newClientHandshake failed:", err)
 	}
 	serverHs := newServerHandshake(nodeID, idKeypair)
+	serverFilter, _ := newReplayFilter()
 
 	// Generate what the client will send to the server.
 	cToS, err := clientHs.generateHandshake()
@@ -53,7 +54,7 @@ func TestHandshakeNtor(t *testing.T) {
 	}
 
 	// Parse the client handshake message.
-	serverSeed, err := serverHs.parseClientHandshake(cToS)
+	serverSeed, err := serverHs.parseClientHandshake(serverFilter, cToS)
 	if err != nil {
 		t.Fatal("serverHandshake.parseClientHandshake() failed", err)
 	}
diff --git a/obfs4.go b/obfs4.go
index 4466375..e9ba2e8 100644
--- a/obfs4.go
+++ b/obfs4.go
@@ -247,7 +247,7 @@ func (c *Obfs4Conn) serverHandshake(nodeID *ntor.NodeID, keypair *ntor.Keypair)
 		c.receiveBuffer.Write(hsBuf[:n])
 
 		var seed []byte
-		seed, err = hs.parseClientHandshake(c.receiveBuffer.Bytes())
+		seed, err = hs.parseClientHandshake(c.listener.filter, c.receiveBuffer.Bytes())
 		if err == ErrMarkNotFoundYet {
 			continue
 		} else if err != nil {
@@ -548,6 +548,8 @@ func DialObfs4(network, address, nodeID, publicKey string) (*Obfs4Conn, error) {
 type Obfs4Listener struct {
 	listener net.Listener
 
+	filter *replayFilter
+
 	keyPair *ntor.Keypair
 	nodeID  *ntor.NodeID
 
@@ -645,6 +647,11 @@ func ListenObfs4(network, laddr, nodeID, privateKey, seed string) (*Obfs4Listene
 		return nil, err
 	}
 
+	l.filter, err = newReplayFilter()
+	if err != nil {
+		return nil, err
+	}
+
 	rng := rand.New(newHashDrbg(l.seed))
 	l.closeDelayBytes = rng.Intn(maxCloseDelayBytes)
 	l.closeDelay = rng.Intn(maxCloseDelay)
diff --git a/replay_filter.go b/replay_filter.go
new file mode 100644
index 0000000..f5c64f8
--- /dev/null
+++ b/replay_filter.go
@@ -0,0 +1,138 @@
+/*
+ * Copyright (c) 2014, Yawning Angel <yawning at torproject dot org>
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ *  * Redistributions of source code must retain the above copyright notice,
+ *    this list of conditions and the following disclaimer.
+ *
+ *  * Redistributions in binary form must reproduce the above copyright notice,
+ *    this list of conditions and the following disclaimer in the documentation
+ *    and/or other materials provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package obfs4
+
+import (
+	"container/list"
+	"crypto/rand"
+	"encoding/binary"
+	"sync"
+
+	"github.com/dchest/siphash"
+)
+
+// maxFilterSize is the maximum capacity of the replay filter.  The busiest
+// bridge I know about processes something along the order of 3000 connections
+// per day.  The maximum timespan any entry can live in the filter is 2 hours,
+// so this value should be sufficient.
+const maxFilterSize = 100 * 1024
+
+// replayFilter is a simple filter designed only to answer if it has seen a
+// given byte sequence before.  It is based around comparing the SipHash-2-4
+// digest of data to match against.  Collisions are treated as positive matches
+// however, the probability of such occurences is negligible.
+type replayFilter struct {
+	lock   sync.Mutex
+	key    [2]uint64
+	filter map[uint64]*filterEntry
+	fifo   *list.List
+}
+
+type filterEntry struct {
+	firstSeen int64
+	hash      uint64
+	element   *list.Element
+}
+
+// newReplayFilter creates a new replayFilter instance.
+func newReplayFilter() (filter *replayFilter, err error) {
+	// Initialize the SipHash-2-4 instance with a random key.
+	var key [16]byte
+	_, err = rand.Read(key[:])
+	if err != nil {
+		return
+	}
+
+	filter = new(replayFilter)
+	filter.key[0] = binary.BigEndian.Uint64(key[0:8])
+	filter.key[1] = binary.BigEndian.Uint64(key[8:16])
+	filter.filter = make(map[uint64]*filterEntry)
+	filter.fifo = list.New()
+
+	return
+}
+
+// testAndSet queries the filter for buf, adds it if it was not present and
+// returns if it has added the entry or not.
+func (f *replayFilter) testAndSet(now int64, buf []byte) bool {
+	hash := siphash.Hash(f.key[0], f.key[1], buf)
+
+	f.lock.Lock()
+	defer f.lock.Unlock()
+
+	f.compactFilter(now)
+
+	entry := f.filter[hash]
+	if entry != nil {
+		return true
+	}
+
+	entry = new(filterEntry)
+	entry.hash = hash
+	entry.firstSeen = now
+	entry.element = f.fifo.PushBack(entry)
+	f.filter[hash] = entry
+
+	return false
+}
+
+// compactFilter purges entries that are too old to be relevant.  If the filter
+// is filled to maxFilterCapacity, it will force purge a single entry.
+func (f *replayFilter) compactFilter(now int64) {
+	e := f.fifo.Front()
+	for e != nil {
+		entry, _ := e.Value.(*filterEntry)
+
+		// If the filter is at max capacity, force purge at least one entry.
+		if f.fifo.Len() < maxFilterSize {
+			deltaT := now - entry.firstSeen
+			if deltaT < 0 {
+				// Aeeeeeee, the system time jumped backwards, potentially by
+				// a lot.  This will eventually self-correct, but "eventually"
+				// could be a long time.  As much as this sucks, jettison the
+				// entire filter.
+				f.filter = make(map[uint64]*filterEntry)
+				f.fifo = list.New()
+				return
+			}
+			if deltaT < 3600*2 {
+				// Why yes, this is 2 hours.  The MAC code includes a hour
+				// resolution timestamp, but to deal with clock skew, it
+				// accepts time +- 1 hour.
+				break
+			}
+		}
+		eNext := e.Next()
+		f.filter[entry.hash] = nil
+		f.fifo.Remove(entry.element)
+		entry.element = nil
+		e = eNext
+	}
+}
+
+/* vim :set ts=4 sw=4 sts=4 noet : */
diff --git a/replay_filter_test.go b/replay_filter_test.go
new file mode 100644
index 0000000..09337c0
--- /dev/null
+++ b/replay_filter_test.go
@@ -0,0 +1,92 @@
+/*
+ * Copyright (c) 2014, Yawning Angel <yawning at torproject dot org>
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ *  * Redistributions of source code must retain the above copyright notice,
+ *    this list of conditions and the following disclaimer.
+ *
+ *  * Redistributions in binary form must reproduce the above copyright notice,
+ *    this list of conditions and the following disclaimer in the documentation
+ *    and/or other materials provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package obfs4
+
+import (
+	"testing"
+)
+
+func TestReplayFilter(t *testing.T) {
+	f, err := newReplayFilter()
+	if err != nil {
+		t.Fatal("newReplayFilter failed:", err)
+	}
+
+	buf := []byte("This is a test of the Emergency Broadcast System.")
+	var now int64 = 3600
+
+	// testAndSet into empty filter, returns false (not present).
+	set := f.testAndSet(now, buf)
+	if set {
+		t.Fatal("testAndSet empty filter returned true")
+	}
+
+	// testAndSet into filter containing entry, should return true(present).
+	set = f.testAndSet(now, buf)
+	if !set {
+		t.Fatal("testAndSet populated filter (replayed) returned false")
+	}
+
+	buf2 := []byte("This concludes this test of the Emergency Broadcast System.")
+	now += 3600 * 2
+
+	// testAndSet with time advanced.
+	set = f.testAndSet(now, buf2)
+	if set {
+		t.Fatal("testAndSet populated filter, 2nd entry returned true")
+	}
+	set = f.testAndSet(now, buf2)
+	if !set {
+		t.Fatal("testAndSet populated filter, 2nd entry (replayed) returned false")
+	}
+
+	// Ensure that the first entry has been removed by compact.
+	set = f.testAndSet(now, buf)
+	if set {
+		t.Fatal("testAndSet populated filter, compact check returned true")
+	}
+
+	// Ensure that the filter gets reaped if the clock jumps backwards.
+	now = 0
+	set = f.testAndSet(now, buf)
+	if set {
+		t.Fatal("testAndSet populated filter, backward time jump returned true")
+	}
+	if len(f.filter) != 1 {
+		t.Fatal("filter map has a unexpected number of entries:", len(f.filter))
+	}
+	if f.fifo.Len() != 1 {
+		t.Fatal("filter fifo has a unexpected number of entries:", f.fifo.Len())
+	}
+
+	// Ensure that the entry is properly added after reaping.
+	set = f.testAndSet(now, buf)
+	if !set {
+		t.Fatal("testAndSet populated filter, post-backward clock jump (replayed) returned false")
+	}
+}