lanes_test.go 41 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175
  1. package nebula
  2. import (
  3. "bytes"
  4. "log/slog"
  5. "net/netip"
  6. "strings"
  7. "testing"
  8. "time"
  9. "github.com/gaissmai/bart"
  10. "github.com/slackhq/nebula/cert"
  11. "github.com/slackhq/nebula/config"
  12. "github.com/slackhq/nebula/firewall"
  13. "github.com/slackhq/nebula/handshake"
  14. "github.com/slackhq/nebula/header"
  15. "github.com/slackhq/nebula/overlay/overlaytest"
  16. "github.com/slackhq/nebula/overlay/tio"
  17. "github.com/slackhq/nebula/test"
  18. "github.com/slackhq/nebula/udp"
  19. "github.com/stretchr/testify/assert"
  20. "github.com/stretchr/testify/require"
  21. )
  22. var (
  23. testMyAddr = netip.MustParseAddr("10.0.0.1")
  24. testPeerAddr = netip.MustParseAddr("10.0.0.2")
  25. )
  26. // newTestLaneSet builds a lane set from a real handshake result so the sessions
  27. // it derives hold usable keys.
  28. func newTestLaneSet(t *testing.T, r *handshake.Result, myLanes int, peerPorts, peerBase, peerTxLanes uint32) *laneSet {
  29. t.Helper()
  30. r.PeerPortCount = peerPorts
  31. r.PeerBasePort = peerBase
  32. r.PeerTxLanes = peerTxLanes
  33. return newLaneSet(r, myLanes, testMyAddr, testPeerAddr)
  34. }
  35. // laneSessionFor derives lane s's session and installs it, standing in for the
  36. // data-plane call that would normally be the first to need it.
  37. func laneSessionFor(t *testing.T, ls *laneSet, s int) *ConnectionState {
  38. t.Helper()
  39. cs, err := ls.session(s)
  40. require.NoError(t, err)
  41. require.NotNil(t, cs)
  42. return cs
  43. }
  44. func newTestLaneHostInfo(t *testing.T, r *handshake.Result, ls *laneSet) *HostInfo {
  45. t.Helper()
  46. cs, err := newConnectionStateFromResult(r)
  47. require.NoError(t, err)
  48. hi := &HostInfo{
  49. vpnAddrs: []netip.Addr{testPeerAddr},
  50. localIndexId: 100,
  51. remoteIndexId: 200,
  52. remotes: NewRemoteList([]netip.Addr{testPeerAddr}, nil),
  53. HandshakePacket: map[uint8][]byte{},
  54. ConnectionState: cs,
  55. lanes: ls,
  56. }
  57. hi.SetRemote(netip.MustParseAddrPort("192.0.2.1:4242"))
  58. return hi
  59. }
  60. func TestLanePortOffset(t *testing.T) {
  61. a := netip.MustParseAddr("10.0.0.1")
  62. b := netip.MustParseAddr("10.0.0.2")
  63. // Deterministic and in range.
  64. for _, count := range []uint16{1, 2, 3, 4, 16, 256} {
  65. o := lanePortOffset(a, b, count)
  66. assert.Equal(t, o, lanePortOffset(a, b, count), "count %d not deterministic", count)
  67. assert.Less(t, o, count, "count %d out of range", count)
  68. }
  69. assert.Equal(t, uint16(0), lanePortOffset(a, b, 0), "zero port count")
  70. // The two sides' rotations cancel when port counts match, preserving the
  71. // lane-i-reverses-lane-j conntrack pairing.
  72. for _, count := range []uint16{2, 3, 4, 7, 16} {
  73. for i := range 32 {
  74. peer := netip.AddrFrom4([4]byte{192, 0, 2, byte(i)})
  75. oA := lanePortOffset(a, peer, count)
  76. oB := lanePortOffset(peer, a, count)
  77. assert.Equal(t, uint16(0), (oA+oB)%count,
  78. "offsets don't cancel for peer %s count %d", peer, count)
  79. }
  80. }
  81. // Distinct small peers land on distinct rotations of a big peer's range,
  82. // not all on the same first ports.
  83. const bigPeerPorts = 16
  84. distinct := map[uint16]struct{}{}
  85. for i := range 64 {
  86. client := netip.AddrFrom4([4]byte{192, 0, 2, byte(i)})
  87. distinct[lanePortOffset(client, a, bigPeerPorts)] = struct{}{}
  88. }
  89. assert.GreaterOrEqual(t, len(distinct), 8, "64 clients only produced %d distinct offsets", len(distinct))
  90. }
  91. func TestLaneTargetPort(t *testing.T) {
  92. ls := &laneSet{peerBasePort: 4242, peerPortCount: 4}
  93. // No rotation: lane i targets base+i, wrapping past the peer's range.
  94. for i, want := range map[int]uint16{1: 4243, 2: 4244, 3: 4245, 5: 4243} {
  95. assert.Equal(t, want, ls.laneTargetPortLocked(i), "lane %d", i)
  96. }
  97. // Rotation shifts the whole mapping; the wrapped lane lands on the base
  98. // port itself, which is a valid distinct 4-tuple (our source port differs).
  99. ls.portOffset = 3
  100. for i, want := range map[int]uint16{1: 4242, 2: 4243, 3: 4244} {
  101. assert.Equal(t, want, ls.laneTargetPortLocked(i), "rotated lane %d", i)
  102. }
  103. // Fewer peer ports than local lanes: rotation still spreads across all of
  104. // the peer's ports.
  105. ls = &laneSet{peerBasePort: 4242, peerPortCount: 2, portOffset: 1}
  106. assert.Equal(t, uint16(4242), ls.laneTargetPortLocked(1))
  107. assert.Equal(t, uint16(4243), ls.laneTargetPortLocked(2))
  108. }
  109. // A lane's keys are derived, not negotiated, so the whole design rests on the
  110. // two sides landing on the same pair without exchanging anything.
  111. func TestLaneKeyDerivationSymmetry(t *testing.T) {
  112. initR, respR := runTestHandshake(t)
  113. initLS := newTestLaneSet(t, initR, 4, 4, 4242, 4)
  114. respLS := newTestLaneSet(t, respR, 4, 4, 4242, 4)
  115. require.Len(t, initLS.sessions, 4)
  116. assert.Nil(t, initLS.sessions[0].Load(), "lane 0 is the base session, not a derived one")
  117. nb := make([]byte, 12)
  118. for s := 1; s < 4; s++ {
  119. out := header.EncodeLane(make([]byte, 0, mtu), header.Version, header.Message, 0, 200, 1, uint8(s))
  120. ct, err := laneSessionFor(t, initLS, s).eKey.EncryptDanger(out, out, []byte("lane payload"), 1, nb)
  121. require.NoError(t, err)
  122. pt, err := laneSessionFor(t, respLS, s).Decrypt(test.NewLogger(), 1, ct, nb)
  123. require.NoError(t, err, "lane %d keys did not match", s)
  124. assert.Equal(t, []byte("lane payload"), pt)
  125. }
  126. // Distinct lanes get distinct keys: lane 2's session must not open lane 1's
  127. // ciphertext, or the header's lane index would be forgeable in effect.
  128. out := header.EncodeLane(make([]byte, 0, mtu), header.Version, header.Message, 0, 200, 7, 1)
  129. ct, err := laneSessionFor(t, initLS, 1).eKey.EncryptDanger(out, out, []byte("lane payload"), 7, nb)
  130. require.NoError(t, err)
  131. _, err = laneSessionFor(t, respLS, 2).Decrypt(test.NewLogger(), 7, ct, nb)
  132. assert.Error(t, err)
  133. }
  134. func TestNewLaneSetSizing(t *testing.T) {
  135. initR, _ := runTestHandshake(t)
  136. // A peer with no multiport advert gets no lanes at all.
  137. assert.Nil(t, newLaneSet(&handshake.Result{}, 4, testMyAddr, testPeerAddr))
  138. // One lane means only the base tunnel, which is not a lane set.
  139. assert.Nil(t, newLaneSet(&handshake.Result{PeerPortCount: 4, PeerTxLanes: 1}, 1, testMyAddr, testPeerAddr))
  140. // Sessions cover both directions: enough for everything the peer may send,
  141. // even though we may only send on a few.
  142. ls := newTestLaneSet(t, initR, 2, 8, 4242, 6)
  143. assert.Len(t, ls.sessions, 6, "sessions must cover the peer's tx lanes")
  144. assert.Equal(t, 2, ls.txLanes, "we may only send on our own lanes")
  145. // The peer's advert sizes the session table but nothing else. Its lane count
  146. // is its own choice, so it must not be able to make us allocate per-lane tx
  147. // state we will never use.
  148. assert.Len(t, ls.txAddr, 2)
  149. assert.Len(t, ls.demand, 2)
  150. assert.Len(t, ls.probe, 2)
  151. // Nothing is derived up front, for the same reason.
  152. for s := range ls.sessions {
  153. assert.Nil(t, ls.sessions[s].Load(), "lane %d derived before anything needed it", s)
  154. }
  155. // Every lane starts demanded so the first traffic tick probes them all: a
  156. // lane that only comes up after its flows do is a lane the flows never move
  157. // onto, because the kernel has pinned them to the queue they started on.
  158. ls = newTestLaneSet(t, initR, 4, 4, 4242, 4)
  159. for s := 1; s < ls.txLanes; s++ {
  160. assert.True(t, ls.demand[s].Load(), "lane %d not demanded at creation", s)
  161. }
  162. // Our tx lanes are clamped to the ports the peer actually bound: a lane
  163. // aimed past the peer's range would land on some unrelated socket.
  164. ls = newTestLaneSet(t, initR, 8, 3, 4242, 8)
  165. assert.Equal(t, 3, ls.txLanes)
  166. assert.Len(t, ls.sessions, 8)
  167. }
  168. // The handshake log lines carry the negotiated lanes, so an operator can tell a
  169. // peer that got none from one that was never asked.
  170. func TestLaneLogAttr(t *testing.T) {
  171. initR, _ := runTestHandshake(t)
  172. // Not running multiport: an empty attr, which slog drops entirely.
  173. assert.Equal(t, slog.Attr{}, laneLogAttr(0, nil))
  174. // Running multiport against a peer that isn't: zeros, not silence.
  175. assert.Equal(t, m{"tx": 0, "sessions": 0}, laneLogAttr(4, nil).Value.Any())
  176. ls := newTestLaneSet(t, initR, 2, 8, 4242, 6)
  177. attr := laneLogAttr(2, ls)
  178. assert.Equal(t, "lanes", attr.Key)
  179. assert.Equal(t, m{
  180. "tx": 2,
  181. "sessions": 6,
  182. "peerBasePort": uint16(4242),
  183. "peerPorts": uint16(8),
  184. "portOffset": ls.portOffset,
  185. }, attr.Value.Any())
  186. }
  187. // The RX path must not cache a session for a lane until a packet on it has
  188. // actually decrypted, or a spoofer naming lanes at random could make us hold a
  189. // replay window and two cipher states per lane without authenticating anything.
  190. func TestLaneSessionRxDerivation(t *testing.T) {
  191. initR, respR := runTestHandshake(t)
  192. respLS := newTestLaneSet(t, respR, 4, 4, 4242, 4)
  193. hi := newTestLaneHostInfo(t, respR, respLS)
  194. ci, cached, err := hi.laneSession(2)
  195. require.NoError(t, err)
  196. require.NotNil(t, ci)
  197. assert.False(t, cached, "the first packet on a lane derives, it does not hit")
  198. assert.Nil(t, respLS.sessions[2].Load(), "an unauthenticated packet must not install a session")
  199. // The lane the peer really is using decrypts, and that is what installs it.
  200. nb := make([]byte, 12)
  201. out := header.EncodeLane(make([]byte, 0, mtu), header.Version, header.Message, 0, 200, 1, 2)
  202. ct, err := laneSessionFor(t, newTestLaneSet(t, initR, 4, 4, 4242, 4), 2).
  203. eKey.EncryptDanger(out, out, []byte("real lane traffic"), 1, nb)
  204. require.NoError(t, err)
  205. pt, err := ci.Decrypt(test.NewLogger(), 1, ct, nb)
  206. require.NoError(t, err)
  207. assert.Equal(t, []byte("real lane traffic"), pt)
  208. respLS.installSession(test.NewLogger(), 2, ci, 1)
  209. assert.Same(t, ci, respLS.sessions[2].Load())
  210. // Now it is a hit, and the replay window the decrypt above advanced is the
  211. // one the next packet sees.
  212. got, cached, err := hi.laneSession(2)
  213. require.NoError(t, err)
  214. assert.True(t, cached)
  215. assert.Same(t, ci, got)
  216. // A racing install loses rather than swapping the session out, which would
  217. // throw away the replay window the live one has been accumulating. The loser's
  218. // packet still has to be marked seen on the winner, or dropping its session
  219. // would make that one counter replayable.
  220. other, err := newLaneConnectionState(&respLS.material, 2)
  221. require.NoError(t, err)
  222. require.True(t, ci.window.Check(test.NewLogger(), 7), "counter 7 seen before the race")
  223. respLS.installSession(test.NewLogger(), 2, other, 7)
  224. assert.Same(t, ci, respLS.sessions[2].Load())
  225. assert.False(t, ci.window.Check(test.NewLogger(), 7),
  226. "the loser's counter was not carried to the surviving window")
  227. }
  228. func TestLaneTxGate(t *testing.T) {
  229. initR, _ := runTestHandshake(t)
  230. ls := newTestLaneSet(t, initR, 4, 4, 4242, 4)
  231. for s := range ls.demand {
  232. // Creation seeds demand on every lane; start from cold so this test can
  233. // tell what the TX path itself raises.
  234. ls.demand[s].Store(false)
  235. }
  236. // A down lane hands back nothing and raises demand, which is what gets it
  237. // re-probed after a demotion.
  238. ci, addr := ls.txLane(1)
  239. assert.Nil(t, ci)
  240. assert.False(t, addr.IsValid())
  241. assert.True(t, ls.demand[1].Load())
  242. assert.False(t, ls.demand[2].Load(), "demand raised on an untouched lane")
  243. // An up lane with no session derived yet cannot happen — the probe that
  244. // promoted it derived one — but it must fall back rather than send in the
  245. // clear if it ever does.
  246. want := netip.MustParseAddrPort("192.0.2.1:4243")
  247. ls.txAddr[1].Store(&want)
  248. ci, _ = ls.txLane(1)
  249. assert.Nil(t, ci)
  250. // Promotion publishes the session and destination together.
  251. sess := laneSessionFor(t, ls, 1)
  252. ls.demand[1].Store(false)
  253. ci, addr = ls.txLane(1)
  254. assert.Same(t, sess, ci)
  255. assert.Equal(t, want, addr)
  256. assert.False(t, ls.demand[1].Load(), "a hit must not raise demand")
  257. // Lane 0 is the base tunnel and lanes at or above txLanes are receive-only.
  258. ci, _ = ls.txLane(0)
  259. assert.Nil(t, ci)
  260. ci, _ = ls.txLane(4)
  261. assert.Nil(t, ci)
  262. }
  263. func TestLaneSessionLookup(t *testing.T) {
  264. initR, _ := runTestHandshake(t)
  265. ls := newTestLaneSet(t, initR, 3, 4, 4242, 3)
  266. hi := newTestLaneHostInfo(t, initR, ls)
  267. assertNoLane := func(s uint8, msg string) {
  268. t.Helper()
  269. ci, cached, err := hi.laneSession(s)
  270. require.NoError(t, err)
  271. assert.Nil(t, ci, msg)
  272. assert.False(t, cached, msg)
  273. }
  274. assertNoLane(0, "lane 0 is the base session")
  275. assertNoLane(3, "a lane beyond what this tunnel covers")
  276. assertNoLane(255, "a lane beyond what this tunnel covers")
  277. // An already-derived lane is returned as a hit.
  278. sess := laneSessionFor(t, ls, 2)
  279. ci, cached, err := hi.laneSession(2)
  280. require.NoError(t, err)
  281. assert.Same(t, sess, ci)
  282. assert.True(t, cached)
  283. // A peer without lanes answers nil for every lane rather than panicking.
  284. bare := &HostInfo{}
  285. ci, _, err = bare.laneSession(1)
  286. require.NoError(t, err)
  287. assert.Nil(t, ci)
  288. }
  289. func TestMaxMessageCounter(t *testing.T) {
  290. initR, _ := runTestHandshake(t)
  291. ls := newTestLaneSet(t, initR, 3, 4, 4242, 3)
  292. hi := newTestLaneHostInfo(t, initR, ls)
  293. hi.ConnectionState.messageCounter.Store(5)
  294. assert.Equal(t, uint64(5), hi.maxMessageCounter())
  295. // A lane past the base is what the rehandshake threshold has to notice: the
  296. // base counter would sit still while the lane burns through its nonces.
  297. laneSessionFor(t, ls, 2).messageCounter.Store(9000)
  298. assert.Equal(t, uint64(9000), hi.maxMessageCounter())
  299. assert.Equal(t, uint64(0), (&HostInfo{}).maxMessageCounter())
  300. }
  301. func TestLaneRetryDelay(t *testing.T) {
  302. assert.Equal(t, laneRetryBase, laneRetryDelay(0))
  303. assert.Equal(t, 2*laneRetryBase, laneRetryDelay(1))
  304. assert.Equal(t, laneRetryMax, laneRetryDelay(laneMaxFails), "backoff must saturate")
  305. }
  306. func newLaneTestInterface(hostMap *HostMap) *Interface {
  307. l := test.NewLogger()
  308. lh := newTestLighthouse()
  309. cs := &CertState{
  310. initiatingVersion: cert.Version1,
  311. privateKey: []byte{},
  312. v1Cert: &dummyCert{version: cert.Version1},
  313. v1Credential: nil,
  314. }
  315. ifce := &Interface{
  316. hostMap: hostMap,
  317. inside: &overlaytest.NoopTun{},
  318. outside: &udp.NoopConn{},
  319. firewall: &Firewall{},
  320. lightHouse: lh,
  321. pki: &PKI{},
  322. handshakeManager: NewHandshakeManager(l, hostMap, lh, &udp.NoopConn{}, defaultHandshakeConfig),
  323. myVpnNetworksTable: new(bart.Lite),
  324. messageMetrics: newMessageMetricsOnlyRecvError(),
  325. writers: []udp.Conn{&udp.NoopConn{}, &udp.NoopConn{}, &udp.NoopConn{}, &udp.NoopConn{}},
  326. // One socket per port keeps writers indexed by lane, which is what most of
  327. // these tests want; the tests that care about the group layout set their own.
  328. routinesPerPort: 1,
  329. l: l,
  330. }
  331. ifce.pki.cs.Store(cs)
  332. conf := config.NewC(l)
  333. punchy := NewPunchyFromConfig(l, conf, nil)
  334. cm := newConnectionManagerFromConfig(l, conf, hostMap, punchy)
  335. cm.intf = ifce
  336. ifce.connectionManager = cm
  337. ifce.handshakeManager.f = ifce
  338. return ifce
  339. }
  340. // The full TX lifecycle of a lane: demand -> probe -> ack -> up, then an
  341. // unanswered keepalive -> demoted.
  342. func TestLaneProbeLifecycle(t *testing.T) {
  343. hostMap := newHostMap(test.NewLogger())
  344. ifce := newLaneTestInterface(hostMap)
  345. initR, _ := runTestHandshake(t)
  346. ls := newTestLaneSet(t, initR, 4, 4, 5353, 4)
  347. hi := newTestLaneHostInfo(t, initR, ls)
  348. nb := make([]byte, 12)
  349. out := make([]byte, mtu)
  350. now := time.Now()
  351. // No demand: nothing is probed, so a peer we barely talk to costs nothing
  352. // beyond its base tunnel. (Creation seeds demand on every lane, so clear it
  353. // to get at the no-demand case.)
  354. for s := range ls.demand {
  355. ls.demand[s].Store(false)
  356. }
  357. ifce.probeLanes(hi, now, nb, out)
  358. ls.mu.Lock()
  359. for s := 1; s < 4; s++ {
  360. assert.True(t, ls.probe[s].sentAt.IsZero(), "lane %d probed without demand", s)
  361. }
  362. ls.mu.Unlock()
  363. // Demand on lane 1 alone probes lane 1 alone, aimed at the peer's port for
  364. // that lane.
  365. ls.demand[1].Store(true)
  366. ifce.probeLanes(hi, now, nb, out)
  367. ls.mu.Lock()
  368. require.False(t, ls.probe[1].sentAt.IsZero(), "demand did not produce a probe")
  369. assert.True(t, ls.probe[2].sentAt.IsZero())
  370. wantPort := ls.laneTargetPortLocked(1)
  371. assert.Equal(t, netip.AddrPortFrom(netip.MustParseAddr("192.0.2.1"), wantPort), ls.probe[1].target)
  372. gen := ls.probe[1].gen
  373. assert.False(t, ls.demand[1].Load(), "the probe did not consume the demand")
  374. ls.mu.Unlock()
  375. // The lane stays down until the ack lands, and a stale generation cannot
  376. // bring it up.
  377. assert.Nil(t, ls.txAddr[1].Load())
  378. _, promoted := ls.noteAck(1, gen+1, now)
  379. assert.False(t, promoted, "an ack for a superseded probe promoted the lane")
  380. assert.Nil(t, ls.txAddr[1].Load())
  381. // The matching ack promotes it, and the destination is the probed target.
  382. ackTarget, promoted := ls.noteAck(1, gen, now)
  383. assert.True(t, promoted)
  384. assert.Equal(t, netip.AddrPortFrom(netip.MustParseAddr("192.0.2.1"), wantPort), ackTarget)
  385. addr := ls.txAddr[1].Load()
  386. require.NotNil(t, addr)
  387. assert.Equal(t, netip.AddrPortFrom(netip.MustParseAddr("192.0.2.1"), wantPort), *addr)
  388. // A second ack is a keepalive, not a promotion.
  389. ls.mu.Lock()
  390. ls.probe[1].sentAt = now
  391. ls.mu.Unlock()
  392. _, promoted = ls.noteAck(1, gen, now)
  393. assert.False(t, promoted)
  394. assert.NotNil(t, ls.txAddr[1].Load())
  395. // An up lane is left alone until the keepalive comes due.
  396. ifce.probeLanes(hi, now, nb, out)
  397. ls.mu.Lock()
  398. assert.True(t, ls.probe[1].sentAt.IsZero(), "an up lane was re-probed early")
  399. ls.mu.Unlock()
  400. // Past the keepalive it re-proves its path...
  401. now = now.Add(laneKeepalive + time.Second)
  402. ifce.probeLanes(hi, now, nb, out)
  403. ls.mu.Lock()
  404. require.False(t, ls.probe[1].sentAt.IsZero(), "keepalive did not probe")
  405. ls.mu.Unlock()
  406. assert.NotNil(t, ls.txAddr[1].Load(), "lane demoted before its probe aged out")
  407. // ...and an unanswered keepalive demotes it with backoff, so the routine
  408. // falls back to the base tunnel.
  409. now = now.Add(laneProbeTimeout + time.Second)
  410. ifce.probeLanes(hi, now, nb, out)
  411. assert.Nil(t, ls.txAddr[1].Load(), "unanswered keepalive did not demote the lane")
  412. ls.mu.Lock()
  413. assert.Equal(t, uint8(1), ls.probe[1].fails)
  414. assert.True(t, ls.probe[1].retryAt.After(now))
  415. ls.mu.Unlock()
  416. // The backoff holds even with fresh demand.
  417. ls.demand[1].Store(true)
  418. ifce.probeLanes(hi, now, nb, out)
  419. ls.mu.Lock()
  420. assert.True(t, ls.probe[1].sentAt.IsZero(), "backoff was ignored")
  421. ls.mu.Unlock()
  422. }
  423. // A lane aims at its own peer port and nothing else. There is no fallback to the
  424. // peer's base port: sharing that destination would cost the peer the receive
  425. // spread lanes exist to create, so an unreachable lane port just means this lane
  426. // stays down and its flows ride the base tunnel.
  427. func TestLaneProbeAlwaysTargetsItsOwnPort(t *testing.T) {
  428. hostMap := newHostMap(test.NewLogger())
  429. ifce := newLaneTestInterface(hostMap)
  430. initR, _ := runTestHandshake(t)
  431. ls := newTestLaneSet(t, initR, 4, 4, 5353, 4)
  432. hi := newTestLaneHostInfo(t, initR, ls)
  433. nb := make([]byte, 12)
  434. out := make([]byte, mtu)
  435. now := time.Now()
  436. ls.mu.Lock()
  437. lanePort := ls.laneTargetPortLocked(1)
  438. ls.mu.Unlock()
  439. require.NotEqual(t, ls.peerBasePort, lanePort, "this test needs a lane that isn't aimed at the base port")
  440. // Repeated failures never move the target off the lane's own port.
  441. for i := 0; i < 4; i++ {
  442. ls.demand[1].Store(true)
  443. ifce.probeLanes(hi, now, nb, out)
  444. ls.mu.Lock()
  445. require.False(t, ls.probe[1].sentAt.IsZero(), "the lane was not probed")
  446. assert.Equal(t, lanePort, ls.probe[1].target.Port(), "probe %d left the lane port", i)
  447. gen := ls.probe[1].gen
  448. ls.mu.Unlock()
  449. // Age the probe out, then wait out the backoff for the next attempt.
  450. now = now.Add(laneProbeTimeout + time.Second)
  451. ifce.probeLanes(hi, now, nb, out)
  452. require.Nil(t, ls.txAddr[1].Load())
  453. assert.False(t, promotedByAck(ls, 1, gen, now), "an aged-out probe was still promotable")
  454. now = now.Add(laneRetryMax + time.Second)
  455. }
  456. // It comes up on that port, and the keepalive re-proves the same one.
  457. ls.demand[1].Store(true)
  458. ifce.probeLanes(hi, now, nb, out)
  459. ls.mu.Lock()
  460. gen := ls.probe[1].gen
  461. ls.mu.Unlock()
  462. target, promoted := ls.noteAck(1, gen, now)
  463. require.True(t, promoted)
  464. assert.Equal(t, lanePort, target.Port())
  465. now = now.Add(laneKeepalive + time.Second)
  466. ifce.probeLanes(hi, now, nb, out)
  467. ls.mu.Lock()
  468. require.False(t, ls.probe[1].sentAt.IsZero(), "the keepalive did not probe")
  469. assert.Equal(t, lanePort, ls.probe[1].target.Port(), "the keepalive left the lane port")
  470. ls.mu.Unlock()
  471. // And a demotion returns to probing that same port.
  472. now = now.Add(laneProbeTimeout + time.Second)
  473. ifce.probeLanes(hi, now, nb, out)
  474. require.Nil(t, ls.txAddr[1].Load(), "the unanswered keepalive did not demote the lane")
  475. now = now.Add(laneRetryMax + time.Second)
  476. ls.demand[1].Store(true)
  477. ifce.probeLanes(hi, now, nb, out)
  478. ls.mu.Lock()
  479. require.False(t, ls.probe[1].sentAt.IsZero(), "the lane was not retried")
  480. assert.Equal(t, lanePort, ls.probe[1].target.Port())
  481. ls.mu.Unlock()
  482. }
  483. // promotedByAck is noteAck's boolean alone, for assertions that only care whether
  484. // an ack could bring the lane up.
  485. func promotedByAck(ls *laneSet, s int, gen uint8, now time.Time) bool {
  486. _, promoted := ls.noteAck(s, gen, now)
  487. return promoted
  488. }
  489. // A roam is a new path with no derivable relationship to the old lane ports, so
  490. // every lane has to be rebuilt rather than moved.
  491. func TestLaneProbeRoamResets(t *testing.T) {
  492. hostMap := newHostMap(test.NewLogger())
  493. ifce := newLaneTestInterface(hostMap)
  494. initR, _ := runTestHandshake(t)
  495. ls := newTestLaneSet(t, initR, 4, 4, 5353, 4)
  496. hi := newTestLaneHostInfo(t, initR, ls)
  497. now := time.Now()
  498. nb := make([]byte, 12)
  499. out := make([]byte, mtu)
  500. ls.demand[1].Store(true)
  501. ifce.probeLanes(hi, now, nb, out)
  502. ls.mu.Lock()
  503. gen := ls.probe[1].gen
  504. ls.mu.Unlock()
  505. _, promoted := ls.noteAck(1, gen, now)
  506. require.True(t, promoted)
  507. require.NotNil(t, ls.txAddr[1].Load())
  508. // New remote address: the lane is taken down, not retargeted.
  509. hi.SetRemote(netip.MustParseAddrPort("198.51.100.7:4242"))
  510. ifce.probeLanes(hi, now, nb, out)
  511. assert.Nil(t, ls.txAddr[1].Load(), "lane survived a roam")
  512. ls.mu.Lock()
  513. assert.Zero(t, ls.probe[1].fails, "a roam is not a lane failure")
  514. assert.Equal(t, netip.MustParseAddr("198.51.100.7"), ls.peerAddr)
  515. ls.mu.Unlock()
  516. // A late ack for the probe sent before the roam must not promote the lane
  517. // on the new path, so the generation has to carry across the reset.
  518. ls.demand[1].Store(true)
  519. ifce.probeLanes(hi, now, nb, out)
  520. _, promoted = ls.noteAck(1, gen, now)
  521. assert.False(t, promoted, "an ack from before the roam promoted the lane")
  522. assert.Nil(t, ls.txAddr[1].Load())
  523. // Relayed (no direct remote) means no lanes at all.
  524. ls.demand[1].Store(true)
  525. ifce.probeLanes(hi, now, nb, out)
  526. ls.mu.Lock()
  527. gen = ls.probe[1].gen
  528. ls.mu.Unlock()
  529. _, promoted = ls.noteAck(1, gen, now)
  530. require.True(t, promoted)
  531. hi.SetRemote(netip.AddrPort{})
  532. ifce.probeLanes(hi, now, nb, out)
  533. assert.Nil(t, ls.txAddr[1].Load(), "lane survived losing the direct path")
  534. }
  535. // A peer answers lane traffic it can't place with a recv_error from the lane
  536. // socket it arrived on, so the spoof guard must accept one from an up lane's
  537. // target and still refuse one from anywhere else.
  538. func TestHandleRecvErrorFromLane(t *testing.T) {
  539. hostMap := newHostMap(test.NewLogger())
  540. ifce := newLaneTestInterface(hostMap)
  541. initR, _ := runTestHandshake(t)
  542. ls := newTestLaneSet(t, initR, 4, 4, 4242, 4)
  543. hi := newTestLaneHostInfo(t, initR, ls)
  544. hostMap.unlockedAddHostInfo(hi, ifce)
  545. now := time.Now()
  546. ifce.probeLanes(hi, now, make([]byte, 12), make([]byte, mtu))
  547. ls.mu.Lock()
  548. gen := ls.probe[1].gen
  549. ls.mu.Unlock()
  550. laneAddr, promoted := ls.noteAck(1, gen, now)
  551. require.True(t, promoted)
  552. require.NotEqual(t, hi.GetRemote(), laneAddr)
  553. h := &header.H{Type: header.RecvError, RemoteIndex: hi.remoteIndexId}
  554. ifce.handleRecvError(netip.MustParseAddrPort("192.0.2.1:9999"), h)
  555. require.NotNil(t, hostMap.QueryReverseIndex(hi.remoteIndexId), "a recv_error from an unknown port closed the tunnel")
  556. ifce.handleRecvError(laneAddr, h)
  557. assert.Nil(t, hostMap.QueryReverseIndex(hi.remoteIndexId), "a recv_error from an up lane's target was ignored")
  558. }
  559. // An idle tunnel is not probed, so its lanes' keepalives can't keep it alive.
  560. // Probing resumes on the first tick that sees traffic.
  561. func TestIdleTunnelSkipsLaneProbes(t *testing.T) {
  562. hostMap := newHostMap(test.NewLogger())
  563. ifce := newLaneTestInterface(hostMap)
  564. initR, _ := runTestHandshake(t)
  565. ls := newTestLaneSet(t, initR, 4, 4, 4242, 4)
  566. hi := newTestLaneHostInfo(t, initR, ls)
  567. hostMap.unlockedAddHostInfo(hi, ifce)
  568. // An empty pool fails every cert, which with pki.disconnect_invalid off
  569. // leaves the tunnel up.
  570. ifce.pki.caPool.Store(cert.NewCAPool())
  571. cm := ifce.connectionManager
  572. p := []byte("")
  573. nb := make([]byte, 12)
  574. out := make([]byte, mtu)
  575. probed := func() bool {
  576. ls.mu.Lock()
  577. defer ls.mu.Unlock()
  578. return !ls.probe[1].sentAt.IsZero()
  579. }
  580. // Adding the tunnel records outbound traffic; drop it so the tick sees idle.
  581. hi.takeTraffic()
  582. cm.doTrafficCheck(hi.localIndexId, p, nb, out, time.Now())
  583. assert.False(t, probed(), "an idle tunnel was probed")
  584. cm.In(hi)
  585. cm.doTrafficCheck(hi.localIndexId, p, nb, out, time.Now())
  586. assert.True(t, probed(), "a tunnel with traffic was not probed")
  587. }
  588. // A lane probe is answered on the base tunnel, echoing the header's lane, so a
  589. // peer cannot get us to vouch for a lane it never probed.
  590. func TestHandleLaneProbe(t *testing.T) {
  591. hostMap := newHostMap(test.NewLogger())
  592. ifce := newLaneTestInterface(hostMap)
  593. initR, respR := runTestHandshake(t)
  594. ls := newTestLaneSet(t, respR, 4, 4, 4242, 4)
  595. hi := newTestLaneHostInfo(t, respR, ls)
  596. rxc := &rxContext{nb: make([]byte, 12), scratch: make([]byte, mtu)}
  597. sent := &recordingUdpConn{}
  598. ifce.writers = []udp.Conn{sent, &udp.NoopConn{}, &udp.NoopConn{}, &udp.NoopConn{}}
  599. // The payload's lane is ignored in favour of the header's.
  600. ifce.handleLaneProbe(hi, 2, []byte{3, 42}, rxc)
  601. require.Len(t, sent.bufs, 1, "the ack must ride the base tunnel's socket")
  602. h := &header.H{}
  603. require.NoError(t, h.Parse(sent.bufs[0]))
  604. assert.Equal(t, header.Test, h.Type)
  605. assert.Equal(t, header.LaneProbeAck, h.Subtype)
  606. assert.Equal(t, uint8(0), h.Lane(), "the ack is base-tunnel traffic")
  607. assert.False(t, hi.sentSinceCheck(), "a probe ack must not count as tunnel traffic")
  608. pt, err := laneSessionFor(t, newTestLaneSet(t, initR, 4, 4, 4242, 4), 1).dKey.DecryptDanger(
  609. nil, sent.bufs[0][:header.Len], sent.bufs[0][header.Len:], h.MessageCounter, make([]byte, 12))
  610. _ = pt
  611. assert.Error(t, err, "the ack must not be readable with a lane key")
  612. // A probe claiming lane 0 or with a truncated payload is answered with
  613. // nothing at all.
  614. sent.bufs = nil
  615. ifce.handleLaneProbe(hi, 0, []byte{1, 2}, rxc)
  616. ifce.handleLaneProbe(hi, 1, []byte{1}, rxc)
  617. assert.Empty(t, sent.bufs)
  618. }
  619. func TestHandleLaneProbeAck(t *testing.T) {
  620. hostMap := newHostMap(test.NewLogger())
  621. ifce := newLaneTestInterface(hostMap)
  622. initR, _ := runTestHandshake(t)
  623. ls := newTestLaneSet(t, initR, 4, 4, 5353, 4)
  624. hi := newTestLaneHostInfo(t, initR, ls)
  625. now := time.Now()
  626. ls.demand[1].Store(true)
  627. ifce.probeLanes(hi, now, make([]byte, 12), make([]byte, mtu))
  628. ls.mu.Lock()
  629. gen := ls.probe[1].gen
  630. ls.mu.Unlock()
  631. // A short or out-of-range ack is ignored, and a peer without lanes does not
  632. // panic the handler.
  633. ifce.handleLaneProbeAck(hi, []byte{1})
  634. ifce.handleLaneProbeAck(hi, []byte{99, gen})
  635. ifce.handleLaneProbeAck(&HostInfo{}, []byte{1, gen})
  636. assert.Nil(t, ls.txAddr[1].Load())
  637. ifce.handleLaneProbeAck(hi, []byte{1, gen})
  638. assert.NotNil(t, ls.txAddr[1].Load())
  639. }
  640. // recordingUdpConn records what was written to it, one datagram per write.
  641. type recordingUdpConn struct {
  642. udp.NoopConn
  643. bufs [][]byte
  644. dsts []netip.AddrPort
  645. }
  646. func (c *recordingUdpConn) WriteTo(b []byte, addr netip.AddrPort) error {
  647. c.bufs = append(c.bufs, append([]byte(nil), b...))
  648. c.dsts = append(c.dsts, addr)
  649. return nil
  650. }
  651. // recordingBatchConn is a udp.Conn that records the batches flushed to it, so a
  652. // test can see which socket a packet left on.
  653. type recordingBatchConn struct {
  654. udp.NoopConn
  655. bufs [][]byte
  656. dsts []netip.AddrPort
  657. }
  658. func (c *recordingBatchConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort) (int, error) {
  659. for i := range bufs {
  660. c.bufs = append(c.bufs, append([]byte(nil), bufs[i]...))
  661. c.dsts = append(c.dsts, addrs[i])
  662. }
  663. return len(bufs), nil
  664. }
  665. // laneOfFlow is what the TX path computes to pick a lane, without needing a
  666. // promoted lane to observe it.
  667. func laneOfFlow(ls *laneSet, p *firewall.Packet) int {
  668. return int((laneFlowHash(p) + uint32(ls.laneBias)) % uint32(ls.txLanes))
  669. }
  670. func testFlow(port uint16) *firewall.Packet {
  671. return &firewall.Packet{
  672. LocalAddr: testMyAddr,
  673. RemoteAddr: testPeerAddr,
  674. LocalPort: port,
  675. RemotePort: 443,
  676. Protocol: 6,
  677. }
  678. }
  679. // flowForLane finds a flow that hashes onto lane s, so a test can aim traffic at
  680. // a specific lane.
  681. func flowForLane(t *testing.T, ls *laneSet, s int) *firewall.Packet {
  682. t.Helper()
  683. for port := uint16(1024); port < 4096; port++ {
  684. p := testFlow(port)
  685. if laneOfFlow(ls, p) == s {
  686. return p
  687. }
  688. }
  689. t.Fatalf("no flow hashes onto lane %d", s)
  690. return nil
  691. }
  692. func TestSendInsideMessageLaneSwap(t *testing.T) {
  693. hostMap := newHostMap(test.NewLogger())
  694. ifce := newLaneTestInterface(hostMap)
  695. ifce.multiport = true
  696. ifce.laneCount = 4
  697. conns := []*recordingBatchConn{{}, {}, {}, {}}
  698. ifce.writers = []udp.Conn{conns[0], conns[1], conns[2], conns[3]}
  699. initR, respR := runTestHandshake(t)
  700. ls := newTestLaneSet(t, initR, 4, 4, 5353, 4)
  701. hi := newTestLaneHostInfo(t, initR, ls)
  702. peerLS := newTestLaneSet(t, respR, 4, 4, 4242, 4)
  703. tx := ifce.newTxQueue(1)
  704. pkt := tio.Packet{Bytes: []byte{0x45, 0, 0, 4, 1, 2, 3, 4}}
  705. nb := make([]byte, 12)
  706. flow1 := flowForLane(t, ls, 1)
  707. for s := range ls.demand {
  708. ls.demand[s].Store(false)
  709. }
  710. // A down lane rides the base tunnel and asks for a probe.
  711. ifce.sendInsideMessage(hi, pkt, flow1, nb, tx)
  712. tx.flush(ifce)
  713. require.Len(t, conns[0].bufs, 1)
  714. assert.Empty(t, conns[1].bufs)
  715. assert.True(t, ls.demand[1].Load(), "a miss did not raise demand")
  716. assert.False(t, ls.demand[2].Load(), "demand raised on a lane nothing hashed onto")
  717. h := &header.H{}
  718. require.NoError(t, h.Parse(conns[0].bufs[0]))
  719. assert.Equal(t, uint8(0), h.Lane())
  720. // Once the lane is up, a flow hashing onto it rides the lane session, the
  721. // lane socket and the lane's destination, tagged with the lane index.
  722. // Promotion normally happens on the ack of a probe, which is also what
  723. // derived the session, so stand both up here.
  724. laneRemote := netip.MustParseAddrPort("192.0.2.1:5354")
  725. laneSessionFor(t, ls, 1)
  726. ls.txAddr[1].Store(&laneRemote)
  727. ls.demand[1].Store(false)
  728. ifce.sendInsideMessage(hi, pkt, flow1, nb, tx)
  729. tx.flush(ifce)
  730. require.Len(t, conns[1].bufs, 1)
  731. assert.Equal(t, laneRemote, conns[1].dsts[0])
  732. assert.False(t, ls.demand[1].Load(), "a hit raised demand")
  733. require.NoError(t, h.Parse(conns[1].bufs[0]))
  734. assert.Equal(t, uint8(1), h.Lane())
  735. assert.Equal(t, hi.remoteIndexId, h.RemoteIndex)
  736. // And the peer's derived lane-1 session is what opens it.
  737. pt, err := laneSessionFor(t, peerLS, 1).Decrypt(test.NewLogger(), h.MessageCounter, conns[1].bufs[0], nb)
  738. require.NoError(t, err)
  739. assert.Equal(t, pkt.Bytes, pt)
  740. // Another routine sending the same flow rides the same lane socket and
  741. // session: the lane follows the flow, not the routine.
  742. tx2 := ifce.newTxQueue(2)
  743. ifce.sendInsideMessage(hi, pkt, flow1, nb, tx2)
  744. tx2.flush(ifce)
  745. require.Len(t, conns[1].bufs, 2)
  746. require.NoError(t, h.Parse(conns[1].bufs[1]))
  747. assert.Equal(t, uint8(1), h.Lane())
  748. // A flow that hashes onto lane 0 stays on the base tunnel even with lane 1
  749. // up: a full share of flows belongs there.
  750. ifce.sendInsideMessage(hi, pkt, flowForLane(t, ls, 0), nb, tx)
  751. tx.flush(ifce)
  752. require.Len(t, conns[0].bufs, 2)
  753. assert.Equal(t, hi.GetRemote(), conns[0].dsts[1])
  754. require.NoError(t, h.Parse(conns[0].bufs[1]))
  755. assert.Equal(t, uint8(0), h.Lane())
  756. // Demotion falls back to the base tunnel on the very next packet.
  757. ls.txAddr[1].Store(nil)
  758. ifce.sendInsideMessage(hi, pkt, flow1, nb, tx)
  759. tx.flush(ifce)
  760. require.Len(t, conns[0].bufs, 3)
  761. require.Len(t, conns[1].bufs, 2)
  762. }
  763. // Without multiport nothing about the send path changes: every socket shares one
  764. // port under SO_REUSEPORT, so a routine keeps writing to its own, and there are
  765. // no lane batches at all. A peer that negotiated no lanes is the same story on a
  766. // node that does run multiport, except that base traffic owes its 4-tuple to
  767. // socket 0.
  768. func TestTxQueueWithoutLanes(t *testing.T) {
  769. hostMap := newHostMap(test.NewLogger())
  770. ifce := newLaneTestInterface(hostMap)
  771. conns := []*recordingBatchConn{{}, {}, {}, {}}
  772. ifce.writers = []udp.Conn{conns[0], conns[1], conns[2], conns[3]}
  773. initR, _ := runTestHandshake(t)
  774. hi := newTestLaneHostInfo(t, initR, nil)
  775. pkt := tio.Packet{Bytes: []byte{0x45, 0, 0, 4, 1, 2, 3, 4}}
  776. nb := make([]byte, 12)
  777. flow := testFlow(1234)
  778. tx := ifce.newTxQueue(2)
  779. assert.Empty(t, tx.lane, "a queue with no lanes must not hold lane batches")
  780. require.Len(t, tx.live, 1)
  781. ifce.sendInsideMessage(hi, pkt, flow, nb, tx)
  782. tx.flush(ifce)
  783. assert.Len(t, conns[2].bufs, 1, "routine 2 did not send on its own socket")
  784. assert.Empty(t, conns[0].bufs)
  785. // Multiport on, but this peer has no lanes: socket 0, lane 0, one batch.
  786. ifce.multiport = true
  787. ifce.laneCount = 4
  788. tx = ifce.newTxQueue(2)
  789. require.Len(t, tx.live, 1)
  790. ifce.sendInsideMessage(hi, pkt, flow, nb, tx)
  791. tx.flush(ifce)
  792. require.Len(t, conns[0].bufs, 1, "base traffic left a socket other than 0")
  793. assert.Len(t, conns[2].bufs, 1)
  794. h := &header.H{}
  795. require.NoError(t, h.Parse(conns[0].bufs[0]))
  796. assert.Equal(t, uint8(0), h.Lane())
  797. assert.Len(t, tx.live, 1, "a lane batch was built for a peer with no lanes")
  798. }
  799. // `routines` is per port, so a port's sockets are a group sharing it through
  800. // SO_REUSEPORT and every routine has a sibling of its own on every port. Pin the
  801. // arithmetic that turns (queue, lane) into one of them.
  802. func TestLaneSockGroupLayout(t *testing.T) {
  803. hostMap := newHostMap(test.NewLogger())
  804. ifce := newLaneTestInterface(hostMap)
  805. // Two routines per port over three ports: writers[s*2+r] is the r'th socket
  806. // on port listen.port+s.
  807. ifce.routinesPerPort = 2
  808. ifce.routines = 6
  809. ifce.laneCount = 3
  810. // Off, every socket is on the one port, so a routine keeps writing to its own
  811. // and any of them can carry any lane.
  812. for q := 0; q < ifce.routines; q++ {
  813. assert.Equal(t, q, ifce.egressSock(q))
  814. for s := 0; s < ifce.laneCount; s++ {
  815. assert.Equal(t, q, ifce.laneSock(q, s))
  816. }
  817. }
  818. ifce.multiport = true
  819. for q := 0; q < ifce.routines; q++ {
  820. // Base traffic is lane 0's port, and each routine has its own socket
  821. // there rather than sharing one with the other five.
  822. assert.Equal(t, q%2, ifce.egressSock(q), "queue %d", q)
  823. for s := 0; s < ifce.laneCount; s++ {
  824. assert.Equal(t, s*2+q%2, ifce.laneSock(q, s), "queue %d lane %d", q, s)
  825. }
  826. }
  827. // Sibling routines land on different sockets of the same port, so a lane's
  828. // port is served by its whole group and not by one socket.
  829. assert.NotEqual(t, ifce.laneSock(0, 2), ifce.laneSock(1, 2))
  830. assert.Equal(t, ifce.laneSock(0, 2), ifce.laneSock(2, 2), "queues 0 and 2 share a group position")
  831. }
  832. // The lane batches a routine builds must flush to that routine's own sockets, not
  833. // to whatever socket happens to be indexed by the lane.
  834. func TestTxQueueLaneBatchesFollowTheGroup(t *testing.T) {
  835. hostMap := newHostMap(test.NewLogger())
  836. ifce := newLaneTestInterface(hostMap)
  837. ifce.multiport = true
  838. ifce.routinesPerPort = 2
  839. ifce.routines = 4
  840. ifce.laneCount = 2
  841. conns := []*recordingBatchConn{{}, {}, {}, {}}
  842. ifce.writers = []udp.Conn{conns[0], conns[1], conns[2], conns[3]}
  843. initR, _ := runTestHandshake(t)
  844. ls := newTestLaneSet(t, initR, 2, 2, 5353, 2)
  845. hi := newTestLaneHostInfo(t, initR, ls)
  846. laneSessionFor(t, ls, 1)
  847. laneRemote := netip.MustParseAddrPort("192.0.2.1:5354")
  848. ls.txAddr[1].Store(&laneRemote)
  849. pkt := tio.Packet{Bytes: []byte{0x45, 0, 0, 4, 1, 2, 3, 4}}
  850. nb := make([]byte, 12)
  851. flow1 := flowForLane(t, ls, 1)
  852. // Routine 1 is the second socket in each group, so its lane 1 traffic leaves
  853. // writers[1*2+1].
  854. tx := ifce.newTxQueue(1)
  855. ifce.sendInsideMessage(hi, pkt, flow1, nb, tx)
  856. tx.flush(ifce)
  857. require.Len(t, conns[3].bufs, 1, "lane 1 did not leave routine 1's socket on the lane port")
  858. assert.Equal(t, laneRemote, conns[3].dsts[0])
  859. assert.Empty(t, conns[2].bufs)
  860. // Routine 0 sends the same flow out the other socket on that same port.
  861. tx0 := ifce.newTxQueue(0)
  862. ifce.sendInsideMessage(hi, pkt, flow1, nb, tx0)
  863. tx0.flush(ifce)
  864. require.Len(t, conns[2].bufs, 1, "sibling routine shared a socket instead of its own")
  865. assert.Len(t, conns[3].bufs, 1)
  866. }
  867. // One routine has to be able to reach every lane, since the kernel decides which
  868. // routine sees a flow and it may well hand one routine everything.
  869. func TestTxLaneForFlowSpread(t *testing.T) {
  870. initR, _ := runTestHandshake(t)
  871. ls := newTestLaneSet(t, initR, 4, 4, 5353, 4)
  872. seen := map[int]int{}
  873. for port := uint16(1024); port < 1224; port++ {
  874. s := laneOfFlow(ls, testFlow(port))
  875. require.Less(t, s, ls.txLanes)
  876. seen[s]++
  877. }
  878. assert.Len(t, seen, 4, "200 flows did not reach all four lanes: %v", seen)
  879. for s, n := range seen {
  880. assert.Greater(t, n, 10, "lane %d got a negligible share of flows", s)
  881. }
  882. // A flow's lane is a function of its 5-tuple alone, so it never moves
  883. // mid-flow.
  884. p := testFlow(1234)
  885. assert.Equal(t, laneOfFlow(ls, p), laneOfFlow(ls, p))
  886. // A tunnel with nothing but the base lane always answers lane 0.
  887. one := newTestLaneSet(t, initR, 1, 1, 5353, 4)
  888. s, ci, addr := one.txLaneForFlow(p)
  889. assert.Zero(t, s)
  890. assert.Nil(t, ci)
  891. assert.False(t, addr.IsValid())
  892. s, ci, _ = (*laneSet)(nil).txLaneForFlow(p)
  893. assert.Zero(t, s)
  894. assert.Nil(t, ci)
  895. }
  896. // The two ends of a flow have to pick partner lanes, or the flow's two
  897. // directions use unrelated 4-tuples instead of exact reverses and neither side's
  898. // traffic arrives through the conntrack entry the other's probe opened.
  899. func TestLaneFlowSymmetry(t *testing.T) {
  900. initR, respR := runTestHandshake(t)
  901. // Our side is the low address; the peer builds the same pair reversed.
  902. const ourBase, peerBase = 4242, 5353
  903. ourLS := newTestLaneSet(t, initR, 4, 4, peerBase, 4)
  904. respR.PeerPortCount, respR.PeerBasePort, respR.PeerTxLanes = 4, ourBase, 4
  905. peerLS := newLaneSet(respR, 4, testPeerAddr, testMyAddr)
  906. require.NotNil(t, peerLS)
  907. require.Equal(t, uint16(0), ourLS.laneBias, "the low address hashes straight")
  908. partnered := 0
  909. for port := uint16(1024); port < 1124; port++ {
  910. p := testFlow(port)
  911. reverse := &firewall.Packet{
  912. LocalAddr: p.RemoteAddr, RemoteAddr: p.LocalAddr,
  913. LocalPort: p.RemotePort, RemotePort: p.LocalPort,
  914. Protocol: p.Protocol,
  915. }
  916. ours := laneOfFlow(ourLS, p)
  917. theirs := laneOfFlow(peerLS, reverse)
  918. require.Equal(t, (ours+int(ourLS.portOffset))%4, theirs,
  919. "flow on port %d: our lane %d is not partnered with their lane %d", port, ours, theirs)
  920. if ours == 0 || theirs == 0 {
  921. // The lane that lands on the other side's base port has no partner
  922. // lane; that flow rides one side's base tunnel. See lanePortOffset.
  923. continue
  924. }
  925. partnered++
  926. // Our source and destination ports are the reverse of theirs.
  927. ourLS.mu.Lock()
  928. peerLS.mu.Lock()
  929. assert.Equal(t, uint16(peerBase+theirs), ourLS.laneTargetPortLocked(ours), "flow on port %d", port)
  930. assert.Equal(t, uint16(ourBase+ours), peerLS.laneTargetPortLocked(theirs), "flow on port %d", port)
  931. peerLS.mu.Unlock()
  932. ourLS.mu.Unlock()
  933. }
  934. assert.Greater(t, partnered, 0, "no flow landed on a partnered lane")
  935. }
  936. // Regression: deleting a hostinfo whose pending entry is NOT the one recorded
  937. // in vpnIps must not evict a concurrently pending handshake for that address.
  938. func TestHandshakeManagerVpnIpsIdentityDelete(t *testing.T) {
  939. l := test.NewLogger()
  940. hostMap := newHostMap(l)
  941. lh := newTestLighthouse()
  942. hm := NewHandshakeManager(l, hostMap, lh, &udp.NoopConn{}, defaultHandshakeConfig)
  943. vpnIp := netip.MustParseAddr("172.1.1.4")
  944. pendingBase := hm.StartHandshake(vpnIp, nil)
  945. require.NotNil(t, pendingBase)
  946. other := &HostInfo{vpnAddrs: []netip.Addr{vpnIp}, localIndexId: 999}
  947. hm.DeleteHostInfo(other)
  948. // The pending base handshake must still be tracked.
  949. assert.Equal(t, pendingBase, hm.QueryVpnAddr(vpnIp))
  950. // And deleting the actual owner still works.
  951. hm.DeleteHostInfo(pendingBase)
  952. assert.Nil(t, hm.QueryVpnAddr(vpnIp))
  953. }
  954. func TestValidLaneAdvert(t *testing.T) {
  955. assert.True(t, validLaneAdvert(4242, 4))
  956. assert.True(t, validLaneAdvert(65535, 1))
  957. assert.True(t, validLaneAdvert(65532, 4), "a range ending on 65535 is valid")
  958. assert.False(t, validLaneAdvert(0, 4), "no base port")
  959. assert.False(t, validLaneAdvert(4242, 0), "no ports")
  960. assert.False(t, validLaneAdvert(65533, 4), "the range runs past 65535")
  961. }
  962. // A peer advertising a port range that would wrap gets a plain tunnel, and a
  963. // warning saying why, rather than lanes aimed at low ports.
  964. func TestMaybeAllocLanesRejectsInvalidRange(t *testing.T) {
  965. hostMap := newHostMap(test.NewLogger())
  966. ifce := newLaneTestInterface(hostMap)
  967. ifce.myVpnAddrs = []netip.Addr{testMyAddr}
  968. hm := ifce.handshakeManager
  969. hm.config.laneCount = 4
  970. var buf bytes.Buffer
  971. hm.l = test.NewLoggerWithOutputAndLevel(&buf, slog.LevelWarn)
  972. initR, _ := runTestHandshake(t)
  973. hi := &HostInfo{vpnAddrs: []netip.Addr{testPeerAddr}}
  974. initR.PeerBasePort, initR.PeerPortCount, initR.PeerTxLanes = 65530, 16, 4
  975. hm.maybeAllocLanes(hi, initR)
  976. assert.Nil(t, hi.lanes)
  977. assert.Contains(t, buf.String(), "invalid port range")
  978. hi = &HostInfo{vpnAddrs: []netip.Addr{testPeerAddr}}
  979. initR.PeerBasePort = 4242
  980. hm.maybeAllocLanes(hi, initR)
  981. assert.NotNil(t, hi.lanes)
  982. }
  983. // A lane that has never come up warns once after laneWarnFails unanswered
  984. // probes, and not again while it keeps failing.
  985. func TestLaneNotComingUpWarnsOnce(t *testing.T) {
  986. hostMap := newHostMap(test.NewLogger())
  987. ifce := newLaneTestInterface(hostMap)
  988. var buf bytes.Buffer
  989. ifce.l = test.NewLoggerWithOutputAndLevel(&buf, slog.LevelWarn)
  990. initR, _ := runTestHandshake(t)
  991. ls := newTestLaneSet(t, initR, 2, 2, 4242, 2)
  992. hi := newTestLaneHostInfo(t, initR, ls)
  993. nb := make([]byte, 12)
  994. out := make([]byte, mtu)
  995. now := time.Now()
  996. const msg = "Multiport lane not coming up"
  997. for i := 1; i <= laneWarnFails+3; i++ {
  998. ls.demand[1].Store(true)
  999. ifce.probeLanes(hi, now, nb, out)
  1000. now = now.Add(laneProbeTimeout)
  1001. ifce.probeLanes(hi, now, nb, out)
  1002. now = now.Add(laneRetryMax)
  1003. if i < laneWarnFails {
  1004. require.NotContains(t, buf.String(), msg, "warned after %d failures", i)
  1005. }
  1006. }
  1007. assert.Equal(t, 1, strings.Count(buf.String(), msg))
  1008. }