diff --git a/docs/pq-handshake.md b/docs/pq-handshake.md index adbfb1c1..bbb0ae82 100644 --- a/docs/pq-handshake.md +++ b/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 - msg2 when a duplicate msg1 completes) are rate limited to one per handshake per retry - interval, to avoid TFTP's Sorcerer's Apprentice amplification. +- Only timers cause transmission, with one reactive exception: the responder resends its + cached msg2 when a copy of msg1 completes again. An initiator sends msg1 to every address + 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 - amplification budget. + msg2 symbols on the path the repeat came by and is held to the 3x amplification + 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`. diff --git a/e2e/pq_hybrid_test.go b/e2e/pq_hybrid_test.go index f3a1949e..694bd9e9 100644 --- a/e2e/pq_hybrid_test.go +++ b/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() +} diff --git a/handshake/fragment/fragment_test.go b/handshake/fragment/fragment_test.go index e606af5a..a71a7445 100644 --- a/handshake/fragment/fragment_test.go +++ b/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))) +} diff --git a/handshake/fragment/table.go b/handshake/fragment/table.go index 5af147eb..565861a4 100644 --- a/handshake/fragment/table.go +++ b/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 - msgLen uint16 - k int - seen [fec.MaxSymbols]bool - fresh int - received int - lastSeen time.Time + hash [sha256.Size]byte + msgLen uint16 + k int + seen [fec.MaxSymbols]bool + fresh int + received int + 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 diff --git a/handshake_manager.go b/handshake_manager.go index 96704ad7..2d7a3d87 100644 --- a/handshake_manager.go +++ b/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...) diff --git a/handshake_manager_pq.go b/handshake_manager_pq.go index a4370d29..f6beee03 100644 --- a/handshake_manager_pq.go +++ b/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 -// cached response (cached) offers fresh symbols and is rate limited to one per -// retry interval. received is how many bytes the triggering msg1 took, which -// bounds what may be sent back before the address is validated. -func (hm *HandshakeManager) sendFragmentedResponse(via ViaSender, msg []byte, hostinfo *HostInfo, cached bool, received int) error { - now := time.Now() - if cached { - last := hostinfo.handshakeResendAt.Load() - if now.UnixNano()-last < int64(hm.config.tryInterval) || - !hostinfo.handshakeResendAt.CompareAndSwap(last, now.UnixNano()) { - return errors.New("handshake response resend rate limited") - } - } else { - hostinfo.handshakeResendAt.Store(now.UnixNano()) - } - +// sendFragmentedResponse sends the responder's msg2 as fragments, each send +// offering symbols the hostinfo has not sent before. The clock on resends is +// per path and lives in the fragment table: a copy of msg1 arriving by a second +// path (the other address family, a relay) is answered on that path at once, +// which carries the handshake when the first path only works one way, and a +// retry on a path is reported as a repeat at most once per retry interval. +// received is how many bytes the triggering msg1 took, which bounds what may be +// sent back before the address is validated. +func (hm *HandshakeManager) sendFragmentedResponse(via ViaSender, msg []byte, hostinfo *HostInfo, received int) error { var h header.H if err := h.Parse(msg); err != nil { return err diff --git a/handshake_manager_pq_test.go b/handshake_manager_pq_test.go index 8da03755..f50262aa 100644 --- a/handshake_manager_pq_test.go +++ b/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) { - require.Error(t, hm.sendFragmentedResponse(via, msg, hostinfo, true, 10*pktLen)) - assert.Empty(t, conn.take()) - }) - - 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)) + t.Run("a repeat is answered at once with fresh symbols", func(t *testing.T) { + // No clock gates a resend: a repeat already cost the initiator K fresh + // symbols on that path, and the amplification budget bounds the answer. + require.NoError(t, hm.sendFragmentedResponse(via, msg, hostinfo, 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, true, pktLen/2)) + require.Error(t, hm.sendFragmentedResponse(via, msg, hostinfo, pktLen/2)) assert.Empty(t, conn.take()) }) } diff --git a/hostmap.go b/hostmap.go index 7427053a..613e5f70 100644 --- a/hostmap.go +++ b/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. - handshakeResendAt atomic.Int64 // unix nanos of the last send + // not sent before. Once complete, a hybrid responder confirms again for a + // 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.