Clock cached msg2 resends per path in the fragment table, so a copy of msg1 arriving by a relay or a second address is answered even when the direct path only works one way

docs/pq-handshake.md

@@ -346,9 +346,17 @@ Implemented in `handshake/fragment` (`Sender`, `Assembly`, and the capped `Table
 
 ### Retransmission and amplification
 
-- Only timers cause transmission. Reactive resends (the responder resending its cached
+- Only timers cause transmission, with one reactive exception: the responder resends its
-  msg2 when a duplicate msg1 completes) are rate limited to one per handshake per retry
+  cached msg2 when a copy of msg1 completes again. An initiator sends msg1 to every address
-  interval, to avoid TFTP's Sorcerer's Apprentice amplification.
+  it knows and through its relays, the responder reassembles each path separately, and each
+  path's first completion is answered on that path at once. A path that only works one way,
+  say a responder answering from the wrong interface, is then covered by the others in the
+  same attempt, and the initiator keeps the union of the msg2 symbols that reach it. After
+  that a path is answered again at most once per retry interval, and only once K symbols not
+  seen before on it arrive: duplicates do nothing, and there are 256 symbols per message, so
+  a resend costs the initiator K packets on that path and the answer is held to the 3x
+  budget below. That is TFTP's Sorcerer's Apprentice bound, per path rather than per
+  handshake.
 - Duplicate msg1 detection compares the decoded message to the cached one **before**
   running Noise or certificate verification.
 - Before an address is validated, a responder sends at most 3x the bytes it received from
@@ -483,17 +491,18 @@ in every scenario above, and within three at least 99.99%.
     certificate work: the trailing repair fragments of an attempt rebuild a msg1 its data
     fragments already completed, and a retrying initiator offers fresh symbols for the
     same msg1. A repeat goes straight to the cached-response resend, which offers fresh
-    msg2 symbols, is rate limited to one per retry interval and is held to the 3x
+    msg2 symbols on the path the repeat came by and is held to the 3x amplification
-    amplification budget.
+    budget.
   - **Relays** carry fragments unchanged; repair fragments trail via a timer.
 - The pre-authentication table is capped at 4 MiB, 1024 entries, 4 per source (IPv4 /32,
   IPv6 /64, or relay channel), 16 KiB messages and a 3 s idle timeout. Cookies are
   demanded from half full.
 - Tests: codec and fragment tests and fuzzers, manager unit tests
-  for the response budget and rate limit, and e2e tests for a clean handshake, loss of
+  for the response budget, table tests for the per-path repeat clock, and e2e tests for a clean handshake, loss of
   data fragments in both directions, fallback to a peer that does not accept PQ, a lost
   msg2 recovered by fresh symbols, sticky mode under an attacker dropping PQ packets,
-  relays, index cleanup on completion, fallback and timeout, and a responder flooded from
+  relays, a direct msg2 lost on every attempt and answered through a relay instead, index
+  cleanup on completion, fallback and timeout, and a responder flooded from
   spoofed sources demanding and accepting a cookie. Unit tests cover cookie binding and
   rotation, the never-amplify rule, the initiator's rate limited resend, and reload. ML-KEM also
   passes under `GOEXPERIMENT=boringcrypto` and `GODEBUG=fips140=on`.

e2e/pq_hybrid_test.go

@@ -589,3 +589,63 @@ func TestHybridWrongResponder(t *testing.T) {
 		})
 	}
 }
+
+// TestHybridMsg2LostDirect loses every msg2 fragment the responder sends
+// directly, as a responder answering from the wrong interface would. msg1 also
+// reaches it through a relay, and that copy must be answered on the relay as
+// soon as it completes, not refused because the direct copy was just answered.
+// The relay takes a few retry ticks to come up in this harness, so the check is
+// that the first copy of msg1 through the relay is enough once it does.
+func TestHybridMsg2LostDirect(t *testing.T) {
+	t.Parallel()
+	ca := newHybridCA(t)
+	useRelays := pqRequire()
+	useRelays["relay"] = m{"use_relays": true}
+	amRelay := pqRequire()
+	amRelay["relay"] = m{"am_relay": true}
+	myControl, myVpn, _, _ := newSimpleServer(cert.Version3, ca.ca, ca.key, "me     ", "10.128.0.1/24", useRelays)
+	relayControl, relayVpn, relayUdp, _ := newSimpleServer(cert.Version3, ca.ca, ca.key, "relay  ", "10.128.0.128/24", amRelay)
+	theirControl, theirVpn, theirUdp, _ := newSimpleServer(cert.Version3, ca.ca, ca.key, "them   ", "10.128.0.2/24", useRelays)
+
+	myControl.InjectLightHouseAddr(theirVpn[0].Addr(), theirUdp)
+	myControl.InjectLightHouseAddr(relayVpn[0].Addr(), relayUdp)
+	myControl.InjectRelays(theirVpn[0].Addr(), []netip.Addr{relayVpn[0].Addr()})
+	relayControl.InjectLightHouseAddr(theirVpn[0].Addr(), theirUdp)
+
+	r := router.NewR(t, myControl, relayControl, theirControl)
+	defer r.RenderFlow()
+	myControl.Start()
+	relayControl.Start()
+	theirControl.Start()
+
+	var relayedMsg1, directMsg2, relayedMsg2 int
+	myControl.InjectTunPacket(BuildTunUDPPacket(theirVpn[0].Addr(), 80, myVpn[0].Addr(), 80, []byte("Hi from me")))
+	require.True(t, routeUntilData(t, r, theirControl, 10*time.Second, func(p *udp.Packet, receiver *nebula.Control) router.ExitType {
+		h, fh := handshakeOf(p)
+		if h == nil || fh == nil {
+			return router.KeepRouting
+		}
+		outer := &header.H{}
+		relayed := outer.Parse(p.Data) == nil && outer.Type == header.Message && outer.Subtype == header.MessageRelay
+		switch {
+		case h.MessageCounter == 1 && relayed && receiver == theirControl:
+			relayedMsg1++
+		case h.MessageCounter == 2 && !relayed && p.From == theirUdp && receiver == myControl:
+			directMsg2++
+			return router.Drop
+		case h.MessageCounter == 2 && relayed && receiver == myControl:
+			relayedMsg2++
+		}
+		return router.KeepRouting
+	}), "tunnel never came up")
+
+	assertUdpPacket(t, []byte("Hi from me"), theirControl.GetFromTun(true), myVpn[0].Addr(), theirVpn[0].Addr(), 80, 80)
+	assert.GreaterOrEqual(t, directMsg2, 9, "the direct answer was sent and lost")
+	assert.GreaterOrEqual(t, relayedMsg2, 9, "the relayed copy of msg1 got its own answer")
+	assert.LessOrEqual(t, relayedMsg1, 7+3, "the first copy of msg1 through the relay is answered, no relayed retry is needed")
+	r.RenderHostmaps("Final hostmaps", myControl, relayControl, theirControl)
+
+	myControl.Stop()
+	relayControl.Stop()
+	theirControl.Stop()
+}

handshake/fragment/fragment_test.go

@@ -597,3 +597,84 @@ func TestTableAnswered(t *testing.T) {
 	assert.Empty(t, tb.answered)
 	tb.mu.Unlock()
 }
+
+// TestTableRepeatInterval checks that an answered message is reported as a
+// repeat at most once per RepeatInterval on one key, while another key, a
+// second path the same message arrives by, keeps its own clock.
+func TestTableRepeatInterval(t *testing.T) {
+	r := rand.New(rand.NewPCG(4, 4))
+	cfg := testConfig()
+	cfg.RepeatInterval = 100 * time.Millisecond
+	tb := NewTable[testKey, string](cfg)
+	now := time.Unix(1000, 0)
+	msg := randMsg(r, 8009) // K=7, 3 repair
+
+	attempt := func(s *Sender) [][]byte {
+		data, repair, err := s.Attempt(DefaultRepair(s.Layout().K))
+		require.NoError(t, err)
+		return append(data, repair...)
+	}
+	repeats := func(key testKey, pkts [][]byte) int {
+		n := 0
+		for _, p := range pkts {
+			_, h, sym := parse(t, p)
+			res, err := tb.Add(key, key.src, now, h, sym)
+			require.NoError(t, err)
+			if res != nil {
+				require.True(t, res.Repeat)
+				n++
+			}
+		}
+		return n
+	}
+	answer := func(key testKey, s *Sender) {
+		var res *Result
+		for _, p := range attempt(s) {
+			_, h, sym := parse(t, p)
+			got, err := tb.Add(key, key.src, now, h, sym)
+			require.NoError(t, err)
+			if got != nil {
+				res = got
+			}
+		}
+		require.NotNil(t, res)
+		tb.Answered(key, res, now)
+	}
+
+	// Two paths carry the same message. Each has its own sender here because
+	// the initiator offers the same symbols on every path.
+	direct := testKey{"direct", 1}
+	relayed := testKey{"relay", 1}
+	sd, err := NewSender(msg, testSubtype, 0, 1, 1)
+	require.NoError(t, err)
+	sr, err := NewSender(msg, testSubtype, 0, 1, 1)
+	require.NoError(t, err)
+	answer(direct, sd)
+	answer(relayed, sr)
+
+	// A retry right after the answer carries K fresh symbols, but the clock on
+	// that path has not run out.
+	now = now.Add(10 * time.Millisecond)
+	assert.Zero(t, repeats(direct, attempt(sd)), "a repeat inside the interval is not reported")
+
+	// Once it has, the first fresh symbol of the next retry is enough: the
+	// fresh symbols kept counting.
+	now = now.Add(100 * time.Millisecond)
+	retry := attempt(sd)
+	assert.Equal(t, 1, repeats(direct, retry[:1]), "the held repeat is reported at once")
+	assert.Zero(t, repeats(direct, retry[1:]), "and only once per retry")
+
+	// The other path has its own clock, which only its own answer started.
+	assert.Equal(t, 1, repeats(relayed, attempt(sr)), "a repeat on another path is independent")
+	assert.Zero(t, repeats(relayed, attempt(sr)), "and is itself clocked")
+
+	// Without an interval every K fresh symbols is a repeat, which is what the
+	// other tests rely on.
+	tb = NewTable[testKey, string](testConfig())
+	s2, err := NewSender(msg, testSubtype, 0, 1, 2)
+	require.NoError(t, err)
+	k2 := testKey{"a", 2}
+	answer(k2, s2)
+	assert.Equal(t, 1, repeats(k2, attempt(s2)))
+	assert.Equal(t, 1, repeats(k2, attempt(s2)))
+}

handshake/fragment/table.go

@@ -24,6 +24,10 @@ type TableConfig struct {
 	MaxPerSource int           // messages in progress per source
 	MaxMsgLen    int           // largest message accepted
 	TTL          time.Duration // an entry idle this long is dropped
+
+	// RepeatInterval is how often, at most, an answered message is reported as a
+	// repeat on one key. Zero reports every K fresh symbols.
+	RepeatInterval time.Duration
 }
 
 // Table reassembles messages from senders that have not authenticated yet, the
@@ -38,7 +42,10 @@ type TableConfig struct {
 // still arriving for it do not start it over. Without that, the repair
 // fragments trailing a message whose data fragments already completed it would
 // open a new entry that, for K above the repair count, never completes: it holds
-// symbols and a source's quota until it expires.
+// symbols and a source's quota until it expires. A retry of an answered message
+// is reported as a repeat, at most once per RepeatInterval on that key. Keys
+// include the path a message arrives by, so the same message arriving by two
+// paths is two keys with two clocks.
 //
 // A Table is safe for concurrent use.
 type Table[K, S comparable] struct {
@@ -55,13 +62,14 @@ type Table[K, S comparable] struct {
 // indexes have been seen, so a retry's fresh symbols can be told apart from
 // stragglers of an attempt already decoded.
 type answeredEntry struct {
-	hash     [sha256.Size]byte
+	hash       [sha256.Size]byte
-	msgLen   uint16
+	msgLen     uint16
-	k        int
+	k          int
-	seen     [fec.MaxSymbols]bool
+	seen       [fec.MaxSymbols]bool
-	fresh    int
+	fresh      int
-	received int
+	received   int
-	lastSeen time.Time
+	lastSeen   time.Time
+	lastRepeat time.Time // when it was answered, or last reported as a repeat
 }
 
 type tableEntry[K, S comparable] struct {
@@ -107,7 +115,7 @@ func (t *Table[K, S]) Add(key K, source S, now time.Time, h Header, sym []byte)
 	t.mu.Lock()
 	if a, ok := t.answered[key]; ok {
 		if now.Sub(a.lastSeen) <= t.cfg.TTL {
-			res := a.add(now, h, sym)
+			res := a.add(now, h, sym, t.cfg.RepeatInterval)
 			t.mu.Unlock()
 			return res, nil
 		}
@@ -165,7 +173,7 @@ func (t *Table[K, S]) Add(key K, source S, now time.Time, h Header, sym []byte)
 	}, nil
 }
 
-func (a *answeredEntry) add(now time.Time, h Header, sym []byte) *Result {
+func (a *answeredEntry) add(now time.Time, h Header, sym []byte, interval time.Duration) *Result {
 	if h.MsgLen != a.msgLen {
 		return nil
 	}
@@ -176,12 +184,16 @@ func (a *answeredEntry) add(now time.Time, h Header, sym []byte) *Result {
 	}
 	a.seen[h.Index] = true
 	a.fresh++
-	if a.fresh < a.k {
+	if a.fresh < a.k || now.Sub(a.lastRepeat) < interval {
+		// Not enough fresh symbols yet, or this key was answered within the
+		// interval. The fresh symbols keep counting, so the first one to arrive
+		// once the interval is up reports the repeat.
 		return nil
 	}
 	r := &Result{Hash: a.hash, Received: a.received, Repeat: true}
 	a.fresh = 0
 	a.received = 0
+	a.lastRepeat = now
 	return r
 }
 
@@ -205,7 +217,7 @@ func (t *Table[K, S]) Answered(key K, res *Result, now time.Time) {
 			break
 		}
 	}
-	t.answered[key] = &answeredEntry{hash: res.Hash, msgLen: res.msgLen, k: res.k, seen: res.seen, lastSeen: now}
+	t.answered[key] = &answeredEntry{hash: res.Hash, msgLen: res.msgLen, k: res.k, seen: res.seen, lastSeen: now, lastRepeat: now}
 }
 
 // Forget drops what Answered recorded for key, so its fragments are reassembled

handshake_manager.go

@@ -140,6 +140,10 @@ func (hh *HandshakeHostInfo) cachePacket(l *slog.Logger, t header.MessageType, s
 }
 
 func NewHandshakeManager(l *slog.Logger, mainHostMap *HostMap, lightHouse *LightHouse, outside udp.Conn, config HandshakeConfig) *HandshakeManager {
+	// A retrying initiator is answered again at most once per retry interval on
+	// each path its msg1 arrives by.
+	tableCfg := pqTableConfig
+	tableCfg.RepeatInterval = config.tryInterval
 	hm := &HandshakeManager{
 		vpnIps:                 map[netip.Addr]*HandshakeHostInfo{},
 		indexes:                map[uint32]*HandshakeHostInfo{},
@@ -153,7 +157,7 @@ func NewHandshakeManager(l *slog.Logger, mainHostMap *HostMap, lightHouse *Light
 		metricInitiated:        metrics.GetOrRegisterCounter("handshake_manager.initiated", nil),
 		metricTimedOut:         metrics.GetOrRegisterCounter("handshake_manager.timed_out", nil),
 		l:                      l,
-		pqTable:                fragment.NewTable[pqFragKey, pqFragSource](pqTableConfig),
+		pqTable:                fragment.NewTable[pqFragKey, pqFragSource](tableCfg),
 		pqAnswered:             newPQAnswered(hsTimeout(config.retries, config.tryInterval)),
 		hybrid:                 newHybridState(),
 	}
@@ -1247,7 +1251,7 @@ func (hm *HandshakeManager) sendHandshakeResponse(via ViaSender, msg []byte, hos
 
 	if header.IsFragmented(subtype) {
 		fields := append(logFields, "from", via)
-		if err := hm.sendFragmentedResponse(via, msg, hostinfo, cached, received); err != nil {
+		if err := hm.sendFragmentedResponse(via, msg, hostinfo, received); err != nil {
 			f.l.Debug("Did not send handshake message", append(fields, "error", err)...)
 		} else {
 			f.l.Info("Handshake message sent", fields...)

handshake_manager_pq.go

@@ -462,7 +462,7 @@ func (hm *HandshakeManager) handleFragment(via ViaSender, packet []byte, h *head
 			return
 		}
 		if existing := hm.pqAnswered.lookup(res.Hash, time.Now()); existing != nil {
-			// Already answered: resend, rate limited, with fresh symbols.
+			// Already answered: resend with fresh symbols on the path this copy came by.
 			if msg := existing.HandshakePacket[handshakePacketStage2]; msg != nil {
 				hm.sendHandshakeResponse(via, msg, existing, true, res.Received)
 			}
@@ -557,22 +557,15 @@ func (hm *HandshakeManager) sendFragmentedAfter(now, later [][]byte, send func([
 	return first
 }
 
-// sendFragmentedResponse sends the responder's msg2 as fragments. A resend of a
+// sendFragmentedResponse sends the responder's msg2 as fragments, each send
-// cached response (cached) offers fresh symbols and is rate limited to one per
+// offering symbols the hostinfo has not sent before. The clock on resends is
-// retry interval. received is how many bytes the triggering msg1 took, which
+// per path and lives in the fragment table: a copy of msg1 arriving by a second
-// bounds what may be sent back before the address is validated.
+// path (the other address family, a relay) is answered on that path at once,
-func (hm *HandshakeManager) sendFragmentedResponse(via ViaSender, msg []byte, hostinfo *HostInfo, cached bool, received int) error {
+// which carries the handshake when the first path only works one way, and a
-	now := time.Now()
+// retry on a path is reported as a repeat at most once per retry interval.
-	if cached {
+// received is how many bytes the triggering msg1 took, which bounds what may be
-		last := hostinfo.handshakeResendAt.Load()
+// sent back before the address is validated.
-		if now.UnixNano()-last < int64(hm.config.tryInterval) ||
+func (hm *HandshakeManager) sendFragmentedResponse(via ViaSender, msg []byte, hostinfo *HostInfo, received int) error {
-			!hostinfo.handshakeResendAt.CompareAndSwap(last, now.UnixNano()) {
-			return errors.New("handshake response resend rate limited")
-		}
-	} else {
-		hostinfo.handshakeResendAt.Store(now.UnixNano())
-	}
-
 	var h header.H
 	if err := h.Parse(msg); err != nil {
 		return err

handshake_manager_pq_test.go

@@ -106,35 +106,25 @@ func TestSendFragmentedResponse(t *testing.T) {
 	const pktLen = header.Len + fragment.HeaderLen + 1000
 
 	t.Run("first send", func(t *testing.T) {
-		require.NoError(t, hm.sendFragmentedResponse(via, msg, hostinfo, false, 0))
+		require.NoError(t, hm.sendFragmentedResponse(via, msg, hostinfo, 0))
 		assert.Equal(t, []int{0, 1, 2, 3}, fragmentIndexes(t, conn.take(), 77, 5))
 	})
 
-	t.Run("resends are rate limited", func(t *testing.T) {
+	t.Run("a repeat is answered at once with fresh symbols", func(t *testing.T) {
-		require.Error(t, hm.sendFragmentedResponse(via, msg, hostinfo, true, 10*pktLen))
+		// No clock gates a resend: a repeat already cost the initiator K fresh
-		assert.Empty(t, conn.take())
+		// symbols on that path, and the amplification budget bounds the answer.
-	})
+		require.NoError(t, hm.sendFragmentedResponse(via, msg, hostinfo, 10*pktLen))
-
-	allowResend := func() {
-		hostinfo.handshakeResendAt.Store(time.Now().Add(-time.Second).UnixNano())
-	}
-
-	t.Run("resend offers fresh symbols", func(t *testing.T) {
-		allowResend()
-		require.NoError(t, hm.sendFragmentedResponse(via, msg, hostinfo, true, 10*pktLen))
 		assert.Equal(t, []int{4, 5, 6, 7}, fragmentIndexes(t, conn.take(), 77, 5))
 	})
 
 	t.Run("amplification budget trims repair", func(t *testing.T) {
-		allowResend()
 		// 3x of a little over one packet allows 3 packets: both data and 1 repair.
-		require.NoError(t, hm.sendFragmentedResponse(via, msg, hostinfo, true, pktLen+10))
+		require.NoError(t, hm.sendFragmentedResponse(via, msg, hostinfo, pktLen+10))
 		assert.Equal(t, []int{8, 9, 10}, fragmentIndexes(t, conn.take(), 77, 5))
 	})
 
 	t.Run("amplification budget below the data refuses", func(t *testing.T) {
-		allowResend()
+		require.Error(t, hm.sendFragmentedResponse(via, msg, hostinfo, pktLen/2))
-		require.Error(t, hm.sendFragmentedResponse(via, msg, hostinfo, true, pktLen/2))
 		assert.Empty(t, conn.take())
 	})
 }

hostmap.go

@@ -261,8 +261,9 @@ type HostInfo struct {
 	HandshakePacket map[uint8][]byte
 
 	// A responder resending a fragmented handshake response offers symbols it has
-	// not sent before, at most once per retry interval.
+	// not sent before. Once complete, a hybrid responder confirms again for a
-	handshakeResendAt   atomic.Int64  // unix nanos of the last send
+	// resent msg3 at most once per retry interval.
+	handshakeResendAt   atomic.Int64  // unix nanos of the last confirmation
 	handshakeNextSymbol atomic.Uint32 // next fragment symbol index to offer
 
 	// nextLHQuery is the earliest we can ask the lighthouse for new information.