inside.go 23 KB

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