1
0

lanes.go 28 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739
  1. package nebula
  2. import (
  3. "context"
  4. "hash/fnv"
  5. "log/slog"
  6. "net/netip"
  7. "sync"
  8. "sync/atomic"
  9. "time"
  10. "github.com/flynn/noise"
  11. "github.com/rcrowley/go-metrics"
  12. "github.com/slackhq/nebula/cert"
  13. "github.com/slackhq/nebula/firewall"
  14. "github.com/slackhq/nebula/handshake"
  15. "github.com/slackhq/nebula/header"
  16. "github.com/slackhq/nebula/noiseutil"
  17. )
  18. // Multiport lanes give one tunnel several underlay 5-tuples, so its traffic
  19. // spreads over ECMP paths, NIC receive queues and per-flow policers instead of
  20. // funnelling through a single flow. Each inside flow picks a lane by hashing its
  21. // own 5-tuple, so the spread doesn't depend on how a kernel steers tun queues
  22. // (see txQueue).
  23. //
  24. // A lane is not a second tunnel: it is an extra session on the same HostInfo.
  25. // Noise leaves us with A.eKey == B.dKey, so both sides expand the same two keys
  26. // with the same per-lane label and land on a matched pair without exchanging
  27. // anything. A lane therefore costs no handshake, has no half-established state,
  28. // and dies exactly when its base tunnel does. Which lane a packet belongs to
  29. // travels in the nebula header, inside the AEAD's associated data.
  30. //
  31. // Lane 0 is the base tunnel itself: HostInfo.ConnectionState, the base port, and
  32. // the peer's real remote address, and it carries its share of flows like any
  33. // other. Lane s > 0 egresses a socket on listen.port+s (see laneSock) toward the
  34. // peer's advertised port range. Receiving on a lane
  35. // needs no permission — the keys are derivable the moment the base handshake
  36. // completes — but sending on one needs proof the new 5-tuple actually works,
  37. // since nothing else would notice a middlebox quietly dropping it. So a lane
  38. // stays down until a probe on it is acked, and falls back to the base tunnel the
  39. // moment it stops being acked.
  40. //
  41. // Both directions are pay-per-use. A lane session is derived on the first packet
  42. // that needs it, because how many lanes exist is partly the peer's call: it
  43. // advertises how many it sends on, and we have to be able to receive all of
  44. // them. Deriving them all up front would let a peer advertising the maximum cost
  45. // us a replay window and two cipher states per lane, per tunnel, for lanes it
  46. // may never send on.
  47. const (
  48. // laneKeyInfo is the HKDF label prefix for lane key expansion. Changing it
  49. // means older builds derive different keys and drop our lane traffic; the
  50. // base tunnel would keep working, so the failure would be a silent loss of
  51. // lanes rather than of connectivity.
  52. laneKeyInfo = "nebula multiport lane v1"
  53. // laneRetryBase and laneRetryMax bound the backoff between probes of a lane
  54. // that will not come up, so a peer whose lane ports are firewalled costs one
  55. // packet a minute rather than one per traffic tick.
  56. laneRetryBase = 5 * time.Second
  57. laneRetryMax = 60 * time.Second
  58. // laneMaxFails caps the failure counter; the backoff saturates well before.
  59. laneMaxFails = 8
  60. // laneProbeTimeout is how long a probe may go unacked before it counts as a
  61. // failure. It is shorter than the connection manager's check interval on
  62. // purpose: an outstanding probe is judged on the next tick either way, and a
  63. // longer timeout would only delay that by a whole tick.
  64. laneProbeTimeout = 2 * time.Second
  65. // laneKeepalive is how often a lane that is up re-proves its path. Traffic
  66. // on a lane is not evidence the lane works — that is the whole reason lanes
  67. // need probing — so a lane that silently breaks is only caught here.
  68. laneKeepalive = 30 * time.Second
  69. )
  70. // laneSet holds a peer's lane sessions and the state deciding which lanes may
  71. // carry traffic. It is built when the base handshake completes and never
  72. // resized, so the slices and their lengths are immutable; mu guards the fields
  73. // under it, and sessions/txAddr/demand are atomics read by the data plane
  74. // without it.
  75. type laneSet struct {
  76. // sessions[s] holds lane s's session once something has needed it, and nil
  77. // until then. sessions[0] is never populated: lane 0 is the base tunnel's own
  78. // ConnectionState. The length is immutable, so the data plane bounds-checks
  79. // and loads with no locking.
  80. sessions []atomic.Pointer[ConnectionState]
  81. // material is what a lane session is derived from, kept because sessions are
  82. // derived lazily and the handshake result is long gone by then.
  83. material laneMaterial
  84. // txAddr[s] holds lane s's remote address while the lane is proven usable
  85. // and nil otherwise. This single atomic is both the TX gate and the
  86. // destination, so a routine that loads non-nil has everything it needs.
  87. // Sized txLanes: lanes above that never send.
  88. txAddr []atomic.Pointer[netip.AddrPort]
  89. // demand[s] is raised at creation, and again by the TX path whenever a flow
  90. // hashes onto lane s while it is down. Probing is demand-driven, so a peer we
  91. // never send to costs nothing beyond its base tunnel no matter how many lanes
  92. // are configured, and a lane that keeps failing is only retried while
  93. // something still wants it. Sized txLanes.
  94. demand []atomic.Bool
  95. // txLanes is how many lanes we may send on — our lane count clamped to the
  96. // ports the peer bound — and so the modulus a flow's hash is reduced by.
  97. // Lanes from txLanes up can only receive, which is how a peer with more
  98. // routines than us still spreads its own traffic.
  99. // Immutable, and the length of every TX-side slice here.
  100. txLanes int
  101. mu sync.Mutex
  102. // peerPortCount and peerBasePort are the peer's advertised port range and
  103. // portOffset is this pair's rotation within it: lane s targets
  104. // peerBasePort + ((s + portOffset) % peerPortCount).
  105. peerPortCount uint16
  106. peerBasePort uint16
  107. portOffset uint16
  108. // laneBias rotates the flow hash before it picks a lane, so the two sides of
  109. // a flow land on lanes that are each other's partner rather than at
  110. // independent points in the range. See newLaneSet.
  111. laneBias uint16
  112. // peerAddr is the address the current lane targets were built from. The
  113. // peer's lane ports have no derivable relationship to a new NAT mapping, so
  114. // a roam invalidates every lane rather than moving it.
  115. peerAddr netip.Addr
  116. // probe[s] is lane s's probe and backoff state. Sized txLanes.
  117. probe []laneProbeState
  118. }
  119. // laneMaterial is everything a lane session is derived from. The two base keys
  120. // are the same secret the base tunnel's own cipher states already hold — the
  121. // noiseutil.CipherState interface just doesn't hand them back, so a lane set
  122. // keeps its own copy rather than a reference to the session.
  123. type laneMaterial struct {
  124. eKey, dKey [32]byte
  125. cipher noise.CipherFunc
  126. myCert cert.Certificate
  127. peerCert *cert.CachedCertificate
  128. initiator bool
  129. }
  130. type laneProbeState struct {
  131. // gen is the generation of the last probe sent, echoed in the ack so a late
  132. // ack cannot promote a lane on the strength of a superseded probe.
  133. gen uint8
  134. // fails is the consecutive failure count driving retryAt.
  135. fails uint8
  136. // sentAt is when the outstanding probe went out, zero when none is pending.
  137. sentAt time.Time
  138. // target is where the outstanding probe went, promoted to txAddr on ack. It
  139. // outlives the probe, which is what lets a demotion log the address that
  140. // stopped answering.
  141. target netip.AddrPort
  142. // lastAck is when the lane was last confirmed usable, driving the keepalive.
  143. lastAck time.Time
  144. // retryAt is the earliest we may probe this lane again.
  145. retryAt time.Time
  146. }
  147. // newLaneSet sets up the lanes for a freshly completed base handshake. It
  148. // returns nil when the pair has no lane beyond the base tunnel, which is the
  149. // normal answer for a peer running without multiport. No session is derived
  150. // here; each is derived on the first packet that needs it.
  151. func newLaneSet(r *handshake.Result, myLanes int, myAddr, peerAddr netip.Addr) *laneSet {
  152. // PeerPortCount and PeerBasePort are already bounded to uint16 by the
  153. // handshake payload parser. A zero port count is a peer that did not
  154. // advertise multiport at all, so there is no lane to be had in either
  155. // direction.
  156. peerPorts := uint16(r.PeerPortCount)
  157. if peerPorts == 0 {
  158. return nil
  159. }
  160. // Sessions have to cover both directions: we send on our lanes and receive
  161. // on the peer's, and one derived session serves both ends of a lane index.
  162. // Only the session table is sized by the peer's advertised count; the TX-side
  163. // state is sized by what we will actually send on.
  164. n := min(max(myLanes, int(r.PeerTxLanes)), header.MaxLane+1)
  165. if n < 2 {
  166. return nil
  167. }
  168. txLanes := min(myLanes, int(peerPorts), n)
  169. offset := lanePortOffset(myAddr, peerAddr, peerPorts)
  170. // A flow picks its lane from a hash both sides compute identically, so with
  171. // the ranges lined up the two directions of a flow would pick the same lane
  172. // index — and lane s targets the peer's lane (s + portOffset), not lane s. The
  173. // high-addressed side rotates its choice by the low side's offset, which is
  174. // its own negated, so the two directions land on partner lanes and their
  175. // 4-tuples are exact reverses: each side's traffic then arrives through the
  176. // conntrack entry the other's probe opened. With mismatched ranges there are
  177. // no partner lanes to find, so don't pretend: hash straight.
  178. bias := uint16(0)
  179. if txLanes == int(peerPorts) && peerAddr.Less(myAddr) {
  180. bias = (peerPorts - offset) % peerPorts
  181. }
  182. ls := &laneSet{
  183. sessions: make([]atomic.Pointer[ConnectionState], n),
  184. material: laneMaterial{
  185. eKey: r.EKey.UnsafeKey(),
  186. dKey: r.DKey.UnsafeKey(),
  187. cipher: r.Cipher,
  188. myCert: r.MyCert,
  189. peerCert: r.RemoteCert,
  190. initiator: r.Initiator,
  191. },
  192. txAddr: make([]atomic.Pointer[netip.AddrPort], txLanes),
  193. demand: make([]atomic.Bool, txLanes),
  194. probe: make([]laneProbeState, txLanes),
  195. txLanes: txLanes,
  196. peerPortCount: peerPorts,
  197. peerBasePort: uint16(r.PeerBasePort),
  198. portOffset: offset,
  199. laneBias: bias,
  200. }
  201. // Every lane starts out demanded, so the first traffic tick on this tunnel
  202. // probes all of them at once instead of waiting for a flow to hash onto each.
  203. // Lanes have to be up *before* the flows are, not after: the peer writes a
  204. // flow's inbound packets to the tun queue matching the socket they arrived on,
  205. // and the kernel remembers that for as long as the flow stays busy. A flow that
  206. // starts while the lanes are still down therefore gets pinned to queue 0 on
  207. // both hosts for its whole life. This costs one probe per lane on any tunnel
  208. // with traffic; an idle tunnel is never probed, so it still costs nothing.
  209. for s := 1; s < txLanes; s++ {
  210. ls.demand[s].Store(true)
  211. }
  212. return ls
  213. }
  214. // laneLogAttr summarizes what a tunnel negotiated, for the handshake log lines.
  215. // It is an empty attr, which slog drops, on a node not running multiport. A peer
  216. // that negotiated no lanes still logs, with zeros: "we offered and got nothing"
  217. // is exactly what you want to see when you expected lanes and have none.
  218. func laneLogAttr(myLanes int, ls *laneSet) slog.Attr {
  219. if myLanes == 0 {
  220. return slog.Attr{}
  221. }
  222. if ls == nil {
  223. return slog.Any("lanes", m{"tx": 0, "sessions": 0})
  224. }
  225. // Every field read here is immutable once the set is built.
  226. return slog.Any("lanes", m{
  227. "tx": ls.txLanes,
  228. "sessions": len(ls.sessions),
  229. "peerBasePort": ls.peerBasePort,
  230. "peerPorts": ls.peerPortCount,
  231. "portOffset": ls.portOffset,
  232. })
  233. }
  234. // emitLaneStats reports how many lanes are carrying traffic and how many tunnels
  235. // have any. Both are counted by walking the hostmap, because a counter kept at
  236. // promotion and demotion would drift upward forever: a tunnel torn down while its
  237. // lanes are up never demotes them. Only a node running multiport pays for the
  238. // walk.
  239. func (f *Interface) emitLaneStats(up, tunnels metrics.Gauge) {
  240. var nUp, nTunnels int64
  241. f.hostMap.ForEachIndex(func(hostinfo *HostInfo) {
  242. ls := hostinfo.lanes
  243. if ls == nil {
  244. return
  245. }
  246. nTunnels++
  247. for s := 1; s < ls.txLanes; s++ {
  248. if ls.txAddr[s].Load() != nil {
  249. nUp++
  250. }
  251. }
  252. })
  253. up.Update(nUp)
  254. tunnels.Update(nTunnels)
  255. }
  256. // lanePortOffset returns the rotation applied to this pair's lane target ports,
  257. // in [0, peerPortCount). Without it every low-routine peer would aim its few
  258. // lanes at a big peer's first few ports, concentrating the big peer's receive
  259. // work on a couple of sockets; the hash spreads pairs across the whole range.
  260. //
  261. // Both sides hash the same sorted vpn-address pair and the higher address
  262. // negates the result, so when port counts match the two sides' rotations
  263. // cancel: our lane s's 4-tuple stays the reverse of the peer's lane s, and each
  264. // side's probe opens the conntrack entry the other's arrives through. (The one
  265. // lane a nonzero rotation lands on the peer's base port has no partner lane, so
  266. // behind a port-restricted NAT it is the one lane that may never come up.)
  267. func lanePortOffset(myAddr, peerAddr netip.Addr, peerPortCount uint16) uint16 {
  268. if peerPortCount == 0 {
  269. return 0
  270. }
  271. lo, hi := myAddr, peerAddr
  272. if hi.Less(lo) {
  273. lo, hi = hi, lo
  274. }
  275. h := fnv.New32a()
  276. b := lo.As16()
  277. h.Write(b[:])
  278. b = hi.As16()
  279. h.Write(b[:])
  280. o := uint16(h.Sum32() % uint32(peerPortCount))
  281. if myAddr == hi {
  282. o = (peerPortCount - o) % peerPortCount
  283. }
  284. return o
  285. }
  286. // laneSession returns the session to decrypt a lane s packet with, deriving one
  287. // if this is the first packet to claim that lane. A nil session with no error
  288. // means this tunnel has no lane s at all.
  289. //
  290. // cached reports whether the session was already in the table. A fresh one is
  291. // deliberately left out of it: anyone who can spoof this tunnel's local index
  292. // can name any lane, and installing on sight would let them make us hold a
  293. // replay window and two cipher states per lane without authenticating anything.
  294. // The caller must install with installSession once the packet decrypts, which is
  295. // the first moment the lane is known to be real.
  296. func (i *HostInfo) laneSession(s uint8) (ci *ConnectionState, cached bool, err error) {
  297. ls := i.lanes
  298. if ls == nil || s == 0 || int(s) >= len(ls.sessions) {
  299. return nil, false, nil
  300. }
  301. if cs := ls.sessions[s].Load(); cs != nil {
  302. return cs, true, nil
  303. }
  304. cs, err := newLaneConnectionState(&ls.material, s)
  305. if err != nil {
  306. return nil, false, err
  307. }
  308. return cs, false, nil
  309. }
  310. // installSession publishes a session derived by laneSession, so the next packet
  311. // on the lane doesn't have to derive it again. cs must have already decrypted the
  312. // packet at messageCounter.
  313. //
  314. // Two routines can race on a lane's first packet and derive a session each. The
  315. // loser's is dropped, and with it the replay-window entry for the packet it just
  316. // accepted, so hand that counter to the session that survives — the keys are
  317. // identical, so it is the same window in every respect that matters.
  318. func (ls *laneSet) installSession(l *slog.Logger, s uint8, cs *ConnectionState, messageCounter uint64) {
  319. if ls.sessions[s].CompareAndSwap(nil, cs) {
  320. return
  321. }
  322. ls.sessions[s].Load().noteSeen(l, messageCounter)
  323. }
  324. // session returns lane s's session for our own use, deriving and installing it
  325. // if it doesn't exist yet. Unlike the RX path this needs no proof the lane is
  326. // real: we only ask for lanes we chose to send on. s must be a lane this set
  327. // covers.
  328. func (ls *laneSet) session(s int) (*ConnectionState, error) {
  329. if cs := ls.sessions[s].Load(); cs != nil {
  330. return cs, nil
  331. }
  332. cs, err := newLaneConnectionState(&ls.material, uint8(s))
  333. if err != nil {
  334. return nil, err
  335. }
  336. if !ls.sessions[s].CompareAndSwap(nil, cs) {
  337. return ls.sessions[s].Load(), nil
  338. }
  339. return cs, nil
  340. }
  341. // maxMessageCounter returns the highest counter across the base session and
  342. // every lane session. Data rides the lanes, so the base counter alone would
  343. // never reach the rehandshake or exhaustion thresholds and the lane keys would
  344. // be used past their data-volume margin. Rolling the base tunnel replaces the
  345. // lane keys with it, since lanes are derived from it.
  346. func (i *HostInfo) maxMessageCounter() uint64 {
  347. if i.ConnectionState == nil {
  348. return 0
  349. }
  350. c := i.ConnectionState.messageCounter.Load()
  351. if ls := i.lanes; ls != nil {
  352. for s := range ls.sessions {
  353. cs := ls.sessions[s].Load()
  354. if cs == nil {
  355. // Never derived, so it has never sent anything either.
  356. continue
  357. }
  358. if lc := cs.messageCounter.Load(); lc > c {
  359. c = lc
  360. }
  361. }
  362. }
  363. return c
  364. }
  365. // txLane returns the session and destination for lane s, or a nil session when
  366. // the lane is down and the caller must use the base tunnel. A miss raises
  367. // demand, which is what gets a down lane probed again, so we pay for a lane
  368. // exactly where real traffic wanted one. Callers on the data plane come through
  369. // txLaneForFlow.
  370. func (ls *laneSet) txLane(s int) (*ConnectionState, netip.AddrPort) {
  371. if ls == nil || s <= 0 || s >= ls.txLanes {
  372. return nil, netip.AddrPort{}
  373. }
  374. if addr := ls.txAddr[s].Load(); addr != nil {
  375. // The lane is only up because a probe was acked on it, and that probe
  376. // derived the session, so this load cannot miss. Fall back rather than
  377. // derive here anyway: this is the hot path and a nil is not worth an HKDF.
  378. if cs := ls.sessions[s].Load(); cs != nil {
  379. return cs, *addr
  380. }
  381. return nil, netip.AddrPort{}
  382. }
  383. // Load-guarded so the common case of a lane that will not come up is a
  384. // plain read and cannot ping-pong the cache line these flags share.
  385. if !ls.demand[s].Load() {
  386. ls.demand[s].Store(true)
  387. }
  388. return nil, netip.AddrPort{}
  389. }
  390. // txLaneForFlow picks the lane a flow rides and returns it with its session and
  391. // destination, or lane 0 and a nil session when the flow belongs on the base
  392. // tunnel — either because the hash landed on lane 0 or because the lane it
  393. // landed on is down.
  394. //
  395. // The lane comes from the flow rather than from the sending routine so that lane
  396. // use doesn't depend on how the kernel steers tun queues; see txQueue. It is a
  397. // pure function of the 5-tuple, so a flow stays on one lane for its life: no
  398. // per-packet reordering, and one lane's replay window sees one set of flows.
  399. func (ls *laneSet) txLaneForFlow(p *firewall.Packet) (int, *ConnectionState, netip.AddrPort) {
  400. if ls == nil || ls.txLanes < 2 {
  401. return 0, nil, netip.AddrPort{}
  402. }
  403. s := int((laneFlowHash(p) + uint32(ls.laneBias)) % uint32(ls.txLanes))
  404. if s == 0 {
  405. // Lane 0 is the base tunnel, and a full share of flows belongs on it.
  406. return 0, nil, netip.AddrPort{}
  407. }
  408. cs, addr := ls.txLane(s)
  409. return s, cs, addr
  410. }
  411. // laneFlowHash hashes a 5-tuple to the same value from either end of the flow,
  412. // which is what lets both peers pick partner lanes for it (see laneBias). FNV-1a
  413. // by hand rather than through hash/fnv: this runs per packet, and the interface
  414. // there would escape the addresses to the heap.
  415. func laneFlowHash(p *firewall.Packet) uint32 {
  416. // Order the two endpoints so the direction of travel cannot change the hash.
  417. aAddr, aPort := p.LocalAddr, p.LocalPort
  418. bAddr, bPort := p.RemoteAddr, p.RemotePort
  419. if bAddr.Less(aAddr) || (aAddr == bAddr && bPort < aPort) {
  420. aAddr, aPort, bAddr, bPort = bAddr, bPort, aAddr, aPort
  421. }
  422. const prime = 16777619
  423. h := uint32(2166136261)
  424. x, y := aAddr.As16(), bAddr.As16()
  425. for i := range x {
  426. h = (h ^ uint32(x[i])) * prime
  427. h = (h ^ uint32(y[i])) * prime
  428. }
  429. for _, b := range [5]byte{byte(aPort >> 8), byte(aPort), byte(bPort >> 8), byte(bPort), p.Protocol} {
  430. h = (h ^ uint32(b)) * prime
  431. }
  432. return h
  433. }
  434. // laneTargetPortLocked returns the peer port lane s aims at. Only meaningful
  435. // when peerPortCount is nonzero, which txLanes > 0 guarantees.
  436. func (ls *laneSet) laneTargetPortLocked(s int) uint16 {
  437. return ls.peerBasePort + uint16((s+int(ls.portOffset))%int(ls.peerPortCount))
  438. }
  439. // laneTargetLocked returns where lane s's probes go: the peer port this lane is
  440. // paired with, on the peer's current direct address.
  441. //
  442. // A lane only ever aims at its own port. There is no fallback to the peer's base
  443. // port when the lane port doesn't answer — a lane sharing the base port's
  444. // destination gains only a source port of its own, while costing the peer the
  445. // receive spread that is the whole point, so a lane that can't reach its port
  446. // stays down and its traffic rides the base tunnel.
  447. func (ls *laneSet) laneTargetLocked(s int, addr netip.Addr) netip.AddrPort {
  448. return netip.AddrPortFrom(addr, ls.laneTargetPortLocked(s))
  449. }
  450. // laneRetryDelay is the backoff after fails consecutive probe failures.
  451. func laneRetryDelay(fails uint8) time.Duration {
  452. d := laneRetryBase << min(fails, 4)
  453. if d > laneRetryMax {
  454. d = laneRetryMax
  455. }
  456. return d
  457. }
  458. // noteAck records an acked probe for lane s, promoting the lane if it was down.
  459. // gen must match the outstanding probe. Reports the target the lane came up on,
  460. // and whether this ack is what promoted it.
  461. func (ls *laneSet) noteAck(s int, gen uint8, now time.Time) (netip.AddrPort, bool) {
  462. if s <= 0 || s >= len(ls.probe) {
  463. return netip.AddrPort{}, false
  464. }
  465. ls.mu.Lock()
  466. defer ls.mu.Unlock()
  467. p := &ls.probe[s]
  468. if p.sentAt.IsZero() || p.gen != gen {
  469. // No probe outstanding, or an ack for a probe we have already given up
  470. // on. Either way it says nothing about the lane's current path.
  471. return netip.AddrPort{}, false
  472. }
  473. p.sentAt = time.Time{}
  474. p.lastAck = now
  475. p.fails = 0
  476. p.retryAt = time.Time{}
  477. target := p.target
  478. if ls.txAddr[s].Load() != nil {
  479. // Keepalive for a lane already up.
  480. return target, false
  481. }
  482. ls.txAddr[s].Store(&target)
  483. return target, true
  484. }
  485. // probeLanes runs one lane maintenance pass for a peer: it demotes lanes whose
  486. // probe went unanswered, re-proves lanes that have been up a while without one,
  487. // and probes down lanes the data plane asked for. Driven by the connection
  488. // manager's per-tunnel traffic tick, which skips a tunnel that is idle in both
  489. // directions.
  490. func (f *Interface) probeLanes(hostinfo *HostInfo, now time.Time, nb, out []byte) {
  491. ls := hostinfo.lanes
  492. if ls == nil || ls.txLanes < 2 {
  493. return
  494. }
  495. remote := hostinfo.GetRemote()
  496. ls.mu.Lock()
  497. defer ls.mu.Unlock()
  498. if !remote.IsValid() {
  499. // Relayed, or otherwise without a direct path. Lanes are direct-only,
  500. // so drop them all; a later tick rebuilds if a direct path returns.
  501. ls.resetLocked()
  502. return
  503. }
  504. if ls.peerAddr != remote.Addr() {
  505. if ls.peerAddr.IsValid() {
  506. // A roam is a new path, not a failure: forget the lanes built on
  507. // the old one and let demand re-probe from a clean backoff. On the
  508. // first pass there is nothing built yet, so just record the address.
  509. ls.resetLocked()
  510. }
  511. ls.peerAddr = remote.Addr()
  512. }
  513. for s := 1; s < ls.txLanes; s++ {
  514. p := &ls.probe[s]
  515. up := ls.txAddr[s].Load() != nil
  516. if !p.sentAt.IsZero() {
  517. if now.Sub(p.sentAt) < laneProbeTimeout {
  518. continue
  519. }
  520. // An aged-out probe is a failure whether it was bringing the lane
  521. // up or keeping it up.
  522. p.sentAt = time.Time{}
  523. p.fails = min(p.fails+1, laneMaxFails)
  524. p.retryAt = now.Add(laneRetryDelay(p.fails))
  525. if up {
  526. ls.txAddr[s].Store(nil)
  527. hostinfo.logger(f.l).Info("Multiport lane demoted, probe unanswered", "lane", s, "udpAddr", p.target)
  528. }
  529. continue
  530. }
  531. if up {
  532. if now.Sub(p.lastAck) < laneKeepalive {
  533. continue
  534. }
  535. } else if now.Before(p.retryAt) || !ls.demand[s].Swap(false) {
  536. continue
  537. }
  538. p.gen++
  539. p.target = ls.laneTargetLocked(s, remote.Addr())
  540. if f.sendLaneProbe(hostinfo, s, p.gen, p.target, nb, out) {
  541. p.sentAt = now
  542. } else {
  543. p.fails = min(p.fails+1, laneMaxFails)
  544. p.retryAt = now.Add(laneRetryDelay(p.fails))
  545. }
  546. }
  547. }
  548. // resetLocked takes every lane down and clears its probe state, without
  549. // counting it as a failure.
  550. //
  551. // The probe generation survives the reset. Starting it over would let a late ack
  552. // for a probe sent before the reset match the first probe sent after it, and
  553. // promote the lane to a target nothing has proven.
  554. //
  555. // Demand is deliberately left standing: it records that a routine has real
  556. // traffic for this peer, which a roam or a relay detour does not change. Keeping
  557. // it re-probes the lanes that were actually carrying data as soon as a path
  558. // exists again, while a lane whose routine has gone quiet stays down.
  559. func (ls *laneSet) resetLocked() {
  560. for s := 1; s < len(ls.probe); s++ {
  561. ls.txAddr[s].Store(nil)
  562. ls.probe[s] = laneProbeState{gen: ls.probe[s].gen}
  563. }
  564. }
  565. // isLaneProbe reports whether t/st is a lane probe or its ack. Probing is upkeep
  566. // the lanes do for themselves, not traffic: counting it would let a lane's own
  567. // keepalive keep an idle tunnel from ever being dropped.
  568. func isLaneProbe(t header.MessageType, st header.MessageSubType) bool {
  569. return t == header.Test && (st == header.LaneProbe || st == header.LaneProbeAck)
  570. }
  571. // isLaneRemote reports whether addr is the target of one of our up lanes. The
  572. // peer answers a lane packet from the socket it arrived on, so a recv_error for
  573. // lane traffic comes from that lane's target rather than the base remote.
  574. func (ls *laneSet) isLaneRemote(addr netip.AddrPort) bool {
  575. if ls == nil {
  576. return false
  577. }
  578. for s := 1; s < ls.txLanes; s++ {
  579. if a := ls.txAddr[s].Load(); a != nil && *a == addr {
  580. return true
  581. }
  582. }
  583. return false
  584. }
  585. // sendLaneProbe sends a probe on lane s to addr from a socket on lane s's port.
  586. // The probe is an ordinary Test packet encrypted with the lane's session, so an
  587. // ack proves the whole lane: our source port reached the peer, its reply reached
  588. // us, and the keys we derived for this lane match the ones it derived. Reports
  589. // whether the probe made it onto the wire.
  590. //
  591. // It goes out the first socket on the port, since the connection manager runs
  592. // this and has no queue of its own; every socket on the port has the same address,
  593. // so the probe proves the path for whichever one the data plane picks.
  594. func (f *Interface) sendLaneProbe(hostinfo *HostInfo, s int, gen uint8, addr netip.AddrPort, nb, out []byte) bool {
  595. // The first probe on a lane is what derives its session.
  596. ci, err := hostinfo.lanes.session(s)
  597. if err != nil {
  598. hostinfo.logger(f.l).Error("Failed to derive multiport lane session", "error", err, "lane", s)
  599. return false
  600. }
  601. if ci == nil || ci.eKey == nil {
  602. return false
  603. }
  604. if noiseutil.EncryptLockNeeded {
  605. ci.writeLock.Lock()
  606. }
  607. c, ok := ci.NextMessageCounter()
  608. if !ok {
  609. if noiseutil.EncryptLockNeeded {
  610. ci.writeLock.Unlock()
  611. }
  612. f.dropExhausted(hostinfo, c, "Dropping multiport lane probe, lane message counter is exhausted")
  613. return false
  614. }
  615. b := header.EncodeLane(out[:0], header.Version, header.Test, header.LaneProbe, hostinfo.remoteIndexId, c, uint8(s))
  616. b, err = ci.eKey.EncryptDanger(b, b, []byte{uint8(s), gen}, c, nb)
  617. if noiseutil.EncryptLockNeeded {
  618. ci.writeLock.Unlock()
  619. }
  620. if err != nil {
  621. hostinfo.logger(f.l).Error("Failed to encrypt multiport lane probe", "error", err, "lane", s)
  622. return false
  623. }
  624. f.messageMetrics.Tx(header.Test, header.LaneProbe, 1)
  625. if err := f.writers[f.laneSock(0, s)].WriteTo(b, addr); err != nil {
  626. hostinfo.logger(f.l).Error("Failed to send multiport lane probe", "error", err, "lane", s, "udpAddr", addr)
  627. return false
  628. }
  629. if f.l.Enabled(context.Background(), slog.LevelDebug) {
  630. hostinfo.logger(f.l).Debug("Multiport lane probe sent", "lane", s, "gen", gen, "udpAddr", addr)
  631. }
  632. return true
  633. }
  634. // handleLaneProbe answers a peer's lane probe. The ack rides the base tunnel on
  635. // purpose: a probe proves the peer's lane s works in its send direction, and
  636. // answering on our own lane s would make the result depend on a second path
  637. // that may be broken independently.
  638. func (f *Interface) handleLaneProbe(hostinfo *HostInfo, lane uint8, payload []byte, rxc *rxContext) {
  639. if lane == 0 || len(payload) < 2 {
  640. return
  641. }
  642. // Echo the header's lane rather than the payload's, so a peer cannot get us
  643. // to vouch for a lane it did not actually probe.
  644. f.send(header.Test, header.LaneProbeAck, hostinfo.ConnectionState, hostinfo,
  645. []byte{lane, payload[1]}, rxc.nb, rxc.scratch[:0])
  646. }
  647. // handleLaneProbeAck promotes the lane a peer just acked.
  648. func (f *Interface) handleLaneProbeAck(hostinfo *HostInfo, payload []byte) {
  649. ls := hostinfo.lanes
  650. if ls == nil || len(payload) < 2 {
  651. return
  652. }
  653. // The target is worth logging next to the demotion that names the same
  654. // address, so a flapping lane can be read off the logs.
  655. if target, promoted := ls.noteAck(int(payload[0]), payload[1], time.Now()); promoted {
  656. hostinfo.logger(f.l).Info("Multiport lane up", "lane", payload[0], "udpAddr", target)
  657. }
  658. }