nebula/interface.go

package nebula

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"net/netip"
	"runtime"
	"slices"
	"sync"
	"sync/atomic"
	"time"

	"github.com/gaissmai/bart"
	"github.com/rcrowley/go-metrics"
	"github.com/slackhq/nebula/util"

	"github.com/slackhq/nebula/cert"
	"github.com/slackhq/nebula/config"
	"github.com/slackhq/nebula/firewall"
	"github.com/slackhq/nebula/header"
	"github.com/slackhq/nebula/overlay"
	"github.com/slackhq/nebula/overlay/batch"
	"github.com/slackhq/nebula/overlay/tio"
	"github.com/slackhq/nebula/udp"
)

const mtu = 9001

type InterfaceConfig struct {
	HostMap            *HostMap
	Outside            udp.Conn
	Inside             overlay.Device
	pki                *PKI
	Cipher             string
	Firewall           *Firewall
	DnsServer          *dnsServer
	HandshakeManager   *HandshakeManager
	lightHouse         *LightHouse
	connectionManager  *connectionManager
	DropLocalBroadcast bool
	DropMulticast      bool
	routines           int
	MessageMetrics     *MessageMetrics
	version            string
	relayManager       *relayManager
	punchy             *Punchy

	tryPromoteEvery uint32
	reQueryEvery    uint32
	reQueryWait     time.Duration

	ConntrackCacheTimeout time.Duration

	// CpuAffinity, when non-empty, names the CPUs each TUN reader goroutine
	// should pin to. Queue i pins to CpuAffinity[i % len(CpuAffinity)] —
	// shorter lists than `routines` cycle. Empty list keeps the default
	// pin-to-(i % NumCPU) behavior. Only consulted when PinThreads is true.
	CpuAffinity []int
	// PinThreads controls whether each TUN reader OS thread is pinned to a
	// single CPU (via tun.pin_threads, default true). Pinning keeps each
	// goroutine's sendmmsg on one XPS-selected NIC TX ring so per-flow
	// packets stay ordered on the wire.
	PinThreads bool

	l *slog.Logger
}

type Interface struct {
	hostMap               *HostMap
	outside               udp.Conn
	inside                overlay.Device
	pki                   *PKI
	firewall              *Firewall
	connectionManager     *connectionManager
	handshakeManager      *HandshakeManager
	dnsServer             *dnsServer
	createTime            time.Time
	lightHouse            *LightHouse
	myBroadcastAddrsTable *bart.Lite
	myVpnAddrs            []netip.Addr // A list of addresses assigned to us via our certificate
	myVpnAddrsTable       *bart.Lite
	myVpnNetworks         []netip.Prefix // A list of networks assigned to us via our certificate
	myVpnNetworksTable    *bart.Lite
	dropLocalBroadcast    bool
	dropMulticast         bool
	routines              int
	disconnectInvalid     atomic.Bool
	closed                atomic.Bool
	// cpuAffinity, when non-empty, names the CPUs each TUN reader goroutine
	// should pin to. Queue i pins to cpuAffinity[i % len(cpuAffinity)].
	// Empty falls back to the default pin-to-(allowed CPU) behavior.
	// Only consulted when pinThreads is true.
	cpuAffinity []int
	// pinThreads controls whether listenIn pins each TUN reader OS thread to
	// a CPU at all (tun.pin_threads, default true). When false, threads are
	// left free to migrate as on stock nebula.
	pinThreads bool
	// ecnEnabled gates RFC 6040 underlay ECN propagation. When true,
	// inside.go copies the inner ECN onto the outer carrier on encap and
	// decryptToTun folds outer CE into the inner header on decap. Toggle
	// via tunnels.ecn (default true).
	ecnEnabled   atomic.Bool
	relayManager *relayManager

	tryPromoteEvery atomic.Uint32
	reQueryEvery    atomic.Uint32
	reQueryWait     atomic.Int64

	sendRecvErrorConfig   recvErrorConfig
	acceptRecvErrorConfig recvErrorConfig

	// rebindCount is used to decide if an active tunnel should trigger a punch notification through a lighthouse
	rebindCount int8
	version     string

	conntrackCacheTimeout time.Duration

	ctx     context.Context
	writers []udp.Conn
	queues  []tio.Queue
	// batchers is one per tun queue, wrapping queues[i].
	// decryptToTun sends plaintext into the batch.RxBatcher;
	// listenOut calls its Flush at the end of each UDP recvmmsg batch.
	batchers []batch.RxBatcher
	wg       sync.WaitGroup

	// fatalErr holds the first unexpected reader error that caused shutdown.
	// nil means "no fatal error" (yet)
	fatalErr atomic.Pointer[error]
	// triggerShutdown is a function that will be run exactly once, when onFatal swaps something non-nil into fatalErr
	triggerShutdown func()

	metricHandshakes    metrics.Histogram
	messageMetrics      *MessageMetrics
	cachedPacketMetrics *cachedPacketMetrics

	l *slog.Logger
}

type EncWriter interface {
	SendVia(via *HostInfo,
		relay *Relay,
		ad,
		nb,
		out []byte,
		nocopy bool,
	)
	SendMessageToVpnAddr(t header.MessageType, st header.MessageSubType, vpnAddr netip.Addr, p, nb, out []byte)
	SendMessageToHostInfo(t header.MessageType, st header.MessageSubType, hostinfo *HostInfo, p, nb, out []byte)
	Handshake(vpnAddr netip.Addr)
	GetHostInfo(vpnAddr netip.Addr) *HostInfo
	GetCertState() *CertState
}

type recvErrorConfig uint8

const (
	recvErrorAlways recvErrorConfig = iota
	recvErrorNever
	recvErrorPrivate
)

func (s recvErrorConfig) ShouldRecvError(endpoint netip.AddrPort) bool {
	switch s {
	case recvErrorPrivate:
		return endpoint.Addr().IsPrivate()
	case recvErrorAlways:
		return true
	case recvErrorNever:
		return false
	default:
		panic(fmt.Errorf("invalid recvErrorConfig value: %d", s))
	}
}

func (s recvErrorConfig) String() string {
	switch s {
	case recvErrorAlways:
		return "always"
	case recvErrorNever:
		return "never"
	case recvErrorPrivate:
		return "private"
	default:
		return fmt.Sprintf("invalid(%d)", s)
	}
}

func NewInterface(ctx context.Context, c *InterfaceConfig) (*Interface, error) {
	if c.Outside == nil {
		return nil, errors.New("no outside connection")
	}
	if c.Inside == nil {
		return nil, errors.New("no inside interface (tun)")
	}
	if c.pki == nil {
		return nil, errors.New("no certificate state")
	}
	if c.Firewall == nil {
		return nil, errors.New("no firewall rules")
	}
	if c.connectionManager == nil {
		return nil, errors.New("no connection manager")
	}

	cs := c.pki.getCertState()
	ifce := &Interface{
		ctx:                   ctx,
		pki:                   c.pki,
		hostMap:               c.HostMap,
		outside:               c.Outside,
		inside:                c.Inside,
		firewall:              c.Firewall,
		dnsServer:             c.DnsServer,
		handshakeManager:      c.HandshakeManager,
		createTime:            time.Now(),
		lightHouse:            c.lightHouse,
		dropLocalBroadcast:    c.DropLocalBroadcast,
		dropMulticast:         c.DropMulticast,
		routines:              c.routines,
		version:               c.version,
		writers:               make([]udp.Conn, c.routines),
		batchers:              make([]batch.RxBatcher, c.routines),
		myVpnNetworks:         cs.myVpnNetworks,
		myVpnNetworksTable:    cs.myVpnNetworksTable,
		myVpnAddrs:            cs.myVpnAddrs,
		myVpnAddrsTable:       cs.myVpnAddrsTable,
		myBroadcastAddrsTable: cs.myVpnBroadcastAddrsTable,
		relayManager:          c.relayManager,
		connectionManager:     c.connectionManager,
		conntrackCacheTimeout: c.ConntrackCacheTimeout,
		cpuAffinity:           c.CpuAffinity,
		pinThreads:            c.PinThreads,

		metricHandshakes: metrics.GetOrRegisterHistogram("handshakes", nil, metrics.NewExpDecaySample(1028, 0.015)),
		messageMetrics:   c.MessageMetrics,
		cachedPacketMetrics: &cachedPacketMetrics{
			sent:    metrics.GetOrRegisterCounter("hostinfo.cached_packets.sent", nil),
			dropped: metrics.GetOrRegisterCounter("hostinfo.cached_packets.dropped", nil),
		},

		l: c.l,
	}

	ifce.tryPromoteEvery.Store(c.tryPromoteEvery)
	ifce.reQueryEvery.Store(c.reQueryEvery)
	ifce.reQueryWait.Store(int64(c.reQueryWait))

	ifce.connectionManager.intf = ifce

	// Held until Close so waiting on the interface blocks until the resources are actually released
	ifce.wg.Add(1)

	return ifce, nil
}

// activate creates the interface on the host. After the interface is created, any
// other services that want to bind listeners to its IP may do so successfully. However,
// the interface isn't going to process anything until run() is called.
func (f *Interface) activate() error {
	// actually turn on tun dev

	addr, err := f.outside.LocalAddr()
	if err != nil {
		f.l.Error("Failed to get udp listen address", "error", err)
	}

	f.l.Info("Nebula interface is active",
		"interface", f.inside.Name(),
		"networks", f.myVpnNetworks,
		"build", f.version,
		"udpAddr", addr,
		"boringcrypto", boringEnabled(),
	)

	if f.routines > 1 && !f.outside.SupportsMultipleReaders() {
		f.routines = 1
		f.l.Warn("multiple udp readers are not supported on this platform, falling back to a single routine")
	}

	// Prepare the tun queues. A device that can't open that many hands back
	// fewer (a single queue on platforms without multiqueue support) and we
	// size the reader routines to what we actually got.
	queues, err := f.inside.Queues(f.routines)
	if err != nil {
		return err
	}
	if len(queues) < f.routines {
		f.l.Warn("tun multiqueue is not supported on this platform, falling back to fewer routines",
			"requested", f.routines, "opened", len(queues))
		f.routines = len(queues)
	}
	f.queues = queues

	metrics.GetOrRegisterGauge("routines", nil).Update(int64(f.routines))

	for i := range f.queues {
		caps := tio.QueueCapabilities(f.queues[i])
		if caps.TSO || caps.USO {
			// Multi-lane: TCP gets coalesced when TSO is on, UDP when USO
			// is on, everything else (and either lane disabled) falls
			// through to passthrough so non-IP / non-TCP-UDP traffic still
			// reaches the TUN.
			arena := batch.NewArena(batch.DefaultMultiArenaCap)
			f.batchers[i] = batch.NewMultiCoalescer(f.queues[i], f.l, arena, caps.TSO, caps.USO)
		} else {
			arena := batch.NewArena(batch.DefaultPassthroughArenaCap)
			f.batchers[i] = batch.NewPassthrough(f.queues[i], arena)
		}
	}

	// On error the caller owns the cleanup, Control.Start cancels the service context
	// before releasing our resources so a waiter never observes a live context
	if err = f.inside.Activate(); err != nil {
		return err
	}

	return nil
}

func (f *Interface) run() {
	// Launch n queues to read packets from udp
	for i := 0; i < f.routines; i++ {
		f.wg.Go(func() {
			f.listenOut(i)
		})
	}

	// Launch n queues to read packets from tun dev
	for i := 0; i < f.routines; i++ {
		f.wg.Go(func() {
			f.listenIn(f.queues[i], i)
		})
	}

}

func (f *Interface) wait() error {
	f.wg.Wait()
	if e := f.fatalErr.Load(); e != nil {
		return *e
	}
	return nil
}

// onFatal stores the first fatal reader error, and calls triggerShutdown if it was the first one
func (f *Interface) onFatal(err error) {
	swapped := f.fatalErr.CompareAndSwap(nil, &err)
	if !swapped {
		return
	}
	if f.triggerShutdown != nil {
		f.triggerShutdown()
	}
}

func (f *Interface) listenOut(i int) {
	var li udp.Conn
	if i > 0 {
		li = f.writers[i]
	} else {
		li = f.outside
	}

	ctCache := firewall.NewConntrackCacheTicker(f.ctx, f.l, f.conntrackCacheTimeout)
	lhh := f.lightHouse.NewRequestHandler()
	h := &header.H{}
	fwPacket := &firewall.Packet{}
	nb := make([]byte, 12, 12)

	listener := func(fromUdpAddr netip.AddrPort, payload []byte, meta udp.RxMeta) {
		plaintext := f.batchers[i].Reserve(len(payload))
		f.readOutsidePackets(ViaSender{UdpAddr: fromUdpAddr}, plaintext[:0], payload, h, fwPacket, lhh, nb, i, ctCache.Get(), meta)
	}

	flusher := func() {
		if err := f.batchers[i].Flush(); err != nil {
			f.l.Error("Failed to flush tun coalescer", "error", err)
		}
	}

	err := li.ListenOut(listener, flusher)

	// An error after teardown began is shutdown noise, the closed flag covers resources
	// Close releases itself and the cancelled ctx covers ones torn down by their owners
	// reacting to it, like the user device pipes
	if err != nil && !f.closed.Load() && f.ctx.Err() == nil {
		f.l.Error("Error while reading inbound packet, closing", "error", err)
		f.onFatal(err)
	}

	f.l.Debug("underlay reader is done", "reader", i)
}

func (f *Interface) listenIn(queue tio.Queue, i int) {
	// Pinning this thread (and goroutine) to a single CPU keeps every sendmmsg from this goroutine going through the
	// same TX ring on the nic, so the wire sees per-flow order. Skip entirely when tun.pin_threads is false.
	if f.pinThreads {
		var cpu int
		if n := len(f.cpuAffinity); n > 0 {
			// Explicit tun.cpu_affinity list wins; parseCpuAffinity already
			// validated the entries against the allowed CPU set.
			cpu = f.cpuAffinity[i%n]
		} else if allowed, err := util.AllowedCPUs(); err == nil && len(allowed) > 0 {
			// Default: spread queues across the CPUs we're actually allowed to
			// run on. Under a cpuset/taskset mask these aren't 0..NumCPU-1, so
			// i % NumCPU would pick unrunnable IDs and every pin would fail.
			cpu = allowed[i%len(allowed)]
		} else {
			cpu = i % runtime.NumCPU()
		}
		if err := util.PinThreadToCPU(cpu); err != nil {
			f.l.Warn("failed to pin tun reader to CPU", "queue", i, "cpu", cpu, "err", err)
		}
	}

	rejectBuf := make([]byte, mtu)
	arenaSize := batch.SendBatchCap * (udp.MTU + 32)
	sb := batch.NewSendBatch(f.writers[i], batch.SendBatchCap, arenaSize)
	fwPacket := &firewall.Packet{}
	nb := make([]byte, 12, 12)

	conntrackCache := firewall.NewConntrackCacheTicker(f.ctx, f.l, f.conntrackCacheTimeout)

	for {
		pkts, err := queue.Read()
		if err != nil {
			// Same shutdown noise handling as listenOut
			if !f.closed.Load() && f.ctx.Err() == nil {
				f.l.Error("Error while reading outbound packet, closing", "error", err, "reader", i)
				f.onFatal(err)
			}
			break
		}

		for _, pkt := range pkts {
			f.consumeInsidePacket(pkt, fwPacket, nb, sb, rejectBuf, i, conntrackCache.Get())
			// Flush incrementally once a full sendmmsg batch has
			// accumulated so the first packets of a deep read drain
			// hit the wire while the rest are still being encrypted.
			if sb.Len() >= batch.SendBatchCap {
				if err := sb.Flush(); err != nil {
					f.l.Error("Failed to write outgoing batch", "error", err, "writer", i)
				}
			}
		}
		if err := sb.Flush(); err != nil {
			f.l.Error("Failed to write outgoing batch", "error", err, "writer", i)
		}
	}

	f.l.Debug("overlay reader is done", "reader", i)
}

func (f *Interface) RegisterConfigChangeCallbacks(c *config.C) {
	c.RegisterReloadCallback(f.reloadFirewall)
	c.RegisterReloadCallback(f.reloadSendRecvError)
	c.RegisterReloadCallback(f.reloadAcceptRecvError)
	c.RegisterReloadCallback(f.reloadDisconnectInvalid)
	c.RegisterReloadCallback(f.reloadMisc)
	c.RegisterReloadCallback(f.reloadEcn)

	for _, udpConn := range f.writers {
		c.RegisterReloadCallback(udpConn.ReloadConfig)
	}
}

func (f *Interface) reloadDisconnectInvalid(c *config.C) {
	initial := c.InitialLoad()
	if initial || c.HasChanged("pki.disconnect_invalid") {
		f.disconnectInvalid.Store(c.GetBool("pki.disconnect_invalid", true))
		if !initial {
			f.l.Info("pki.disconnect_invalid changed", "value", f.disconnectInvalid.Load())
		}
	}
}

func (f *Interface) reloadFirewall(c *config.C) {
	cs := f.pki.getCertState()
	curCert := cs.getCertificate(cert.Version2)
	if curCert == nil {
		curCert = cs.getCertificate(cert.Version1)
	}

	// The firewall builds its routableNetworks set from the certificate's UnsafeNetworks at construction.
	// Check to see if that set has changed, and if so, rebuild the firewall.
	certUnsafeChanged := curCert != nil && !slices.Equal(curCert.UnsafeNetworks(), f.firewall.unsafeNetworks)

	if !c.HasChanged("firewall") && !certUnsafeChanged {
		f.l.Debug("No firewall config change detected")
		return
	}

	fw, err := NewFirewallFromConfig(f.l, cs, c)
	if err != nil {
		f.l.Error("Error while creating firewall during reload", "error", err)
		return
	}

	oldFw := f.firewall
	conntrack := oldFw.Conntrack
	conntrack.Lock()
	defer conntrack.Unlock()

	fw.rulesVersion = oldFw.rulesVersion + 1
	// If rulesVersion is back to zero, we have wrapped all the way around. Be
	// safe and just reset conntrack in this case.
	if fw.rulesVersion == 0 {
		f.l.Warn("firewall rulesVersion has overflowed, resetting conntrack",
			"firewallHashes", fw.GetRuleHashes(),
			"oldFirewallHashes", oldFw.GetRuleHashes(),
			"rulesVersion", fw.rulesVersion,
		)
	} else {
		fw.Conntrack = conntrack
	}

	f.firewall = fw

	oldFw.Destroy()
	f.l.Info("New firewall has been installed",
		"firewallHashes", fw.GetRuleHashes(),
		"oldFirewallHashes", oldFw.GetRuleHashes(),
		"rulesVersion", fw.rulesVersion,
	)
}

func (f *Interface) reloadSendRecvError(c *config.C) {
	if c.InitialLoad() || c.HasChanged("listen.send_recv_error") {
		stringValue := c.GetString("listen.send_recv_error", "always")

		switch stringValue {
		case "always":
			f.sendRecvErrorConfig = recvErrorAlways
		case "never":
			f.sendRecvErrorConfig = recvErrorNever
		case "private":
			f.sendRecvErrorConfig = recvErrorPrivate
		default:
			if c.GetBool("listen.send_recv_error", true) {
				f.sendRecvErrorConfig = recvErrorAlways
			} else {
				f.sendRecvErrorConfig = recvErrorNever
			}
		}

		f.l.Info("Loaded send_recv_error config", "sendRecvError", f.sendRecvErrorConfig.String())
	}
}

func (f *Interface) reloadAcceptRecvError(c *config.C) {
	if c.InitialLoad() || c.HasChanged("listen.accept_recv_error") {
		stringValue := c.GetString("listen.accept_recv_error", "always")

		switch stringValue {
		case "always":
			f.acceptRecvErrorConfig = recvErrorAlways
		case "never":
			f.acceptRecvErrorConfig = recvErrorNever
		case "private":
			f.acceptRecvErrorConfig = recvErrorPrivate
		default:
			if c.GetBool("listen.accept_recv_error", true) {
				f.acceptRecvErrorConfig = recvErrorAlways
			} else {
				f.acceptRecvErrorConfig = recvErrorNever
			}
		}

		f.l.Info("Loaded accept_recv_error config", "acceptRecvError", f.acceptRecvErrorConfig.String())
	}
}

func (f *Interface) reloadMisc(c *config.C) {
	if c.HasChanged("counters.try_promote") {
		n := c.GetUint32("counters.try_promote", defaultPromoteEvery)
		f.tryPromoteEvery.Store(n)
		f.l.Info("counters.try_promote has changed")
	}

	if c.HasChanged("counters.requery_every_packets") {
		n := c.GetUint32("counters.requery_every_packets", defaultReQueryEvery)
		f.reQueryEvery.Store(n)
		f.l.Info("counters.requery_every_packets has changed")
	}

	if c.HasChanged("timers.requery_wait_duration") {
		n := c.GetDuration("timers.requery_wait_duration", defaultReQueryWait)
		f.reQueryWait.Store(int64(n))
		f.l.Info("timers.requery_wait_duration has changed")
	}
}

// reloadEcn syncs Interface.ecnEnabled with the tunnels.ecn config knob.
// Default is enabled (RFC 6040 normal mode); set false on the rare path
// where an underlay middlebox rewrites or drops ECN bits unpredictably.
func (f *Interface) reloadEcn(c *config.C) {
	initial := c.InitialLoad()
	if initial || c.HasChanged("tunnels.ecn") {
		v := c.GetBool("tunnels.ecn", true)
		changed := f.ecnEnabled.Swap(v) != v
		if !initial {
			f.l.Info("tunnels.ecn changed", "enabled", v)
			if changed {
				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)
			}
		}
	}
}

func (f *Interface) emitStats(ctx context.Context, i time.Duration) {
	ticker := time.NewTicker(i)
	defer ticker.Stop()

	udpStats := udp.NewUDPStatsEmitter(f.writers)

	certExpirationGauge := metrics.GetOrRegisterGauge("certificate.ttl_seconds", nil)
	certInitiatingVersion := metrics.GetOrRegisterGauge("certificate.initiating_version", nil)
	certMaxVersion := metrics.GetOrRegisterGauge("certificate.max_version", nil)

	emit := func() {
		f.firewall.EmitStats()
		f.handshakeManager.EmitStats()
		udpStats()

		certState := f.pki.getCertState()
		defaultCrt := certState.GetDefaultCertificate()
		certExpirationGauge.Update(int64(defaultCrt.NotAfter().Sub(time.Now()) / time.Second))
		certInitiatingVersion.Update(int64(defaultCrt.Version()))

		// Report the max certificate version we are capable of using
		if certState.v2Cert != nil {
			certMaxVersion.Update(int64(certState.v2Cert.Version()))
		} else {
			certMaxVersion.Update(int64(certState.v1Cert.Version()))
		}
	}

	// Prime gauges so a Prometheus scrape that lands before the first tick
	// sees real values instead of the zero defaults (issue #907).
	emit()

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			emit()
		}
	}
}

func (f *Interface) GetHostInfo(vpnIp netip.Addr) *HostInfo {
	return f.hostMap.QueryVpnAddr(vpnIp)
}

func (f *Interface) GetCertState() *CertState {
	return f.pki.getCertState()
}

// Close releases the interface's resources: the udp sockets and the tun device.
// It is idempotent and safe to call at any point in the lifecycle, including on an interface that never activated,
// calls after the first return nil without doing anything.
func (f *Interface) Close() error {
	if !f.closed.CompareAndSwap(false, true) {
		return nil
	}

	var errs []error

	// Release the udp readers
	for i, u := range f.writers {
		err := u.Close()
		if err != nil {
			f.l.Error("Error while closing udp socket", "error", err, "writer", i)
			errs = append(errs, err)
		}
	}

	// Release the tun device (closing the tun also closes all readers)
	closeErr := f.inside.Close()
	if closeErr != nil {
		errs = append(errs, closeErr)
	}

	// Release the construction token so waiters know the resources are gone
	f.wg.Done()
	return errors.Join(errs...)
}