inside.go 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663
  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, nb, tx, q)
  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, seg, scratch, nb []byte) []byte {
  115. if noiseutil.EncryptLockNeeded {
  116. ci.writeLock.Lock()
  117. }
  118. c := ci.messageCounter.Add(1)
  119. out := header.Encode(scratch, header.Version, header.Message, 0, hostinfo.remoteIndexId, c)
  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. // hostinfo is always the base tunnel (the hostmap resolves by vpn address);
  143. // when routine q has an established lane to this peer, the direct path swaps
  144. // to the lane's session and socket below. Relay and base traffic stays on
  145. // tx.base (socket 0).
  146. func (f *Interface) sendInsideMessage(hostinfo *HostInfo, pkt tio.Packet, nb []byte, tx *txQueue, q int) {
  147. ci := hostinfo.ConnectionState
  148. if ci.eKey == nil {
  149. return
  150. }
  151. // Base and relay traffic stays on socket 0; the direct path may swap to tx.lane below.
  152. sendBatch := tx.base
  153. // One traffic-out mark covers every segment of the superpacket; doing it
  154. // per segment in sendInsideEncrypt paid an atomic store up to ~45 extra
  155. // times per TSO packet, inside writeLock under boring crypto.
  156. //
  157. // We rebound since this tunnel last sent, ask the lighthouse to get the far side punching at us again
  158. if f.connectionManager.Out(hostinfo) {
  159. f.lightHouse.QueryServer(hostinfo.vpnAddrs[0])
  160. if f.l.Enabled(context.Background(), slog.LevelDebug) {
  161. hostinfo.logger(f.l).Debug("Lighthouse update triggered for punch due to rebind epoch",
  162. "vpnAddrs", hostinfo.vpnAddrs,
  163. )
  164. }
  165. }
  166. remote := hostinfo.GetRemote()
  167. if !remote.IsValid() { //the relay path
  168. //first, find our relay hostinfo:
  169. var relayHostInfo *HostInfo
  170. var relay *Relay
  171. var err error
  172. for _, relayIP := range hostinfo.relayState.CopyRelayIps() {
  173. relayHostInfo, relay, err = f.hostMap.QueryVpnAddrsRelayFor(hostinfo.vpnAddrs, relayIP)
  174. if err != nil {
  175. hostinfo.relayState.DeleteRelay(relayIP)
  176. hostinfo.logger(f.l).Info("sendNoMetrics failed to find HostInfo",
  177. "relay", relayIP,
  178. "error", err,
  179. )
  180. continue
  181. }
  182. break
  183. }
  184. if relayHostInfo == nil || relay == nil {
  185. //failure already logged
  186. return
  187. }
  188. err = tio.SegmentSuperpacket(pkt, func(seg []byte) error {
  189. //relay header + header + plaintext + AEAD tag (16 bytes for both AES-GCM and ChaCha20-Poly1305) + relay tag
  190. scratch := sendBatch.Reserve(header.Len + header.Len + len(seg) + 16 + 16)
  191. innerPacket := f.sendInsideEncrypt(hostinfo, ci, seg, scratch[header.Len:], nb)
  192. if innerPacket == nil {
  193. return nil
  194. }
  195. //now we need to do a relay-encrypt:
  196. toSend, err := f.prepareSendVia(relayHostInfo, relay, innerPacket, nb, scratch, true)
  197. if err != nil {
  198. //already logged
  199. return nil
  200. }
  201. sendBatch.Commit(toSend, relayHostInfo.GetRemote())
  202. return nil
  203. })
  204. if err != nil {
  205. hostinfo.logger(f.l).Error("Failed to segment superpacket for relay send", "error", err)
  206. }
  207. return
  208. }
  209. // Direct path: prefer this routine's lane tunnel when it is established.
  210. // The pointer is only published once the lane's ConnectionState is fully
  211. // populated, so a non-nil Load is always usable. On lane death the slot
  212. // CAS-clears and traffic falls back to the base tunnel instantly.
  213. if ls := hostinfo.lanes; ls != nil && q < len(ls.txLanes) {
  214. if lane := ls.txLanes[q].Load(); lane != nil {
  215. if lci := lane.ConnectionState; lci != nil && lci.eKey != nil {
  216. hostinfo = lane
  217. ci = lci
  218. remote = lane.GetRemote()
  219. sendBatch = tx.lane
  220. }
  221. }
  222. }
  223. err := tio.SegmentSuperpacket(pkt, func(seg []byte) error {
  224. // header + plaintext + AEAD tag (16 bytes for both AES-GCM and ChaCha20-Poly1305)
  225. scratch := sendBatch.Reserve(header.Len + len(seg) + 16)
  226. out := f.sendInsideEncrypt(hostinfo, ci, seg, scratch, nb)
  227. if out == nil {
  228. return nil
  229. }
  230. sendBatch.Commit(out, remote)
  231. return nil
  232. })
  233. if err != nil {
  234. hostinfo.logger(f.l).Error("Failed to segment superpacket for send", "error", err)
  235. }
  236. }
  237. func (f *Interface) rejectInside(packet []byte, out []byte, q int) {
  238. if !f.firewall.OutboundSendReject {
  239. return
  240. }
  241. out = iputil.CreateRejectPacket(packet, out)
  242. if len(out) == 0 {
  243. return
  244. }
  245. _, err := f.queues[q].Write(out)
  246. if err != nil {
  247. f.l.Error("Failed to write to tun", "error", err)
  248. }
  249. }
  250. func (f *Interface) rejectOutside(packet []byte, ci *ConnectionState, hostinfo *HostInfo, nb, rejectBuf []byte, q int) {
  251. if !f.firewall.InboundSendReject {
  252. return
  253. }
  254. // split rejectBuf to make sure we have room to write the plaintext rejection, then encrypt it, without trampling anything
  255. // we can't re-use packet, if we need to send an icmp reject, it won't be long enough.
  256. half := len(rejectBuf) / 2
  257. encryptBuf := rejectBuf[0:0:half] //the first half of rejectBuf's capacity, len set to 0
  258. buildBuf := rejectBuf[half:]
  259. out := iputil.CreateRejectPacket(packet, buildBuf)
  260. if len(out) == 0 {
  261. return
  262. }
  263. if len(out) > iputil.MaxRejectPacketSize {
  264. if f.l.Enabled(context.Background(), slog.LevelInfo) {
  265. f.l.Info("rejectOutside: packet too big, not sending", "packet", packet, "outPacket", out)
  266. }
  267. return
  268. }
  269. f.sendNoMetrics(header.Message, 0, ci, hostinfo, netip.AddrPort{}, out, nb, encryptBuf, q)
  270. }
  271. // 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
  272. // it does not check if it is within our vpn networks!
  273. func (f *Interface) Handshake(vpnAddr netip.Addr) {
  274. f.handshakeManager.GetOrHandshake(vpnAddr, nil)
  275. }
  276. // getOrHandshakeNoRouting returns nil if the vpnAddr is not routable.
  277. // If the 2nd return var is false then the hostinfo is not ready to be used in a tunnel
  278. func (f *Interface) getOrHandshakeNoRouting(vpnAddr netip.Addr, cacheCallback func(*HandshakeHostInfo)) (*HostInfo, bool) {
  279. if f.myVpnNetworksTable.Contains(vpnAddr) {
  280. return f.handshakeManager.GetOrHandshake(vpnAddr, cacheCallback)
  281. }
  282. return nil, false
  283. }
  284. // getOrHandshakeConsiderRouting will try to find the HostInfo to handle this packet, starting a handshake if necessary.
  285. // If the 2nd return var is false then the hostinfo is not ready to be used in a tunnel.
  286. func (f *Interface) getOrHandshakeConsiderRouting(fwPacket *firewall.Packet, cacheCallback func(*HandshakeHostInfo)) (*HostInfo, bool) {
  287. destinationAddr := fwPacket.RemoteAddr
  288. hostinfo, ready := f.getOrHandshakeNoRouting(destinationAddr, cacheCallback)
  289. // Host is inside the mesh, no routing required
  290. if hostinfo != nil {
  291. return hostinfo, ready
  292. }
  293. gateways := f.inside.RoutesFor(destinationAddr)
  294. switch len(gateways) {
  295. case 0:
  296. return nil, false
  297. case 1:
  298. // Single gateway route
  299. return f.handshakeManager.GetOrHandshake(gateways[0].Addr(), cacheCallback)
  300. default:
  301. // Multi gateway route, perform ECMP categorization
  302. gatewayAddr, balancingOk := routing.BalancePacket(fwPacket, gateways)
  303. if !balancingOk {
  304. // This happens if the gateway buckets were not calculated, this _should_ never happen
  305. f.l.Error("Gateway buckets not calculated, fallback from ECMP to random routing. Please report this bug.")
  306. }
  307. var handshakeInfoForChosenGateway *HandshakeHostInfo
  308. var hhReceiver = func(hh *HandshakeHostInfo) {
  309. handshakeInfoForChosenGateway = hh
  310. }
  311. // Store the handshakeHostInfo for later.
  312. // If this node is not reachable we will attempt other nodes, if none are reachable we will
  313. // cache the packet for this gateway.
  314. if hostinfo, ready = f.handshakeManager.GetOrHandshake(gatewayAddr, hhReceiver); ready {
  315. return hostinfo, true
  316. }
  317. // It appears the selected gateway cannot be reached, find another gateway to fallback on.
  318. // The current implementation breaks ECMP but that seems better than no connectivity.
  319. // If ECMP is also required when a gateway is down then connectivity status
  320. // for each gateway needs to be kept and the weights recalculated when they go up or down.
  321. // This would also need to interact with unsafe_route updates through reloading the config or
  322. // use of the use_system_route_table option
  323. if f.l.Enabled(context.Background(), slog.LevelDebug) {
  324. f.l.Debug("Calculated gateway for ECMP not available, attempting other gateways",
  325. "destination", destinationAddr,
  326. "originalGateway", gatewayAddr,
  327. )
  328. }
  329. for i := range gateways {
  330. // Skip the gateway that failed previously
  331. if gateways[i].Addr() == gatewayAddr {
  332. continue
  333. }
  334. // We do not need the HandshakeHostInfo since we cache the packet in the originally chosen gateway
  335. if hostinfo, ready = f.handshakeManager.GetOrHandshake(gateways[i].Addr(), nil); ready {
  336. return hostinfo, true
  337. }
  338. }
  339. // No gateways reachable, cache the packet in the originally chosen gateway
  340. cacheCallback(handshakeInfoForChosenGateway)
  341. return hostinfo, false
  342. }
  343. }
  344. func (f *Interface) sendMessageNow(t header.MessageType, st header.MessageSubType, hostinfo *HostInfo, p, nb, out []byte) {
  345. fp := &firewall.ParsedPacket{}
  346. err := newPacket(p, false, fp)
  347. if err != nil {
  348. f.l.Warn("error while parsing outgoing packet for firewall check", "error", err)
  349. return
  350. }
  351. // check if packet is in outbound fw rules
  352. dropReason := f.firewall.Drop(fp.Packet, false, hostinfo, f.pki.GetCAPool(), nil)
  353. if dropReason != nil {
  354. if f.l.Enabled(context.Background(), slog.LevelDebug) {
  355. f.l.Debug("dropping cached packet",
  356. "fwPacket", fp,
  357. "reason", dropReason,
  358. )
  359. }
  360. return
  361. }
  362. f.sendNoMetrics(header.Message, st, hostinfo.ConnectionState, hostinfo, netip.AddrPort{}, p, nb, out, 0)
  363. }
  364. // SendMessageToVpnAddr handles real addr:port lookup and sends to the current best known address for vpnAddr.
  365. // This function ignores myVpnNetworksTable, and will always attempt to treat the address as a vpnAddr
  366. func (f *Interface) SendMessageToVpnAddr(t header.MessageType, st header.MessageSubType, vpnAddr netip.Addr, p, nb, out []byte) {
  367. hostInfo, ready := f.handshakeManager.GetOrHandshake(vpnAddr, func(hh *HandshakeHostInfo) {
  368. hh.cachePacket(f.l, t, st, p, f.SendMessageToHostInfo, f.cachedPacketMetrics)
  369. })
  370. if hostInfo == nil {
  371. if f.l.Enabled(context.Background(), slog.LevelDebug) {
  372. f.l.Debug("dropping SendMessageToVpnAddr, vpnAddr not in our vpn networks or in unsafe routes",
  373. "vpnAddr", vpnAddr,
  374. )
  375. }
  376. return
  377. }
  378. if !ready {
  379. return
  380. }
  381. f.SendMessageToHostInfo(t, st, hostInfo, p, nb, out)
  382. }
  383. func (f *Interface) SendMessageToHostInfo(t header.MessageType, st header.MessageSubType, hi *HostInfo, p, nb, out []byte) {
  384. f.send(t, st, hi.ConnectionState, hi, p, nb, out)
  385. }
  386. func (f *Interface) send(t header.MessageType, st header.MessageSubType, ci *ConnectionState, hostinfo *HostInfo, p, nb, out []byte) {
  387. f.messageMetrics.Tx(t, st, 1)
  388. f.sendNoMetrics(t, st, ci, hostinfo, netip.AddrPort{}, p, nb, out, 0)
  389. }
  390. func (f *Interface) sendTo(t header.MessageType, st header.MessageSubType, ci *ConnectionState, hostinfo *HostInfo, remote netip.AddrPort, p, nb, out []byte) {
  391. f.messageMetrics.Tx(t, st, 1)
  392. f.sendNoMetrics(t, st, ci, hostinfo, remote, p, nb, out, 0)
  393. }
  394. // dropExhausted records an exhaustion drop and logs once, on the crossing send, for a spent tunnel.
  395. func (f *Interface) dropExhausted(hostinfo *HostInfo, c uint64, msg string) {
  396. f.messageMetrics.TxExhausted(1)
  397. if c == RejectAfterMessages {
  398. hostinfo.logger(f.l).Error(msg)
  399. }
  400. }
  401. func (f *Interface) prepareSendVia(via *HostInfo,
  402. relay *Relay,
  403. ad,
  404. nb,
  405. out []byte,
  406. nocopy bool,
  407. ) ([]byte, error) {
  408. if noiseutil.EncryptLockNeeded {
  409. // NOTE: for goboring AESGCMTLS we need to lock because of the nonce check
  410. via.ConnectionState.writeLock.Lock()
  411. }
  412. c, ok := via.ConnectionState.NextMessageCounter()
  413. if !ok {
  414. if noiseutil.EncryptLockNeeded {
  415. via.ConnectionState.writeLock.Unlock()
  416. }
  417. f.dropExhausted(via, c, "Dropping outbound relay packets, tunnel message counter is exhausted")
  418. return nil, fmt.Errorf("tunnel message counter is exhausted")
  419. }
  420. out = header.Encode(out, header.Version, header.Message, header.MessageRelay, relay.RemoteIndex, c)
  421. f.connectionManager.OutNoRebind(via)
  422. // Authenticate the header and payload, but do not encrypt for this message type.
  423. // The payload consists of the inner, unencrypted Nebula header, as well as the end-to-end encrypted payload.
  424. if len(out)+len(ad)+via.ConnectionState.eKey.Overhead() > cap(out) {
  425. if noiseutil.EncryptLockNeeded {
  426. via.ConnectionState.writeLock.Unlock()
  427. }
  428. via.logger(f.l).Error("SendVia out buffer not large enough for relay",
  429. "outCap", cap(out),
  430. "payloadLen", len(ad),
  431. "headerLen", len(out),
  432. "cipherOverhead", via.ConnectionState.eKey.Overhead(),
  433. )
  434. return nil, io.ErrShortBuffer
  435. }
  436. // The header bytes are written to the 'out' slice; Grow the slice to hold the header and associated data payload.
  437. offset := len(out)
  438. out = out[:offset+len(ad)]
  439. // In one call path, the associated data _is_ already stored in out. In other call paths, the associated data must
  440. // be copied into 'out'.
  441. if !nocopy {
  442. copy(out[offset:], ad)
  443. }
  444. var err error
  445. out, err = via.ConnectionState.eKey.EncryptDanger(out, out, nil, c, nb)
  446. if noiseutil.EncryptLockNeeded {
  447. via.ConnectionState.writeLock.Unlock()
  448. }
  449. if err != nil {
  450. via.logger(f.l).Info("Failed to EncryptDanger in sendVia", "error", err)
  451. return nil, err
  452. }
  453. f.connectionManager.RelayUsed(relay.LocalIndex)
  454. return out, nil
  455. }
  456. // SendVia sends a payload through a Relay tunnel. No authentication or encryption is done
  457. // to the payload for the ultimate target host, making this a useful method for sending
  458. // handshake messages to peers through relay tunnels.
  459. // via is the HostInfo through which the message is relayed.
  460. // ad is the plaintext data to authenticate, but not encrypt
  461. // nb is a buffer used to store the nonce value, re-used for performance reasons.
  462. // out is a buffer used to store the result of the Encrypt operation
  463. // q indicates which writer to use to send the packet.
  464. func (f *Interface) SendVia(via *HostInfo, relay *Relay, ad, nb, out []byte, nocopy bool, q int) {
  465. toSend, err := f.prepareSendVia(via, relay, ad, nb, out, nocopy)
  466. if err != nil {
  467. // already logged by prepareSendVia
  468. return
  469. }
  470. // Relay carriers are base tunnels (sockIdx 0); indexing through the carrier keeps the invariant explicit.
  471. err = f.writers[f.egressSock(via, 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. // With multiport the tunnel's own sockIdx is authoritative: a lane must egress the lane's 4-tuple so the peer's
  479. // spoof/roam checks accept keepalives, close packets and rejects and the NAT entry stays warm, and a base or relay
  480. // carrier must egress socket 0 so a vanilla peer never sees per-routine source ports.
  481. //
  482. // Without multiport every socket shares one port under SO_REUSEPORT, so the source address is identical either way and
  483. // we keep q, the socket the triggering packet arrived on, to avoid contending on socket 0's fd.
  484. func (f *Interface) egressSock(hostinfo *HostInfo, q int) int {
  485. if f.multiport {
  486. return hostinfo.sockIdx
  487. }
  488. return q
  489. }
  490. func (f *Interface) sendNoMetrics(t header.MessageType, st header.MessageSubType, ci *ConnectionState, hostinfo *HostInfo, remote netip.AddrPort, p, nb, out []byte, q int) {
  491. if ci.eKey == nil {
  492. return
  493. }
  494. q = f.egressSock(hostinfo, q)
  495. useRelay := !remote.IsValid() && !hostinfo.GetRemote().IsValid()
  496. fullOut := out
  497. if useRelay {
  498. if len(out) < header.Len {
  499. // out always has a capacity of mtu, but not always a length greater than the header.Len.
  500. // Grow it to make sure the next operation works.
  501. out = out[:header.Len]
  502. }
  503. // Save a header's worth of data at the front of the 'out' buffer.
  504. out = out[header.Len:]
  505. }
  506. if noiseutil.EncryptLockNeeded {
  507. // NOTE: for goboring AESGCMTLS we need to lock because of the nonce check
  508. ci.writeLock.Lock()
  509. }
  510. c, ok := ci.NextMessageCounter()
  511. if !ok {
  512. if noiseutil.EncryptLockNeeded {
  513. ci.writeLock.Unlock()
  514. }
  515. f.dropExhausted(hostinfo, c, "Dropping outbound packets, tunnel message counter is exhausted")
  516. return
  517. }
  518. //l.WithField("trace", string(debug.Stack())).Error("out Header ", &Header{Version, t, st, 0, hostinfo.remoteIndexId, c}, p)
  519. out = header.Encode(out, header.Version, t, st, hostinfo.remoteIndexId, c)
  520. // A closing tunnel is torn down right after this, so skip the connection manager entirely: no point recording
  521. // traffic or asking the lighthouse for a punch. Otherwise, if we rebound since this tunnel last sent, ask the
  522. // lighthouse to get the far side punching at us again.
  523. //
  524. // isLane is checked last on purpose: a lane still needs its Out() traffic recorded for the liveness decision, it
  525. // just shouldn't issue the lighthouse query. The base tunnel issues the one query for the peer.
  526. if t != header.CloseTunnel && f.connectionManager.Out(hostinfo) && !hostinfo.isLane() {
  527. f.lightHouse.QueryServer(hostinfo.vpnAddrs[0])
  528. if f.l.Enabled(context.Background(), slog.LevelDebug) {
  529. f.l.Debug("Lighthouse update triggered for punch due to rebind epoch",
  530. "vpnAddrs", hostinfo.vpnAddrs,
  531. )
  532. }
  533. }
  534. var err error
  535. out, err = ci.eKey.EncryptDanger(out, out, p, c, nb)
  536. if noiseutil.EncryptLockNeeded {
  537. ci.writeLock.Unlock()
  538. }
  539. if err != nil {
  540. hostinfo.logger(f.l).Error("Failed to encrypt outgoing packet",
  541. "error", err,
  542. "udpAddr", remote,
  543. "counter", c,
  544. )
  545. return
  546. }
  547. if remote.IsValid() {
  548. err = f.writers[q].WriteTo(out, remote)
  549. if err != nil {
  550. hostinfo.logger(f.l).Error("Failed to write outgoing packet",
  551. "error", err,
  552. "udpAddr", remote,
  553. )
  554. }
  555. } else if hr := hostinfo.GetRemote(); hr.IsValid() {
  556. err = f.writers[q].WriteTo(out, hr)
  557. if err != nil {
  558. hostinfo.logger(f.l).Error("Failed to write outgoing packet",
  559. "error", err,
  560. "udpAddr", hr,
  561. )
  562. }
  563. } else {
  564. if hostinfo.isLane() {
  565. // A lane always has a valid remote (set from its own handshake);
  566. // reaching here means the lane is broken, and lane ciphertext must
  567. // never ride a relay (relays are base-tunnel-only).
  568. hostinfo.logger(f.l).Error("Dropping lane packet with no valid remote")
  569. return
  570. }
  571. // Try to send via a relay
  572. for _, relayIP := range hostinfo.relayState.CopyRelayIps() {
  573. relayHostInfo, relay, err := f.hostMap.QueryVpnAddrsRelayFor(hostinfo.vpnAddrs, relayIP)
  574. if err != nil {
  575. hostinfo.relayState.DeleteRelay(relayIP)
  576. hostinfo.logger(f.l).Info("sendNoMetrics failed to find HostInfo",
  577. "relay", relayIP,
  578. "error", err,
  579. )
  580. continue
  581. }
  582. f.SendVia(relayHostInfo, relay, out, nb, fullOut[:header.Len+len(out)], true, q)
  583. break
  584. }
  585. }
  586. }