inside.go 22 KB

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