diff --git a/docs/pq-handshake.md b/docs/pq-handshake.md new file mode 100644 index 00000000..047aef88 --- /dev/null +++ b/docs/pq-handshake.md @@ -0,0 +1,274 @@ +# Post-quantum handshakes + +Status: draft. Decisions below were made on 2026-10-13; open items are listed at the end. +Phase 1 is implemented (see [Implementation](#implementation-phase-1)). + +## Goals + +1. **Confidentiality against harvest-now-decrypt-later.** Traffic recorded today must stay + secret after a cryptographically relevant quantum computer exists. This needs a + post-quantum key exchange now. +2. **Post-quantum authentication, later.** Needs certificates whose CA signature and host + keys are post-quantum. This requires reissuing certificates, so it ships second. +3. **Never rely on IP fragmentation.** ML-KEM and ML-DSA do not fit in one packet, so the + handshake must carry messages larger than the path MTU itself. + +Non-goals: identity hiding beyond what IX offers today, and signatures at handshake time. +The only signature in the system stays the CA signature inside the certificate. + +## Primitives and sizes + +| Primitive | Size | +|---|---| +| ML-KEM-1024 encapsulation key | 1568 B | +| ML-KEM-1024 ciphertext | 1568 B | +| ML-DSA-87 public key / signature | 2592 B / 4627 B | +| Today's IX msg1 / msg2 (v2 cert, 151 B) | 252 B / 287 B | + +`crypto/mlkem` is available in Go 1.26 (the current `go.mod`); `crypto/mldsa` arrives in +Go 1.27 and is only needed for phase 2. + +## Phase 1: ephemeral ML-KEM on today's certificates + +A new handshake subtype, `ix_psk2_mlkem1024` (2), IX-shaped (2 messages; the responder +completes on msg1 as today), using the flynn/noise `psk2` modifier. The Noise protocol name +becomes `Noise_IXpsk2_...`, which separates it from the classical subtype. + +- **msg1**: the initiator generates a fresh ML-KEM-1024 key pair and places the + encapsulation key in `NebulaHandshakeDetails.KEMKey` (field 9). msg1's payload is mixed + into the Noise transcript, so it cannot be stripped or replaced without msg2 failing. + With the psk modifier Noise also mixes `e` into the cipher key, so msg1's static key and + payload are encrypted, though only under a key derived from the public ephemeral. +- **Responder**: after `ReadMessage(msg1)` it encapsulates to that key, obtaining + `(ct, ss)`, and calls `HandshakeState.SetPresharedKey(psk)` before writing msg2, with + `psk = HKDF-Extract(SHA-256, salt = "nebula ix_psk2_mlkem1024 v1", ikm = ss || ct)`. + The ciphertext is bound in because, unlike the encapsulation key, it is not part of the + Noise transcript. +- **msg2** on the wire is `ct || noise_msg2`. The ciphertext travels outside the Noise + message because the initiator needs `ss` before processing msg2's `psk` token. Any change + to `ct` yields a different `ss` (ML-KEM implicit rejection) and the handshake fails. + +Transport keys depend on both the X25519 DH outputs and the ML-KEM shared secret. +Authentication is unchanged and classical. With a minimal v2 certificate msg1 and msg2 are +about 1.84 KB each, so each travels as 2 data and 2 repair fragments of about 936 bytes. +A KEM key on the classical subtype, or a missing or malformed one on this subtype, is a fatal +protocol error; a forged ciphertext only fails that packet. + +## Phase 2: pqIX hybrid with certificate v3 + +Certificate v3 is hybrid: the classical static key and CA signature plus a static +ML-KEM-1024 key and an ML-DSA-87 CA signature. Host authentication is by KEM rather than by +signature, following PQNoise's pqIX, while keeping the classical DH tokens: + +- **msg1** (I to R): `e, s`, ephemeral ML-KEM key, initiator cert. +- **msg2** (R to I): `e, s`, ciphertext to I's ephemeral KEM key, ciphertext to I's static + KEM key, responder cert. +- **msg3** (I to R): ciphertext to R's static KEM key, plus first data. R authenticates I + when msg3 decrypts; I authenticates R implicitly when R's first reply decrypts. + +The initiator never caches the responder's certificate, so every handshake has three +messages and carries both full certificates. With ML-DSA-87: + +| Message | Size | Data fragments (k) | +|---|---|---| +| msg1 | ~8.2 KB | 7 | +| msg2 | ~9.8 KB | 9 | +| msg3 | ~1.6 KB | 2 | + +flynn/noise cannot mix multiple KEM secrets into the chaining key, so phase 2 needs a +KEM-aware Noise symmetric state written in this repository. Both subtypes coexist. + +Cryptographic review of the hybrid composition will be done by in-house cryptographers. + +## Fragment transport + +### Requirements + +- No IP fragmentation: every datagram's UDP payload is at most 1232 bytes **after** relay + wrapping (16 B header + 16 B tag), the IPv6 minimum MTU budget also used by DNS Flag Day + 2020 and Solana's Turbine. +- Every fragment of a handshake uses the same 5-tuple. +- Bounded pre-authentication state, bounded amplification, selective loss recovery, + tolerance of reordering and duplication, and a strictly validated, fuzzable parser. + +### Wire format + +Every packet of a fragmented handshake subtype is: + +``` +Nebula header (16) Version | Type=Handshake | Subtype | Reserved=0 | RemoteIndex | Counter +Fragment header (8) FragID u32 | MsgLen u16 | Index u8 | Flags u8 +Symbol SymbolSize bytes +``` + +- `Counter` is the Noise message number (1, 2, 3). `RemoteIndex` is 0 for msg1, as today. +- `FragID` is the sender's local index (random, non-zero). +- `MsgLen` is the length of the whole message. Both ends derive the layout from it: + `K = ceil(MsgLen / 1176)`, `SymbolSize = ceil(MsgLen / K)`. Every symbol has the same + size, so a packet is at most 16 + 8 + 1176 = 1200 bytes. +- `Index < K` is a data symbol (a slice of the message, the last one zero padded); + `Index >= K` is a repair symbol. +- `Flags` is reserved for a cookie extension and must be zero. + +### Forward error correction + +Messages are protected with a systematic Reed-Solomon code over GF(2^8), implemented in +`handshake/fec`: + +- The repair rows form a Cauchy matrix, `coefficient(row, col) = 1 / (row XOR col)` with + `row >= K > col`, so every square submatrix is invertible and **any K distinct symbols + reconstruct the message**. Naive identity-over-Vandermonde constructions are not MDS + (the PAR1 flaw and the 1997 Plank tutorial erratum). +- A repair row depends only on its index, so repair symbols are generated on demand and + their indexes never need to be agreed in advance. Up to 256 symbols exist per message. +- Known-answer vectors in `fec_test.go` pin the encoding and were cross-checked against an + independent implementation; changing them is a wire format change. +- Cost for a 9.8 KB message (k=9): about 13 µs for three repair symbols, about 19 µs to + decode with three data symbols missing. + +**The sender chooses m**, the number of repair symbols per attempt. The receiver only needs +any K symbols, so this is policy and can change without a protocol change. The starting +rule is `m = max(2, ceil(K/3))`. + +**Retries send fresh symbols.** Attempt 0 sends indexes `0 .. K+m-1`; each retry sends +the next `K+m` unused indexes. The receiver merges all distinct symbols across attempts, +so no received fragment is wasted. Indexes are a deterministic function of the message, +so retransmits stay idempotent: the responder caches msg2 itself and regenerates any +symbol. + +**Repair symbols trail the data.** Data symbols go out back to back; repair symbols follow +after a short delay, so a queue tail-drop at the end of the data run does not also take the +repair. In simulation this lifts small messages substantially (k=2 at 10% loss, burst 8: +91.4% to 98.2% first-attempt success). Pacing every fragment was also evaluated; it gains +about 5 more points for k=7..9 on bursty paths, which is not worth the extra machinery, +and the retry with fresh symbols covers the remainder. + +### Receiver rules + +Implemented in `handshake/fragment` (`Sender`, `Assembly`, and the capped `Table`). + +- Fragments are stored individually as they arrive and charged against a memory cap. + Nothing is allocated from a claimed size alone (FragmentSmack, game-engine "fragment + bombs"). +- `MsgLen` must be identical for every fragment of a `FragID`; a mismatch drops the whole + entry. msg1 is capped at `MsgLen <= 16384` (K <= 14). +- The decoder keeps at most K distinct symbols, so an entry never exceeds about MsgLen bytes + however many repair symbols arrive. +- When K symbols are present the message is decoded and handed to the handshake `Machine` + as a synthetic `[header | message]`. The Machine never sees fragments. +- A completed entry is forgotten immediately. A late duplicate fragment starts a new entry + and, once enough arrive, reconstructs the same message again; the handshake layer's + duplicate detection recognises it. +- If Noise rejects a decoded message, the entry is discarded and the next attempt + supplies fresh symbols. (Possible refinement: keep a few spare symbols so an injected + bad symbol can be identified by re-encoding instead of waiting for a retry.) +- Pre-authentication msg1 entries live in a table keyed by (source address or relay index, + FragID) with a total memory cap, per-source quotas (/32 for IPv4, /64 for IPv6), a + sliding TTL of about 2 s refreshed by progress, and random eviction under pressure. + msg2 and msg3 reassembly attaches to the existing pending handshake found by RemoteIndex. + +### 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. +- 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 + it (QUIC's rule). Normal handshakes sit near 1.1-1.2x. +- In phase 2 the initiator retransmits msg3 until any authenticated packet arrives from + the responder; the responder sends a small confirmation on completion and holds tunnel + traffic until msg3 verifies. + +### Denial of service + +The responder can no longer complete statelessly on msg1, since msg1 arrives in pieces. +Defenses are the capped table above plus a **cookie under load**: when the table is under +pressure, a fragment from an unvalidated address is answered with a stateless cookie +(a MAC over source address, FragID and a time bucket), and the initiator resends with the +cookie. This costs a round trip only while under attack, in the style of WireGuard's +mac2, QUIC Retry and the SCTP state cookie. A CA-keyed pre-filter MAC was considered and +rejected: networks commonly trust many CAs. + +## Rollout + +- Separate settings for initiating and accepting PQ handshakes, both off by default: + `handshakes.pq.initiate`, `handshakes.pq.accept`. `handshakes.pq.repair_delay` (default + 10ms) sets how long repair fragments trail the data fragments. +- Old nodes silently drop unknown subtypes, so discovery is speculative: an initiator with + `initiate` enabled alternates PQ and classical attempts on successive retries, keeping + both pending, and the first to complete wins. +- `handshakes.pq.sticky` (default off): once a peer has completed a PQ handshake, never fall + back to classical for it again during this process lifetime. This resists active + downgrade after first contact. +- No lighthouse capability advertisement. + +## Simulation results + +Gilbert-Elliott channel (bad state drops everything), losses expressed as average rate and +mean burst length in packets. First-attempt success for the whole handshake with +`m = max(2, ceil(K/3))` and repair delayed 16 packet slots, compared with no FEC: + +| Handshake | Loss | Burst 1 | Burst 2 | Burst 4 | Burst 8 | +|---|---|---|---|---|---| +| phase 1 (2, 2) no FEC | 5% | 81.4% | 85.6% | 87.9% | 89.0% | +| phase 1 (2, 2) FEC | 5% | 99.9% | 99.4% | 99.3% | 98.6% | +| phase 1 (2, 2) FEC | 10% | 99.3% | 97.5% | 97.4% | 96.3% | +| phase 2 (7, 9, 2) no FEC | 5% | 39.8% | 57.5% | 70.4% | 77.7% | +| phase 2 (7, 9, 2) FEC | 5% | 99.6% | 94.2% | 90.4% | 88.8% | +| phase 2 (7, 9, 2) FEC | 10% | 95.9% | 84.6% | 79.0% | 77.3% | + +With fresh repair symbols, phase 2 completes within two attempts at least 99.1% of the time +in every scenario above, and within three at least 99.99%. + +## Prior art + +- **WireGuard / Rosenpass**: fit every message in one packet; cookies and biscuits for + statelessness. Not possible at our sizes. +- **IKEv2**: RFC 7383 fragments only after keys exist, with each fragment authenticated; + RFC 9242/9370 move PQ key exchange into intermediate exchanges for that reason. +- **QUIC / DTLS 1.3**: offset-based reassembly with ACKs; QUIC's 3x amplification limit and + 1200-byte floor. +- **Solana Turbine**: Reed-Solomon over 1232-byte UDP packets through untrusted relays; + moved to Merkle-authenticated shreds after unauthenticated repair symbols enabled + poisoning. +- **kcp-go, UDPspeeder, CM256**: RS FEC over UDP; parity must scale up for small groups. +- **gQUIC FEC (removed)**: FEC does not pay for bulk transfer under bursty loss; it does + for short, latency-critical transfers such as handshakes. +- **TFTP, NFS duplicate request cache, memcached UDP, IP reassembly bugs**: retransmit + storms, ID-wrap misassembly, reflection amplification, and allocation from claimed sizes. + +## Implementation (phase 1) + +- `handshake`: the `ix_psk2_mlkem1024` subtype (`patterns.go`, `kem.go`), the `KEMKey` + payload field, and the psk2 plumbing in `Machine`. +- `handshake/fec`, `handshake/fragment`: the code and the fragment transport. +- `handshake_manager_pq.go` and the manager: + - **Initiator.** Attempts alternate PQ and classical, starting with PQ (sticky peers get + only PQ). The PQ attempt has its own local index registered alongside the classical + one; whichever response completes first wins, and the other index is released + (`settleIndex`). A timed out or deleted pending handshake releases both. + - **Responder.** msg1 fragments go into the capped pre-authentication table. A decoded + msg1 is looked up by hash among recently answered ones before any Noise, KEM or + 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. + - **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. +- Tests: Noise-level KEM 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 + 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, and index cleanup on completion, fallback and timeout. ML-KEM and HKDF also + pass under `GOEXPERIMENT=boringcrypto` and `GODEBUG=fips140=on`. + +## Open items + +1. Cookie wire format and the load threshold that triggers it. +2. Whether the decoder keeps spare symbols to identify injected symbols. +3. msg3 confirmation message format (phase 2). +4. Configuration reload: the `handshakes.pq` settings are read at startup only, like the + rest of the `handshakes` section. diff --git a/e2e/pq_handshakes_test.go b/e2e/pq_handshakes_test.go new file mode 100644 index 00000000..0519e38f --- /dev/null +++ b/e2e/pq_handshakes_test.go @@ -0,0 +1,351 @@ +//go:build e2e_testing + +package e2e + +import ( + "net/netip" + "sync" + "testing" + "time" + + "github.com/slackhq/nebula" + "github.com/slackhq/nebula/cert" + "github.com/slackhq/nebula/cert_test" + "github.com/slackhq/nebula/e2e/router" + "github.com/slackhq/nebula/handshake/fragment" + "github.com/slackhq/nebula/header" + "github.com/slackhq/nebula/udp" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func pqConfig(initiate, accept, sticky bool) m { + return m{"handshakes": m{"pq": m{ + "initiate": initiate, + "accept": accept, + "sticky": sticky, + "repair_delay": "0s", + }}} +} + +// handshakeOf returns the handshake header p carries, looking inside relay +// wrapping, plus the fragment header when the handshake is fragmented. +func handshakeOf(p *udp.Packet) (*header.H, *fragment.Header) { + b := p.Data + h := &header.H{} + if h.Parse(b) != nil { + return nil, nil + } + if h.Type == header.Message && h.Subtype == header.MessageRelay { + // Relays authenticate but do not encrypt the inner packet. + b = b[header.Len:] + if h.Parse(b) != nil { + return nil, nil + } + } + if h.Type != header.Handshake { + return nil, nil + } + if !header.IsFragmented(h.Subtype) { + return h, nil + } + // A relayed packet still has the relay's AEAD tag on the end. + for _, tag := range []int{0, 16} { + if len(b) < header.Len+tag { + break + } + if fh, _, err := fragment.Parse(b[header.Len : len(b)-tag]); err == nil { + return h, &fh + } + } + return h, nil +} + +// handshakeTally counts handshake packets by style as the router moves them. +type handshakeTally struct { + sync.Mutex + pq, classical int + maxPQLen int +} + +func (ht *handshakeTally) observe(p *udp.Packet) { + h, _ := handshakeOf(p) + if h == nil { + return + } + ht.Lock() + defer ht.Unlock() + if h.Subtype == header.HandshakeIXPSK2MLKEM1024 { + ht.pq++ + ht.maxPQLen = max(ht.maxPQLen, len(p.Data)) + } else { + ht.classical++ + } +} + +// isTunnelData reports whether p is ordinary tunnel traffic headed for to, +// directly or as the last hop of a relay. +func isTunnelData(p *udp.Packet, receiver, to *nebula.Control) bool { + h := &header.H{} + if receiver != to || h.Parse(p.Data) != nil || h.Type != header.Message { + return false + } + if h.Subtype == header.MessageRelay { + if h.Parse(p.Data[header.Len:]) != nil { + return false + } + } + return h.Type == header.Message && h.Subtype == header.MessageNone +} + +func newPQPair(t *testing.T, mine, theirs m) (myControl, theirControl *nebula.Control, myVpn, theirVpn []netip.Prefix, myUdp, theirUdp netip.AddrPort, r *router.R) { + t.Helper() + ca, _, caKey, _ := cert_test.NewTestCaCert(cert.Version2, cert.Curve_CURVE25519, time.Now(), time.Now().Add(10*time.Minute), nil, nil, []string{}) + myControl, myVpn, myUdp, _ = newSimpleServer(cert.Version2, ca, caKey, "me", "10.128.0.1/24", mine) + theirControl, theirVpn, theirUdp, _ = newSimpleServer(cert.Version2, ca, caKey, "them", "10.128.0.2/24", theirs) + myControl.InjectLightHouseAddr(theirVpn[0].Addr(), theirUdp) + r = router.NewR(t, myControl, theirControl) + myControl.Start() + theirControl.Start() + return +} + +func TestPQHandshake(t *testing.T) { + t.Parallel() + // They only accept, which is all a PQ initiator needs. + myControl, theirControl, myVpn, theirVpn, myUdp, theirUdp, r := newPQPair(t, pqConfig(true, true, false), pqConfig(false, true, false)) + defer r.RenderFlow() + + var tally handshakeTally + myControl.InjectTunPacket(BuildTunUDPPacket(theirVpn[0].Addr(), 80, myVpn[0].Addr(), 80, []byte("Hi from me"))) + require.True(t, r.RouteForAllExitFuncOrTimeout(5*time.Second, func(p *udp.Packet, receiver *nebula.Control) router.ExitType { + tally.observe(p) + if isTunnelData(p, receiver, theirControl) { + return router.RouteAndExit + } + 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.Zero(t, tally.classical, "no classical handshake should be needed") + assert.GreaterOrEqual(t, tally.pq, 4, "both messages travel as several fragments") + assert.LessOrEqual(t, tally.maxPQLen, fragment.MaxPacketLen) + + assertHostInfoPair(t, myUdp, theirUdp, myVpn, theirVpn, myControl, theirControl) + assertTunnel(t, myVpn[0].Addr(), theirVpn[0].Addr(), myControl, theirControl, r) + assert.Empty(t, myControl.ListHostmapIndexes(true), "no pending index may outlive the handshake") + assert.Empty(t, theirControl.ListHostmapIndexes(true)) + + myControl.Stop() + theirControl.Stop() +} + +// TestPQHandshakeLoss drops data fragments of both messages; the repair +// fragments must carry the handshake without a retry. +func TestPQHandshakeLoss(t *testing.T) { + t.Parallel() + myControl, theirControl, myVpn, theirVpn, _, _, r := newPQPair(t, pqConfig(true, true, false), pqConfig(false, true, false)) + defer r.RenderFlow() + + var tally handshakeTally + dropped := 0 + myControl.InjectTunPacket(BuildTunUDPPacket(theirVpn[0].Addr(), 80, myVpn[0].Addr(), 80, []byte("Hi from me"))) + require.True(t, r.RouteForAllExitFuncOrTimeout(5*time.Second, func(p *udp.Packet, receiver *nebula.Control) router.ExitType { + if h, fh := handshakeOf(p); fh != nil { + // Every data fragment of msg1, and the first of msg2. + if (h.MessageCounter == 1 && int(fh.Index) < 2) || (h.MessageCounter == 2 && fh.Index == 0) { + dropped++ + return router.Drop + } + } + tally.observe(p) + if isTunnelData(p, receiver, theirControl) { + return router.RouteAndExit + } + 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.Equal(t, 3, dropped) + assert.Zero(t, tally.classical, "the first attempt should succeed from repair fragments") + assertTunnel(t, myVpn[0].Addr(), theirVpn[0].Addr(), myControl, theirControl, r) + + myControl.Stop() + theirControl.Stop() +} + +// TestPQFallback has a peer that does not accept PQ handshakes. The initiator's +// next attempt is classical and the tunnel comes up anyway. +func TestPQFallback(t *testing.T) { + t.Parallel() + myControl, theirControl, myVpn, theirVpn, myUdp, theirUdp, r := newPQPair(t, pqConfig(true, true, false), nil) + defer r.RenderFlow() + + var tally handshakeTally + myControl.InjectTunPacket(BuildTunUDPPacket(theirVpn[0].Addr(), 80, myVpn[0].Addr(), 80, []byte("Hi from me"))) + require.True(t, r.RouteForAllExitFuncOrTimeout(5*time.Second, func(p *udp.Packet, receiver *nebula.Control) router.ExitType { + tally.observe(p) + if isTunnelData(p, receiver, theirControl) { + return router.RouteAndExit + } + 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.Positive(t, tally.pq, "a PQ attempt comes first") + assert.Positive(t, tally.classical, "then a classical one") + + assertHostInfoPair(t, myUdp, theirUdp, myVpn, theirVpn, myControl, theirControl) + assertTunnel(t, myVpn[0].Addr(), theirVpn[0].Addr(), myControl, theirControl, r) + assert.Empty(t, myControl.ListHostmapIndexes(true), "the losing PQ attempt's index must be released") + + myControl.Stop() + theirControl.Stop() +} + +// TestPQResendFreshSymbols loses the responder's whole first msg2. The +// initiator's next PQ attempt offers fresh msg1 symbols, the responder +// recognises the repeat and resends msg2 with symbols it has not sent before. +func TestPQResendFreshSymbols(t *testing.T) { + t.Parallel() + myControl, theirControl, myVpn, theirVpn, _, _, r := newPQPair(t, pqConfig(true, true, false), pqConfig(false, true, false)) + defer r.RenderFlow() + + var firstMsg2, freshMsg2, freshMsg1 int + myControl.InjectTunPacket(BuildTunUDPPacket(theirVpn[0].Addr(), 80, myVpn[0].Addr(), 80, []byte("Hi from me"))) + require.True(t, r.RouteForAllExitFuncOrTimeout(10*time.Second, func(p *udp.Packet, receiver *nebula.Control) router.ExitType { + h, fh := handshakeOf(p) + switch { + case h != nil && fh == nil: + // Keep the classical fallback out of the way. + return router.Drop + case fh != nil && h.MessageCounter == 2 && fh.Index < 4: + firstMsg2++ + return router.Drop + case fh != nil && h.MessageCounter == 2: + freshMsg2++ + case fh != nil && h.MessageCounter == 1 && fh.Index >= 4: + freshMsg1++ + } + if isTunnelData(p, receiver, theirControl) { + return router.RouteAndExit + } + 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.Equal(t, 4, firstMsg2, "the first msg2 is 2 data and 2 repair fragments") + assert.Positive(t, freshMsg1, "the retry offers fresh msg1 symbols") + assert.Positive(t, freshMsg2, "the resend offers fresh msg2 symbols") + assertTunnel(t, myVpn[0].Addr(), theirVpn[0].Addr(), myControl, theirControl, r) + + myControl.Stop() + theirControl.Stop() +} + +// TestPQSticky checks a sticky initiator never offers a classical handshake to a +// peer it has already completed a PQ handshake with. +func TestPQSticky(t *testing.T) { + t.Parallel() + myControl, theirControl, myVpn, theirVpn, _, _, r := newPQPair(t, pqConfig(true, true, true), pqConfig(false, true, false)) + defer r.RenderFlow() + + myControl.InjectTunPacket(BuildTunUDPPacket(theirVpn[0].Addr(), 80, myVpn[0].Addr(), 80, []byte("Hi from me"))) + require.True(t, r.RouteForAllExitFuncOrTimeout(5*time.Second, func(p *udp.Packet, receiver *nebula.Control) router.ExitType { + if isTunnelData(p, receiver, theirControl) { + return router.RouteAndExit + } + return router.KeepRouting + }), "tunnel never came up") + theirControl.GetFromTun(true) + + t.Log("Rehandshake while an attacker drops every PQ packet") + var tally handshakeTally + myControl.ReHandshake(theirVpn[0].Addr()) + r.RouteForAllExitFuncOrTimeout(time.Second, func(p *udp.Packet, receiver *nebula.Control) router.ExitType { + tally.observe(p) + if h, _ := handshakeOf(p); h != nil && h.Subtype == header.HandshakeIXPSK2MLKEM1024 { + return router.Drop + } + return router.KeepRouting + }) + assert.Positive(t, tally.pq, "PQ attempts keep being made") + assert.Zero(t, tally.classical, "sticky must never fall back") + + myControl.Stop() + theirControl.Stop() +} + +func TestPQRelays(t *testing.T) { + t.Parallel() + ca, _, caKey, _ := cert_test.NewTestCaCert(cert.Version1, cert.Curve_CURVE25519, time.Now(), time.Now().Add(10*time.Minute), nil, nil, []string{}) + useRelays := pqConfig(true, true, false) + useRelays["relay"] = m{"use_relays": true} + amRelay := pqConfig(true, true, false) + amRelay["relay"] = m{"am_relay": true} + myControl, myVpnIpNet, _, _ := newSimpleServer(cert.Version1, ca, caKey, "me ", "10.128.0.1/24", useRelays) + relayControl, relayVpnIpNet, relayUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "relay ", "10.128.0.128/24", amRelay) + theirControl, theirVpnIpNet, theirUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "them ", "10.128.0.2/24", useRelays) + + myControl.InjectLightHouseAddr(relayVpnIpNet[0].Addr(), relayUdpAddr) + myControl.InjectRelays(theirVpnIpNet[0].Addr(), []netip.Addr{relayVpnIpNet[0].Addr()}) + relayControl.InjectLightHouseAddr(theirVpnIpNet[0].Addr(), theirUdpAddr) + + r := router.NewR(t, myControl, relayControl, theirControl) + defer r.RenderFlow() + + myControl.Start() + relayControl.Start() + theirControl.Start() + + var tally handshakeTally + var relayedPQ int + myControl.InjectTunPacket(BuildTunUDPPacket(theirVpnIpNet[0].Addr(), 80, myVpnIpNet[0].Addr(), 80, []byte("Hi from me"))) + require.True(t, r.RouteForAllExitFuncOrTimeout(10*time.Second, func(p *udp.Packet, receiver *nebula.Control) router.ExitType { + tally.observe(p) + outer := &header.H{} + if outer.Parse(p.Data) == nil && outer.Type == header.Message && outer.Subtype == header.MessageRelay { + if h, fh := handshakeOf(p); h != nil && fh != nil { + relayedPQ++ + assert.LessOrEqual(t, len(p.Data), fragment.MaxPacketLen+32, "a relayed fragment fits the IPv6 minimum MTU") + } + } + if isTunnelData(p, receiver, theirControl) { + return router.RouteAndExit + } + return router.KeepRouting + }), "tunnel never came up") + + assertUdpPacket(t, []byte("Hi from me"), theirControl.GetFromTun(true), myVpnIpNet[0].Addr(), theirVpnIpNet[0].Addr(), 80, 80) + assert.Positive(t, relayedPQ, "the end to end handshake is PQ through the relay") + r.RenderHostmaps("Final hostmaps", myControl, relayControl, theirControl) + + myControl.Stop() + relayControl.Stop() + theirControl.Stop() +} + +// TestPQTimeoutReleasesIndexes lets a handshake with both a PQ and a classical +// attempt pending time out, and checks neither index is left behind. +func TestPQTimeoutReleasesIndexes(t *testing.T) { + t.Parallel() + mine := pqConfig(true, true, false) + mine["handshakes"].(m)["retries"] = 2 + myControl, theirControl, myVpn, theirVpn, _, _, r := newPQPair(t, mine, pqConfig(false, true, false)) + defer r.RenderFlow() + + var tally handshakeTally + myControl.InjectTunPacket(BuildTunUDPPacket(theirVpn[0].Addr(), 80, myVpn[0].Addr(), 80, []byte("Hi from me"))) + r.RouteForAllExitFuncOrTimeout(1500*time.Millisecond, func(p *udp.Packet, receiver *nebula.Control) router.ExitType { + tally.observe(p) + return router.Drop + }) + + assert.Positive(t, tally.pq) + assert.Positive(t, tally.classical) + assert.Nil(t, myControl.GetHostInfoByVpnAddr(theirVpn[0].Addr(), true), "the handshake should have timed out") + assert.Empty(t, myControl.ListHostmapIndexes(true)) + + myControl.Stop() + theirControl.Stop() +} diff --git a/examples/config.yml b/examples/config.yml index 0d1a9180..670e3f63 100644 --- a/examples/config.yml +++ b/examples/config.yml @@ -418,6 +418,21 @@ logging: # after receiving the response for lighthouse queries #trigger_buffer: 64 + # pq enables handshakes that mix an ephemeral ML-KEM-1024 key exchange into the session keys, protecting + # recorded traffic against a future quantum computer. Authentication is unchanged. These handshakes are too + # large for one packet and travel as fragments with forward error correction. + #pq: + # initiate starts post-quantum handshakes. Attempts alternate between post-quantum and classical until one is + # answered, so peers that do not support them still connect. + #initiate: false + # accept answers post-quantum handshakes from peers. + #accept: false + # sticky never falls back to a classical handshake with a peer that has completed a post-quantum one since + # nebula started, which defeats an attacker who drops post-quantum handshakes to force a downgrade. + #sticky: false + # repair_delay is how long the forward error correction fragments trail the rest of a handshake message. + #repair_delay: 10ms + # Tunnel manager settings #tunnels: # drop_inactive controls whether inactive tunnels are maintained or dropped after the inactive_timeout period has diff --git a/handshake_manager.go b/handshake_manager.go index ff80693c..76916725 100644 --- a/handshake_manager.go +++ b/handshake_manager.go @@ -15,6 +15,7 @@ import ( "github.com/rcrowley/go-metrics" "github.com/slackhq/nebula/cert" "github.com/slackhq/nebula/handshake" + "github.com/slackhq/nebula/handshake/fragment" "github.com/slackhq/nebula/header" "github.com/slackhq/nebula/udp" ) @@ -49,6 +50,7 @@ type HandshakeConfig struct { tryInterval time.Duration retries int64 triggerBuffer int + pq pqHandshakeConfig messageMetrics *MessageMetrics } @@ -74,6 +76,13 @@ type HandshakeManager struct { // can be used to trigger outbound handshake for the given vpnIp trigger chan netip.Addr + + // pqTable reassembles post-quantum msg1s from senders that have not authenticated yet. + pqTable *fragment.Table[pqFragKey, pqFragSource] + // pqPeers holds vpn addresses that completed a post-quantum handshake, for sticky mode. + pqPeers sync.Map + // pqAnswered recognises repeats of post-quantum msg1s already answered. + pqAnswered *pqAnswered } type HandshakeHostInfo struct { @@ -89,6 +98,7 @@ type HandshakeHostInfo struct { hostinfo *HostInfo machine *handshake.Machine // The handshake state machine, set during stage 0 (initiator) or beginHandshake (responder multi-message) + pq *pqAttempt // The initiator's post-quantum attempt, pending alongside the classical one } func (hh *HandshakeHostInfo) cachePacket(l *slog.Logger, t header.MessageType, st header.MessageSubType, packet []byte, f packetCallback, m *cachedPacketMetrics) { @@ -130,6 +140,8 @@ 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), + pqAnswered: newPQAnswered(hsTimeout(config.retries, config.tryInterval)), } } @@ -145,6 +157,8 @@ func (hm *HandshakeManager) Run(ctx context.Context) { hm.handleOutbound(vpnIP, true) case now := <-clockSource.C: hm.NextOutboundHandshakeTimerTick(now) + hm.pqTable.Expire(now) + hm.pqAnswered.expire(now) } } } @@ -154,7 +168,7 @@ func (hm *HandshakeManager) HandleIncoming(via ViaSender, packet []byte, h *head // don't yet support) are dropped here rather than silently routed through // the IX path. Add a case when introducing a new pattern. switch h.Subtype { - case header.HandshakeIXPSK0: + case header.HandshakeIXPSK0, pqSubtype: // supported default: hm.l.Debug("dropping handshake with unsupported subtype", @@ -170,6 +184,11 @@ func (hm *HandshakeManager) HandleIncoming(via ViaSender, packet []byte, h *head } } + if header.IsFragmented(h.Subtype) { + hm.handleFragment(via, packet, h) + return + } + // First message of a new handshake. The wire format requires RemoteIndex // to be zero here (the initiator has no responder index to fill in yet), // and generateIndex never allocates 0, so any non-zero RemoteIndex on a @@ -181,7 +200,7 @@ func (hm *HandshakeManager) HandleIncoming(via ViaSender, packet []byte, h *head "from", via, "remoteIndex", h.RemoteIndex) return } - hm.beginHandshake(via, packet, h) + hm.beginHandshake(via, packet, h, 0) return } @@ -230,6 +249,9 @@ func (hm *HandshakeManager) handleOutbound(vpnIp netip.Addr, lighthouseTriggered "style": header.SubTypeName(header.Handshake, hh.machine.Subtype()), }) } + if hh.pq != nil { + fields = append(fields, "pqInitiatorIndex", hh.pq.index) + } hh.hostinfo.logger(hm.l).Info("Handshake timed out", fields...) hm.metricTimedOut.Inc(1) hm.DeleteHostInfo(hostinfo) @@ -239,25 +261,44 @@ func (hm *HandshakeManager) handleOutbound(vpnIp netip.Addr, lighthouseTriggered // Increment the counter to increase our delay, linear backoff hh.counter++ - // Check if we have a handshake packet to transmit yet - if !hh.ready { - if !hm.buildStage0Packet(hh) { + // Pick this attempt's style and make sure its first message exists. A PQ + // attempt sends msg1 as fragments: now goes out immediately, later after the + // repair delay, and every attempt offers fresh repair symbols. + var now, later [][]byte + var machine *handshake.Machine + if hm.usePQ(hh, vpnIp) { + if hh.pq == nil && !hm.buildPQAttempt(hh) { hm.OutboundHandshakeTimer.Add(vpnIp, hm.config.tryInterval*time.Duration(hh.counter)) return } + var err error + now, later, err = hh.pq.sender.Attempt(fragment.DefaultRepair(hh.pq.sender.Layout().K)) + if err != nil { + hostinfo.logger(hm.l).Error("Failed to fragment handshake", "error", err) + hm.OutboundHandshakeTimer.Add(vpnIp, hm.config.tryInterval*time.Duration(hh.counter)) + return + } + machine = hh.pq.machine + } else { + if !hh.ready { + if !hm.buildStage0Packet(hh) { + hm.OutboundHandshakeTimer.Add(vpnIp, hm.config.tryInterval*time.Duration(hh.counter)) + return + } + } + // TODO: this hardcodes "always retransmit stage 0", which is correct for + // IX (the initiator only ever sends one packet, msg1) but wrong the + // moment a 3+ message pattern lands. The retry loop should resend the + // most recent outgoing message, not always stage 0. That implies + // HandshakeHostInfo tracking a single "currentOutbound" packet (bytes + + // header metadata) that gets replaced as the handshake progresses, + // instead of indexing into HandshakePacket. + now = [][]byte{hostinfo.HandshakePacket[handshakePacketStage0]} + machine = hh.machine } - - // TODO: this hardcodes "always retransmit stage 0", which is correct for - // IX (the initiator only ever sends one packet, msg1) but wrong the - // moment a 3+ message pattern lands. The retry loop should resend the - // most recent outgoing message, not always stage 0. That implies - // HandshakeHostInfo tracking a single "currentOutbound" packet (bytes + - // header metadata) that gets replaced as the handshake progresses, - // instead of indexing into HandshakePacket. - stage0 := hostinfo.HandshakePacket[handshakePacketStage0] hsFields := m{ - "stage": uint64(hh.machine.MessageIndex()), - "style": header.SubTypeName(header.Handshake, hh.machine.Subtype()), + "stage": uint64(machine.MessageIndex()), + "style": header.SubTypeName(header.Handshake, machine.Subtype()), } remotes := hostinfo.remotes.CopyAddrs(hm.mainHostMap.GetPreferredRanges()) @@ -285,8 +326,10 @@ func (hm *HandshakeManager) handleOutbound(vpnIp netip.Addr, lighthouseTriggered // Send the handshake to all known ips, stage 2 takes care of assigning the hostinfo.remote based on the first to reply var sentTo []netip.AddrPort hostinfo.remotes.ForEach(hm.mainHostMap.GetPreferredRanges(), func(addr netip.AddrPort, _ bool) { - hm.messageMetrics.Tx(header.Handshake, hh.machine.Subtype(), 1) - err := hm.outside.WriteTo(stage0, addr) + hm.messageMetrics.Tx(header.Handshake, machine.Subtype(), int64(len(now)+len(later))) + err := hm.sendFragmented(now, later, func(p []byte) error { + return hm.outside.WriteTo(p, addr) + }) if err != nil { // These repeat every attempt, so match the success log below and only shout when the remotes changed level := slog.LevelDebug @@ -322,7 +365,7 @@ func (hm *HandshakeManager) handleOutbound(vpnIp netip.Addr, lighthouseTriggered ) } - hm.f.relayManager.StartRelays(hm.f, vpnIp, hh, stage0) + hm.f.relayManager.StartRelays(hm.f, vpnIp, hh, now, later) // If a lighthouse triggered this attempt then we are still in the timer wheel and do not need to re-add if !lighthouseTriggered { @@ -532,6 +575,10 @@ func (hm *HandshakeManager) DeleteHostInfo(hostinfo *HostInfo) { func (hm *HandshakeManager) unlockedDeleteHostInfo(hostinfo *HostInfo) { for _, addr := range hostinfo.vpnAddrs { if cur, ok := hm.vpnIps[addr]; ok && cur.hostinfo == hostinfo { + // A PQ attempt holds its own index alongside the hostinfo's. + if idx := cur.pqIndex(); idx != 0 && hm.indexes[idx] == cur { + delete(hm.indexes, idx) + } delete(hm.vpnIps, addr) } } @@ -642,9 +689,8 @@ func hsTimeout(tries int64, interval time.Duration) time.Duration { return time.Duration(tries / 2 * ((2 * int64(interval)) + (tries-1)*int64(interval))) } -// buildStage0Packet creates the initial handshake packet for the initiator. -func (hm *HandshakeManager) buildStage0Packet(hh *HandshakeHostInfo) bool { - cs := hm.f.pki.getCertState() +// initiatingVersion picks the certificate version to initiate a handshake with. +func (hm *HandshakeManager) initiatingVersion(hh *HandshakeHostInfo, cs *CertState) cert.Version { v := cs.DefaultVersion() if hh.initiatingVersionOverride != cert.VersionPre1 { v = hh.initiatingVersionOverride @@ -656,6 +702,13 @@ func (hm *HandshakeManager) buildStage0Packet(hh *HandshakeHostInfo) bool { } } } + return v +} + +// buildStage0Packet creates the initial handshake packet for the initiator. +func (hm *HandshakeManager) buildStage0Packet(hh *HandshakeHostInfo) bool { + cs := hm.f.pki.getCertState() + v := hm.initiatingVersion(hh, cs) cred := cs.GetCredential(v) if cred == nil { @@ -694,7 +747,10 @@ func (hm *HandshakeManager) buildStage0Packet(hh *HandshakeHostInfo) bool { // beginHandshake handles an incoming handshake packet that doesn't match any // existing pending handshake. It creates a new responder Machine and processes // the first message. -func (hm *HandshakeManager) beginHandshake(via ViaSender, packet []byte, h *header.H) { +// +// received is the number of bytes a fragmented msg1 arrived in, which bounds the +// response sent back before the address is validated; 0 for unfragmented ones. +func (hm *HandshakeManager) beginHandshake(via ViaSender, packet []byte, h *header.H, received int) { f := hm.f cs := f.pki.getCertState() @@ -708,7 +764,7 @@ func (hm *HandshakeManager) beginHandshake(via ViaSender, packet []byte, h *head machine, err := handshake.NewMachine( v, cs.GetCredential, hm.certVerifier(), func() (uint32, error) { return generateIndex(f.l) }, - false, header.HandshakeIXPSK0, + false, h.Subtype, ) if err != nil { f.l.Error("Failed to create handshake machine", "from", via, "error", err) @@ -798,11 +854,15 @@ func (hm *HandshakeManager) beginHandshake(via ViaSender, packet []byte, h *head existing, err := hm.CheckAndComplete(hostinfo, handshakePacketStage0, f) if err != nil { - hm.handleCheckAndCompleteError(err, existing, hostinfo, via) + hm.handleCheckAndCompleteError(err, existing, hostinfo, via, machine.Subtype(), received) return } - hm.sendHandshakeResponse(via, response, hostinfo, false) + hm.sendHandshakeResponse(via, response, hostinfo, false, received) + if header.IsFragmented(machine.Subtype()) { + hm.pqAnswered.add(packet[header.Len:], hostinfo, time.Now()) + } + hm.notePQPeer(machine.Subtype(), vpnAddrs) hostinfo.remotes.RefreshFromHandshake(vpnAddrs) // Don't wait for UpdateWorker @@ -813,23 +873,32 @@ func (hm *HandshakeManager) beginHandshake(via ViaSender, packet []byte, h *head // continueHandshake feeds an incoming packet to an existing pending handshake Machine. func (hm *HandshakeManager) continueHandshake(via ViaSender, hh *HandshakeHostInfo, packet []byte) { - f := hm.f - hh.Lock() defer hh.Unlock() - // Re-verify hh is still tracked. Between queryIndex returning and us taking - // hh.Lock, handleOutbound may have timed out and deleted it. Once we hold - // hh.Lock no other deleter can race our index: handleOutbound also takes - // hh.Lock first, and handleRecvError targets a main-hostmap entry with a - // different localIndexId. - hm.RLock() - cur, ok := hm.indexes[hh.hostinfo.localIndexId] - hm.RUnlock() - if !ok || cur != hh { + if !hm.stillPending(hh, hh.hostinfo.localIndexId) { return } + hm.processResponse(via, hh, hh.machine, packet) +} +// stillPending re-verifies hh is still tracked under index. Between queryIndex +// returning and the caller taking hh.Lock, handleOutbound may have timed out and +// deleted it. Once hh.Lock is held no other deleter can race the index: +// handleOutbound also takes hh.Lock first, and handleRecvError targets a +// main-hostmap entry with a different localIndexId. The caller must hold hh.Lock. +func (hm *HandshakeManager) stillPending(hh *HandshakeHostInfo, index uint32) bool { + hm.RLock() + defer hm.RUnlock() + cur, ok := hm.indexes[index] + return ok && cur == hh +} + +// processResponse runs a responder's message through one of the initiator's +// pending attempts and completes the tunnel when the handshake finishes. The +// caller must hold hh.Lock. +func (hm *HandshakeManager) processResponse(via ViaSender, hh *HandshakeHostInfo, machine *handshake.Machine, packet []byte) { + f := hm.f hostinfo := hh.hostinfo if !via.IsRelayed { if !f.lightHouse.GetRemoteAllowList().AllowAll(hostinfo.vpnAddrs, via.UdpAddr.Addr()) { @@ -839,7 +908,6 @@ func (hm *HandshakeManager) continueHandshake(via ViaSender, hh *HandshakeHostIn } } - machine := hh.machine if machine == nil { f.l.Error("No handshake machine available for continuation", "vpnAddrs", hostinfo.vpnAddrs, "from", via) @@ -862,7 +930,7 @@ func (hm *HandshakeManager) continueHandshake(via ViaSender, hh *HandshakeHostIn } if response != nil { - hm.sendHandshakeResponse(via, response, hostinfo, false) + hm.sendHandshakeResponse(via, response, hostinfo, false, 0) } if result == nil { @@ -975,7 +1043,10 @@ func (hm *HandshakeManager) continueHandshake(via ViaSender, hh *HandshakeHostIn hostinfo.vpnAddrs = vpnAddrs hostinfo.buildNetworks(f.myVpnNetworksTable, remoteCert.Certificate) + // The classical and PQ attempts each hold an index; keep the one that won. + hm.settleIndex(hh, result.LocalIndex) hm.Complete(hostinfo, f) + hm.notePQPeer(machine.Subtype(), vpnAddrs) if len(hh.packetStore) > 0 { if f.l.Enabled(context.Background(), slog.LevelDebug) { @@ -1054,20 +1125,23 @@ func (hm *HandshakeManager) validatePeerCert(via ViaSender, remoteCert *cert.Cac // cached is true when msg is a stored response being retransmitted because // the peer's stage-1 retransmit landed (the ErrAlreadySeen path); false on a // fresh response. -func (hm *HandshakeManager) sendHandshakeResponse(via ViaSender, msg []byte, hostinfo *HostInfo, cached bool) { +// +// received is the number of bytes a fragmented msg1 arrived in; it bounds what a +// fragmented response may send back. It is ignored for unfragmented subtypes. +func (hm *HandshakeManager) sendHandshakeResponse(via ViaSender, msg []byte, hostinfo *HostInfo, cached bool, received int) { if msg == nil { return } f := hm.f - f.messageMetrics.Tx(header.Handshake, header.MessageSubType(msg[1]), 1) + subtype := header.MessageSubType(msg[1]) // Common log fields. peerCert may be nil during intermediate // multi-message flows (handshake hasn't completed yet); skip the cert // block if so. logFields := []any{ "vpnAddrs", hostinfo.vpnAddrs, - "handshake", m{"stage": uint64(2), "style": header.SubTypeName(header.Handshake, header.HandshakeIXPSK0)}, + "handshake", m{"stage": uint64(2), "style": header.SubTypeName(header.Handshake, subtype)}, "cached", cached, "initiatorIndex", hostinfo.remoteIndexId, "responderIndex", hostinfo.localIndexId, @@ -1081,6 +1155,17 @@ 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 { + f.l.Debug("Did not send handshake message", append(fields, "error", err)...) + } else { + f.l.Info("Handshake message sent", fields...) + } + return + } + + f.messageMetrics.Tx(header.Handshake, subtype, 1) if !via.IsRelayed { fields := append(logFields, "from", via) err := f.outside.WriteTo(msg, via.UdpAddr) @@ -1107,10 +1192,10 @@ func (hm *HandshakeManager) sendHandshakeResponse(via ViaSender, msg []byte, hos // This only fires from the responder-side beginHandshake path, after the // peer cert has been validated and ConnectionState populated, so peerCert // is always non-nil for the cases that log it. -func (hm *HandshakeManager) handleCheckAndCompleteError(err error, existing, hostinfo *HostInfo, via ViaSender) { +func (hm *HandshakeManager) handleCheckAndCompleteError(err error, existing, hostinfo *HostInfo, via ViaSender, subtype header.MessageSubType, received int) { f := hm.f peerCert := hostinfo.ConnectionState.peerCert - hsFields := m{"stage": uint64(1), "style": header.SubTypeName(header.Handshake, header.HandshakeIXPSK0)} + hsFields := m{"stage": uint64(1), "style": header.SubTypeName(header.Handshake, subtype)} switch err { case ErrAlreadySeen: @@ -1121,7 +1206,7 @@ func (hm *HandshakeManager) handleCheckAndCompleteError(err error, existing, hos // ephemeral keys; a freshly-built one would have different keys and break // the tunnel even though both sides "completed" the handshake. if msg := existing.HandshakePacket[handshakePacketStage2]; msg != nil { - hm.sendHandshakeResponse(via, msg, existing, true) + hm.sendHandshakeResponse(via, msg, existing, true, received) } case ErrExistingHostInfo: diff --git a/handshake_manager_pq.go b/handshake_manager_pq.go new file mode 100644 index 00000000..71557ba5 --- /dev/null +++ b/handshake_manager_pq.go @@ -0,0 +1,446 @@ +package nebula + +import ( + "context" + "crypto/sha256" + "errors" + "log/slog" + "net/netip" + "sync" + "time" + + "github.com/slackhq/nebula/handshake" + "github.com/slackhq/nebula/handshake/fragment" + "github.com/slackhq/nebula/header" +) + +const ( + // pqSubtype is the post-quantum handshake this node speaks. + pqSubtype = header.HandshakeIXPSK2MLKEM1024 + + // DefaultPQRepairDelay is how long repair fragments trail the data fragments. + DefaultPQRepairDelay = 10 * time.Millisecond + + // pqAmplification bounds what a responder sends in reply to an unvalidated + // address, as a multiple of what it received from it. + pqAmplification = 3 +) + +// pqTableConfig bounds the responder's reassembly of msg1 fragments from senders +// that have not authenticated yet. +var pqTableConfig = fragment.TableConfig{ + MaxBytes: 4 << 20, + MaxEntries: 1024, + MaxPerSource: 4, + MaxMsgLen: 16384, + TTL: 3 * time.Second, +} + +type pqHandshakeConfig struct { + initiate bool // start PQ handshakes + accept bool // answer PQ handshakes + sticky bool // never fall back to classical for a peer that completed PQ + repairDelay time.Duration // delay between data and repair fragments +} + +// pqAttempt is the initiator's post-quantum attempt for a pending handshake. It +// lives alongside the classical attempt in HandshakeHostInfo and has its own +// local index, so whichever answer arrives first wins. +type pqAttempt struct { + machine *handshake.Machine + index uint32 + sender *fragment.Sender + msg2 *fragment.Assembly +} + +// pqFragKey identifies one msg1 being reassembled. +type pqFragKey struct { + addr netip.AddrPort + relay uint32 + fragID uint32 +} + +// pqFragSource groups msg1 reassemblies for the per-source quota: an IPv4 /32, +// an IPv6 /64, or one relay channel. +type pqFragSource struct { + prefix netip.Prefix + relay uint32 +} + +func pqFragKeyFor(via ViaSender, fragID uint32) (pqFragKey, pqFragSource) { + if via.IsRelayed { + return pqFragKey{relay: via.relay.LocalIndex, fragID: fragID}, + pqFragSource{relay: via.relay.LocalIndex} + } + addr := via.UdpAddr.Addr().Unmap() + bits := 64 + if addr.Is4() { + bits = 32 + } + prefix, _ := addr.Prefix(bits) + return pqFragKey{addr: via.UdpAddr, fragID: fragID}, pqFragSource{prefix: prefix} +} + +// reassembled rebuilds the packet the handshake Machine expects from the header +// of a fragment and the decoded message. +func reassembled(h *header.H, msg []byte) []byte { + p := make([]byte, header.Len, header.Len+len(msg)) + header.Encode(p, header.Version, header.Handshake, h.Subtype, h.RemoteIndex, h.MessageCounter) + return append(p, msg...) +} + +// usePQ decides whether this outbound attempt is post-quantum. Attempts +// alternate, starting with PQ, because a peer that does not speak PQ silently +// drops it. A peer that already completed a PQ handshake never gets a classical +// attempt when sticky is on. +func (hm *HandshakeManager) usePQ(hh *HandshakeHostInfo, vpnIp netip.Addr) bool { + if !hm.config.pq.initiate { + return false + } + if hm.config.pq.sticky { + if _, ok := hm.pqPeers.Load(vpnIp); ok { + return true + } + } + return hh.counter%2 == 1 +} + +// buildPQAttempt creates the initiator's PQ Machine and msg1 sender. +func (hm *HandshakeManager) buildPQAttempt(hh *HandshakeHostInfo) bool { + cs := hm.f.pki.getCertState() + v := hm.initiatingVersion(hh, cs) + if cs.GetCredential(v) == nil { + hm.f.l.Error("Unable to handshake with host because no certificate is available", + "vpnAddrs", hh.hostinfo.vpnAddrs, "certVersion", v) + return false + } + + pq := &pqAttempt{} + machine, err := handshake.NewMachine( + v, cs.GetCredential, + hm.certVerifier(), func() (uint32, error) { return hm.allocatePQIndex(hh, pq) }, + true, pqSubtype, + ) + if err != nil { + hm.f.l.Error("Failed to create handshake machine", + "vpnAddrs", hh.hostinfo.vpnAddrs, "error", err) + return false + } + + msg, err := machine.Initiate(nil) + if err != nil { + hm.f.l.Error("Failed to initiate handshake", + "vpnAddrs", hh.hostinfo.vpnAddrs, "error", err) + hm.releasePQIndex(hh, pq) + return false + } + + sender, err := fragment.NewSender(msg[header.Len:], pqSubtype, 0, 1, pq.index) + if err != nil { + hm.f.l.Error("Failed to fragment handshake", + "vpnAddrs", hh.hostinfo.vpnAddrs, "error", err) + hm.releasePQIndex(hh, pq) + return false + } + + pq.machine = machine + pq.sender = sender + hh.pq = pq + return true +} + +// allocatePQIndex registers a local index for the PQ attempt. Unlike +// allocateIndex it leaves hostinfo.localIndexId alone; that is settled when one +// of the attempts completes. +func (hm *HandshakeManager) allocatePQIndex(hh *HandshakeHostInfo, pq *pqAttempt) (uint32, error) { + hm.mainHostMap.RLock() + defer hm.mainHostMap.RUnlock() + hm.Lock() + defer hm.Unlock() + + for range 32 { + index, err := generateIndex(hm.l) + if err != nil { + return 0, err + } + + _, inPending := hm.indexes[index] + _, inMain := hm.mainHostMap.Indexes[index] + + if !inMain && !inPending { + pq.index = index + hm.indexes[index] = hh + return index, nil + } + } + + return 0, errors.New("failed to generate unique localIndexId") +} + +func (hm *HandshakeManager) releasePQIndex(hh *HandshakeHostInfo, pq *pqAttempt) { + hm.Lock() + defer hm.Unlock() + if pq.index != 0 && hm.indexes[pq.index] == hh { + delete(hm.indexes, pq.index) + } +} + +// settleIndex makes winner the hostinfo's local index and releases the index of +// the attempt that lost. +func (hm *HandshakeManager) settleIndex(hh *HandshakeHostInfo, winner uint32) { + hm.Lock() + defer hm.Unlock() + for _, idx := range []uint32{hh.hostinfo.localIndexId, hh.pqIndex()} { + if idx != 0 && idx != winner && hm.indexes[idx] == hh { + delete(hm.indexes, idx) + } + } + hh.hostinfo.localIndexId = winner +} + +func (hh *HandshakeHostInfo) pqIndex() uint32 { + if hh.pq == nil { + return 0 + } + return hh.pq.index +} + +// handleFragment routes one fragment of a PQ handshake message. +func (hm *HandshakeManager) handleFragment(via ViaSender, packet []byte, h *header.H) { + fh, sym, err := fragment.Parse(packet[header.Len:]) + if err != nil { + hm.l.Debug("dropping malformed handshake fragment", "from", via, "error", err) + return + } + + switch h.MessageCounter { + case 1: + if !hm.config.pq.accept { + hm.l.Debug("dropping post-quantum handshake, not accepting them", "from", via) + return + } + if h.RemoteIndex != 0 { + hm.l.Debug("dropping stage-1 handshake with non-zero RemoteIndex", + "from", via, "remoteIndex", h.RemoteIndex) + return + } + key, src := pqFragKeyFor(via, fh.FragID) + res, err := hm.pqTable.Add(key, src, time.Now(), fh, sym) + if err != nil { + hm.l.Debug("dropping handshake fragment", "from", via, "error", err) + return + } + if res == nil { + return + } + if existing := hm.pqAnswered.lookup(res.Msg, time.Now()); existing != nil { + // Already answered: resend, rate limited, with fresh symbols. + if msg := existing.HandshakePacket[handshakePacketStage2]; msg != nil { + hm.sendHandshakeResponse(via, msg, existing, true, res.Received) + } + return + } + hm.beginHandshake(via, reassembled(h, res.Msg), h, res.Received) + + case 2: + if hh := hm.queryIndex(h.RemoteIndex); hh != nil { + hm.continuePQ(via, hh, h, fh, sym) + } + } +} + +// continuePQ collects msg2 fragments for the initiator's PQ attempt. +func (hm *HandshakeManager) continuePQ(via ViaSender, hh *HandshakeHostInfo, h *header.H, fh fragment.Header, sym []byte) { + hh.Lock() + defer hh.Unlock() + + pq := hh.pq + if pq == nil || pq.index != h.RemoteIndex || !hm.stillPending(hh, pq.index) { + return + } + + if pq.msg2 == nil { + a, err := fragment.NewAssembly(fh) + if err != nil { + return + } + pq.msg2 = a + } + ready, err := pq.msg2.Add(fh, sym) + if errors.Is(err, fragment.ErrMismatch) { + // The responder's FragID and MsgLen never change, so this is a stray or a + // forgery. Start over from this fragment rather than wedge. + pq.msg2 = nil + return + } + if err != nil || !ready { + return + } + + msg, err := pq.msg2.Decode() + pq.msg2 = nil + if err != nil { + hm.l.Debug("failed to decode handshake fragments", "from", via, "error", err) + return + } + hm.processResponse(via, hh, pq.machine, reassembled(h, msg)) +} + +// sendFragmented sends packets to a peer: now immediately and later after the +// repair delay. It returns the first error from the immediate sends; errors from +// delayed sends are only logged. +func (hm *HandshakeManager) sendFragmented(now, later [][]byte, send func([]byte) error) error { + var first error + for _, p := range now { + if err := send(p); err != nil && first == nil { + first = err + } + } + sendLater := func() { + for _, p := range later { + if err := send(p); err != nil { + hm.l.Debug("Failed to send handshake repair fragment", "error", err) + } + } + } + switch { + case len(later) == 0: + case hm.config.pq.repairDelay <= 0: + sendLater() + default: + time.AfterFunc(hm.config.pq.repairDelay, sendLater) + } + 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()) + } + + var h header.H + if err := h.Parse(msg); err != nil { + return err + } + s, err := fragment.NewSender(msg[header.Len:], h.Subtype, h.RemoteIndex, h.MessageCounter, hostinfo.localIndexId) + if err != nil { + return err + } + + l := s.Layout() + repair := fragment.DefaultRepair(l.K) + if received > 0 { + pktLen := header.Len + fragment.HeaderLen + l.SymbolSize + budget := pqAmplification * received / pktLen + if budget < l.K { + return errors.New("handshake response exceeds the amplification budget") + } + repair = min(repair, budget-l.K) + } + + n := uint32(l.K + repair) + s.SetNext(int(hostinfo.handshakeNextSymbol.Add(n) - n)) + pkts, later, err := s.Attempt(repair) + if err != nil { + return err + } + + f := hm.f + var send func([]byte) error + if !via.IsRelayed { + send = func(p []byte) error { + return f.outside.WriteTo(p, via.UdpAddr) + } + } else { + if via.relay == nil { + return errors.New("both addr and via.relay are nil") + } + hostinfo.relayState.InsertRelayTo(via.relayHI.vpnAddrs[0]) + via.relayHI.relayState.UpdateRelayForByIdxState(via.relay.LocalIndex, Established) + send = func(p []byte) error { + f.SendVia(via.relayHI, via.relay, p, make([]byte, 12), make([]byte, mtu), false, 0) + return nil + } + } + f.messageMetrics.Tx(header.Handshake, h.Subtype, int64(len(pkts)+len(later))) + return hm.sendFragmented(pkts, later, send) +} + +// notePQPeer remembers peers that completed a PQ handshake, for sticky mode. +func (hm *HandshakeManager) notePQPeer(subtype header.MessageSubType, vpnAddrs []netip.Addr) { + if !hm.config.pq.sticky || subtype != pqSubtype { + return + } + for _, a := range vpnAddrs { + hm.pqPeers.Store(a, struct{}{}) + } + if hm.l.Enabled(context.Background(), slog.LevelDebug) { + hm.l.Debug("Remembering post-quantum peer", "vpnAddrs", vpnAddrs) + } +} + +// pqAnswered remembers the msg1s a responder recently answered, keyed by hash, +// so a repeat is recognised before any noise, KEM or certificate work. Repeats +// are routine: the trailing repair fragments of an attempt rebuild a msg1 that +// its data fragments already completed, and a retrying initiator offers fresh +// symbols for the same msg1. Entries only exist for completed handshakes, which +// need a valid certificate, and live as long as an initiator keeps retrying. +type pqAnswered struct { + sync.Mutex + ttl time.Duration + entries map[[sha256.Size]byte]pqAnsweredEntry +} + +type pqAnsweredEntry struct { + hostinfo *HostInfo + at time.Time +} + +const pqAnsweredMax = 4096 + +func newPQAnswered(ttl time.Duration) *pqAnswered { + return &pqAnswered{ttl: ttl, entries: map[[sha256.Size]byte]pqAnsweredEntry{}} +} + +func (a *pqAnswered) add(msg1 []byte, hostinfo *HostInfo, now time.Time) { + a.Lock() + defer a.Unlock() + if len(a.entries) >= pqAnsweredMax { + for k := range a.entries { + delete(a.entries, k) + break + } + } + a.entries[sha256.Sum256(msg1)] = pqAnsweredEntry{hostinfo: hostinfo, at: now} +} + +func (a *pqAnswered) lookup(msg1 []byte, now time.Time) *HostInfo { + a.Lock() + defer a.Unlock() + e, ok := a.entries[sha256.Sum256(msg1)] + if !ok || now.Sub(e.at) > a.ttl { + return nil + } + return e.hostinfo +} + +func (a *pqAnswered) expire(now time.Time) { + a.Lock() + defer a.Unlock() + for k, e := range a.entries { + if now.Sub(e.at) > a.ttl { + delete(a.entries, k) + } + } +} diff --git a/handshake_manager_pq_test.go b/handshake_manager_pq_test.go new file mode 100644 index 00000000..832429c1 --- /dev/null +++ b/handshake_manager_pq_test.go @@ -0,0 +1,155 @@ +package nebula + +import ( + "math/rand/v2" + "net/netip" + "sync" + "testing" + "time" + + "github.com/slackhq/nebula/handshake/fragment" + "github.com/slackhq/nebula/header" + "github.com/slackhq/nebula/test" + "github.com/slackhq/nebula/udp" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// recordingConn captures what is written to it. +type recordingConn struct { + udp.NoopConn + sync.Mutex + sent [][]byte +} + +func (c *recordingConn) WriteTo(b []byte, _ netip.AddrPort) error { + c.Lock() + defer c.Unlock() + c.sent = append(c.sent, append([]byte(nil), b...)) + return nil +} + +func (c *recordingConn) take() [][]byte { + c.Lock() + defer c.Unlock() + s := c.sent + c.sent = nil + return s +} + +func newPQTestManager(t *testing.T) (*HandshakeManager, *recordingConn) { + l := test.NewLogger() + conn := &recordingConn{} + cfg := defaultHandshakeConfig + cfg.pq = pqHandshakeConfig{initiate: true, accept: true} + hm := NewHandshakeManager(l, newHostMap(l), newTestLighthouse(), conn, cfg) + hm.f = &Interface{ + handshakeManager: hm, + outside: conn, + messageMetrics: newMessageMetricsOnlyRecvError(), + l: l, + } + return hm, conn +} + +// fragmentIndexes checks every packet is a fragment of the expected response and +// returns their symbol indexes. +func fragmentIndexes(t *testing.T, pkts [][]byte, fragID, remoteIndex uint32) []int { + t.Helper() + var out []int + for _, p := range pkts { + require.LessOrEqual(t, len(p), fragment.MaxPacketLen) + var h header.H + require.NoError(t, h.Parse(p)) + assert.Equal(t, pqSubtype, h.Subtype) + assert.Equal(t, remoteIndex, h.RemoteIndex) + assert.Equal(t, uint64(2), h.MessageCounter) + fh, _, err := fragment.Parse(p[header.Len:]) + require.NoError(t, err) + assert.Equal(t, fragID, fh.FragID) + out = append(out, int(fh.Index)) + } + return out +} + +func TestSendFragmentedResponse(t *testing.T) { + hm, conn := newPQTestManager(t) + via := ViaSender{UdpAddr: netip.MustParseAddrPort("192.0.2.1:4242")} + hostinfo := &HostInfo{localIndexId: 77} + + // A 2000 byte message is 2 symbols of 1000, so packets of 1024 bytes. + msg := make([]byte, header.Len, header.Len+2000) + header.Encode(msg, header.Version, header.Handshake, pqSubtype, 5, 2) + r := rand.New(rand.NewPCG(1, 1)) + for range 2000 { + msg = append(msg, byte(r.Uint32())) + } + 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)) + 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)) + 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)) + 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)) + assert.Empty(t, conn.take()) + }) +} + +func TestPQAnswered(t *testing.T) { + now := time.Unix(1000, 0) + a := newPQAnswered(5 * time.Second) + hi := &HostInfo{} + + assert.Nil(t, a.lookup([]byte("msg1"), now)) + a.add([]byte("msg1"), hi, now) + assert.Same(t, hi, a.lookup([]byte("msg1"), now.Add(time.Second))) + assert.Nil(t, a.lookup([]byte("other"), now)) + assert.Nil(t, a.lookup([]byte("msg1"), now.Add(6*time.Second)), "stale entries are not answered") + + a.expire(now.Add(6 * time.Second)) + assert.Empty(t, a.entries) + + for i := range pqAnsweredMax + 10 { + a.add([]byte{byte(i), byte(i >> 8)}, hi, now) + } + assert.Len(t, a.entries, pqAnsweredMax) +} + +func TestPQFragKeyFor(t *testing.T) { + k1, s1 := pqFragKeyFor(ViaSender{UdpAddr: netip.MustParseAddrPort("192.0.2.1:4242")}, 9) + k2, s2 := pqFragKeyFor(ViaSender{UdpAddr: netip.MustParseAddrPort("192.0.2.1:4243")}, 9) + assert.NotEqual(t, k1, k2, "different ports are different messages") + assert.Equal(t, s1, s2, "but share an IPv4 /32 quota") + + _, s3 := pqFragKeyFor(ViaSender{UdpAddr: netip.MustParseAddrPort("[2001:db8::1]:1")}, 9) + _, s4 := pqFragKeyFor(ViaSender{UdpAddr: netip.MustParseAddrPort("[2001:db8::ffff]:1")}, 9) + assert.Equal(t, s3, s4, "an IPv6 /64 shares a quota") + + _, s5 := pqFragKeyFor(ViaSender{UdpAddr: netip.MustParseAddrPort("[::ffff:192.0.2.1]:1")}, 9) + assert.Equal(t, s1, s5, "v4-mapped addresses count as IPv4") +} diff --git a/hostmap.go b/hostmap.go index 71814a2f..7427053a 100644 --- a/hostmap.go +++ b/hostmap.go @@ -260,6 +260,11 @@ type HostInfo struct { // We need these to avoid replayed handshake packets creating new hostinfos which causes churn 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 + handshakeNextSymbol atomic.Uint32 // next fragment symbol index to offer + // nextLHQuery is the earliest we can ask the lighthouse for new information. // This is used to limit lighthouse re-queries in chatty clients nextLHQuery atomic.Int64 diff --git a/main.go b/main.go index a8d01779..cc91615b 100644 --- a/main.go +++ b/main.go @@ -232,6 +232,12 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev retries: int64(c.GetInt("handshakes.retries", DefaultHandshakeRetries)), triggerBuffer: c.GetInt("handshakes.trigger_buffer", DefaultHandshakeTriggerBuffer), messageMetrics: messageMetrics, + pq: pqHandshakeConfig{ + initiate: c.GetBool("handshakes.pq.initiate", false), + accept: c.GetBool("handshakes.pq.accept", false), + sticky: c.GetBool("handshakes.pq.sticky", false), + repairDelay: c.GetDuration("handshakes.pq.repair_delay", DefaultPQRepairDelay), + }, } handshakeManager := NewHandshakeManager(l, hostMap, lightHouse, udpConns[0], handshakeConfig) diff --git a/relay_manager.go b/relay_manager.go index 46d7a2bb..a6e3f22c 100644 --- a/relay_manager.go +++ b/relay_manager.go @@ -57,8 +57,9 @@ func (rm *relayManager) GetUseRelays() bool { // StartRelays drives the relay-establishment side of an outbound handshake attempt. // For each candidate relay it either kicks off a handshake to the relay, sends a CreateRelayRequest, retransmits // one that may have been lost, or, once the relay is Established, forwards the in-progress -// stage 0 handshake packet for vpnIp through it. -func (rm *relayManager) StartRelays(f *Interface, vpnIp netip.Addr, hh *HandshakeHostInfo, stage0 []byte) { +// stage 0 handshake packets for vpnIp through it: now immediately and later after the +// handshake repair delay. +func (rm *relayManager) StartRelays(f *Interface, vpnIp netip.Addr, hh *HandshakeHostInfo, now, later [][]byte) { hostinfo := hh.hostinfo if !rm.GetUseRelays() || len(hostinfo.remotes.relays) == 0 { hh.lastRelays = nil @@ -161,7 +162,10 @@ func (rm *relayManager) StartRelays(f *Interface, vpnIp netip.Addr, hh *Handshak switch existingRelay.State { case Established: hl.Log(context.Background(), level, "Send handshake via relay", "relay", relay.String()) - f.SendVia(relayHostInfo, existingRelay, stage0, make([]byte, 12), make([]byte, mtu), false, 0) + _ = f.handshakeManager.sendFragmented(now, later, func(p []byte) error { + f.SendVia(relayHostInfo, existingRelay, p, make([]byte, 12), make([]byte, mtu), false, 0) + return nil + }) case Disestablished: // Mark this relay as 'requested' relayHostInfo.relayState.UpdateRelayForByIpState(vpnIp, Requested) diff --git a/relay_manager_test.go b/relay_manager_test.go index 8da38940..9e9dd99c 100644 --- a/relay_manager_test.go +++ b/relay_manager_test.go @@ -50,7 +50,7 @@ func TestStartRelaysLogDedupe(t *testing.T) { var buf bytes.Buffer rm := newRM(&buf) hh := newHH() - rm.StartRelays(f, vpnIp, hh, nil) + rm.StartRelays(f, vpnIp, hh, nil, nil) assert.Equal(t, []netip.Addr{vpnIp}, hh.lastRelays, "lastRelays should record the relay set we just attempted") assert.Contains(t, buf.String(), "level=INFO "+msg, "expected Info level on first attempt") }) @@ -59,10 +59,10 @@ func TestStartRelaysLogDedupe(t *testing.T) { var buf bytes.Buffer rm := newRM(&buf) hh := newHH() - rm.StartRelays(f, vpnIp, hh, nil) + rm.StartRelays(f, vpnIp, hh, nil, nil) first := append([]netip.Addr(nil), hh.lastRelays...) buf.Reset() - rm.StartRelays(f, vpnIp, hh, nil) + rm.StartRelays(f, vpnIp, hh, nil, nil) assert.Equal(t, first, hh.lastRelays) assert.Contains(t, buf.String(), "level=DEBUG "+msg, "expected Debug level on identical retry") assert.NotContains(t, buf.String(), "level=INFO "+msg, "Info should not fire on identical retry") @@ -72,13 +72,13 @@ func TestStartRelaysLogDedupe(t *testing.T) { var buf bytes.Buffer rm := newRM(&buf) hh := newHH() - rm.StartRelays(f, vpnIp, hh, nil) + rm.StartRelays(f, vpnIp, hh, nil, nil) buf.Reset() // The lighthouse handed us a new set this round. hh.hostinfo.remotes.relays = []netip.Addr{vpnIp, otherRelay} - rm.StartRelays(f, vpnIp, hh, nil) + rm.StartRelays(f, vpnIp, hh, nil, nil) assert.Equal(t, []netip.Addr{vpnIp, otherRelay}, hh.lastRelays) assert.Contains(t, buf.String(), "level=INFO "+msg, "expected Info when the relay list changes") }) @@ -90,7 +90,7 @@ func TestStartRelaysLogDedupe(t *testing.T) { hh := newHH() hh.lastRelays = []netip.Addr{vpnIp} - rm.StartRelays(f, vpnIp, hh, nil) + rm.StartRelays(f, vpnIp, hh, nil, nil) assert.Nil(t, hh.lastRelays, "with relays disabled lastRelays should be cleared") assert.NotContains(t, buf.String(), msg, "should not log when we shortcut out") })