interface.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723
  1. package nebula
  2. import (
  3. "context"
  4. "crypto/fips140"
  5. "errors"
  6. "fmt"
  7. "log/slog"
  8. "net/netip"
  9. "runtime"
  10. "slices"
  11. "sync"
  12. "sync/atomic"
  13. "time"
  14. "github.com/gaissmai/bart"
  15. "github.com/rcrowley/go-metrics"
  16. "github.com/slackhq/nebula/util"
  17. "github.com/slackhq/nebula/cert"
  18. "github.com/slackhq/nebula/config"
  19. "github.com/slackhq/nebula/firewall"
  20. "github.com/slackhq/nebula/header"
  21. "github.com/slackhq/nebula/overlay"
  22. "github.com/slackhq/nebula/overlay/batch"
  23. "github.com/slackhq/nebula/overlay/tio"
  24. "github.com/slackhq/nebula/ratelimit"
  25. "github.com/slackhq/nebula/udp"
  26. )
  27. const mtu = 9001
  28. type InterfaceConfig struct {
  29. HostMap *HostMap
  30. Outside udp.Conn
  31. Inside overlay.Device
  32. pki *PKI
  33. Cipher string
  34. Firewall *Firewall
  35. DnsServer *dnsServer
  36. HandshakeManager *HandshakeManager
  37. lightHouse *LightHouse
  38. connectionManager *connectionManager
  39. DropLocalBroadcast bool
  40. DropMulticast bool
  41. routines int
  42. MessageMetrics *MessageMetrics
  43. version string
  44. relayManager *relayManager
  45. punchy *Punchy
  46. tryPromoteEvery uint32
  47. reQueryEvery uint32
  48. reQueryWait time.Duration
  49. ConntrackCacheTimeout time.Duration
  50. // CpuAffinity, when non-empty, names the CPUs each TUN reader goroutine
  51. // should pin to. Queue i pins to CpuAffinity[i % len(CpuAffinity)] —
  52. // shorter lists than `routines` cycle. Empty list keeps the default
  53. // pin-to-(i % NumCPU) behavior. Only consulted when PinThreads is true.
  54. CpuAffinity []int
  55. // PinThreads controls whether each TUN reader OS thread is pinned to a
  56. // single CPU (via tun.pin_threads, default true). Pinning keeps each
  57. // goroutine's sendmmsg on one XPS-selected NIC TX ring so per-flow
  58. // packets stay ordered on the wire.
  59. PinThreads bool
  60. staged stagedConfig
  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. disconnectInvalid atomic.Bool
  83. closed atomic.Bool
  84. // cpuAffinity, when non-empty, names the CPUs each TUN reader goroutine
  85. // should pin to. Queue i pins to cpuAffinity[i % len(cpuAffinity)].
  86. // Empty falls back to the default pin-to-(allowed CPU) behavior.
  87. // Only consulted when pinThreads is true.
  88. cpuAffinity []int
  89. // pinThreads controls whether listenIn pins each TUN reader OS thread to
  90. // a CPU at all (tun.pin_threads, default true). When false, threads are
  91. // left free to migrate as on stock nebula.
  92. pinThreads bool
  93. staged stagedConfig
  94. relayManager *relayManager
  95. tryPromoteEvery atomic.Uint32
  96. reQueryEvery atomic.Uint32
  97. reQueryWait atomic.Int64
  98. sendRecvErrorConfig recvErrorConfig
  99. acceptRecvErrorConfig recvErrorConfig
  100. // Bumped on every udp rebind, tunnels compare it to decide they need a punch from the far side
  101. rebindEpoch atomic.Uint32
  102. version string
  103. conntrackCacheTimeout time.Duration
  104. ctx context.Context
  105. writers []udp.Conn
  106. queues []tio.Queue
  107. // batchers is one per tun queue, wrapping queues[i]. readOutsidePackets
  108. // commits plaintext into the batcher; the plaintext is decrypted
  109. // in place inside the UDP receive buffers, so listenOut must call Flush
  110. // at the end of each UDP recvmmsg batch, before those buffers are
  111. // reused (every udp.Conn ListenOut guarantees that ordering).
  112. batchers []*batch.MultiCoalescer
  113. wg sync.WaitGroup
  114. udpTxErrLimiter *ratelimit.Limiter // udp direct transmit error log limiter
  115. udpRelayTxErrLimiter *ratelimit.Limiter // udp relay transmit error log limiter
  116. txOversizeLimiter *ratelimit.Limiter // oversized outbound packet drop log limiter
  117. // fatalErr holds the first unexpected reader error that caused shutdown.
  118. // nil means "no fatal error" (yet)
  119. fatalErr atomic.Pointer[error]
  120. // triggerShutdown is a function that will be run exactly once, when onFatal swaps something non-nil into fatalErr
  121. triggerShutdown func()
  122. metricHandshakes metrics.Histogram
  123. messageMetrics *MessageMetrics
  124. cachedPacketMetrics *cachedPacketMetrics
  125. metricTxDropped metrics.Counter
  126. l *slog.Logger
  127. }
  128. type EncWriter interface {
  129. SendVia(via *HostInfo, relay *Relay, ad, nb, out []byte, nocopy bool, q int)
  130. SendMessageToVpnAddr(t header.MessageType, st header.MessageSubType, vpnAddr netip.Addr, p, nb, out []byte)
  131. SendMessageToHostInfo(t header.MessageType, st header.MessageSubType, hostinfo *HostInfo, p, nb, out []byte)
  132. Handshake(vpnAddr netip.Addr)
  133. GetHostInfo(vpnAddr netip.Addr) *HostInfo
  134. GetCertState() *CertState
  135. }
  136. type recvErrorConfig uint8
  137. const (
  138. recvErrorAlways recvErrorConfig = iota
  139. recvErrorNever
  140. recvErrorPrivate
  141. )
  142. func (s recvErrorConfig) ShouldRecvError(endpoint netip.AddrPort) bool {
  143. switch s {
  144. case recvErrorPrivate:
  145. return endpoint.Addr().IsPrivate()
  146. case recvErrorAlways:
  147. return true
  148. case recvErrorNever:
  149. return false
  150. default:
  151. panic(fmt.Errorf("invalid recvErrorConfig value: %d", s))
  152. }
  153. }
  154. func (s recvErrorConfig) String() string {
  155. switch s {
  156. case recvErrorAlways:
  157. return "always"
  158. case recvErrorNever:
  159. return "never"
  160. case recvErrorPrivate:
  161. return "private"
  162. default:
  163. return fmt.Sprintf("invalid(%d)", s)
  164. }
  165. }
  166. func NewInterface(ctx context.Context, c *InterfaceConfig) (*Interface, error) {
  167. if c.Outside == nil {
  168. return nil, errors.New("no outside connection")
  169. }
  170. if c.Inside == nil {
  171. return nil, errors.New("no inside interface (tun)")
  172. }
  173. if c.pki == nil {
  174. return nil, errors.New("no certificate state")
  175. }
  176. if c.Firewall == nil {
  177. return nil, errors.New("no firewall rules")
  178. }
  179. if c.connectionManager == nil {
  180. return nil, errors.New("no connection manager")
  181. }
  182. udpTxErrLimiter, err := ratelimit.New(1, time.Second)
  183. if err != nil {
  184. return nil, err
  185. }
  186. udpRelayTxErrLimiter, err := ratelimit.New(1, time.Second)
  187. if err != nil {
  188. return nil, err
  189. }
  190. txOversizeLimiter, err := ratelimit.New(1, time.Second)
  191. if err != nil {
  192. return nil, err
  193. }
  194. if c.routines <= 1 {
  195. c.PinThreads = false //pinning is not useful unless there's more than one tun reader
  196. }
  197. cs := c.pki.getCertState()
  198. ifce := &Interface{
  199. ctx: ctx,
  200. pki: c.pki,
  201. hostMap: c.HostMap,
  202. outside: c.Outside,
  203. inside: c.Inside,
  204. firewall: c.Firewall,
  205. dnsServer: c.DnsServer,
  206. handshakeManager: c.HandshakeManager,
  207. createTime: time.Now(),
  208. lightHouse: c.lightHouse,
  209. dropLocalBroadcast: c.DropLocalBroadcast,
  210. dropMulticast: c.DropMulticast,
  211. routines: c.routines,
  212. version: c.version,
  213. writers: make([]udp.Conn, c.routines),
  214. batchers: make([]*batch.MultiCoalescer, c.routines),
  215. myVpnNetworks: cs.myVpnNetworks,
  216. myVpnNetworksTable: cs.myVpnNetworksTable,
  217. myVpnAddrs: cs.myVpnAddrs,
  218. myVpnAddrsTable: cs.myVpnAddrsTable,
  219. myBroadcastAddrsTable: cs.myVpnBroadcastAddrsTable,
  220. relayManager: c.relayManager,
  221. connectionManager: c.connectionManager,
  222. conntrackCacheTimeout: c.ConntrackCacheTimeout,
  223. cpuAffinity: c.CpuAffinity,
  224. pinThreads: c.PinThreads,
  225. staged: c.staged,
  226. udpTxErrLimiter: udpTxErrLimiter,
  227. udpRelayTxErrLimiter: udpRelayTxErrLimiter,
  228. txOversizeLimiter: txOversizeLimiter,
  229. metricHandshakes: metrics.GetOrRegisterHistogram("handshakes", nil, metrics.NewExpDecaySample(1028, 0.015)),
  230. metricTxDropped: metrics.GetOrRegisterCounter("udp.tx.dropped", nil),
  231. messageMetrics: c.MessageMetrics,
  232. cachedPacketMetrics: &cachedPacketMetrics{
  233. sent: metrics.GetOrRegisterCounter("hostinfo.cached_packets.sent", nil),
  234. dropped: metrics.GetOrRegisterCounter("hostinfo.cached_packets.dropped", nil),
  235. },
  236. l: c.l,
  237. }
  238. ifce.tryPromoteEvery.Store(c.tryPromoteEvery)
  239. ifce.reQueryEvery.Store(c.reQueryEvery)
  240. ifce.reQueryWait.Store(int64(c.reQueryWait))
  241. ifce.connectionManager.intf = ifce
  242. // Held until Close so waiting on the interface blocks until the resources are actually released
  243. ifce.wg.Add(1)
  244. return ifce, nil
  245. }
  246. // activate creates the interface on the host. After the interface is created, any
  247. // other services that want to bind listeners to its IP may do so successfully. However,
  248. // the interface isn't going to process anything until run() is called.
  249. func (f *Interface) activate() error {
  250. // actually turn on tun dev
  251. addr, err := f.outside.LocalAddr()
  252. if err != nil {
  253. f.l.Error("Failed to get udp listen address", "error", err)
  254. }
  255. f.l.Info("Nebula interface is active",
  256. "interface", f.inside.Name(),
  257. "networks", f.myVpnNetworks,
  258. "build", f.version,
  259. "udpAddr", addr,
  260. "boringcrypto", boringEnabled(),
  261. "fips140Version", fips140.Version(),
  262. "fips140Enabled", fips140.Enabled(),
  263. "fips140Enforced", fips140.Enforced(),
  264. )
  265. if f.routines > 1 && !f.outside.SupportsMultipleReaders() {
  266. f.routines = 1
  267. f.l.Warn("multiple udp readers are not supported on this platform, falling back to a single routine")
  268. }
  269. // Prepare the tun queues. A device that can't open that many hands back
  270. // fewer (a single queue on platforms without multiqueue support) and we
  271. // size the reader routines to what we actually got.
  272. queues, err := f.inside.Queues(f.routines)
  273. if err != nil {
  274. return err
  275. }
  276. if len(queues) < f.routines {
  277. // TODO: this clamp is only safe because it is unreachable when the
  278. // udp side has multiple readers (linux Queues opens exactly n or
  279. // errors; every other platform already clamped routines to 1 above).
  280. // If a platform ever returns fewer queues than routines with
  281. // SO_REUSEPORT sockets already bound, the surplus sockets get no
  282. // listenOut and the kernel blackholes every flow it hashes to them —
  283. // fail loudly or close the extra sockets instead.
  284. f.l.Warn("tun multiqueue is not supported on this platform, falling back to fewer routines",
  285. "requested", f.routines, "opened", len(queues))
  286. f.routines = len(queues)
  287. }
  288. f.queues = queues
  289. metrics.GetOrRegisterGauge("routines", nil).Update(int64(f.routines))
  290. for i := range f.queues {
  291. f.batchers[i] = batch.NewMultiCoalescer(f.queues[i], f.l)
  292. }
  293. // On error the caller owns the cleanup, Control.Start cancels the service context
  294. // before releasing our resources so a waiter never observes a live context
  295. if err = f.inside.Activate(); err != nil {
  296. return err
  297. }
  298. return nil
  299. }
  300. func (f *Interface) run() {
  301. // Launch n queues to read packets from udp
  302. for i := 0; i < f.routines; i++ {
  303. f.wg.Go(func() {
  304. f.listenOut(i)
  305. })
  306. }
  307. // Launch n queues to read packets from tun dev
  308. for i := 0; i < f.routines; i++ {
  309. f.wg.Go(func() {
  310. f.listenIn(f.queues[i], i)
  311. })
  312. }
  313. }
  314. func (f *Interface) wait() error {
  315. f.wg.Wait()
  316. if e := f.fatalErr.Load(); e != nil {
  317. return *e
  318. }
  319. return nil
  320. }
  321. // onFatal stores the first fatal reader error, and calls triggerShutdown if it was the first one
  322. func (f *Interface) onFatal(err error) {
  323. swapped := f.fatalErr.CompareAndSwap(nil, &err)
  324. if !swapped {
  325. return
  326. }
  327. if f.triggerShutdown != nil {
  328. f.triggerShutdown()
  329. }
  330. }
  331. type rxContext struct {
  332. q int
  333. scratch []byte
  334. // nb is a re-usable nonce buffer for decrypt calls to use
  335. nb []byte
  336. h *header.H
  337. fwPacket *firewall.ParsedPacket
  338. hostmapCache map[uint32]*HostInfo
  339. lhh *LightHouseHandler
  340. ctCache *firewall.ConntrackCacheTicker
  341. // decrypted is staged receive's result for the packet being handled, decrypted ahead of time on a worker
  342. decrypted *rxItem
  343. }
  344. func newRxContext(f *Interface, q int) *rxContext {
  345. return &rxContext{
  346. q: q,
  347. scratch: make([]byte, mtu),
  348. nb: make([]byte, 12, 12),
  349. h: &header.H{},
  350. fwPacket: &firewall.ParsedPacket{},
  351. hostmapCache: map[uint32]*HostInfo{},
  352. lhh: f.lightHouse.NewRequestHandler(),
  353. ctCache: firewall.NewConntrackCacheTicker(f.ctx, f.l, f.conntrackCacheTimeout),
  354. }
  355. }
  356. func (f *Interface) listenOut(i int) {
  357. var li udp.Conn
  358. if i > 0 {
  359. li = f.writers[i]
  360. } else {
  361. li = f.outside
  362. }
  363. rxc := newRxContext(f, i)
  364. listener := func(fromUdpAddr netip.AddrPort, payload []byte) {
  365. f.readOutsidePackets(ViaSender{UdpAddr: fromUdpAddr}, payload, rxc)
  366. }
  367. flusher := func(bufs [][]byte) [][]byte {
  368. f.flushOutside(rxc)
  369. return bufs
  370. }
  371. var err error
  372. if hc, ok := li.(handoffConn); ok && f.staged.workers > 0 {
  373. // Every stage on its own goroutine, see flow_rx.go. rxc becomes the ordered writer's context.
  374. err = newStagedRx(f, i, rxc, hc).listen()
  375. } else {
  376. if f.staged.workers > 0 {
  377. f.l.Info("udp conn cannot hand off receive batches, receive stays inline", "reader", i)
  378. }
  379. err = li.ListenOut(listener, flusher)
  380. }
  381. // An error after teardown began is shutdown noise, the closed flag covers resources
  382. // Close releases itself and the cancelled ctx covers ones torn down by their owners
  383. // reacting to it, like the user device pipes
  384. if err != nil && !f.closed.Load() && f.ctx.Err() == nil {
  385. f.l.Error("Error while reading inbound packet, closing", "error", err)
  386. f.onFatal(err)
  387. }
  388. f.l.Debug("underlay reader is done", "reader", i)
  389. }
  390. func (f *Interface) pinThisThread(i int) {
  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. func (f *Interface) listenIn(queue tio.Queue, i int) {
  409. // Pinning this thread (and goroutine) to a single CPU keeps every sendmmsg from this goroutine going through the
  410. // same TX ring on the nic, so the wire sees per-flow order. Skip entirely when tun.pin_threads is false.
  411. if f.pinThreads {
  412. f.pinThisThread(i)
  413. }
  414. rejectBuf := make([]byte, mtu)
  415. arenaSize := batch.SendBatchCap * (udp.MTU + 32)
  416. sb := batch.NewSendBatch(f.writers[i], batch.SendBatchCap, arenaSize)
  417. fwPacket := &firewall.ParsedPacket{}
  418. nb := make([]byte, 12, 12)
  419. conntrackCache := firewall.NewConntrackCacheTicker(f.ctx, f.l, f.conntrackCacheTimeout)
  420. for {
  421. pkts, err := queue.Read()
  422. if err != nil {
  423. // Same shutdown noise handling as listenOut
  424. if !f.closed.Load() && f.ctx.Err() == nil {
  425. f.l.Error("Error while reading outbound packet, closing", "error", err, "reader", i)
  426. f.onFatal(err)
  427. }
  428. break
  429. }
  430. for _, pkt := range pkts {
  431. f.consumeInsidePacket(pkt, fwPacket, nb, sb, rejectBuf, i, conntrackCache.Get())
  432. // Flush incrementally once a full sendmmsg batch has
  433. // accumulated so the first packets of a deep read drain
  434. // hit the wire while the rest are still being encrypted.
  435. if sb.Len() >= batch.SendBatchCap {
  436. f.flushSendBatch(sb, i)
  437. }
  438. }
  439. f.flushSendBatch(sb, i)
  440. }
  441. f.l.Debug("overlay reader is done", "reader", i)
  442. }
  443. // flushSendBatch drains sb to the underlay and accounts for anything it could not deliver. A shortfall means
  444. // specific destinations were undeliverable (a stale remote, a reject rule), which the backend logs per peer at
  445. // debug; here it is only a counter, so one unreachable peer cannot spam a log line per batch.
  446. func (f *Interface) flushSendBatch(sb *batch.SendBatch, q int) {
  447. queued := sb.Len()
  448. written, err := sb.Flush()
  449. if err != nil {
  450. f.l.Error("Failed to write outgoing batch", "error", err, "writer", q)
  451. }
  452. if dropped := queued - written; dropped > 0 {
  453. f.metricTxDropped.Inc(int64(dropped))
  454. }
  455. }
  456. func (f *Interface) RegisterConfigChangeCallbacks(c *config.C) {
  457. c.RegisterReloadCallback(f.reloadFirewall)
  458. c.RegisterReloadCallback(f.reloadSendRecvError)
  459. c.RegisterReloadCallback(f.reloadAcceptRecvError)
  460. c.RegisterReloadCallback(f.reloadDisconnectInvalid)
  461. c.RegisterReloadCallback(f.reloadMisc)
  462. for _, udpConn := range f.writers {
  463. c.RegisterReloadCallback(udpConn.ReloadConfig)
  464. }
  465. }
  466. func (f *Interface) reloadDisconnectInvalid(c *config.C) {
  467. initial := c.InitialLoad()
  468. if initial || c.HasChanged("pki.disconnect_invalid") {
  469. f.disconnectInvalid.Store(c.GetBool("pki.disconnect_invalid", true))
  470. if !initial {
  471. f.l.Info("pki.disconnect_invalid changed", "value", f.disconnectInvalid.Load())
  472. }
  473. }
  474. }
  475. func (f *Interface) reloadFirewall(c *config.C) {
  476. cs := f.pki.getCertState()
  477. curCert := cs.getCertificate(cert.Version2)
  478. if curCert == nil {
  479. curCert = cs.getCertificate(cert.Version1)
  480. }
  481. // The firewall builds its routableNetworks set from the certificate's UnsafeNetworks at construction.
  482. // Check to see if that set has changed, and if so, rebuild the firewall.
  483. certUnsafeChanged := curCert != nil && !slices.Equal(curCert.UnsafeNetworks(), f.firewall.unsafeNetworks)
  484. if !c.HasChanged("firewall") && !certUnsafeChanged {
  485. f.l.Debug("No firewall config change detected")
  486. return
  487. }
  488. fw, err := NewFirewallFromConfig(f.l, cs, c)
  489. if err != nil {
  490. f.l.Error("Error while creating firewall during reload", "error", err)
  491. return
  492. }
  493. oldFw := f.firewall
  494. conntrack := oldFw.Conntrack
  495. conntrack.Lock()
  496. defer conntrack.Unlock()
  497. fw.rulesVersion = oldFw.rulesVersion + 1
  498. // If rulesVersion is back to zero, we have wrapped all the way around. Be
  499. // safe and just reset conntrack in this case.
  500. if fw.rulesVersion == 0 {
  501. f.l.Warn("firewall rulesVersion has overflowed, resetting conntrack",
  502. "firewallHashes", fw.GetRuleHashes(),
  503. "oldFirewallHashes", oldFw.GetRuleHashes(),
  504. "rulesVersion", fw.rulesVersion,
  505. )
  506. } else {
  507. fw.Conntrack = conntrack
  508. }
  509. f.firewall = fw
  510. oldFw.Destroy()
  511. f.l.Info("New firewall has been installed",
  512. "firewallHashes", fw.GetRuleHashes(),
  513. "oldFirewallHashes", oldFw.GetRuleHashes(),
  514. "rulesVersion", fw.rulesVersion,
  515. )
  516. }
  517. func (f *Interface) reloadSendRecvError(c *config.C) {
  518. if c.InitialLoad() || c.HasChanged("listen.send_recv_error") {
  519. stringValue := c.GetString("listen.send_recv_error", "always")
  520. switch stringValue {
  521. case "always":
  522. f.sendRecvErrorConfig = recvErrorAlways
  523. case "never":
  524. f.sendRecvErrorConfig = recvErrorNever
  525. case "private":
  526. f.sendRecvErrorConfig = recvErrorPrivate
  527. default:
  528. if c.GetBool("listen.send_recv_error", true) {
  529. f.sendRecvErrorConfig = recvErrorAlways
  530. } else {
  531. f.sendRecvErrorConfig = recvErrorNever
  532. }
  533. }
  534. f.l.Info("Loaded send_recv_error config", "sendRecvError", f.sendRecvErrorConfig.String())
  535. }
  536. }
  537. func (f *Interface) reloadAcceptRecvError(c *config.C) {
  538. if c.InitialLoad() || c.HasChanged("listen.accept_recv_error") {
  539. stringValue := c.GetString("listen.accept_recv_error", "always")
  540. switch stringValue {
  541. case "always":
  542. f.acceptRecvErrorConfig = recvErrorAlways
  543. case "never":
  544. f.acceptRecvErrorConfig = recvErrorNever
  545. case "private":
  546. f.acceptRecvErrorConfig = recvErrorPrivate
  547. default:
  548. if c.GetBool("listen.accept_recv_error", true) {
  549. f.acceptRecvErrorConfig = recvErrorAlways
  550. } else {
  551. f.acceptRecvErrorConfig = recvErrorNever
  552. }
  553. }
  554. f.l.Info("Loaded accept_recv_error config", "acceptRecvError", f.acceptRecvErrorConfig.String())
  555. }
  556. }
  557. func (f *Interface) reloadMisc(c *config.C) {
  558. if c.HasChanged("counters.try_promote") {
  559. n := c.GetUint32("counters.try_promote", defaultPromoteEvery)
  560. f.tryPromoteEvery.Store(n)
  561. f.l.Info("counters.try_promote has changed")
  562. }
  563. if c.HasChanged("counters.requery_every_packets") {
  564. n := c.GetUint32("counters.requery_every_packets", defaultReQueryEvery)
  565. f.reQueryEvery.Store(n)
  566. f.l.Info("counters.requery_every_packets has changed")
  567. }
  568. if c.HasChanged("timers.requery_wait_duration") {
  569. n := c.GetDuration("timers.requery_wait_duration", defaultReQueryWait)
  570. f.reQueryWait.Store(int64(n))
  571. f.l.Info("timers.requery_wait_duration has changed")
  572. }
  573. }
  574. // statsEmitter updates the interface's gauges, the stats server runs it on its capture loop.
  575. func (f *Interface) statsEmitter() func() {
  576. udpStats := udp.NewUDPStatsEmitter(f.writers)
  577. certExpirationGauge := metrics.GetOrRegisterGauge("certificate.ttl_seconds", nil)
  578. certInitiatingVersion := metrics.GetOrRegisterGauge("certificate.initiating_version", nil)
  579. certMaxVersion := metrics.GetOrRegisterGauge("certificate.max_version", nil)
  580. return func() {
  581. f.firewall.EmitStats()
  582. f.handshakeManager.EmitStats()
  583. udpStats()
  584. certState := f.pki.getCertState()
  585. defaultCrt := certState.GetDefaultCertificate()
  586. certExpirationGauge.Update(int64(defaultCrt.NotAfter().Sub(time.Now()) / time.Second))
  587. certInitiatingVersion.Update(int64(defaultCrt.Version()))
  588. // Report the max certificate version we are capable of using
  589. if certState.v2Cert != nil {
  590. certMaxVersion.Update(int64(certState.v2Cert.Version()))
  591. } else {
  592. certMaxVersion.Update(int64(certState.v1Cert.Version()))
  593. }
  594. }
  595. }
  596. func (f *Interface) GetHostInfo(vpnIp netip.Addr) *HostInfo {
  597. return f.hostMap.QueryVpnAddr(vpnIp)
  598. }
  599. func (f *Interface) GetCertState() *CertState {
  600. return f.pki.getCertState()
  601. }
  602. // Close releases the interface's resources: the udp sockets and the tun device.
  603. // It is idempotent and safe to call at any point in the lifecycle, including on an interface that never activated,
  604. // calls after the first return nil without doing anything.
  605. func (f *Interface) Close() error {
  606. if !f.closed.CompareAndSwap(false, true) {
  607. return nil
  608. }
  609. var errs []error
  610. // Release the udp readers
  611. for i, u := range f.writers {
  612. err := u.Close()
  613. if err != nil {
  614. f.l.Error("Error while closing udp socket", "error", err, "writer", i)
  615. errs = append(errs, err)
  616. }
  617. }
  618. // Release the tun device (closing the tun also closes all readers)
  619. closeErr := f.inside.Close()
  620. if closeErr != nil {
  621. errs = append(errs, closeErr)
  622. }
  623. // Release the construction token so waiters know the resources are gone
  624. f.wg.Done()
  625. return errors.Join(errs...)
  626. }