interface.go 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737
  1. package nebula
  2. import (
  3. "context"
  4. "errors"
  5. "fmt"
  6. "log/slog"
  7. "net/netip"
  8. "runtime"
  9. "slices"
  10. "sync"
  11. "sync/atomic"
  12. "time"
  13. "github.com/gaissmai/bart"
  14. "github.com/rcrowley/go-metrics"
  15. "github.com/slackhq/nebula/util"
  16. "github.com/slackhq/nebula/cert"
  17. "github.com/slackhq/nebula/config"
  18. "github.com/slackhq/nebula/firewall"
  19. "github.com/slackhq/nebula/header"
  20. "github.com/slackhq/nebula/overlay"
  21. "github.com/slackhq/nebula/overlay/batch"
  22. "github.com/slackhq/nebula/overlay/tio"
  23. "github.com/slackhq/nebula/udp"
  24. )
  25. const mtu = 9001
  26. type InterfaceConfig struct {
  27. HostMap *HostMap
  28. Outside udp.Conn
  29. Inside overlay.Device
  30. pki *PKI
  31. Cipher string
  32. Firewall *Firewall
  33. DnsServer *dnsServer
  34. HandshakeManager *HandshakeManager
  35. lightHouse *LightHouse
  36. connectionManager *connectionManager
  37. DropLocalBroadcast bool
  38. DropMulticast bool
  39. routines int
  40. // Multiport means writers[i] is bound to listen.port+i (not a shared
  41. // SO_REUSEPORT port) and lane tunnels are negotiated with capable peers.
  42. Multiport bool
  43. MessageMetrics *MessageMetrics
  44. version string
  45. relayManager *relayManager
  46. punchy *Punchy
  47. tryPromoteEvery uint32
  48. reQueryEvery uint32
  49. reQueryWait time.Duration
  50. ConntrackCacheTimeout time.Duration
  51. // CpuAffinity, when non-empty, names the CPUs each TUN reader goroutine
  52. // should pin to. Queue i pins to CpuAffinity[i % len(CpuAffinity)] —
  53. // shorter lists than `routines` cycle. Empty list keeps the default
  54. // pin-to-(i % NumCPU) behavior. Only consulted when PinThreads is true.
  55. CpuAffinity []int
  56. // PinThreads controls whether each TUN reader OS thread is pinned to a
  57. // single CPU (via tun.pin_threads, default true). Pinning keeps each
  58. // goroutine's sendmmsg on one XPS-selected NIC TX ring so per-flow
  59. // packets stay ordered on the wire.
  60. PinThreads bool
  61. l *slog.Logger
  62. }
  63. type Interface struct {
  64. hostMap *HostMap
  65. outside udp.Conn
  66. inside overlay.Device
  67. pki *PKI
  68. firewall *Firewall
  69. connectionManager *connectionManager
  70. handshakeManager *HandshakeManager
  71. dnsServer *dnsServer
  72. createTime time.Time
  73. lightHouse *LightHouse
  74. myBroadcastAddrsTable *bart.Lite
  75. myVpnAddrs []netip.Addr // A list of addresses assigned to us via our certificate
  76. myVpnAddrsTable *bart.Lite
  77. myVpnNetworks []netip.Prefix // A list of networks assigned to us via our certificate
  78. myVpnNetworksTable *bart.Lite
  79. dropLocalBroadcast bool
  80. dropMulticast bool
  81. routines int
  82. multiport bool
  83. disconnectInvalid atomic.Bool
  84. closed atomic.Bool
  85. // cpuAffinity, when non-empty, names the CPUs each TUN reader goroutine
  86. // should pin to. Queue i pins to cpuAffinity[i % len(cpuAffinity)].
  87. // Empty falls back to the default pin-to-(allowed CPU) behavior.
  88. // Only consulted when pinThreads is true.
  89. cpuAffinity []int
  90. // pinThreads controls whether listenIn pins each TUN reader OS thread to
  91. // a CPU at all (tun.pin_threads, default true). When false, threads are
  92. // left free to migrate as on stock nebula.
  93. pinThreads bool
  94. // ecnEnabled gates RFC 6040 underlay ECN propagation. When true,
  95. // inside.go copies the inner ECN onto the outer carrier on encap and
  96. // decryptToTun folds outer CE into the inner header on decap. Toggle
  97. // via tunnels.ecn (default true).
  98. ecnEnabled atomic.Bool
  99. relayManager *relayManager
  100. tryPromoteEvery atomic.Uint32
  101. reQueryEvery atomic.Uint32
  102. reQueryWait atomic.Int64
  103. sendRecvErrorConfig recvErrorConfig
  104. acceptRecvErrorConfig recvErrorConfig
  105. // rebindCount is used to decide if an active tunnel should trigger a punch notification through a lighthouse
  106. rebindCount int8
  107. version string
  108. conntrackCacheTimeout time.Duration
  109. ctx context.Context
  110. writers []udp.Conn
  111. queues []tio.Queue
  112. // batchers is one per tun queue, wrapping queues[i]. readOutsidePackets
  113. // commits plaintext into the batch.RxBatcher; the plaintext is decrypted
  114. // in place inside the UDP receive buffers, so listenOut must call Flush
  115. // at the end of each UDP recvmmsg batch, before those buffers are
  116. // reused (every udp.Conn ListenOut guarantees that ordering).
  117. batchers []batch.RxBatcher
  118. wg sync.WaitGroup
  119. // fatalErr holds the first unexpected reader error that caused shutdown.
  120. // nil means "no fatal error" (yet)
  121. fatalErr atomic.Pointer[error]
  122. // triggerShutdown is a function that will be run exactly once, when onFatal swaps something non-nil into fatalErr
  123. triggerShutdown func()
  124. metricHandshakes metrics.Histogram
  125. messageMetrics *MessageMetrics
  126. cachedPacketMetrics *cachedPacketMetrics
  127. l *slog.Logger
  128. }
  129. type EncWriter interface {
  130. SendVia(via *HostInfo,
  131. relay *Relay,
  132. ad,
  133. nb,
  134. out []byte,
  135. nocopy bool,
  136. )
  137. SendMessageToVpnAddr(t header.MessageType, st header.MessageSubType, vpnAddr netip.Addr, p, nb, out []byte)
  138. SendMessageToHostInfo(t header.MessageType, st header.MessageSubType, hostinfo *HostInfo, p, nb, out []byte)
  139. Handshake(vpnAddr netip.Addr)
  140. GetHostInfo(vpnAddr netip.Addr) *HostInfo
  141. GetCertState() *CertState
  142. }
  143. type recvErrorConfig uint8
  144. const (
  145. recvErrorAlways recvErrorConfig = iota
  146. recvErrorNever
  147. recvErrorPrivate
  148. )
  149. func (s recvErrorConfig) ShouldRecvError(endpoint netip.AddrPort) bool {
  150. switch s {
  151. case recvErrorPrivate:
  152. return endpoint.Addr().IsPrivate()
  153. case recvErrorAlways:
  154. return true
  155. case recvErrorNever:
  156. return false
  157. default:
  158. panic(fmt.Errorf("invalid recvErrorConfig value: %d", s))
  159. }
  160. }
  161. func (s recvErrorConfig) String() string {
  162. switch s {
  163. case recvErrorAlways:
  164. return "always"
  165. case recvErrorNever:
  166. return "never"
  167. case recvErrorPrivate:
  168. return "private"
  169. default:
  170. return fmt.Sprintf("invalid(%d)", s)
  171. }
  172. }
  173. func NewInterface(ctx context.Context, c *InterfaceConfig) (*Interface, error) {
  174. if c.Outside == nil {
  175. return nil, errors.New("no outside connection")
  176. }
  177. if c.Inside == nil {
  178. return nil, errors.New("no inside interface (tun)")
  179. }
  180. if c.pki == nil {
  181. return nil, errors.New("no certificate state")
  182. }
  183. if c.Firewall == nil {
  184. return nil, errors.New("no firewall rules")
  185. }
  186. if c.connectionManager == nil {
  187. return nil, errors.New("no connection manager")
  188. }
  189. cs := c.pki.getCertState()
  190. ifce := &Interface{
  191. ctx: ctx,
  192. pki: c.pki,
  193. hostMap: c.HostMap,
  194. outside: c.Outside,
  195. inside: c.Inside,
  196. firewall: c.Firewall,
  197. dnsServer: c.DnsServer,
  198. handshakeManager: c.HandshakeManager,
  199. createTime: time.Now(),
  200. lightHouse: c.lightHouse,
  201. dropLocalBroadcast: c.DropLocalBroadcast,
  202. dropMulticast: c.DropMulticast,
  203. routines: c.routines,
  204. multiport: c.Multiport,
  205. version: c.version,
  206. writers: make([]udp.Conn, c.routines),
  207. batchers: make([]batch.RxBatcher, c.routines),
  208. myVpnNetworks: cs.myVpnNetworks,
  209. myVpnNetworksTable: cs.myVpnNetworksTable,
  210. myVpnAddrs: cs.myVpnAddrs,
  211. myVpnAddrsTable: cs.myVpnAddrsTable,
  212. myBroadcastAddrsTable: cs.myVpnBroadcastAddrsTable,
  213. relayManager: c.relayManager,
  214. connectionManager: c.connectionManager,
  215. conntrackCacheTimeout: c.ConntrackCacheTimeout,
  216. cpuAffinity: c.CpuAffinity,
  217. pinThreads: c.PinThreads,
  218. metricHandshakes: metrics.GetOrRegisterHistogram("handshakes", nil, metrics.NewExpDecaySample(1028, 0.015)),
  219. messageMetrics: c.MessageMetrics,
  220. cachedPacketMetrics: &cachedPacketMetrics{
  221. sent: metrics.GetOrRegisterCounter("hostinfo.cached_packets.sent", nil),
  222. dropped: metrics.GetOrRegisterCounter("hostinfo.cached_packets.dropped", nil),
  223. },
  224. l: c.l,
  225. }
  226. ifce.tryPromoteEvery.Store(c.tryPromoteEvery)
  227. ifce.reQueryEvery.Store(c.reQueryEvery)
  228. ifce.reQueryWait.Store(int64(c.reQueryWait))
  229. ifce.connectionManager.intf = ifce
  230. // Held until Close so waiting on the interface blocks until the resources are actually released
  231. ifce.wg.Add(1)
  232. return ifce, nil
  233. }
  234. // activate creates the interface on the host. After the interface is created, any
  235. // other services that want to bind listeners to its IP may do so successfully. However,
  236. // the interface isn't going to process anything until run() is called.
  237. func (f *Interface) activate() error {
  238. // actually turn on tun dev
  239. addr, err := f.outside.LocalAddr()
  240. if err != nil {
  241. f.l.Error("Failed to get udp listen address", "error", err)
  242. }
  243. f.l.Info("Nebula interface is active",
  244. "interface", f.inside.Name(),
  245. "networks", f.myVpnNetworks,
  246. "build", f.version,
  247. "udpAddr", addr,
  248. "boringcrypto", boringEnabled(),
  249. )
  250. // Under multiport each socket has exactly one reader on its own port, so
  251. // the shared-port multi-reader capability is irrelevant (and main.go
  252. // already hard-errored on unsupported platforms).
  253. if f.routines > 1 && !f.multiport && !f.outside.SupportsMultipleReaders() {
  254. f.routines = 1
  255. f.l.Warn("multiple udp readers are not supported on this platform, falling back to a single routine")
  256. }
  257. // Prepare the tun queues. A device that can't open that many hands back
  258. // fewer (a single queue on platforms without multiqueue support) and we
  259. // size the reader routines to what we actually got.
  260. queues, err := f.inside.Queues(f.routines)
  261. if err != nil {
  262. return err
  263. }
  264. if len(queues) < f.routines {
  265. if f.multiport {
  266. // The lane sockets are already bound one-per-routine; shrinking
  267. // the routine count would leave bound ports with no reader.
  268. return fmt.Errorf("multiport requires %d tun queues, device provided %d", f.routines, len(queues))
  269. }
  270. f.l.Warn("tun multiqueue is not supported on this platform, falling back to fewer routines",
  271. "requested", f.routines, "opened", len(queues))
  272. f.routines = len(queues)
  273. }
  274. f.queues = queues
  275. metrics.GetOrRegisterGauge("routines", nil).Update(int64(f.routines))
  276. for i := range f.queues {
  277. caps := tio.QueueCapabilities(f.queues[i])
  278. if caps.TSO || caps.USO {
  279. // Multi-lane: TCP gets coalesced when TSO is on, UDP when USO
  280. // is on, everything else (and either lane disabled) falls
  281. // through to passthrough so non-IP / non-TCP-UDP traffic still
  282. // reaches the TUN.
  283. f.batchers[i] = batch.NewMultiCoalescer(f.queues[i], f.l, caps.TSO, caps.USO)
  284. } else {
  285. f.batchers[i] = batch.NewPassthrough(f.queues[i])
  286. }
  287. }
  288. // On error the caller owns the cleanup, Control.Start cancels the service context
  289. // before releasing our resources so a waiter never observes a live context
  290. if err = f.inside.Activate(); err != nil {
  291. return err
  292. }
  293. return nil
  294. }
  295. func (f *Interface) run() {
  296. // Launch n queues to read packets from udp
  297. for i := 0; i < f.routines; i++ {
  298. f.wg.Go(func() {
  299. f.listenOut(i)
  300. })
  301. }
  302. // Launch n queues to read packets from tun dev
  303. for i := 0; i < f.routines; i++ {
  304. f.wg.Go(func() {
  305. f.listenIn(f.queues[i], i)
  306. })
  307. }
  308. }
  309. func (f *Interface) wait() error {
  310. f.wg.Wait()
  311. if e := f.fatalErr.Load(); e != nil {
  312. return *e
  313. }
  314. return nil
  315. }
  316. // onFatal stores the first fatal reader error, and calls triggerShutdown if it was the first one
  317. func (f *Interface) onFatal(err error) {
  318. swapped := f.fatalErr.CompareAndSwap(nil, &err)
  319. if !swapped {
  320. return
  321. }
  322. if f.triggerShutdown != nil {
  323. f.triggerShutdown()
  324. }
  325. }
  326. func (f *Interface) listenOut(i int) {
  327. var li udp.Conn
  328. if i > 0 {
  329. li = f.writers[i]
  330. } else {
  331. li = f.outside
  332. }
  333. ctCache := firewall.NewConntrackCacheTicker(f.ctx, f.l, f.conntrackCacheTimeout)
  334. lhh := f.lightHouse.NewRequestHandler()
  335. h := &header.H{}
  336. fwPacket := &firewall.Packet{}
  337. nb := make([]byte, 12, 12)
  338. scratch := make([]byte, mtu)
  339. listener := func(fromUdpAddr netip.AddrPort, payload []byte, meta udp.RxMeta) {
  340. f.readOutsidePackets(ViaSender{UdpAddr: fromUdpAddr, SockIdx: i}, scratch, payload, h, fwPacket, lhh, nb, i, ctCache.Get(), meta)
  341. }
  342. flusher := func() {
  343. if err := f.batchers[i].Flush(); err != nil {
  344. f.l.Error("Failed to flush tun coalescer", "error", err)
  345. }
  346. }
  347. err := li.ListenOut(listener, flusher)
  348. // An error after teardown began is shutdown noise, the closed flag covers resources
  349. // Close releases itself and the cancelled ctx covers ones torn down by their owners
  350. // reacting to it, like the user device pipes
  351. if err != nil && !f.closed.Load() && f.ctx.Err() == nil {
  352. f.l.Error("Error while reading inbound packet, closing", "error", err)
  353. f.onFatal(err)
  354. }
  355. f.l.Debug("underlay reader is done", "reader", i)
  356. }
  357. // txQueue is the per-routine TX state owned by one listenIn goroutine. lane
  358. // is bound to the routine's own socket and carries lane-tunnel data; base is
  359. // bound to socket 0 and carries base-tunnel and relay data, which must keep
  360. // the base source port (a vanilla peer would otherwise see per-routine source
  361. // ports and roam-thrash). The two alias when multiport is off or on routine 0.
  362. // Concurrent sendmmsg on the shared socket-0 fd is safe: a flow is pinned to
  363. // one routine by tun steering, so per-flow wire order still holds.
  364. type txQueue struct {
  365. lane *batch.SendBatch
  366. base *batch.SendBatch
  367. }
  368. func (tx *txQueue) full() bool {
  369. if tx.lane.Len() >= batch.SendBatchCap {
  370. return true
  371. }
  372. return tx.base != tx.lane && tx.base.Len() >= batch.SendBatchCap
  373. }
  374. // flush drains base before lane so that when a flow moves from the base
  375. // tunnel onto a freshly established lane mid-window, its packets still leave
  376. // this host in encryption order.
  377. func (tx *txQueue) flush(l *slog.Logger, i int) {
  378. if tx.base != tx.lane {
  379. if err := tx.base.Flush(); err != nil {
  380. l.Error("Failed to write outgoing batch", "error", err, "writer", 0)
  381. }
  382. }
  383. if err := tx.lane.Flush(); err != nil {
  384. l.Error("Failed to write outgoing batch", "error", err, "writer", i)
  385. }
  386. }
  387. func (f *Interface) listenIn(queue tio.Queue, i int) {
  388. // Pinning this thread (and goroutine) to a single CPU keeps every sendmmsg from this goroutine going through the
  389. // same TX ring on the nic, so the wire sees per-flow order. Skip entirely when tun.pin_threads is false.
  390. if f.pinThreads {
  391. var cpu int
  392. if n := len(f.cpuAffinity); n > 0 {
  393. // Explicit tun.cpu_affinity list wins; parseCpuAffinity already
  394. // validated the entries against the allowed CPU set.
  395. cpu = f.cpuAffinity[i%n]
  396. } else if allowed, err := util.AllowedCPUs(); err == nil && len(allowed) > 0 {
  397. // Default: spread queues across the CPUs we're actually allowed to
  398. // run on. Under a cpuset/taskset mask these aren't 0..NumCPU-1, so
  399. // i % NumCPU would pick unrunnable IDs and every pin would fail.
  400. cpu = allowed[i%len(allowed)]
  401. } else {
  402. cpu = i % runtime.NumCPU()
  403. }
  404. if err := util.PinThreadToCPU(cpu); err != nil {
  405. f.l.Warn("failed to pin tun reader to CPU", "queue", i, "cpu", cpu, "err", err)
  406. }
  407. }
  408. rejectBuf := make([]byte, mtu)
  409. arenaSize := batch.SendBatchCap * (udp.MTU + 32)
  410. sb := batch.NewSendBatch(f.writers[i], batch.SendBatchCap, arenaSize)
  411. tx := &txQueue{lane: sb, base: sb}
  412. if f.multiport && i != 0 {
  413. tx.base = batch.NewSendBatch(f.writers[0], batch.SendBatchCap, arenaSize)
  414. }
  415. fwPacket := &firewall.Packet{}
  416. nb := make([]byte, 12, 12)
  417. conntrackCache := firewall.NewConntrackCacheTicker(f.ctx, f.l, f.conntrackCacheTimeout)
  418. for {
  419. pkts, err := queue.Read()
  420. if err != nil {
  421. // Same shutdown noise handling as listenOut
  422. if !f.closed.Load() && f.ctx.Err() == nil {
  423. f.l.Error("Error while reading outbound packet, closing", "error", err, "reader", i)
  424. f.onFatal(err)
  425. }
  426. break
  427. }
  428. for _, pkt := range pkts {
  429. f.consumeInsidePacket(pkt, fwPacket, nb, tx, rejectBuf, i, conntrackCache.Get())
  430. // Flush incrementally once a full sendmmsg batch has
  431. // accumulated so the first packets of a deep read drain
  432. // hit the wire while the rest are still being encrypted.
  433. if tx.full() {
  434. tx.flush(f.l, i)
  435. }
  436. }
  437. tx.flush(f.l, i)
  438. }
  439. f.l.Debug("overlay reader is done", "reader", i)
  440. }
  441. func (f *Interface) RegisterConfigChangeCallbacks(c *config.C) {
  442. c.RegisterReloadCallback(f.reloadFirewall)
  443. c.RegisterReloadCallback(f.reloadSendRecvError)
  444. c.RegisterReloadCallback(f.reloadAcceptRecvError)
  445. c.RegisterReloadCallback(f.reloadDisconnectInvalid)
  446. c.RegisterReloadCallback(f.reloadMisc)
  447. c.RegisterReloadCallback(f.reloadEcn)
  448. for _, udpConn := range f.writers {
  449. c.RegisterReloadCallback(udpConn.ReloadConfig)
  450. }
  451. }
  452. func (f *Interface) reloadDisconnectInvalid(c *config.C) {
  453. initial := c.InitialLoad()
  454. if initial || c.HasChanged("pki.disconnect_invalid") {
  455. f.disconnectInvalid.Store(c.GetBool("pki.disconnect_invalid", true))
  456. if !initial {
  457. f.l.Info("pki.disconnect_invalid changed", "value", f.disconnectInvalid.Load())
  458. }
  459. }
  460. }
  461. func (f *Interface) reloadFirewall(c *config.C) {
  462. cs := f.pki.getCertState()
  463. curCert := cs.getCertificate(cert.Version2)
  464. if curCert == nil {
  465. curCert = cs.getCertificate(cert.Version1)
  466. }
  467. // The firewall builds its routableNetworks set from the certificate's UnsafeNetworks at construction.
  468. // Check to see if that set has changed, and if so, rebuild the firewall.
  469. certUnsafeChanged := curCert != nil && !slices.Equal(curCert.UnsafeNetworks(), f.firewall.unsafeNetworks)
  470. if !c.HasChanged("firewall") && !certUnsafeChanged {
  471. f.l.Debug("No firewall config change detected")
  472. return
  473. }
  474. fw, err := NewFirewallFromConfig(f.l, cs, c)
  475. if err != nil {
  476. f.l.Error("Error while creating firewall during reload", "error", err)
  477. return
  478. }
  479. oldFw := f.firewall
  480. conntrack := oldFw.Conntrack
  481. conntrack.Lock()
  482. defer conntrack.Unlock()
  483. fw.rulesVersion = oldFw.rulesVersion + 1
  484. // If rulesVersion is back to zero, we have wrapped all the way around. Be
  485. // safe and just reset conntrack in this case.
  486. if fw.rulesVersion == 0 {
  487. f.l.Warn("firewall rulesVersion has overflowed, resetting conntrack",
  488. "firewallHashes", fw.GetRuleHashes(),
  489. "oldFirewallHashes", oldFw.GetRuleHashes(),
  490. "rulesVersion", fw.rulesVersion,
  491. )
  492. } else {
  493. fw.Conntrack = conntrack
  494. }
  495. f.firewall = fw
  496. oldFw.Destroy()
  497. f.l.Info("New firewall has been installed",
  498. "firewallHashes", fw.GetRuleHashes(),
  499. "oldFirewallHashes", oldFw.GetRuleHashes(),
  500. "rulesVersion", fw.rulesVersion,
  501. )
  502. }
  503. func (f *Interface) reloadSendRecvError(c *config.C) {
  504. if c.InitialLoad() || c.HasChanged("listen.send_recv_error") {
  505. stringValue := c.GetString("listen.send_recv_error", "always")
  506. switch stringValue {
  507. case "always":
  508. f.sendRecvErrorConfig = recvErrorAlways
  509. case "never":
  510. f.sendRecvErrorConfig = recvErrorNever
  511. case "private":
  512. f.sendRecvErrorConfig = recvErrorPrivate
  513. default:
  514. if c.GetBool("listen.send_recv_error", true) {
  515. f.sendRecvErrorConfig = recvErrorAlways
  516. } else {
  517. f.sendRecvErrorConfig = recvErrorNever
  518. }
  519. }
  520. f.l.Info("Loaded send_recv_error config", "sendRecvError", f.sendRecvErrorConfig.String())
  521. }
  522. }
  523. func (f *Interface) reloadAcceptRecvError(c *config.C) {
  524. if c.InitialLoad() || c.HasChanged("listen.accept_recv_error") {
  525. stringValue := c.GetString("listen.accept_recv_error", "always")
  526. switch stringValue {
  527. case "always":
  528. f.acceptRecvErrorConfig = recvErrorAlways
  529. case "never":
  530. f.acceptRecvErrorConfig = recvErrorNever
  531. case "private":
  532. f.acceptRecvErrorConfig = recvErrorPrivate
  533. default:
  534. if c.GetBool("listen.accept_recv_error", true) {
  535. f.acceptRecvErrorConfig = recvErrorAlways
  536. } else {
  537. f.acceptRecvErrorConfig = recvErrorNever
  538. }
  539. }
  540. f.l.Info("Loaded accept_recv_error config", "acceptRecvError", f.acceptRecvErrorConfig.String())
  541. }
  542. }
  543. func (f *Interface) reloadMisc(c *config.C) {
  544. if c.HasChanged("counters.try_promote") {
  545. n := c.GetUint32("counters.try_promote", defaultPromoteEvery)
  546. f.tryPromoteEvery.Store(n)
  547. f.l.Info("counters.try_promote has changed")
  548. }
  549. if c.HasChanged("counters.requery_every_packets") {
  550. n := c.GetUint32("counters.requery_every_packets", defaultReQueryEvery)
  551. f.reQueryEvery.Store(n)
  552. f.l.Info("counters.requery_every_packets has changed")
  553. }
  554. if c.HasChanged("timers.requery_wait_duration") {
  555. n := c.GetDuration("timers.requery_wait_duration", defaultReQueryWait)
  556. f.reQueryWait.Store(int64(n))
  557. f.l.Info("timers.requery_wait_duration has changed")
  558. }
  559. }
  560. // reloadEcn syncs Interface.ecnEnabled with the tunnels.ecn config knob.
  561. // Default is enabled (RFC 6040 normal mode); set false on the rare path
  562. // where an underlay middlebox rewrites or drops ECN bits unpredictably.
  563. func (f *Interface) reloadEcn(c *config.C) {
  564. initial := c.InitialLoad()
  565. if initial || c.HasChanged("tunnels.ecn") {
  566. v := c.GetBool("tunnels.ecn", true)
  567. changed := f.ecnEnabled.Swap(v) != v
  568. if !initial {
  569. f.l.Info("tunnels.ecn changed", "enabled", v)
  570. if changed {
  571. f.l.Warn("tunnels.ecn datapath toggled, but route-level ECN negotiation (RTAX_FEATURE_ECN) retains its previous state until nebula is restarted", "enabled", v)
  572. }
  573. }
  574. }
  575. }
  576. func (f *Interface) emitStats(ctx context.Context, i time.Duration) {
  577. ticker := time.NewTicker(i)
  578. defer ticker.Stop()
  579. udpStats := udp.NewUDPStatsEmitter(f.writers)
  580. certExpirationGauge := metrics.GetOrRegisterGauge("certificate.ttl_seconds", nil)
  581. certInitiatingVersion := metrics.GetOrRegisterGauge("certificate.initiating_version", nil)
  582. certMaxVersion := metrics.GetOrRegisterGauge("certificate.max_version", nil)
  583. emit := func() {
  584. f.firewall.EmitStats()
  585. f.handshakeManager.EmitStats()
  586. udpStats()
  587. certState := f.pki.getCertState()
  588. defaultCrt := certState.GetDefaultCertificate()
  589. certExpirationGauge.Update(int64(defaultCrt.NotAfter().Sub(time.Now()) / time.Second))
  590. certInitiatingVersion.Update(int64(defaultCrt.Version()))
  591. // Report the max certificate version we are capable of using
  592. if certState.v2Cert != nil {
  593. certMaxVersion.Update(int64(certState.v2Cert.Version()))
  594. } else {
  595. certMaxVersion.Update(int64(certState.v1Cert.Version()))
  596. }
  597. }
  598. // Prime gauges so a Prometheus scrape that lands before the first tick
  599. // sees real values instead of the zero defaults (issue #907).
  600. emit()
  601. for {
  602. select {
  603. case <-ctx.Done():
  604. return
  605. case <-ticker.C:
  606. emit()
  607. }
  608. }
  609. }
  610. func (f *Interface) GetHostInfo(vpnIp netip.Addr) *HostInfo {
  611. return f.hostMap.QueryVpnAddr(vpnIp)
  612. }
  613. func (f *Interface) GetCertState() *CertState {
  614. return f.pki.getCertState()
  615. }
  616. // Close releases the interface's resources: the udp sockets and the tun device.
  617. // It is idempotent and safe to call at any point in the lifecycle, including on an interface that never activated,
  618. // calls after the first return nil without doing anything.
  619. func (f *Interface) Close() error {
  620. if !f.closed.CompareAndSwap(false, true) {
  621. return nil
  622. }
  623. var errs []error
  624. // Release the udp readers
  625. for i, u := range f.writers {
  626. err := u.Close()
  627. if err != nil {
  628. f.l.Error("Error while closing udp socket", "error", err, "writer", i)
  629. errs = append(errs, err)
  630. }
  631. }
  632. // Release the tun device (closing the tun also closes all readers)
  633. closeErr := f.inside.Close()
  634. if closeErr != nil {
  635. errs = append(errs, closeErr)
  636. }
  637. // Release the construction token so waiters know the resources are gone
  638. f.wg.Done()
  639. return errors.Join(errs...)
  640. }