inside.go 24 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671
  1. package nebula
  2. import (
  3. "context"
  4. "fmt"
  5. "io"
  6. "log/slog"
  7. "net/netip"
  8. "github.com/slackhq/nebula/firewall"
  9. "github.com/slackhq/nebula/header"
  10. "github.com/slackhq/nebula/iputil"
  11. "github.com/slackhq/nebula/noiseutil"
  12. "github.com/slackhq/nebula/overlay/tio"
  13. "github.com/slackhq/nebula/routing"
  14. )
  15. func (f *Interface) consumeInsidePacket(pkt tio.Packet, fwPacket *firewall.ParsedPacket, nb []byte, tx *txQueue, rejectBuf []byte, q int, localCache firewall.ConntrackCache) {
  16. // borrowed: pkt.Bytes is owned by the originating tio.Queue and is
  17. // only valid until the next Read on that queue. Every consumer below
  18. // (parse, self-forward, handshake cache, sendInsideMessage) reads it
  19. // synchronously; do not retain pkt outside this call. If a future
  20. // caller needs to keep the packet, use pkt.Clone() to detach it from
  21. // the borrow.
  22. //
  23. // pkt.Bytes is either one IP datagram (GSO zero) or a TSO/USO
  24. // superpacket. In both cases the L3+L4 headers at the start describe
  25. // the same 5-tuple every segment will share, so a single newPacket /
  26. // firewall check covers the whole superpacket.
  27. packet := pkt.Bytes
  28. err := newPacket(packet, false, fwPacket)
  29. if err != nil {
  30. if f.l.Enabled(context.Background(), slog.LevelDebug) {
  31. f.l.Debug("Error while validating outbound packet",
  32. "packet", packet,
  33. "error", err,
  34. )
  35. }
  36. return
  37. }
  38. // Ignore local broadcast packets
  39. if f.dropLocalBroadcast {
  40. if f.myBroadcastAddrsTable.Contains(fwPacket.RemoteAddr) {
  41. return
  42. }
  43. }
  44. if f.myVpnAddrsTable.Contains(fwPacket.RemoteAddr) {
  45. // Immediately forward packets from self to self.
  46. // This should only happen on Darwin-based and FreeBSD hosts, which
  47. // routes packets from the Nebula addr to the Nebula addr through the Nebula
  48. // TUN device.
  49. if immediatelyForwardToSelf {
  50. // Write copies into the kernel queue synchronously, so seg's lifetime ends at return.
  51. // A self-forwarded superpacket would be re-handed to the
  52. // kernel as one giant blob; segment first so the loopback
  53. // path sees one IP datagram per Write.
  54. err := tio.SegmentSuperpacket(pkt, func(seg []byte) error {
  55. // The kernel may have left the transport checksum for hardware
  56. // offload to finish; nothing between here and the tun will.
  57. iputil.SetTransportChecksum(seg)
  58. _, werr := f.queues[q].Write(seg)
  59. return werr
  60. })
  61. if err != nil {
  62. f.l.Error("Failed to forward to tun", "error", err)
  63. }
  64. }
  65. // Otherwise, drop. On linux, we should never see these packets - Linux
  66. // routes packets from the nebula addr to the nebula addr through the loopback device.
  67. return
  68. }
  69. // Ignore multicast packets
  70. if f.dropMulticast && fwPacket.RemoteAddr.IsMulticast() {
  71. return
  72. }
  73. hostinfo, ready := f.getOrHandshakeConsiderRouting(&fwPacket.Packet, func(hh *HandshakeHostInfo) {
  74. // borrowed: SegmentSuperpacket builds each segment in the kernel-supplied pkt
  75. // bytes underneath. cachePacket explicitly copies its argument (handshake_manager.go cachePacket),
  76. // so retaining segments past the loop is safe.
  77. err := tio.SegmentSuperpacket(pkt, func(seg []byte) error {
  78. hh.cachePacket(f.l, header.Message, 0, seg, f.sendMessageNow, f.cachedPacketMetrics)
  79. return nil
  80. })
  81. if err != nil && f.l.Enabled(context.Background(), slog.LevelDebug) {
  82. f.l.Debug("Failed to segment superpacket for handshake cache",
  83. "error", err,
  84. "vpnAddr", fwPacket.RemoteAddr,
  85. )
  86. }
  87. })
  88. if hostinfo == nil {
  89. f.rejectInside(packet, rejectBuf, q)
  90. if f.l.Enabled(context.Background(), slog.LevelDebug) {
  91. f.l.Debug("dropping outbound packet, vpnAddr not in our vpn networks or in unsafe networks",
  92. "vpnAddr", fwPacket.RemoteAddr,
  93. "fwPacket", fwPacket,
  94. )
  95. }
  96. return
  97. }
  98. if !ready {
  99. return
  100. }
  101. dropReason := f.firewall.Drop(fwPacket.Packet, false, hostinfo, f.pki.GetCAPool(), localCache)
  102. if dropReason == nil {
  103. f.sendInsideMessage(hostinfo, pkt, &fwPacket.Packet, nb, tx)
  104. } else {
  105. f.rejectInside(packet, rejectBuf, q)
  106. if f.l.Enabled(context.Background(), slog.LevelDebug) {
  107. hostinfo.logger(f.l).Debug("dropping outbound packet",
  108. "fwPacket", fwPacket,
  109. "reason", dropReason,
  110. )
  111. }
  112. }
  113. }
  114. func (f *Interface) sendInsideEncrypt(hostinfo *HostInfo, ci *ConnectionState, lane uint8, seg, scratch, nb []byte) []byte {
  115. if noiseutil.EncryptLockNeeded {
  116. ci.writeLock.Lock()
  117. }
  118. c := ci.messageCounter.Add(1)
  119. out := header.EncodeLane(scratch, header.Version, header.Message, 0, hostinfo.remoteIndexId, c, lane)
  120. out, encErr := ci.eKey.EncryptDanger(out, out, seg, c, nb)
  121. if noiseutil.EncryptLockNeeded {
  122. ci.writeLock.Unlock()
  123. }
  124. if encErr != nil {
  125. hostinfo.logger(f.l).Error("Failed to encrypt outgoing packet",
  126. "error", encErr,
  127. "udpAddr", hostinfo.GetRemote(),
  128. "counter", c,
  129. )
  130. // Skip this segment; the rest of the superpacket can still go out. TCP will retransmit anything we drop here.
  131. return nil
  132. }
  133. return out
  134. }
  135. // sendInsideMessage encrypts a firewall-approved inside packet (or every
  136. // segment of a TSO/USO superpacket) into the caller's batch slot for
  137. // later sendmmsg flush. Segmentation is fused with encryption here so the
  138. // kernel-supplied superpacket bytes never get written into a separate
  139. // scratch arena: SegmentSuperpacket builds each segment's plaintext in
  140. // segScratch[:segLen] in turn, and we encrypt directly into a fresh SendBatch slot.
  141. //
  142. // When this flow has a usable multiport lane to this peer, the direct path swaps
  143. // to that lane's session and socket below. Relay and base traffic stays on
  144. // tx.base, this routine's socket on the base port.
  145. func (f *Interface) sendInsideMessage(hostinfo *HostInfo, pkt tio.Packet, fwPacket *firewall.Packet, nb []byte, tx *txQueue) {
  146. ci := hostinfo.ConnectionState
  147. if ci.eKey == nil {
  148. return
  149. }
  150. // Base and relay traffic stays on the base port; the direct path may swap to tx.lane below.
  151. sendBatch := tx.base
  152. // One traffic-out mark covers every segment of the superpacket; doing it
  153. // per segment in sendInsideEncrypt paid an atomic store up to ~45 extra
  154. // times per TSO packet, inside writeLock under boring crypto.
  155. //
  156. // We rebound since this tunnel last sent, ask the lighthouse to get the far side punching at us again
  157. if f.connectionManager.Out(hostinfo) {
  158. f.lightHouse.QueryServer(hostinfo.vpnAddrs[0])
  159. if f.l.Enabled(context.Background(), slog.LevelDebug) {
  160. hostinfo.logger(f.l).Debug("Lighthouse update triggered for punch due to rebind epoch",
  161. "vpnAddrs", hostinfo.vpnAddrs,
  162. )
  163. }
  164. }
  165. remote := hostinfo.GetRemote()
  166. if !remote.IsValid() { //the relay path
  167. //first, find our relay hostinfo:
  168. var relayHostInfo *HostInfo
  169. var relay *Relay
  170. var err error
  171. for _, relayIP := range hostinfo.relayState.CopyRelayIps() {
  172. relayHostInfo, relay, err = f.hostMap.QueryVpnAddrsRelayFor(hostinfo.vpnAddrs, relayIP)
  173. if err != nil {
  174. hostinfo.relayState.DeleteRelay(relayIP)
  175. hostinfo.logger(f.l).Info("sendNoMetrics failed to find HostInfo",
  176. "relay", relayIP,
  177. "error", err,
  178. )
  179. continue
  180. }
  181. break
  182. }
  183. if relayHostInfo == nil || relay == nil {
  184. //failure already logged
  185. return
  186. }
  187. err = tio.SegmentSuperpacket(pkt, func(seg []byte) error {
  188. //relay header + header + plaintext + AEAD tag (16 bytes for both AES-GCM and ChaCha20-Poly1305) + relay tag
  189. scratch := sendBatch.Reserve(header.Len + header.Len + len(seg) + 16 + 16)
  190. innerPacket := f.sendInsideEncrypt(hostinfo, ci, 0, seg, scratch[header.Len:], nb)
  191. if innerPacket == nil {
  192. return nil
  193. }
  194. //now we need to do a relay-encrypt:
  195. toSend, err := f.prepareSendVia(relayHostInfo, relay, innerPacket, nb, scratch, true)
  196. if err != nil {
  197. //already logged
  198. return nil
  199. }
  200. sendBatch.Commit(toSend, relayHostInfo.GetRemote())
  201. return nil
  202. })
  203. if err != nil {
  204. hostinfo.logger(f.l).Error("Failed to segment superpacket for relay send", "error", err)
  205. }
  206. return
  207. }
  208. // Direct path: prefer this flow's multiport lane once it is proven usable.
  209. // txLaneForFlow hands back the lane's session and destination together, so
  210. // there is no window where one is set and the other is not, and a demotion
  211. // drops us back onto the base tunnel on the very next packet.
  212. //
  213. // A miss is also how a lane gets re-probed after a demotion: txLane raises
  214. // demand, which the connection manager's next tick on this tunnel picks up.
  215. // Until the lane is up the traffic rides the base tunnel, the same fallback
  216. // a demoted lane uses.
  217. lane := uint8(0)
  218. if s, lci, laneRemote := hostinfo.lanes.txLaneForFlow(fwPacket); lci != nil {
  219. lane = uint8(s)
  220. ci = lci
  221. remote = laneRemote
  222. sendBatch = tx.laneBatch(f, s)
  223. }
  224. err := tio.SegmentSuperpacket(pkt, func(seg []byte) error {
  225. // header + plaintext + AEAD tag (16 bytes for both AES-GCM and ChaCha20-Poly1305)
  226. scratch := sendBatch.Reserve(header.Len + len(seg) + 16)
  227. out := f.sendInsideEncrypt(hostinfo, ci, lane, seg, scratch, nb)
  228. if out == nil {
  229. return nil
  230. }
  231. sendBatch.Commit(out, remote)
  232. return nil
  233. })
  234. if err != nil {
  235. hostinfo.logger(f.l).Error("Failed to segment superpacket for send", "error", err)
  236. }
  237. }
  238. func (f *Interface) rejectInside(packet []byte, out []byte, q int) {
  239. if !f.firewall.OutboundSendReject {
  240. return
  241. }
  242. out = iputil.CreateRejectPacket(packet, out)
  243. if len(out) == 0 {
  244. return
  245. }
  246. _, err := f.queues[q].Write(out)
  247. if err != nil {
  248. f.l.Error("Failed to write to tun", "error", err)
  249. }
  250. }
  251. func (f *Interface) rejectOutside(packet []byte, ci *ConnectionState, hostinfo *HostInfo, nb, rejectBuf []byte, q int) {
  252. if !f.firewall.InboundSendReject {
  253. return
  254. }
  255. // split rejectBuf to make sure we have room to write the plaintext rejection, then encrypt it, without trampling anything
  256. // we can't re-use packet, if we need to send an icmp reject, it won't be long enough.
  257. half := len(rejectBuf) / 2
  258. encryptBuf := rejectBuf[0:0:half] //the first half of rejectBuf's capacity, len set to 0
  259. buildBuf := rejectBuf[half:]
  260. out := iputil.CreateRejectPacket(packet, buildBuf)
  261. if len(out) == 0 {
  262. return
  263. }
  264. if len(out) > iputil.MaxRejectPacketSize {
  265. if f.l.Enabled(context.Background(), slog.LevelInfo) {
  266. f.l.Info("rejectOutside: packet too big, not sending", "packet", packet, "outPacket", out)
  267. }
  268. return
  269. }
  270. f.sendNoMetrics(header.Message, 0, ci, hostinfo, netip.AddrPort{}, out, nb, encryptBuf, q)
  271. }
  272. // Handshake will attempt to initiate a tunnel with the provided vpn address. This is a no-op if the tunnel is already established or being established
  273. // it does not check if it is within our vpn networks!
  274. func (f *Interface) Handshake(vpnAddr netip.Addr) {
  275. f.handshakeManager.GetOrHandshake(vpnAddr, nil)
  276. }
  277. // getOrHandshakeNoRouting returns nil if the vpnAddr is not routable.
  278. // If the 2nd return var is false then the hostinfo is not ready to be used in a tunnel
  279. func (f *Interface) getOrHandshakeNoRouting(vpnAddr netip.Addr, cacheCallback func(*HandshakeHostInfo)) (*HostInfo, bool) {
  280. if f.myVpnNetworksTable.Contains(vpnAddr) {
  281. return f.handshakeManager.GetOrHandshake(vpnAddr, cacheCallback)
  282. }
  283. return nil, false
  284. }
  285. // getOrHandshakeConsiderRouting will try to find the HostInfo to handle this packet, starting a handshake if necessary.
  286. // If the 2nd return var is false then the hostinfo is not ready to be used in a tunnel.
  287. func (f *Interface) getOrHandshakeConsiderRouting(fwPacket *firewall.Packet, cacheCallback func(*HandshakeHostInfo)) (*HostInfo, bool) {
  288. destinationAddr := fwPacket.RemoteAddr
  289. hostinfo, ready := f.getOrHandshakeNoRouting(destinationAddr, cacheCallback)
  290. // Host is inside the mesh, no routing required
  291. if hostinfo != nil {
  292. return hostinfo, ready
  293. }
  294. gateways := f.inside.RoutesFor(destinationAddr)
  295. switch len(gateways) {
  296. case 0:
  297. return nil, false
  298. case 1:
  299. // Single gateway route
  300. return f.handshakeManager.GetOrHandshake(gateways[0].Addr(), cacheCallback)
  301. default:
  302. // Multi gateway route, perform ECMP categorization
  303. gatewayAddr, balancingOk := routing.BalancePacket(fwPacket, gateways)
  304. if !balancingOk {
  305. // This happens if the gateway buckets were not calculated, this _should_ never happen
  306. f.l.Error("Gateway buckets not calculated, fallback from ECMP to random routing. Please report this bug.")
  307. }
  308. var handshakeInfoForChosenGateway *HandshakeHostInfo
  309. var hhReceiver = func(hh *HandshakeHostInfo) {
  310. handshakeInfoForChosenGateway = hh
  311. }
  312. // Store the handshakeHostInfo for later.
  313. // If this node is not reachable we will attempt other nodes, if none are reachable we will
  314. // cache the packet for this gateway.
  315. if hostinfo, ready = f.handshakeManager.GetOrHandshake(gatewayAddr, hhReceiver); ready {
  316. return hostinfo, true
  317. }
  318. // It appears the selected gateway cannot be reached, find another gateway to fallback on.
  319. // The current implementation breaks ECMP but that seems better than no connectivity.
  320. // If ECMP is also required when a gateway is down then connectivity status
  321. // for each gateway needs to be kept and the weights recalculated when they go up or down.
  322. // This would also need to interact with unsafe_route updates through reloading the config or
  323. // use of the use_system_route_table option
  324. if f.l.Enabled(context.Background(), slog.LevelDebug) {
  325. f.l.Debug("Calculated gateway for ECMP not available, attempting other gateways",
  326. "destination", destinationAddr,
  327. "originalGateway", gatewayAddr,
  328. )
  329. }
  330. for i := range gateways {
  331. // Skip the gateway that failed previously
  332. if gateways[i].Addr() == gatewayAddr {
  333. continue
  334. }
  335. // We do not need the HandshakeHostInfo since we cache the packet in the originally chosen gateway
  336. if hostinfo, ready = f.handshakeManager.GetOrHandshake(gateways[i].Addr(), nil); ready {
  337. return hostinfo, true
  338. }
  339. }
  340. // No gateways reachable, cache the packet in the originally chosen gateway
  341. cacheCallback(handshakeInfoForChosenGateway)
  342. return hostinfo, false
  343. }
  344. }
  345. func (f *Interface) sendMessageNow(t header.MessageType, st header.MessageSubType, hostinfo *HostInfo, p, nb, out []byte) {
  346. fp := &firewall.ParsedPacket{}
  347. err := newPacket(p, false, fp)
  348. if err != nil {
  349. f.l.Warn("error while parsing outgoing packet for firewall check", "error", err)
  350. return
  351. }
  352. // check if packet is in outbound fw rules
  353. dropReason := f.firewall.Drop(fp.Packet, false, hostinfo, f.pki.GetCAPool(), nil)
  354. if dropReason != nil {
  355. if f.l.Enabled(context.Background(), slog.LevelDebug) {
  356. f.l.Debug("dropping cached packet",
  357. "fwPacket", fp,
  358. "reason", dropReason,
  359. )
  360. }
  361. return
  362. }
  363. f.sendNoMetrics(header.Message, st, hostinfo.ConnectionState, hostinfo, netip.AddrPort{}, p, nb, out, 0)
  364. }
  365. // SendMessageToVpnAddr handles real addr:port lookup and sends to the current best known address for vpnAddr.
  366. // This function ignores myVpnNetworksTable, and will always attempt to treat the address as a vpnAddr
  367. func (f *Interface) SendMessageToVpnAddr(t header.MessageType, st header.MessageSubType, vpnAddr netip.Addr, p, nb, out []byte) {
  368. hostInfo, ready := f.handshakeManager.GetOrHandshake(vpnAddr, func(hh *HandshakeHostInfo) {
  369. hh.cachePacket(f.l, t, st, p, f.SendMessageToHostInfo, f.cachedPacketMetrics)
  370. })
  371. if hostInfo == nil {
  372. if f.l.Enabled(context.Background(), slog.LevelDebug) {
  373. f.l.Debug("dropping SendMessageToVpnAddr, vpnAddr not in our vpn networks or in unsafe routes",
  374. "vpnAddr", vpnAddr,
  375. )
  376. }
  377. return
  378. }
  379. if !ready {
  380. return
  381. }
  382. f.SendMessageToHostInfo(t, st, hostInfo, p, nb, out)
  383. }
  384. func (f *Interface) SendMessageToHostInfo(t header.MessageType, st header.MessageSubType, hi *HostInfo, p, nb, out []byte) {
  385. f.send(t, st, hi.ConnectionState, hi, p, nb, out)
  386. }
  387. func (f *Interface) send(t header.MessageType, st header.MessageSubType, ci *ConnectionState, hostinfo *HostInfo, p, nb, out []byte) {
  388. f.messageMetrics.Tx(t, st, 1)
  389. f.sendNoMetrics(t, st, ci, hostinfo, netip.AddrPort{}, p, nb, out, 0)
  390. }
  391. func (f *Interface) sendTo(t header.MessageType, st header.MessageSubType, ci *ConnectionState, hostinfo *HostInfo, remote netip.AddrPort, p, nb, out []byte) {
  392. f.messageMetrics.Tx(t, st, 1)
  393. f.sendNoMetrics(t, st, ci, hostinfo, remote, p, nb, out, 0)
  394. }
  395. // dropExhausted records an exhaustion drop and logs once, on the crossing send, for a spent tunnel.
  396. func (f *Interface) dropExhausted(hostinfo *HostInfo, c uint64, msg string) {
  397. f.messageMetrics.TxExhausted(1)
  398. if c == RejectAfterMessages {
  399. hostinfo.logger(f.l).Error(msg)
  400. }
  401. }
  402. func (f *Interface) prepareSendVia(via *HostInfo,
  403. relay *Relay,
  404. ad,
  405. nb,
  406. out []byte,
  407. nocopy bool,
  408. ) ([]byte, error) {
  409. if noiseutil.EncryptLockNeeded {
  410. // NOTE: for goboring AESGCMTLS we need to lock because of the nonce check
  411. via.ConnectionState.writeLock.Lock()
  412. }
  413. c, ok := via.ConnectionState.NextMessageCounter()
  414. if !ok {
  415. if noiseutil.EncryptLockNeeded {
  416. via.ConnectionState.writeLock.Unlock()
  417. }
  418. f.dropExhausted(via, c, "Dropping outbound relay packets, tunnel message counter is exhausted")
  419. return nil, fmt.Errorf("tunnel message counter is exhausted")
  420. }
  421. out = header.Encode(out, header.Version, header.Message, header.MessageRelay, relay.RemoteIndex, c)
  422. f.connectionManager.OutNoRebind(via)
  423. // Authenticate the header and payload, but do not encrypt for this message type.
  424. // The payload consists of the inner, unencrypted Nebula header, as well as the end-to-end encrypted payload.
  425. if len(out)+len(ad)+via.ConnectionState.eKey.Overhead() > cap(out) {
  426. if noiseutil.EncryptLockNeeded {
  427. via.ConnectionState.writeLock.Unlock()
  428. }
  429. via.logger(f.l).Error("SendVia out buffer not large enough for relay",
  430. "outCap", cap(out),
  431. "payloadLen", len(ad),
  432. "headerLen", len(out),
  433. "cipherOverhead", via.ConnectionState.eKey.Overhead(),
  434. )
  435. return nil, io.ErrShortBuffer
  436. }
  437. // The header bytes are written to the 'out' slice; Grow the slice to hold the header and associated data payload.
  438. offset := len(out)
  439. out = out[:offset+len(ad)]
  440. // In one call path, the associated data _is_ already stored in out. In other call paths, the associated data must
  441. // be copied into 'out'.
  442. if !nocopy {
  443. copy(out[offset:], ad)
  444. }
  445. var err error
  446. out, err = via.ConnectionState.eKey.EncryptDanger(out, out, nil, c, nb)
  447. if noiseutil.EncryptLockNeeded {
  448. via.ConnectionState.writeLock.Unlock()
  449. }
  450. if err != nil {
  451. via.logger(f.l).Info("Failed to EncryptDanger in sendVia", "error", err)
  452. return nil, err
  453. }
  454. f.connectionManager.RelayUsed(relay.LocalIndex)
  455. return out, nil
  456. }
  457. // SendVia sends a payload through a Relay tunnel. No authentication or encryption is done
  458. // to the payload for the ultimate target host, making this a useful method for sending
  459. // handshake messages to peers through relay tunnels.
  460. // via is the HostInfo through which the message is relayed.
  461. // ad is the plaintext data to authenticate, but not encrypt
  462. // nb is a buffer used to store the nonce value, re-used for performance reasons.
  463. // out is a buffer used to store the result of the Encrypt operation
  464. // q indicates which writer to use to send the packet.
  465. func (f *Interface) SendVia(via *HostInfo, relay *Relay, ad, nb, out []byte, nocopy bool, q int) {
  466. toSend, err := f.prepareSendVia(via, relay, ad, nb, out, nocopy)
  467. if err != nil {
  468. // already logged by prepareSendVia
  469. return
  470. }
  471. err = f.writers[f.egressSock(q)].WriteTo(toSend, via.GetRemote())
  472. if err != nil {
  473. via.logger(f.l).Info("Failed to WriteTo in sendVia", "error", err)
  474. }
  475. }
  476. // egressSock picks the socket a tunnel packet leaves from.
  477. //
  478. // Everything that is not lane data plane leaves from the base port: handshakes, keepalives, close packets, rejects and
  479. // relay carriers all belong to the base tunnel's 4-tuple, which is the only one a peer's spoof/roam checks and a
  480. // vanilla peer's expectations know about. Lane data goes through laneSock instead and never comes here.
  481. //
  482. // Which socket on the base port doesn't matter — they share an address, so they produce identical packets — so keep to
  483. // this routine's own share of the group and leave the rest of it uncontended. Without multiport that is q itself, since
  484. // every socket is on the base port.
  485. func (f *Interface) egressSock(q int) int {
  486. return f.laneSock(q, 0)
  487. }
  488. // laneSock returns the index in writers of a socket bound to lane s's port, for a
  489. // routine that reads queue q.
  490. //
  491. // Under multiport the sockets are laid out port-major — writers[s*routinesPerPort
  492. // + r] is the r'th socket on port listen.port+s — so every routine has a sibling
  493. // socket on every port and the arithmetic is a lane index away. Routines pick the
  494. // sibling matching their own position in their group, which spreads the writers
  495. // for one port over that port's whole group rather than funnelling them onto its
  496. // first socket. It is a pure function of (q, s), so a flow always leaves from the
  497. // same socket and cannot reorder itself across two of them.
  498. //
  499. // Without multiport there is one port and every socket is on it, so any lane
  500. // resolves to q's own socket.
  501. func (f *Interface) laneSock(q, s int) int {
  502. if !f.multiport {
  503. return q
  504. }
  505. return s*f.routinesPerPort + q%f.routinesPerPort
  506. }
  507. func (f *Interface) sendNoMetrics(t header.MessageType, st header.MessageSubType, ci *ConnectionState, hostinfo *HostInfo, remote netip.AddrPort, p, nb, out []byte, q int) {
  508. if ci.eKey == nil {
  509. return
  510. }
  511. q = f.egressSock(q)
  512. useRelay := !remote.IsValid() && !hostinfo.GetRemote().IsValid()
  513. fullOut := out
  514. if useRelay {
  515. if len(out) < header.Len {
  516. // out always has a capacity of mtu, but not always a length greater than the header.Len.
  517. // Grow it to make sure the next operation works.
  518. out = out[:header.Len]
  519. }
  520. // Save a header's worth of data at the front of the 'out' buffer.
  521. out = out[header.Len:]
  522. }
  523. if noiseutil.EncryptLockNeeded {
  524. // NOTE: for goboring AESGCMTLS we need to lock because of the nonce check
  525. ci.writeLock.Lock()
  526. }
  527. c, ok := ci.NextMessageCounter()
  528. if !ok {
  529. if noiseutil.EncryptLockNeeded {
  530. ci.writeLock.Unlock()
  531. }
  532. f.dropExhausted(hostinfo, c, "Dropping outbound packets, tunnel message counter is exhausted")
  533. return
  534. }
  535. //l.WithField("trace", string(debug.Stack())).Error("out Header ", &Header{Version, t, st, 0, hostinfo.remoteIndexId, c}, p)
  536. out = header.Encode(out, header.Version, t, st, hostinfo.remoteIndexId, c)
  537. // A closing tunnel is torn down right after this, so skip the connection manager entirely: no point recording
  538. // traffic or asking the lighthouse for a punch. Lane probe acks are not traffic either. Otherwise, if we rebound
  539. // since this tunnel last sent, ask the lighthouse to get the far side punching at us again.
  540. if t != header.CloseTunnel && !isLaneProbe(t, st) && f.connectionManager.Out(hostinfo) {
  541. f.lightHouse.QueryServer(hostinfo.vpnAddrs[0])
  542. if f.l.Enabled(context.Background(), slog.LevelDebug) {
  543. f.l.Debug("Lighthouse update triggered for punch due to rebind epoch",
  544. "vpnAddrs", hostinfo.vpnAddrs,
  545. )
  546. }
  547. }
  548. var err error
  549. out, err = ci.eKey.EncryptDanger(out, out, p, c, nb)
  550. if noiseutil.EncryptLockNeeded {
  551. ci.writeLock.Unlock()
  552. }
  553. if err != nil {
  554. hostinfo.logger(f.l).Error("Failed to encrypt outgoing packet",
  555. "error", err,
  556. "udpAddr", remote,
  557. "counter", c,
  558. )
  559. return
  560. }
  561. if remote.IsValid() {
  562. err = f.writers[q].WriteTo(out, remote)
  563. if err != nil {
  564. hostinfo.logger(f.l).Error("Failed to write outgoing packet",
  565. "error", err,
  566. "udpAddr", remote,
  567. )
  568. }
  569. } else if hr := hostinfo.GetRemote(); hr.IsValid() {
  570. err = f.writers[q].WriteTo(out, hr)
  571. if err != nil {
  572. hostinfo.logger(f.l).Error("Failed to write outgoing packet",
  573. "error", err,
  574. "udpAddr", hr,
  575. )
  576. }
  577. } else {
  578. // Try to send via a relay
  579. for _, relayIP := range hostinfo.relayState.CopyRelayIps() {
  580. relayHostInfo, relay, err := f.hostMap.QueryVpnAddrsRelayFor(hostinfo.vpnAddrs, relayIP)
  581. if err != nil {
  582. hostinfo.relayState.DeleteRelay(relayIP)
  583. hostinfo.logger(f.l).Info("sendNoMetrics failed to find HostInfo",
  584. "relay", relayIP,
  585. "error", err,
  586. )
  587. continue
  588. }
  589. f.SendVia(relayHostInfo, relay, out, nb, fullOut[:header.Len+len(out)], true, q)
  590. break
  591. }
  592. }
  593. }