Merge branch 'main' into fix/upload-server

client/internal/dns/service_listener.go

@@ -6,6 +6,7 @@ import (
 	"net"
 	"net/netip"
 	"runtime"
+	"slices"
 	"strconv"
 	"sync"
 	"time"
@@ -17,17 +18,20 @@ import (
 	nberrors "github.com/netbirdio/netbird/client/errors"
 
 	firewall "github.com/netbirdio/netbird/client/firewall/manager"
-	"github.com/netbirdio/netbird/client/internal/ebpf"
-	ebpfMgr "github.com/netbirdio/netbird/client/internal/ebpf/manager"
 )
 
 const (
 	customPort = 5053
+	// randomPortAttempts bounds the search for a port free on both protocols.
+	randomPortAttempts = 5
 )
 
 var (
 	defaultIP = netip.MustParseAddr("127.0.0.1")
 	customIP  = netip.MustParseAddr("127.0.0.153")
+
+	// dnatProtocols are the protocols the port 53 redirect covers.
+	dnatProtocols = []firewall.Protocol{firewall.ProtocolUDP, firewall.ProtocolTCP}
 )
 
 type serviceViaListener struct {
@@ -40,9 +44,20 @@ type serviceViaListener struct {
 	listenPort        uint16
 	listenerIsRunning bool
 	listenerFlagLock  sync.Mutex
-	ebpfService       ebpfMgr.Manager
 	firewall          Firewall
-	tcpDNATConfigured bool
+	// dnatRules holds the port 53 redirects that are installed and not yet
+	// removed, so a removal that fails can be retried.
+	dnatRules []dnatRule
+}
+
+// dnatRule is a port 53 redirect as it was installed. The target is kept with
+// the rule because the listener can come back on a different address or port,
+// and a retried removal has to name the address and port the rule was added
+// with, not the ones in use now.
+type dnatRule struct {
+	protocol firewall.Protocol
+	ip       netip.Addr
+	port     uint16
 }
 
 func newServiceViaListener(wgIface WGIface, customAddr *netip.AddrPort, fw Firewall) *serviceViaListener {
@@ -112,34 +127,93 @@ func (s *serviceViaListener) Listen() error {
 		}
 	}()
 
-	// When eBPF redirects UDP port 53 to our listen port, TCP still needs
+	if s.listenPort != DefaultPort {
-	// a DNAT rule because eBPF only handles UDP.
+		s.setupDNAT()
-	if s.ebpfService != nil && s.firewall != nil && s.listenPort != DefaultPort {
-		if err := s.firewall.AddOutputDNAT(s.listenIP, firewall.ProtocolTCP, DefaultPort, s.listenPort); err != nil {
-			log.Warnf("failed to add DNS TCP DNAT rule, TCP DNS on port 53 will not work: %v", err)
-		} else {
-			s.tcpDNATConfigured = true
-			log.Infof("added DNS TCP DNAT rule: %s:%d -> %s:%d", s.listenIP, DefaultPort, s.listenIP, s.listenPort)
-		}
 	}
 
 	return nil
 }
 
+// setupDNAT redirects port 53 to the port the DNS server actually listens on.
+// Both protocols must be redirected or none: RuntimePort reports port 53 only
+// while the full redirect is in place, so a half-configured redirect would
+// advertise a resolver that answers over one protocol.
+func (s *serviceViaListener) setupDNAT() {
+	if s.firewall == nil {
+		log.Errorf("no firewall manager available to redirect DNS port %d to %d, "+
+			"clients pointed at %s will not reach the resolver", DefaultPort, s.listenPort, s.listenIP)
+		return
+	}
+
+	// Clear whatever an earlier removal left behind first. Those rules can point
+	// at an address or port this listener no longer uses, and they are matched
+	// before anything added now, so adding a redirect on top of one would keep
+	// sending port 53 traffic to the previous listener while reporting the
+	// redirect as complete. The rules stay recorded for a later attempt.
+	if err := s.removeDNAT(); err != nil {
+		log.Errorf("failed to remove stale DNS DNAT rules, leaving port %d redirected to the previous listener: %v",
+			DefaultPort, err)
+		return
+	}
+
+	for _, proto := range dnatProtocols {
+		if err := s.firewall.AddOutputDNAT(s.listenIP, proto, DefaultPort, s.listenPort); err != nil {
+			log.Errorf("failed to add DNS %s DNAT rule, DNS on port %d will not work: %v",
+				proto, DefaultPort, err)
+			if err := s.removeDNAT(); err != nil {
+				log.Warnf("failed to roll back DNS DNAT rules, retrying on stop: %v", err)
+			}
+			return
+		}
+		s.dnatRules = append(s.dnatRules, dnatRule{protocol: proto, ip: s.listenIP, port: s.listenPort})
+	}
+
+	log.Infof("added DNS DNAT rules: %s:%d -> %s:%d (UDP + TCP)", s.listenIP, DefaultPort, s.listenIP, s.listenPort)
+}
+
+// removeDNAT removes every installed port 53 redirect. A rule whose removal
+// fails stays recorded so a later setup or Stop retries it, rather than leaving
+// port 53 pointing at a resolver that is no longer listening.
+func (s *serviceViaListener) removeDNAT() error {
+	if s.firewall == nil {
+		return nil
+	}
+
+	var merr *multierror.Error
+	var remaining []dnatRule
+	for _, rule := range s.dnatRules {
+		if err := s.firewall.RemoveOutputDNAT(rule.ip, rule.protocol, DefaultPort, rule.port); err != nil {
+			merr = multierror.Append(merr, fmt.Errorf("remove DNS %s DNAT rule for %s:%d: %w",
+				rule.protocol, rule.ip, rule.port, err))
+			remaining = append(remaining, rule)
+		}
+	}
+	s.dnatRules = remaining
+
+	return nberrors.FormatErrorOrNil(merr)
+}
+
 func (s *serviceViaListener) Stop() error {
 	s.listenerFlagLock.Lock()
 	defer s.listenerFlagLock.Unlock()
 
+	var merr *multierror.Error
+
+	// Redirects are removed even when the listener is already stopped, so that
+	// a removal which failed earlier is retried instead of leaving port 53
+	// pointing at a resolver that no longer listens.
+	if err := s.removeDNAT(); err != nil {
+		merr = multierror.Append(merr, err)
+	}
+
 	if !s.listenerIsRunning {
-		return nil
+		return nberrors.FormatErrorOrNil(merr)
 	}
 	s.listenerIsRunning = false
 
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 
-	var merr *multierror.Error
-
 	if err := s.server.ShutdownContext(ctx); err != nil {
 		merr = multierror.Append(merr, fmt.Errorf("stop DNS UDP server: %w", err))
 	}
@@ -148,19 +222,6 @@ func (s *serviceViaListener) Stop() error {
 		merr = multierror.Append(merr, fmt.Errorf("stop DNS TCP server: %w", err))
 	}
 
-	if s.tcpDNATConfigured && s.firewall != nil {
-		if err := s.firewall.RemoveOutputDNAT(s.listenIP, firewall.ProtocolTCP, DefaultPort, s.listenPort); err != nil {
-			merr = multierror.Append(merr, fmt.Errorf("remove DNS TCP DNAT rule: %w", err))
-		}
-		s.tcpDNATConfigured = false
-	}
-
-	if s.ebpfService != nil {
-		if err := s.ebpfService.FreeDNSFwd(); err != nil {
-			merr = multierror.Append(merr, fmt.Errorf("stop traffic forwarder: %w", err))
-		}
-	}
-
 	return nberrors.FormatErrorOrNil(merr)
 }
 
@@ -177,11 +238,23 @@ func (s *serviceViaListener) RuntimePort() int {
 	s.listenerFlagLock.Lock()
 	defer s.listenerFlagLock.Unlock()
 
-	if s.ebpfService != nil {
+	if s.redirectInstalled() {
 		return DefaultPort
-	} else {
-		return int(s.listenPort)
 	}
+	return int(s.listenPort)
+}
+
+// redirectInstalled reports whether every protocol is redirected from port 53
+// to the address and port the listener currently serves. Rules left over from
+// an earlier listener do not count.
+func (s *serviceViaListener) redirectInstalled() bool {
+	for _, proto := range dnatProtocols {
+		current := dnatRule{protocol: proto, ip: s.listenIP, port: s.listenPort}
+		if !slices.Contains(s.dnatRules, current) {
+			return false
+		}
+	}
+	return true
 }
 
 func (s *serviceViaListener) RuntimeIP() netip.Addr {
@@ -190,30 +263,29 @@ func (s *serviceViaListener) RuntimeIP() netip.Addr {
 
 // evalListenAddress figures out the listen address for the DNS server.
 // IPv4-only: all peers have a v4 overlay address, and DNS config points to v4.
-// First checks port 53 on WG interface or lo, then tries eBPF on a random port,
+// Prefers port 53 on the overlay interface or lo, so no redirect is needed at
-// then falls back to port 5053.
+// all; when it is taken it falls back to port 5053 and then to a random free
+// port, both of which need the port 53 redirect set up by setupDNAT.
 func (s *serviceViaListener) evalListenAddress() (netip.Addr, uint16, error) {
 	if s.customAddr != nil {
 		return s.customAddr.Addr(), s.customAddr.Port(), nil
 	}
 
-	ip, ok := s.testFreePort(DefaultPort)
+	if ip, ok := s.testFreePort(DefaultPort); ok {
-	if ok {
 		return ip, DefaultPort, nil
 	}
 
-	ebpfSrv, port, ok := s.tryToUseeBPF()
+	if ip, ok := s.testFreePort(customPort); ok {
-	if ok {
-		s.ebpfService = ebpfSrv
-		return s.wgInterface.Address().IP, port, nil
-	}
-
-	ip, ok = s.testFreePort(customPort)
-	if ok {
 		return ip, customPort, nil
 	}
 
-	return netip.Addr{}, 0, fmt.Errorf("failed to find a free port for DNS server")
+	ip := s.wgInterface.Address().IP
+	port, err := s.randomFreePort(ip)
+	if err != nil {
+		return netip.Addr{}, 0, fmt.Errorf("find a free port for DNS server: %w", err)
+	}
+
+	return ip, port, nil
 }
 
 func (s *serviceViaListener) testFreePort(port int) (netip.Addr, bool) {
@@ -260,48 +332,25 @@ func (s *serviceViaListener) tryToBind(ip netip.Addr, port int) bool {
 	return true
 }
 
-// tryToUseeBPF decides whether to apply eBPF program to capture DNS traffic on port 53.
+// randomFreePort returns a port that is free on ip for both UDP and TCP, since
-// This is needed because on some operating systems if we start a DNS server not on a default port 53,
+// the DNS server binds both. The probe listeners are closed again, so the port
-// the domain name  resolution won't work. So, in case we are running on Linux and picked a free
+// is only likely, not guaranteed, to still be free when the server binds it.
-// port we should fall back to the eBPF solution that will capture traffic on port 53 and forward
+func (s *serviceViaListener) randomFreePort(ip netip.Addr) (uint16, error) {
-// it to a local DNS server running on the chosen port.
+	for range randomPortAttempts {
-func (s *serviceViaListener) tryToUseeBPF() (ebpfMgr.Manager, uint16, bool) {
+		probeListener, err := net.ListenUDP("udp4", &net.UDPAddr{})
-	if runtime.GOOS != "linux" {
+		if err != nil {
-		return nil, 0, false
+			return 0, fmt.Errorf("bind random port: %w", err)
+		}
+
+		port := uint16(probeListener.LocalAddr().(*net.UDPAddr).Port)
+		if err := probeListener.Close(); err != nil {
+			return 0, fmt.Errorf("free up probed port: %w", err)
+		}
+
+		if s.tryToBind(ip, int(port)) {
+			return port, nil
+		}
 	}
 
-	port, err := s.generateFreePort() //nolint:staticcheck,unused
+	return 0, fmt.Errorf("no port free for UDP and TCP on %s after %d attempts", ip, randomPortAttempts)
-	if err != nil {
-		log.Warnf("failed to generate a free port for eBPF DNS forwarder server: %s", err)
-		return nil, 0, false
-	}
-
-	ebpfSrv := ebpf.GetEbpfManagerInstance()
-	err = ebpfSrv.LoadDNSFwd(s.wgInterface.Address().IP, int(port))
-	if err != nil {
-		log.Warnf("failed to load DNS forwarder eBPF program, error: %s", err)
-		return nil, 0, false
-	}
-
-	return ebpfSrv, port, true
-}
-
-func (s *serviceViaListener) generateFreePort() (uint16, error) {
-	ok := s.tryToBind(s.wgInterface.Address().IP, customPort)
-	if ok {
-		return customPort, nil
-	}
-
-	probeListener, err := net.ListenUDP("udp4", &net.UDPAddr{})
-	if err != nil {
-		log.Debugf("failed to bind random port for DNS: %s", err)
-		return 0, err
-	}
-
-	port := uint16(probeListener.LocalAddr().(*net.UDPAddr).Port)
-	if err = probeListener.Close(); err != nil {
-		log.Debugf("failed to free up DNS port: %s", err)
-		return 0, err
-	}
-	return port, nil
 }

client/internal/dns/service_listener_test.go

@@ -1,6 +1,7 @@
 package dns
 
 import (
+	"errors"
 	"fmt"
 	"net"
 	"net/netip"
@@ -10,6 +11,8 @@ import (
 	"github.com/miekg/dns"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+
+	firewall "github.com/netbirdio/netbird/client/firewall/manager"
 )
 
 func TestServiceViaListener_TCPAndUDP(t *testing.T) {
@@ -84,3 +87,133 @@ func TestServiceViaListener_TCPAndUDP(t *testing.T) {
 	require.NotEmpty(t, tcpResp.Answer)
 	assert.Contains(t, tcpResp.Answer[0].String(), "192.0.2.1", "TCP response should contain expected IP")
 }
+
+type dnatCall struct {
+	rule  dnatRule
+	added bool
+}
+
+// fakeFirewall records DNAT calls and fails the ones named in addErrs/removeErrs.
+type fakeFirewall struct {
+	calls      []dnatCall
+	addErrs    map[firewall.Protocol]error
+	removeErrs map[firewall.Protocol]error
+}
+
+func (f *fakeFirewall) AddOutputDNAT(ip netip.Addr, protocol firewall.Protocol, _, translatedPort uint16) error {
+	if err := f.addErrs[protocol]; err != nil {
+		return err
+	}
+	f.calls = append(f.calls, dnatCall{rule: dnatRule{protocol: protocol, ip: ip, port: translatedPort}, added: true})
+	return nil
+}
+
+func (f *fakeFirewall) RemoveOutputDNAT(ip netip.Addr, protocol firewall.Protocol, _, translatedPort uint16) error {
+	if err := f.removeErrs[protocol]; err != nil {
+		return err
+	}
+	f.calls = append(f.calls, dnatCall{rule: dnatRule{protocol: protocol, ip: ip, port: translatedPort}})
+	return nil
+}
+
+func newDNATTestService(fw Firewall) *serviceViaListener {
+	return &serviceViaListener{
+		listenIP:   netip.MustParseAddr("100.64.0.1"),
+		listenPort: customPort,
+		firewall:   fw,
+	}
+}
+
+func TestSetupDNAT_BothProtocols(t *testing.T) {
+	svc := newDNATTestService(&fakeFirewall{})
+
+	svc.setupDNAT()
+
+	assert.Len(t, svc.dnatRules, len(dnatProtocols))
+	assert.Equal(t, DefaultPort, svc.RuntimePort(), "port 53 is advertised once both redirects are installed")
+}
+
+func TestSetupDNAT_RollsBackPartialRedirect(t *testing.T) {
+	fw := &fakeFirewall{addErrs: map[firewall.Protocol]error{firewall.ProtocolTCP: errors.New("nftables busy")}}
+	svc := newDNATTestService(fw)
+
+	svc.setupDNAT()
+
+	assert.Empty(t, svc.dnatRules, "the UDP redirect installed before the failure must be rolled back")
+	assert.Equal(t, int(svc.listenPort), svc.RuntimePort(), "an incomplete redirect must not advertise port 53")
+	udp := dnatRule{protocol: firewall.ProtocolUDP, ip: svc.listenIP, port: svc.listenPort}
+	assert.Contains(t, fw.calls, dnatCall{rule: udp}, "UDP removal should have been attempted")
+}
+
+// A rollback that fails must keep the rule recorded, so port 53 is not left
+// redirected to a resolver that no longer listens.
+func TestStop_RetriesFailedDNATRemoval(t *testing.T) {
+	fw := &fakeFirewall{
+		addErrs:    map[firewall.Protocol]error{firewall.ProtocolTCP: errors.New("nftables busy")},
+		removeErrs: map[firewall.Protocol]error{firewall.ProtocolUDP: errors.New("nftables busy")},
+	}
+	svc := newDNATTestService(fw)
+
+	svc.setupDNAT()
+	udp := dnatRule{protocol: firewall.ProtocolUDP, ip: svc.listenIP, port: svc.listenPort}
+	require.Equal(t, []dnatRule{udp}, svc.dnatRules, "a failed rollback keeps the rule for a later retry")
+
+	require.Error(t, svc.Stop(), "the failing removal should be reported")
+	require.Equal(t, []dnatRule{udp}, svc.dnatRules)
+
+	delete(fw.removeErrs, firewall.ProtocolUDP)
+	require.NoError(t, svc.Stop(), "a later stop retries the removal")
+	assert.Empty(t, svc.dnatRules)
+}
+
+// A stale rule that cannot be removed is matched before anything added now, so
+// no new redirect may be installed on top of it and port 53 must not be
+// advertised as reaching this listener.
+func TestSetupDNAT_AbortsWhileStaleRuleRemains(t *testing.T) {
+	fw := &fakeFirewall{removeErrs: map[firewall.Protocol]error{firewall.ProtocolUDP: errors.New("nftables busy")}}
+	svc := newDNATTestService(fw)
+	stalePort := svc.listenPort
+
+	svc.setupDNAT()
+	require.Error(t, svc.Stop())
+	staleUDP := dnatRule{protocol: firewall.ProtocolUDP, ip: svc.listenIP, port: stalePort}
+	require.Equal(t, []dnatRule{staleUDP}, svc.dnatRules)
+
+	svc.listenPort = stalePort + 1
+	fw.calls = nil
+
+	svc.setupDNAT()
+
+	assert.Equal(t, []dnatRule{staleUDP}, svc.dnatRules, "the stale rule stays recorded for a later attempt")
+	for _, call := range fw.calls {
+		assert.False(t, call.added, "no redirect may be installed while a stale one is still in place")
+	}
+	assert.Equal(t, int(svc.listenPort), svc.RuntimePort(), "port 53 must not be advertised")
+}
+
+// A rule left behind by a failed removal must be removed with the address and
+// port it was installed with, even when the listener has since moved to another
+// port, and it must not count towards the redirect the new listener advertises.
+func TestSetupDNAT_ClearsStaleRuleAfterPortChange(t *testing.T) {
+	fw := &fakeFirewall{removeErrs: map[firewall.Protocol]error{firewall.ProtocolUDP: errors.New("nftables busy")}}
+	svc := newDNATTestService(fw)
+	stalePort := svc.listenPort
+
+	svc.setupDNAT()
+	require.Error(t, svc.Stop())
+	staleUDP := dnatRule{protocol: firewall.ProtocolUDP, ip: svc.listenIP, port: stalePort}
+	require.Equal(t, []dnatRule{staleUDP}, svc.dnatRules)
+
+	delete(fw.removeErrs, firewall.ProtocolUDP)
+	svc.listenPort = stalePort + 1
+	fw.calls = nil
+
+	svc.setupDNAT()
+
+	assert.Contains(t, fw.calls, dnatCall{rule: staleUDP}, "the stale rule must be removed with its original port")
+	assert.Len(t, svc.dnatRules, len(dnatProtocols))
+	assert.Equal(t, DefaultPort, svc.RuntimePort(), "the new listener is fully redirected")
+	for _, rule := range svc.dnatRules {
+		assert.Equal(t, svc.listenPort, rule.port, "only rules for the current listener remain")
+	}
+}

client/internal/ebpf/ebpf/bpf_bpfeb.go

@@ -1,5 +1,5 @@
 // Code generated by bpf2go; DO NOT EDIT.
-//go:build arm64be || armbe || mips || mips64 || mips64p32 || ppc64 || s390 || s390x || sparc || sparc64
+//go:build mips || mips64 || ppc64 || s390x
 
 package ebpf
 
@@ -47,9 +47,10 @@ func loadBpfObjects(obj interface{}, opts *ebpf.CollectionOptions) error {
 type bpfSpecs struct {
 	bpfProgramSpecs
 	bpfMapSpecs
+	bpfVariableSpecs
 }
 
-// bpfSpecs contains programs before they are loaded into the kernel.
+// bpfProgramSpecs contains programs before they are loaded into the kernel.
 //
 // It can be passed ebpf.CollectionSpec.Assign.
 type bpfProgramSpecs struct {
@@ -61,17 +62,28 @@ type bpfProgramSpecs struct {
 // It can be passed ebpf.CollectionSpec.Assign.
 type bpfMapSpecs struct {
 	NbFeatures           *ebpf.MapSpec `ebpf:"nb_features"`
-	NbMapDnsIp           *ebpf.MapSpec `ebpf:"nb_map_dns_ip"`
-	NbMapDnsPort         *ebpf.MapSpec `ebpf:"nb_map_dns_port"`
 	NbWgProxySettingsMap *ebpf.MapSpec `ebpf:"nb_wg_proxy_settings_map"`
 }
 
+// bpfVariableSpecs contains global variables before they are loaded into the kernel.
+//
+// It can be passed ebpf.CollectionSpec.Assign.
+type bpfVariableSpecs struct {
+	FlagFeatureWgProxy *ebpf.VariableSpec `ebpf:"flag_feature_wg_proxy"`
+	MapKeyFeatures     *ebpf.VariableSpec `ebpf:"map_key_features"`
+	MapKeyProxyPort    *ebpf.VariableSpec `ebpf:"map_key_proxy_port"`
+	MapKeyWgPort       *ebpf.VariableSpec `ebpf:"map_key_wg_port"`
+	ProxyPort          *ebpf.VariableSpec `ebpf:"proxy_port"`
+	WgPort             *ebpf.VariableSpec `ebpf:"wg_port"`
+}
+
 // bpfObjects contains all objects after they have been loaded into the kernel.
 //
 // It can be passed to loadBpfObjects or ebpf.CollectionSpec.LoadAndAssign.
 type bpfObjects struct {
 	bpfPrograms
 	bpfMaps
+	bpfVariables
 }
 
 func (o *bpfObjects) Close() error {
@@ -86,20 +98,28 @@ func (o *bpfObjects) Close() error {
 // It can be passed to loadBpfObjects or ebpf.CollectionSpec.LoadAndAssign.
 type bpfMaps struct {
 	NbFeatures           *ebpf.Map `ebpf:"nb_features"`
-	NbMapDnsIp           *ebpf.Map `ebpf:"nb_map_dns_ip"`
-	NbMapDnsPort         *ebpf.Map `ebpf:"nb_map_dns_port"`
 	NbWgProxySettingsMap *ebpf.Map `ebpf:"nb_wg_proxy_settings_map"`
 }
 
 func (m *bpfMaps) Close() error {
 	return _BpfClose(
 		m.NbFeatures,
-		m.NbMapDnsIp,
-		m.NbMapDnsPort,
 		m.NbWgProxySettingsMap,
 	)
 }
 
+// bpfVariables contains all global variables after they have been loaded into the kernel.
+//
+// It can be passed to loadBpfObjects or ebpf.CollectionSpec.LoadAndAssign.
+type bpfVariables struct {
+	FlagFeatureWgProxy *ebpf.Variable `ebpf:"flag_feature_wg_proxy"`
+	MapKeyFeatures     *ebpf.Variable `ebpf:"map_key_features"`
+	MapKeyProxyPort    *ebpf.Variable `ebpf:"map_key_proxy_port"`
+	MapKeyWgPort       *ebpf.Variable `ebpf:"map_key_wg_port"`
+	ProxyPort          *ebpf.Variable `ebpf:"proxy_port"`
+	WgPort             *ebpf.Variable `ebpf:"wg_port"`
+}
+
 // bpfPrograms contains all programs after they have been loaded into the kernel.
 //
 // It can be passed to loadBpfObjects or ebpf.CollectionSpec.LoadAndAssign.

client/internal/ebpf/ebpf/bpf_bpfel.go

@@ -1,5 +1,5 @@
 // Code generated by bpf2go; DO NOT EDIT.
-//go:build 386 || amd64 || amd64p32 || arm || arm64 || loong64 || mips64le || mips64p32le || mipsle || ppc64le || riscv64
+//go:build 386 || amd64 || arm || arm64 || loong64 || mips64le || mipsle || ppc64le || riscv64 || wasm
 
 package ebpf
 
@@ -47,9 +47,10 @@ func loadBpfObjects(obj interface{}, opts *ebpf.CollectionOptions) error {
 type bpfSpecs struct {
 	bpfProgramSpecs
 	bpfMapSpecs
+	bpfVariableSpecs
 }
 
-// bpfSpecs contains programs before they are loaded into the kernel.
+// bpfProgramSpecs contains programs before they are loaded into the kernel.
 //
 // It can be passed ebpf.CollectionSpec.Assign.
 type bpfProgramSpecs struct {
@@ -61,17 +62,28 @@ type bpfProgramSpecs struct {
 // It can be passed ebpf.CollectionSpec.Assign.
 type bpfMapSpecs struct {
 	NbFeatures           *ebpf.MapSpec `ebpf:"nb_features"`
-	NbMapDnsIp           *ebpf.MapSpec `ebpf:"nb_map_dns_ip"`
-	NbMapDnsPort         *ebpf.MapSpec `ebpf:"nb_map_dns_port"`
 	NbWgProxySettingsMap *ebpf.MapSpec `ebpf:"nb_wg_proxy_settings_map"`
 }
 
+// bpfVariableSpecs contains global variables before they are loaded into the kernel.
+//
+// It can be passed ebpf.CollectionSpec.Assign.
+type bpfVariableSpecs struct {
+	FlagFeatureWgProxy *ebpf.VariableSpec `ebpf:"flag_feature_wg_proxy"`
+	MapKeyFeatures     *ebpf.VariableSpec `ebpf:"map_key_features"`
+	MapKeyProxyPort    *ebpf.VariableSpec `ebpf:"map_key_proxy_port"`
+	MapKeyWgPort       *ebpf.VariableSpec `ebpf:"map_key_wg_port"`
+	ProxyPort          *ebpf.VariableSpec `ebpf:"proxy_port"`
+	WgPort             *ebpf.VariableSpec `ebpf:"wg_port"`
+}
+
 // bpfObjects contains all objects after they have been loaded into the kernel.
 //
 // It can be passed to loadBpfObjects or ebpf.CollectionSpec.LoadAndAssign.
 type bpfObjects struct {
 	bpfPrograms
 	bpfMaps
+	bpfVariables
 }
 
 func (o *bpfObjects) Close() error {
@@ -86,20 +98,28 @@ func (o *bpfObjects) Close() error {
 // It can be passed to loadBpfObjects or ebpf.CollectionSpec.LoadAndAssign.
 type bpfMaps struct {
 	NbFeatures           *ebpf.Map `ebpf:"nb_features"`
-	NbMapDnsIp           *ebpf.Map `ebpf:"nb_map_dns_ip"`
-	NbMapDnsPort         *ebpf.Map `ebpf:"nb_map_dns_port"`
 	NbWgProxySettingsMap *ebpf.Map `ebpf:"nb_wg_proxy_settings_map"`
 }
 
 func (m *bpfMaps) Close() error {
 	return _BpfClose(
 		m.NbFeatures,
-		m.NbMapDnsIp,
-		m.NbMapDnsPort,
 		m.NbWgProxySettingsMap,
 	)
 }
 
+// bpfVariables contains all global variables after they have been loaded into the kernel.
+//
+// It can be passed to loadBpfObjects or ebpf.CollectionSpec.LoadAndAssign.
+type bpfVariables struct {
+	FlagFeatureWgProxy *ebpf.Variable `ebpf:"flag_feature_wg_proxy"`
+	MapKeyFeatures     *ebpf.Variable `ebpf:"map_key_features"`
+	MapKeyProxyPort    *ebpf.Variable `ebpf:"map_key_proxy_port"`
+	MapKeyWgPort       *ebpf.Variable `ebpf:"map_key_wg_port"`
+	ProxyPort          *ebpf.Variable `ebpf:"proxy_port"`
+	WgPort             *ebpf.Variable `ebpf:"wg_port"`
+}
+
 // bpfPrograms contains all programs after they have been loaded into the kernel.
 //
 // It can be passed to loadBpfObjects or ebpf.CollectionSpec.LoadAndAssign.

client/internal/ebpf/ebpf/dns_fwd_linux.go

@@ -1,52 +0,0 @@
-package ebpf
-
-import (
-	"encoding/binary"
-	"fmt"
-	"net/netip"
-
-	log "github.com/sirupsen/logrus"
-)
-
-const (
-	mapKeyDNSIP   uint32 = 0
-	mapKeyDNSPort uint32 = 1
-)
-
-func (tf *GeneralManager) LoadDNSFwd(ip netip.Addr, dnsPort int) error {
-	log.Debugf("load eBPF DNS forwarder, watching addr: %s:53, redirect to port: %d", ip, dnsPort)
-	tf.lock.Lock()
-	defer tf.lock.Unlock()
-
-	err := tf.loadXdp()
-	if err != nil {
-		return err
-	}
-
-	if !ip.Is4() {
-		return fmt.Errorf("eBPF DNS forwarder only supports IPv4, got %s", ip)
-	}
-	ip4 := ip.As4()
-	err = tf.bpfObjs.NbMapDnsIp.Put(mapKeyDNSIP, binary.BigEndian.Uint32(ip4[:]))
-	if err != nil {
-		return err
-	}
-
-	err = tf.bpfObjs.NbMapDnsPort.Put(mapKeyDNSPort, uint16(dnsPort))
-	if err != nil {
-		return err
-	}
-
-	tf.setFeatureFlag(featureFlagDnsForwarder)
-	err = tf.bpfObjs.NbFeatures.Put(mapKeyFeatures, tf.featureFlags)
-	if err != nil {
-		return err
-	}
-	return nil
-}
-
-func (tf *GeneralManager) FreeDNSFwd() error {
-	log.Debugf("free ebpf DNS forwarder")
-	return tf.unsetFeatureFlag(featureFlagDnsForwarder)
-}
-

client/internal/ebpf/ebpf/manager_linux.go

@@ -15,8 +15,7 @@ import (
 const (
 	mapKeyFeatures uint32 = 0
 
-	featureFlagWGProxy      = 0b00000001
+	featureFlagWGProxy = 0b00000001
-	featureFlagDnsForwarder = 0b00000010
 )
 
 var (
@@ -28,9 +27,9 @@ var (
 
 // GeneralManager is used to load multiple eBPF programs with a custom check (if then) done in prog.c
 // The manager simply adds a feature (byte) of each program to a map that is shared between the userspace and kernel.
-// When packet arrives, the C code checks for each feature (if it is set) and executes each enabled program (e.g., dns_fwd.c and wg_proxy.c).
+// When packet arrives, the C code checks for each feature (if it is set) and executes each enabled program (e.g., wg_proxy.c).
 //
-//go:generate go run github.com/cilium/ebpf/cmd/bpf2go -cc clang-14 bpf src/prog.c -- -I /usr/x86_64-linux-gnu/include
+//go:generate go run github.com/cilium/ebpf/cmd/bpf2go -cc clang-14 bpf src/prog.c -- -I /usr/x86_64-linux-gnu/include -include src/bpf_map_def.h
 type GeneralManager struct {
 	lock         sync.Mutex
 	link         link.Link

client/internal/ebpf/ebpf/manager_linux_test.go

@@ -7,33 +7,24 @@ import (
 func TestManager_setFeatureFlag(t *testing.T) {
 	mgr := GeneralManager{}
 	mgr.setFeatureFlag(featureFlagWGProxy)
-	if mgr.featureFlags != 1 {
+	if mgr.featureFlags != featureFlagWGProxy {
 		t.Errorf("invalid feature state")
 	}
 
-	mgr.setFeatureFlag(featureFlagDnsForwarder)
+	mgr.setFeatureFlag(featureFlagWGProxy)
-	if mgr.featureFlags != 3 {
+	if mgr.featureFlags != featureFlagWGProxy {
-		t.Errorf("invalid feature state")
+		t.Errorf("setting a flag twice must be idempotent, got: %d", mgr.featureFlags)
 	}
 }
 
 func TestManager_unsetFeatureFlag(t *testing.T) {
 	mgr := GeneralManager{}
 	mgr.setFeatureFlag(featureFlagWGProxy)
-	mgr.setFeatureFlag(featureFlagDnsForwarder)
 
 	err := mgr.unsetFeatureFlag(featureFlagWGProxy)
 	if err != nil {
 		t.Errorf("unexpected error: %s", err)
 	}
-	if mgr.featureFlags != 2 {
-		t.Errorf("invalid feature state, expected: %d, got: %d", 2, mgr.featureFlags)
-	}
-
-	err = mgr.unsetFeatureFlag(featureFlagDnsForwarder)
-	if err != nil {
-		t.Errorf("unexpected error: %s", err)
-	}
 	if mgr.featureFlags != 0 {
 		t.Errorf("invalid feature state, expected: %d, got: %d", 0, mgr.featureFlags)
 	}

client/internal/ebpf/ebpf/src/bpf_map_def.h

@@ -0,0 +1,16 @@
+// libbpf 1.0 removed struct bpf_map_def, but the programs here keep the legacy
+// map definitions: they load on kernels built without BTF, which BTF-style
+// (SEC(".maps")) definitions do not. Define the struct ourselves so the
+// programs compile against current libbpf headers.
+#ifndef NB_BPF_MAP_DEF_H
+#define NB_BPF_MAP_DEF_H
+
+struct bpf_map_def {
+	unsigned int type;
+	unsigned int key_size;
+	unsigned int value_size;
+	unsigned int max_entries;
+	unsigned int map_flags;
+};
+
+#endif

client/internal/ebpf/ebpf/src/dns_fwd.c

@@ -1,67 +0,0 @@
-const __u32 map_key_dns_ip = 0;
-const __u32 map_key_dns_port = 1;
-
-struct bpf_map_def SEC("maps") nb_map_dns_ip = {
-	.type = BPF_MAP_TYPE_ARRAY,
-	.key_size = sizeof(__u32),
-	.value_size = sizeof(__u32),
-	.max_entries = 10,
-};
-
-struct bpf_map_def SEC("maps") nb_map_dns_port = {
-	.type = BPF_MAP_TYPE_ARRAY,
-	.key_size = sizeof(__u32),
-	.value_size = sizeof(__u16),
-	.max_entries = 10,
-};
-
-__be32 dns_ip = 0;
-__be16 dns_port = 0;
-
-// 13568 is 53 in big endian
-__be16 GENERAL_DNS_PORT = 13568;
-
-bool read_settings() {
-    __u16 *port_value;
-    __u32 *ip_value;
-
-    // read dns ip
-    ip_value = bpf_map_lookup_elem(&nb_map_dns_ip, &map_key_dns_ip);
-    if(!ip_value) {
-        return false;
-    }
-    dns_ip = htonl(*ip_value);
-
-    // read dns port
-    port_value = bpf_map_lookup_elem(&nb_map_dns_port, &map_key_dns_port);
-    if (!port_value) {
-        return false;
-    }
-    dns_port = htons(*port_value);
-    return true;
-}
-
-int xdp_dns_fwd(struct iphdr  *ip, struct udphdr *udp) {
-    if (dns_port == 0) {
-        if(!read_settings()){
-            return XDP_PASS;
-        }
-        // bpf_printk("dns port: %d", ntohs(dns_port));
-        // bpf_printk("dns ip: %d", ntohl(dns_ip));
-    }
-
-    if (udp->dest == GENERAL_DNS_PORT && ip->daddr == dns_ip) {
-        udp->dest = dns_port;
-        // Clear the now-stale checksum; zero means "not computed" for IPv4.
-        udp->check = 0;
-        return XDP_PASS;
-    }
-
-    if (udp->source == dns_port && ip->saddr == dns_ip) {
-        udp->source = GENERAL_DNS_PORT;
-        udp->check = 0;
-        return XDP_PASS;
-    }
-
-    return XDP_PASS;
-}

client/internal/ebpf/ebpf/src/prog.c

@@ -5,11 +5,9 @@
 #include <netinet/in.h>
 #include <linux/bpf.h>
 #include <bpf/bpf_helpers.h>
-#include "dns_fwd.c"
 #include "wg_proxy.c"
 
 const __u16 flag_feature_wg_proxy = 0b01;
-const __u16 flag_feature_dns_fwd = 0b10;
 
 const __u32 map_key_features = 0;
 struct bpf_map_def SEC("maps") nb_features = {
@@ -48,10 +46,6 @@ int nb_xdp_prog(struct xdp_md *ctx) {
        return XDP_PASS;
     }
 
-    if (*features & flag_feature_dns_fwd) {
-        xdp_dns_fwd(ip, udp);
-    }
-
     if (*features & flag_feature_wg_proxy) {
         xdp_wg_proxy(ip, udp);
     }

client/internal/ebpf/ebpf/src/readme.md

@@ -1,8 +1,18 @@
-# DNS forwarder
+# XDP programs
 
-The agent attach the XDP program to the lo device. We can not use fake address in eBPF because the
+`prog.c` is attached to the `lo` device and dispatches to the features enabled in the
-traffic does not appear in the eBPF program. The program capture the traffic on wg_ip:53 and 
+`nb_features` map. The only feature is the WireGuard proxy (`wg_proxy.c`): it rewrites
-overwrite in it the destination port to 5053.
+loopback UDP sent from the WireGuard listen port so it reaches the userspace relay proxy
+port instead, and swaps the peer endpoint port into the source so the proxy can tell
+peers apart.
+
+Maps use the legacy `struct bpf_map_def` form, defined in `bpf_map_def.h` because libbpf
+1.0 removed it. They load on kernels built without BTF, which BTF-style (`SEC(".maps")`)
+definitions do not.
+
+Regenerate the objects with `go generate ./client/internal/ebpf/ebpf/`; it needs
+`clang-14`. Loading a regenerated object needs root, attaching it needs `bpf_link`
+(kernel >= 5.7), and only one XDP program can own `lo` at a time.
 
 # Debug
 

client/internal/ebpf/manager/manager.go

@@ -1,11 +1,7 @@
 package manager
 
-import "net/netip"
+// Manager is used to load multiple eBPF programs. E.g., the WireGuard proxy
-
-// Manager is used to load multiple eBPF programs. E.g., current DNS programs and WireGuard proxy
 type Manager interface {
-	LoadDNSFwd(ip netip.Addr, dnsPort int) error
-	FreeDNSFwd() error
 	LoadWgProxy(proxyPort, wgPort int) error
 	FreeWGProxy() error
 }

client/internal/profilemanager/invoking_user.go

@@ -6,6 +6,7 @@ import (
 	"os/user"
 	"path/filepath"
 	"runtime"
+	"strconv"
 
 	log "github.com/sirupsen/logrus"
 )
@@ -13,17 +14,21 @@ import (
 const envSudoUser = "SUDO_USER"
 
 var (
-	geteuid    = os.Geteuid
+	currentUser = user.Current
-	lookupUser = user.Lookup
+	getegid     = os.Getegid
+	geteuid     = os.Geteuid
+	lookupUser  = user.Lookup
 )
 
 // InvokingUser returns the user a CLI invocation acts for. Under sudo that is
 // the user who ran sudo, not root: privileged flags force commands through
 // sudo, and resolving profiles as root would silently switch the daemon to
-// root's (default) profile instead of the invoking user's. Privilege decisions
+// root's (default) profile instead of the invoking user's. An unmapped positive
-// are not made here — those stay on the kernel credentials of the daemon
+// process UID uses its numeric kernel identity; root, sudo lookup failures, and
-// connection, which SUDO_USER (a plain environment variable) can never
+// unavailable platform identities still fail closed. Privilege decisions stay
-// influence; a forged value only selects a profile root could select anyway.
+// on the kernel credentials of the daemon connection, which SUDO_USER (a plain
+// environment variable) can never influence; a forged value only selects a
+// profile root could select anyway.
 func InvokingUser() (*user.User, error) {
 	if u, ok := sudoInvokingUser(); ok {
 		return u, nil
@@ -35,7 +40,23 @@ func InvokingUser() (*user.User, error) {
 	if sudoActive() {
 		return nil, fmt.Errorf("resolve sudo invoking user %q: refusing to fall back to root", os.Getenv(envSudoUser))
 	}
-	return user.Current()
+	u, err := currentUser()
+	if err == nil {
+		return u, nil
+	}
+
+	uid := geteuid()
+	if uid <= 0 {
+		return nil, err
+	}
+
+	log.Debugf("current user lookup for UID %d: %v; using numeric UID", uid, err)
+	uidString := strconv.Itoa(uid)
+	return &user.User{
+		Username: uidString,
+		Uid:      uidString,
+		Gid:      strconv.Itoa(getegid()),
+	}, nil
 }
 
 // IsPlainRoot reports that the process runs as root with no usable sudo

client/internal/profilemanager/invoking_user_test.go

@@ -2,6 +2,7 @@ package profilemanager
 
 import (
 	"errors"
+	"fmt"
 	"io/fs"
 	"os"
 	"os/user"
@@ -21,7 +22,51 @@ func TestInvokingUserFallsBackToProcessUser(t *testing.T) {
 
 	current, err := user.Current()
 	require.NoError(t, err)
-	assert.Equal(t, current.Username, got.Username)
+	assert.Equal(t, current.Username, got.Username, "invoking user should match the process user without sudo")
+}
+
+func TestInvokingUserFailsClosedWithoutPositiveUID(t *testing.T) {
+	for _, uid := range []int{0, -1} {
+		t.Run(fmt.Sprintf("UID%d", uid), func(t *testing.T) {
+			t.Setenv(envSudoUser, "")
+			lookupErr := errors.New("current user unavailable")
+			fakeUnmappedUser(t, uid, 0, lookupErr)
+
+			got, err := InvokingUser()
+			require.ErrorIs(t, err, lookupErr)
+			assert.Nil(t, got, "root or unavailable UID must not become a synthetic identity")
+		})
+	}
+}
+
+func TestProfileFilePathUsesNumericIdentityForUnmappedNonRoot(t *testing.T) {
+	t.Setenv(envSudoUser, "")
+	fakeUnmappedUser(t, 1001230000, 0, errors.New("user: unknown userid 1001230000"))
+
+	profilesRoot := t.TempDir()
+	origDir := DefaultConfigPathDir
+	origOverride := ConfigDirOverride
+	DefaultConfigPathDir = profilesRoot
+	ConfigDirOverride = ""
+	t.Cleanup(func() {
+		DefaultConfigPathDir = origDir
+		ConfigDirOverride = origOverride
+	})
+
+	profileID := ID("0123456789abcdef0123456789abcdef")
+	got, err := (&Profile{ID: profileID}).FilePath()
+	require.NoError(t, err)
+	assert.Equal(t,
+		filepath.Join(profilesRoot, "1001230000", profileID.String()+".json"),
+		got,
+		"profile path should use the numeric UID namespace",
+	)
+
+	entries, err := os.ReadDir(profilesRoot)
+	require.NoError(t, err)
+	require.Len(t, entries, 1, "only the numeric UID directory should be created")
+	assert.Equal(t, "1001230000", entries[0].Name(), "profile namespace should be numeric")
+	assert.True(t, entries[0].IsDir(), "profile namespace should be a directory")
 }
 
 func TestSudoInvokingUserInactiveWithoutSudoContext(t *testing.T) {
@@ -60,6 +105,13 @@ func TestInvokingUserFailsClosedWhenSudoLookupFails(t *testing.T) {
 	fakeSudo(t, filepath.Join("/home", "misha"))
 	lookupUser = func(string) (*user.User, error) { return nil, errors.New("nss unavailable") }
 
+	origCurrentUser := currentUser
+	currentUser = func() (*user.User, error) {
+		t.Fatal("currentUser must not be called after a sudo lookup failure")
+		return nil, errors.New("currentUser called unexpectedly")
+	}
+	t.Cleanup(func() { currentUser = origCurrentUser })
+
 	got, err := InvokingUser()
 	require.Error(t, err)
 	assert.Nil(t, got, "must not resolve to the root process user")
@@ -215,6 +267,22 @@ func fakeSudo(t *testing.T, home string) {
 	})
 }
 
+func fakeUnmappedUser(t *testing.T, uid, gid int, lookupErr error) {
+	t.Helper()
+
+	origCurrentUser := currentUser
+	origEuid := geteuid
+	origEgid := getegid
+	currentUser = func() (*user.User, error) { return nil, lookupErr }
+	geteuid = func() int { return uid }
+	getegid = func() int { return gid }
+	t.Cleanup(func() {
+		currentUser = origCurrentUser
+		geteuid = origEuid
+		getegid = origEgid
+	})
+}
+
 func assertNoEntries(t *testing.T, root string) {
 	t.Helper()
 	err := filepath.WalkDir(root, func(path string, _ fs.DirEntry, err error) error {

client/mdm/conflicts.go

@@ -1,6 +1,10 @@
 package mdm
 
-import "net/url"
+import (
+	"net/url"
+
+	"github.com/netbirdio/netbird/util"
+)
 
 // PreSharedKeyRedactedSentinel is the redaction mask returned in place of a
 // real pre-shared key; an incoming value equal to it is a round-trip echo,
@@ -44,8 +48,9 @@ func ConflictStringPtr(key string, p *string) ConflictCheck {
 	}
 }
 
-// ConflictURL builds a ConflictCheck for a URL-typed MDM key; both sides are
+// ConflictURL builds a ConflictCheck for a URL-typed MDM key. The two sides are
-// normalized via CanonicalURL before comparison.
+// compared as the endpoints they address, not as strings: see
+// util.SameServiceURL.
 func ConflictURL(key, got string) ConflictCheck {
 	return ConflictCheck{
 		Key: key,
@@ -54,7 +59,7 @@ func ConflictURL(key, got string) ConflictCheck {
 				return true
 			}
 			want, ok := pol.GetString(key)
-			return ok && CanonicalURL(want) == CanonicalURL(got)
+			return ok && util.SameServiceURLStrings(want, got)
 		},
 	}
 }

client/mdm/conflicts_test.go

@@ -0,0 +1,40 @@
+package mdm
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// The same spellings, through the conflict check that decides whether a request
+// is refused. An enforced URL restated in another spelling addresses the very
+// server the policy names, so it must not be reported as a conflict.
+func TestConflictURLComparesEndpoints(t *testing.T) {
+	policy := NewPolicy(map[string]any{KeyManagementURL: "https://mgmt.example.com"})
+	require.True(t, policy.HasKey(KeyManagementURL))
+
+	for _, restated := range []string{
+		"https://mgmt.example.com",
+		"https://mgmt.example.com:443",
+		"https://mgmt.example.com/",
+		"https://MGMT.example.com",
+		"https://mgmt.example.com:0443",
+	} {
+		conflicts := ResolveConflicts(policy, []ConflictCheck{ConflictURL(KeyManagementURL, restated)})
+		assert.Empty(t, conflicts, "%q is the enforced endpoint written differently", restated)
+	}
+
+	for _, diverging := range []string{
+		"https://other.example.com",
+		"http://mgmt.example.com",
+		"https://mgmt.example.com:8443",
+		"https://mgmt.example.com/other",
+	} {
+		conflicts := ResolveConflicts(policy, []ConflictCheck{ConflictURL(KeyManagementURL, diverging)})
+		assert.Equal(t, []string{KeyManagementURL}, conflicts, "%q addresses another endpoint", diverging)
+	}
+
+	// An unset field is not a request to change anything.
+	assert.Empty(t, ResolveConflicts(policy, []ConflictCheck{ConflictURL(KeyManagementURL, "")}))
+}

client/mdm/restrictions.go

@@ -20,6 +20,7 @@ type Fields struct {
 	DisableMetricsCollection bool   `json:"disableMetricsCollection"`
 	SplitTunnelMode          bool   `json:"splitTunnelMode"`
 	SplitTunnelApps          bool   `json:"splitTunnelApps"`
+	RemoteJobsAllowed        bool   `json:"allowRemoteJobs"`
 	DisableAdvancedView      *bool  `json:"disableAdvancedView"`
 }
 
@@ -60,6 +61,7 @@ func BuildRestrictions(policy *Policy) Restrictions {
 	r.MDM.DisableMetricsCollection = policy.HasKey(KeyDisableMetricsCollection)
 	r.MDM.SplitTunnelMode = policy.HasKey(KeySplitTunnelMode)
 	r.MDM.SplitTunnelApps = policy.HasKey(KeySplitTunnelApps)
+	r.MDM.RemoteJobsAllowed = policy.HasKey(KeyRemoteJobsAllowed)
 	if v, ok := policy.GetBool(KeyAllowServerSSH); ok {
 		r.MDM.AllowServerSSH = &v
 	}

management/cmd/root.go

@@ -54,6 +54,15 @@ func Execute() error {
 	return rootCmd.Execute()
 }
 
+// Customize hands the fully built root command to fn so an embedding binary
+// can extend or adjust the command tree — most commonly attaching its own
+// subcommands next to (or under) the built-in ones — before calling Execute.
+// The root command is constructed in this package's init, so Customize may be
+// called from the embedding binary's main at any point before Execute.
+func Customize(fn func(root *cobra.Command)) {
+	fn(rootCmd)
+}
+
 func init() {
 	mgmtCmd.Flags().IntVar(&mgmtPort, "port", 80, "server port to listen on (defaults to 443 if TLS is enabled, 80 otherwise")
 	mgmtCmd.Flags().BoolVar(&disableLegacyManagementPort, "disable-legacy-port", false, "disabling the old legacy port (33073)")

management/cmd/root_test.go

@@ -0,0 +1,42 @@
+package cmd
+
+import (
+	"testing"
+
+	"github.com/spf13/cobra"
+)
+
+// TestCustomize verifies an embedding binary can extend the command tree: a
+// top-level command attached through the hook, and a subcommand attached under
+// the built-in admin group, are both resolvable exactly as Execute would
+// resolve them.
+func TestCustomize(t *testing.T) {
+	topLevel := &cobra.Command{Use: "some-extra", RunE: func(*cobra.Command, []string) error { return nil }}
+	nested := &cobra.Command{Use: "cluster", RunE: func(*cobra.Command, []string) error { return nil }}
+
+	Customize(func(root *cobra.Command) {
+		root.AddCommand(topLevel)
+		for _, c := range root.Commands() {
+			if c.Name() == "admin" {
+				c.AddCommand(nested)
+				return
+			}
+		}
+		t.Fatal("admin command not found in the root tree")
+	})
+	t.Cleanup(func() {
+		rootCmd.RemoveCommand(topLevel)
+		for _, c := range rootCmd.Commands() {
+			if c.Name() == "admin" {
+				c.RemoveCommand(nested)
+			}
+		}
+	})
+
+	if found, _, err := rootCmd.Find([]string{"some-extra"}); err != nil || found != topLevel {
+		t.Fatalf("top-level command not resolvable: found=%v err=%v", found, err)
+	}
+	if found, _, err := rootCmd.Find([]string{"admin", "cluster"}); err != nil || found != nested {
+		t.Fatalf("nested admin subcommand not resolvable: found=%v err=%v", found, err)
+	}
+}

management/internals/modules/reverseproxy/domain/manager/domain_test.go

@@ -66,8 +66,8 @@ func TestExtractClusterFromFreeDomain(t *testing.T) {
 
 func TestExtractClusterFromCustomDomains(t *testing.T) {
 	customDomains := []*domain.Domain{
-		{Domain: "example.com", TargetCluster: "eu1.proxy.netbird.io"},
+		{Domain: "example.com", TargetCluster: "eu1.proxy.netbird.io", Validated: true},
-		{Domain: "proxy.corp.io", TargetCluster: "us1.proxy.netbird.io"},
+		{Domain: "proxy.corp.io", TargetCluster: "us1.proxy.netbird.io", Validated: true},
 	}
 
 	tests := []struct {
@@ -120,19 +120,49 @@ func TestExtractClusterFromCustomDomains(t *testing.T) {
 
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
-			cluster, ok := extractClusterFromCustomDomains(tc.domain, customDomains)
+			cluster, match := extractClusterFromCustomDomains(tc.domain, customDomains)
-			assert.Equal(t, tc.wantOK, ok)
+			if !tc.wantOK {
-			if ok {
+				assert.Equal(t, customDomainNoMatch, match, "unrelated domain should not match any custom domain")
-				assert.Equal(t, tc.wantVal, cluster)
+				return
 			}
+			assert.Equal(t, customDomainValidated, match, "validated custom domain should resolve a cluster")
+			assert.Equal(t, tc.wantVal, cluster)
 		})
 	}
 }
 
+// An unvalidated row must never yield a cluster: the account has not shown it
+// controls the name, so no service may be bound to it.
+func TestExtractClusterFromCustomDomains_UnvalidatedDomainRefused(t *testing.T) {
+	customDomains := []*domain.Domain{
+		{Domain: "example.com", TargetCluster: "eu1.proxy.netbird.io", Validated: false},
+	}
+
+	for _, serviceDomain := range []string{"example.com", "app.example.com"} {
+		t.Run(serviceDomain, func(t *testing.T) {
+			cluster, match := extractClusterFromCustomDomains(serviceDomain, customDomains)
+			assert.Equal(t, customDomainUnvalidated, match, "unvalidated row must be reported as such")
+			assert.Empty(t, cluster, "unvalidated row must not resolve a cluster")
+		})
+	}
+}
+
+// A more specific unvalidated row must not shadow a validated parent domain.
+func TestExtractClusterFromCustomDomains_ValidatedParentWinsOverUnvalidatedChild(t *testing.T) {
+	customDomains := []*domain.Domain{
+		{Domain: "example.com", TargetCluster: "cluster-generic", Validated: true},
+		{Domain: "app.example.com", TargetCluster: "cluster-app", Validated: false},
+	}
+
+	cluster, match := extractClusterFromCustomDomains("app.example.com", customDomains)
+	assert.Equal(t, customDomainValidated, match)
+	assert.Equal(t, "cluster-generic", cluster, "validated parent domain should provide the cluster")
+}
+
 func TestExtractClusterFromCustomDomains_OverlappingDomains(t *testing.T) {
 	customDomains := []*domain.Domain{
-		{Domain: "example.com", TargetCluster: "cluster-generic"},
+		{Domain: "example.com", TargetCluster: "cluster-generic", Validated: true},
-		{Domain: "app.example.com", TargetCluster: "cluster-app"},
+		{Domain: "app.example.com", TargetCluster: "cluster-app", Validated: true},
 	}
 
 	tests := []struct {
@@ -164,8 +194,8 @@ func TestExtractClusterFromCustomDomains_OverlappingDomains(t *testing.T) {
 
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
-			cluster, ok := extractClusterFromCustomDomains(tc.domain, customDomains)
+			cluster, match := extractClusterFromCustomDomains(tc.domain, customDomains)
-			assert.True(t, ok)
+			assert.Equal(t, customDomainValidated, match)
 			assert.Equal(t, tc.wantVal, cluster)
 		})
 	}

management/internals/modules/reverseproxy/domain/manager/manager.go

@@ -26,6 +26,7 @@ type store interface {
 	GetAgentNetworkSettings(ctx context.Context, lockStrength nbstore.LockingStrength, accountID string) (*agentnetworkTypes.Settings, error)
 
 	GetCustomDomain(ctx context.Context, accountID string, domainID string) (*domain.Domain, error)
+	GetCustomDomainByName(ctx context.Context, domainName string) (*domain.Domain, error)
 	ListFreeDomains(ctx context.Context, accountID string) ([]string, error)
 	ListCustomDomains(ctx context.Context, accountID string) ([]*domain.Domain, error)
 	CreateCustomDomain(ctx context.Context, accountID string, domainName string, targetCluster string, validated bool) (*domain.Domain, error)
@@ -150,6 +151,10 @@ func (m Manager) CreateDomain(ctx context.Context, accountID, userID, domainName
 		return nil, fmt.Errorf("target cluster %s is not available", targetCluster)
 	}
 
+	if err := m.checkDomainAvailable(ctx, domainName); err != nil {
+		return nil, err
+	}
+
 	// Attempt an initial validation against the specified cluster only
 	var validated bool
 	if m.validator.IsValid(ctx, domainName, []string{targetCluster}) {
@@ -166,6 +171,23 @@ func (m Manager) CreateDomain(ctx context.Context, accountID, userID, domainName
 	return d, nil
 }
 
+// checkDomainAvailable reports whether the domain is free to claim. The unique
+// index on the column is the real guard; this turns the violation into a
+// conflict the caller can act on instead of a database error, and says nothing
+// about which account holds the domain.
+func (m Manager) checkDomainAvailable(ctx context.Context, domainName string) error {
+	_, err := m.store.GetCustomDomainByName(ctx, domainName)
+	if err == nil {
+		return status.Errorf(status.AlreadyExists, "domain %s is already registered", domainName)
+	}
+
+	if sErr, ok := status.FromError(err); ok && sErr.Type() == status.NotFound {
+		return nil
+	}
+
+	return fmt.Errorf("look up domain: %w", err)
+}
+
 func (m Manager) DeleteDomain(ctx context.Context, accountID, userID, domainID string) error {
 	ok, ctx, err := m.permissionsManager.ValidateUserPermissions(ctx, accountID, userID, modules.Services, operations.Delete)
 	if err != nil {
@@ -203,7 +225,9 @@ func (m Manager) ValidateDomain(ctx context.Context, accountID, userID, domainID
 		log.WithFields(log.Fields{
 			"accountID": accountID,
 			"domainID":  domainID,
-		}).WithError(err).Error("validate domain")
+			"userID":    userID,
+		}).Error("validate domain: permission denied")
+		return
 	}
 
 	log.WithFields(log.Fields{
@@ -298,9 +322,12 @@ func (m Manager) DeriveClusterFromDomain(ctx context.Context, accountID, domain
 		return "", fmt.Errorf("list custom domains: %w", err)
 	}
 
-	targetCluster, valid := extractClusterFromCustomDomains(domain, customDomains)
+	targetCluster, match := extractClusterFromCustomDomains(domain, customDomains)
-	if valid {
+	switch match {
+	case customDomainValidated:
 		return targetCluster, nil
+	case customDomainUnvalidated:
+		return "", status.Errorf(status.PreconditionFailed, "domain %s is not validated", domain)
 	}
 
 	return "", fmt.Errorf("domain %s does not match any available proxy cluster", domain)
@@ -363,19 +390,46 @@ func (m Manager) reservedGatewayAddress(ctx context.Context, accountID string) (
 	return settings.ProxyAddress, nil
 }
 
-func extractClusterFromCustomDomains(serviceDomain string, customDomains []*domain.Domain) (string, bool) {
+// customDomainMatch describes how a service domain relates to the account's
+// custom domain rows.
+type customDomainMatch int
+
+const (
+	customDomainNoMatch customDomainMatch = iota
+	customDomainUnvalidated
+	customDomainValidated
+)
+
+// extractClusterFromCustomDomains finds the longest custom domain covering the
+// service domain and reports its target cluster. Only a validated row yields a
+// cluster: until the CNAME check has passed the account has not shown it
+// controls the name, so no traffic may be routed for it.
+func extractClusterFromCustomDomains(serviceDomain string, customDomains []*domain.Domain) (string, customDomainMatch) {
 	bestCluster := ""
 	bestLen := -1
+	matched := false
 	for _, cd := range customDomains {
 		if serviceDomain != cd.Domain && !strings.HasSuffix(serviceDomain, "."+cd.Domain) {
 			continue
 		}
+		matched = true
+		if !cd.Validated {
+			continue
+		}
 		if l := len(cd.Domain); l > bestLen {
 			bestLen = l
 			bestCluster = cd.TargetCluster
 		}
 	}
-	return bestCluster, bestLen >= 0
+
+	switch {
+	case bestLen >= 0:
+		return bestCluster, customDomainValidated
+	case matched:
+		return "", customDomainUnvalidated
+	default:
+		return "", customDomainNoMatch
+	}
 }
 
 // ExtractClusterFromFreeDomain extracts the cluster address from a free domain.

management/internals/modules/reverseproxy/domain/manager/manager_realstore_test.go

@@ -0,0 +1,326 @@
+package manager
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel/metric/noop"
+
+	"github.com/netbirdio/netbird/management/internals/modules/reverseproxy/domain"
+	proxymanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy/manager"
+	"github.com/netbirdio/netbird/management/server/activity"
+	"github.com/netbirdio/netbird/management/server/mock_server"
+	"github.com/netbirdio/netbird/management/server/permissions"
+	nbstore "github.com/netbirdio/netbird/management/server/store"
+	"github.com/netbirdio/netbird/management/server/types"
+	"github.com/netbirdio/netbird/shared/management/status"
+)
+
+const (
+	testCluster    = "eu.proxy.test"
+	accountA       = "account-a"
+	accountAUser   = "account-a-admin"
+	accountB       = "account-b"
+	accountBUser   = "account-b-admin"
+	accountAMember = "account-a-member"
+)
+
+// stubResolver answers CNAME lookups from a table the test controls, so a
+// domain can point at the cluster or nowhere without touching a real resolver.
+type stubResolver struct {
+	mu     sync.Mutex
+	cnames map[string]string
+}
+
+func (r *stubResolver) LookupCNAME(_ context.Context, host string) (string, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	cname, ok := r.cnames[host]
+	if !ok {
+		return "", fmt.Errorf("lookup %s: no such host", host)
+	}
+	return cname + ".", nil
+}
+
+func (r *stubResolver) set(host, cname string) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.cnames[host] = cname
+}
+
+type domainTestEnv struct {
+	manager  Manager
+	store    nbstore.Store
+	resolver *stubResolver
+}
+
+// setupDomainTest builds the domain manager on a real SQLite store with two
+// accounts and one active public proxy cluster.
+func setupDomainTest(t *testing.T) *domainTestEnv {
+	t.Helper()
+
+	ctx := context.Background()
+	testStore, cleanup, err := nbstore.NewTestStoreFromSQL(ctx, "", t.TempDir())
+	require.NoError(t, err)
+	t.Cleanup(cleanup)
+
+	for accountID, userID := range map[string]string{accountA: accountAUser, accountB: accountBUser} {
+		users := map[string]*types.User{
+			userID: {
+				Id:        userID,
+				AccountID: accountID,
+				Role:      types.UserRoleAdmin,
+			},
+		}
+		if accountID == accountA {
+			// A real member of the account whose role denies Services:Create, so
+			// permission denial is exercised as ok=false rather than as a lookup
+			// error for a user who is not in the account at all.
+			users[accountAMember] = &types.User{
+				Id:        accountAMember,
+				AccountID: accountID,
+				Role:      types.UserRoleUser,
+			}
+		}
+
+		require.NoError(t, testStore.SaveAccount(ctx, &types.Account{
+			Id:        accountID,
+			CreatedBy: userID,
+			Settings:  &types.Settings{},
+			Users:     users,
+		}))
+	}
+
+	proxyMgr, err := proxymanager.NewManager(testStore, noop.NewMeterProvider().Meter(""))
+	require.NoError(t, err)
+
+	_, err = proxyMgr.Connect(ctx, "proxy-1", "session-1", testCluster, "127.0.0.1", nil, nil)
+	require.NoError(t, err)
+
+	resolver := &stubResolver{cnames: make(map[string]string)}
+
+	mgr := Manager{
+		store:              testStore,
+		proxyManager:       proxyMgr,
+		validator:          domain.Validator{Resolver: resolver},
+		permissionsManager: permissions.NewManager(testStore),
+		accountManager: &mock_server.MockAccountManager{
+			StoreEventFunc: func(context.Context, string, string, string, activity.ActivityDescriber, map[string]any) {},
+		},
+	}
+
+	return &domainTestEnv{manager: mgr, store: testStore, resolver: resolver}
+}
+
+// storedDomain reads a domain row back through the store so assertions are made
+// on what was persisted rather than on the value the manager returned.
+func storedDomain(t *testing.T, s nbstore.Store, accountID, domainName string) *domain.Domain {
+	t.Helper()
+
+	domains, err := s.ListCustomDomains(context.Background(), accountID)
+	require.NoError(t, err)
+	for _, d := range domains {
+		if d.Domain == domainName {
+			return d
+		}
+	}
+	return nil
+}
+
+// A domain whose CNAME check fails is stored unvalidated and must not resolve a
+// cluster, which is what service creation gates on.
+func TestCreateDomain_FailedLookupIsNotServable(t *testing.T) {
+	ctx := context.Background()
+	env := setupDomainTest(t)
+
+	created, err := env.manager.CreateDomain(ctx, accountA, accountAUser, "apps.example.com", testCluster)
+	require.NoError(t, err)
+	assert.False(t, created.Validated, "a domain whose CNAME lookup fails must not be created validated")
+
+	stored := storedDomain(t, env.store, accountA, "apps.example.com")
+	require.NotNil(t, stored, "domain row should exist")
+	assert.False(t, stored.Validated, "persisted row must be unvalidated")
+
+	cluster, err := env.manager.DeriveClusterFromDomain(ctx, accountA, "apps.example.com")
+	require.Error(t, err, "an unvalidated domain must not resolve a cluster")
+	assert.Empty(t, cluster)
+	assert.Contains(t, err.Error(), "not validated", "error should tell the caller what to fix")
+
+	sErr, ok := status.FromError(err)
+	require.True(t, ok, "error should be a typed status error")
+	assert.Equal(t, status.PreconditionFailed, sErr.Type())
+
+	_, err = env.manager.DeriveClusterFromDomain(ctx, accountA, "sub.apps.example.com")
+	assert.Error(t, err, "subdomains of an unvalidated custom domain are not servable either")
+}
+
+// A second account claiming a registered domain gets a clean conflict, not a
+// database error surfaced as a 500.
+func TestCreateDomain_DuplicateIsAConflict(t *testing.T) {
+	ctx := context.Background()
+	env := setupDomainTest(t)
+
+	_, err := env.manager.CreateDomain(ctx, accountA, accountAUser, "shared.example.com", testCluster)
+	require.NoError(t, err)
+
+	_, err = env.manager.CreateDomain(ctx, accountB, accountBUser, "shared.example.com", testCluster)
+	require.Error(t, err)
+
+	sErr, ok := status.FromError(err)
+	require.True(t, ok, "conflict must be a typed status error, not a raw database error")
+	assert.Equal(t, status.AlreadyExists, sErr.Type(), "conflict should map to 409, not 500")
+	assert.NotContains(t, sErr.Message, accountA, "the response must not reveal the holding account")
+
+	assert.Nil(t, storedDomain(t, env.store, accountB, "shared.example.com"), "no row should be written on conflict")
+}
+
+// The same account re-adding one of its own domains is a conflict too.
+func TestCreateDomain_SameAccountDuplicateIsAConflict(t *testing.T) {
+	ctx := context.Background()
+	env := setupDomainTest(t)
+
+	_, err := env.manager.CreateDomain(ctx, accountA, accountAUser, "dup.example.com", testCluster)
+	require.NoError(t, err)
+
+	_, err = env.manager.CreateDomain(ctx, accountA, accountAUser, "dup.example.com", testCluster)
+	require.Error(t, err)
+
+	sErr, ok := status.FromError(err)
+	require.True(t, ok)
+	assert.Equal(t, status.AlreadyExists, sErr.Type())
+}
+
+// The negative control: a validated domain still derives its cluster, for the
+// bare name and for subdomains, exactly as before.
+func TestCreateDomain_ValidatedDomainDerivesCluster(t *testing.T) {
+	ctx := context.Background()
+	env := setupDomainTest(t)
+	env.resolver.set("validation.valid.example.com", testCluster)
+
+	created, err := env.manager.CreateDomain(ctx, accountA, accountAUser, "valid.example.com", testCluster)
+	require.NoError(t, err)
+	require.True(t, created.Validated, "a matching CNAME should validate on create")
+
+	cluster, err := env.manager.DeriveClusterFromDomain(ctx, accountA, "valid.example.com")
+	require.NoError(t, err)
+	assert.Equal(t, testCluster, cluster)
+
+	cluster, err = env.manager.DeriveClusterFromDomain(ctx, accountA, "app.valid.example.com")
+	require.NoError(t, err)
+	assert.Equal(t, testCluster, cluster, "subdomains of a validated custom domain resolve too")
+}
+
+// Validating a domain flips the gate: the same lookup that failed before now
+// resolves a cluster.
+func TestValidateDomain_UnlocksClusterDerivation(t *testing.T) {
+	ctx := context.Background()
+	env := setupDomainTest(t)
+
+	created, err := env.manager.CreateDomain(ctx, accountA, accountAUser, "later.example.com", testCluster)
+	require.NoError(t, err)
+	require.False(t, created.Validated)
+
+	_, err = env.manager.DeriveClusterFromDomain(ctx, accountA, "later.example.com")
+	require.Error(t, err)
+
+	env.resolver.set("validation.later.example.com", testCluster)
+	env.manager.ValidateDomain(ctx, accountA, accountAUser, created.ID)
+
+	require.True(t, storedDomain(t, env.store, accountA, "later.example.com").Validated)
+
+	cluster, err := env.manager.DeriveClusterFromDomain(ctx, accountA, "later.example.com")
+	require.NoError(t, err)
+	assert.Equal(t, testCluster, cluster)
+}
+
+// Free cluster domains are unaffected by the custom domain gate.
+func TestDeriveClusterFromDomain_FreeDomainUnaffected(t *testing.T) {
+	ctx := context.Background()
+	env := setupDomainTest(t)
+
+	cluster, err := env.manager.DeriveClusterFromDomain(ctx, accountA, "myapp.abc123."+testCluster)
+	require.NoError(t, err)
+	assert.Equal(t, testCluster, cluster)
+}
+
+// The manager pre-check exists to turn a conflict into a 409, but the unique
+// index on the column is what actually guarantees the domain is claimed once.
+//
+// Two requests can clear the pre-check concurrently and race to the insert.
+// Inserting twice through the store reaches the same code path the loser of
+// that race takes, without the nondeterminism of driving it from goroutines,
+// and the loser must still see a conflict rather than an internal error.
+func TestStore_DuplicateDomainRejectedByIndexAsConflict(t *testing.T) {
+	ctx := context.Background()
+	env := setupDomainTest(t)
+
+	_, err := env.store.CreateCustomDomain(ctx, accountA, "indexed.example.com", testCluster, false)
+	require.NoError(t, err)
+
+	_, err = env.store.CreateCustomDomain(ctx, accountB, "indexed.example.com", testCluster, false)
+	require.Error(t, err, "the unique index must reject the same domain in a second account")
+
+	sErr, ok := status.FromError(err)
+	require.True(t, ok, "the losing insert must return a typed status error")
+	assert.Equal(t, status.AlreadyExists, sErr.Type(), "a lost race is a 409, not a 500")
+}
+
+// Validation is what decides whether a domain routes traffic, so a caller
+// without permission to it must not be able to flip the flag. The check logged
+// the denial and then carried on, which was inert while nothing read Validated
+// and is not once cluster derivation gates on it.
+func TestValidateDomain_PermissionDeniedDoesNotValidate(t *testing.T) {
+	ctx := context.Background()
+	env := setupDomainTest(t)
+
+	created, err := env.manager.CreateDomain(ctx, accountA, accountAUser, "guarded.example.com", testCluster)
+	require.NoError(t, err)
+	require.False(t, created.Validated)
+
+	// The CNAME is in place, so the only thing standing between this caller and
+	// a validated domain is the permission check.
+	env.resolver.set("validation.guarded.example.com", testCluster)
+
+	env.manager.ValidateDomain(ctx, accountA, accountAMember, created.ID)
+
+	stored := storedDomain(t, env.store, accountA, "guarded.example.com")
+	require.NotNil(t, stored)
+	assert.False(t, stored.Validated, "a caller without permission must not validate the domain")
+
+	_, err = env.manager.DeriveClusterFromDomain(ctx, accountA, "guarded.example.com")
+	assert.Error(t, err, "the domain must still be unservable")
+}
+
+// Validation runs asynchronously, so it can finish after the domain was
+// deleted and then write a stale row back. gorm's Save falls back to an insert
+// when an update affects no rows, which would resurrect the domain as
+// validated; UpdateCustomDomain avoids that by selecting explicit columns.
+// This pins that behaviour, since dropping the Select would reintroduce it.
+func TestUpdateCustomDomain_DoesNotResurrectDeletedDomain(t *testing.T) {
+	ctx := context.Background()
+	env := setupDomainTest(t)
+
+	created, err := env.manager.CreateDomain(ctx, accountA, accountAUser, "racy.example.com", testCluster)
+	require.NoError(t, err)
+
+	stale := storedDomain(t, env.store, accountA, "racy.example.com")
+	require.NotNil(t, stale)
+
+	require.NoError(t, env.manager.DeleteDomain(ctx, accountA, accountAUser, created.ID))
+	require.Nil(t, storedDomain(t, env.store, accountA, "racy.example.com"), "the domain should be gone")
+
+	// What an in-flight validation would write once its CNAME check succeeded.
+	// The write has to succeed for the assertion below to mean anything: a
+	// rejected write would leave the domain absent for the wrong reason.
+	stale.Validated = true
+	_, err = env.store.UpdateCustomDomain(ctx, accountA, stale)
+	require.NoError(t, err, "the update itself must succeed, so absence is not just a failed write")
+
+	assert.Nil(t, storedDomain(t, env.store, accountA, "racy.example.com"),
+		"a late validation write must not recreate a deleted domain")
+}

management/internals/modules/reverseproxy/domain/manager/manager_test.go

@@ -184,6 +184,10 @@ func (s *stubStore) GetCustomDomain(context.Context, string, string) (*domain.Do
 	panic("not used in allow-list tests")
 }
 
+func (s *stubStore) GetCustomDomainByName(context.Context, string) (*domain.Domain, error) {
+	panic("not used in allow-list tests")
+}
+
 func (s *stubStore) ListFreeDomains(context.Context, string) ([]string, error) {
 	panic("not used in allow-list tests")
 }

management/internals/modules/reverseproxy/service/manager/domain_validation_test.go

@@ -0,0 +1,127 @@
+package manager
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel/metric/noop"
+
+	domainmanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/domain/manager"
+	proxymanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy/manager"
+	rpservice "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service"
+	"github.com/netbirdio/netbird/management/server/activity"
+	"github.com/netbirdio/netbird/management/server/mock_server"
+	"github.com/netbirdio/netbird/management/server/permissions"
+	"github.com/netbirdio/netbird/management/server/store"
+	"github.com/netbirdio/netbird/shared/management/status"
+)
+
+const validationTestCluster = "eu.proxy.test"
+
+// withRealDomainManager swaps the stub cluster deriver for the real domain
+// manager backed by the same store, so service creation is gated by the actual
+// domain rows rather than by a test double that always agrees.
+func withRealDomainManager(t *testing.T, mgr *Manager, testStore store.Store) {
+	t.Helper()
+
+	ctx := context.Background()
+	proxyMgr, err := proxymanager.NewManager(testStore, noop.NewMeterProvider().Meter(""))
+	require.NoError(t, err)
+
+	_, err = proxyMgr.Connect(ctx, "proxy-1", "session-1", validationTestCluster, "127.0.0.1", nil, nil)
+	require.NoError(t, err)
+
+	accountMgr := &mock_server.MockAccountManager{
+		StoreEventFunc: func(context.Context, string, string, string, activity.ActivityDescriber, map[string]any) {},
+	}
+	mgr.clusterDeriver = domainmanager.NewManager(testStore, proxyMgr, permissions.NewManager(testStore), accountMgr)
+}
+
+func newTestService(domain string) *rpservice.Service {
+	return &rpservice.Service{
+		Name:    "test-service",
+		Domain:  domain,
+		Enabled: true,
+		Mode:    rpservice.ModeHTTP,
+		Targets: []*rpservice.Target{{
+			Host:       "10.0.0.1",
+			Port:       8080,
+			Protocol:   "http",
+			TargetId:   testPeerID,
+			TargetType: "peer",
+			Enabled:    true,
+		}},
+	}
+}
+
+// A service must not bind to a domain the account has not validated, and
+// nothing may be persisted for the attempt.
+func TestCreateService_RefusesUnvalidatedDomain(t *testing.T) {
+	ctx := context.Background()
+	mgr, testStore := setupIntegrationTest(t)
+	withRealDomainManager(t, mgr, testStore)
+
+	_, err := testStore.CreateCustomDomain(ctx, testAccountID, "unproven.example.com", validationTestCluster, false)
+	require.NoError(t, err)
+
+	_, err = mgr.CreateService(ctx, testAccountID, testUserID, newTestService("unproven.example.com"))
+	require.Error(t, err, "an unvalidated domain must not bind a service")
+	assert.Contains(t, err.Error(), "not validated", "the API error should name the actual problem")
+
+	sErr, ok := status.FromError(err)
+	require.True(t, ok, "error should be a typed status error")
+	assert.Equal(t, status.PreconditionFailed, sErr.Type())
+
+	services, err := testStore.GetAccountServices(ctx, store.LockingStrengthNone, testAccountID)
+	require.NoError(t, err)
+	assert.Empty(t, services, "no service row should be written for a refused domain")
+}
+
+// The negative control: a validated domain still binds a service and derives
+// its cluster exactly as before.
+func TestCreateService_ValidatedDomainBindsService(t *testing.T) {
+	ctx := context.Background()
+	mgr, testStore := setupIntegrationTest(t)
+	withRealDomainManager(t, mgr, testStore)
+
+	_, err := testStore.CreateCustomDomain(ctx, testAccountID, "proven.example.com", validationTestCluster, true)
+	require.NoError(t, err)
+
+	created, err := mgr.CreateService(ctx, testAccountID, testUserID, newTestService("app.proven.example.com"))
+	require.NoError(t, err)
+	assert.Equal(t, validationTestCluster, created.ProxyCluster, "service should bind to the domain's target cluster")
+
+	services, err := testStore.GetAccountServices(ctx, store.LockingStrengthNone, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, services, 1, "the service should be persisted")
+	assert.Equal(t, "app.proven.example.com", services[0].Domain)
+}
+
+// An update must not be a way around the creation gate: moving a live service
+// onto an unvalidated domain has to fail rather than silently keep the old
+// cluster and start serving the new hostname.
+func TestUpdateService_RefusesMoveToUnvalidatedDomain(t *testing.T) {
+	ctx := context.Background()
+	mgr, testStore := setupIntegrationTest(t)
+	withRealDomainManager(t, mgr, testStore)
+
+	_, err := testStore.CreateCustomDomain(ctx, testAccountID, "proven.example.com", validationTestCluster, true)
+	require.NoError(t, err)
+	_, err = testStore.CreateCustomDomain(ctx, testAccountID, "unproven.example.com", validationTestCluster, false)
+	require.NoError(t, err)
+
+	created, err := mgr.CreateService(ctx, testAccountID, testUserID, newTestService("app.proven.example.com"))
+	require.NoError(t, err)
+
+	moved := *created
+	moved.Domain = "app.unproven.example.com"
+	_, err = mgr.UpdateService(ctx, testAccountID, testUserID, &moved)
+	require.Error(t, err, "moving to an unvalidated domain must fail")
+	assert.Contains(t, err.Error(), "not validated")
+
+	stored, err := testStore.GetServiceByID(ctx, store.LockingStrengthNone, testAccountID, created.ID)
+	require.NoError(t, err)
+	assert.Equal(t, "app.proven.example.com", stored.Domain, "the service must keep its original domain")
+}

management/internals/modules/reverseproxy/service/manager/manager.go

@@ -606,16 +606,19 @@ func (m *Manager) resolveEffectiveCluster(ctx context.Context, accountID string,
 		return existing.ProxyCluster, nil
 	}
 
-	if m.clusterDeriver != nil {
+	if m.clusterDeriver == nil {
-		derived, err := m.clusterDeriver.DeriveClusterFromDomain(ctx, accountID, svc.Domain)
+		return existing.ProxyCluster, nil
-		if err != nil {
-			log.WithError(err).Warnf("could not derive cluster from domain %s", svc.Domain)
-		} else {
-			return derived, nil
-		}
 	}
 
-	return existing.ProxyCluster, nil
+	// Falling back to the old cluster here would let an update move a service
+	// onto a domain the account has not validated, bypassing the check that
+	// creation makes.
+	derived, err := m.clusterDeriver.DeriveClusterFromDomain(ctx, accountID, svc.Domain)
+	if err != nil {
+		return "", status.Errorf(status.PreconditionFailed, "could not derive cluster from domain %s: %v", svc.Domain, err)
+	}
+
+	return derived, nil
 }
 
 func (m *Manager) executeServiceUpdate(ctx context.Context, transaction store.Store, accountID string, service *service.Service, updateInfo *serviceUpdateInfo, customPorts *bool, effectiveCluster string) error {

management/server/account.go

@@ -719,8 +719,10 @@ func (am *DefaultAccountManager) schedulePeerLoginExpiration(ctx context.Context
 		log.WithContext(ctx).Tracef("peer login expiration job for account %s is already scheduled", accountID)
 		return
 	}
+	// The job outlives the request that arms it, so it must not inherit the request's cancellation.
+	jobCtx := context.WithoutCancel(ctx)
 	if nextRun, ok := am.getNextPeerExpiration(ctx, accountID); ok {
-		go am.peerLoginExpiry.Schedule(ctx, nextRun, accountID, am.peerLoginExpirationJob(ctx, accountID))
+		go am.peerLoginExpiry.Schedule(jobCtx, nextRun, accountID, am.peerLoginExpirationJob(jobCtx, accountID))
 	}
 }
 
@@ -752,8 +754,9 @@ func (am *DefaultAccountManager) peerInactivityExpirationJob(ctx context.Context
 // checkAndSchedulePeerInactivityExpiration periodically checks for inactive peers to end their sessions
 func (am *DefaultAccountManager) checkAndSchedulePeerInactivityExpiration(ctx context.Context, accountID string) {
 	am.peerInactivityExpiry.Cancel(ctx, []string{accountID})
+	jobCtx := context.WithoutCancel(ctx)
 	if nextRun, ok := am.getNextInactivePeerExpiration(ctx, accountID); ok {
-		go am.peerInactivityExpiry.Schedule(ctx, nextRun, accountID, am.peerInactivityExpirationJob(ctx, accountID))
+		go am.peerInactivityExpiry.Schedule(jobCtx, nextRun, accountID, am.peerInactivityExpirationJob(jobCtx, accountID))
 	}
 }
 

management/server/account_test.go

@@ -1920,6 +1920,154 @@ func TestDefaultAccountManager_MarkPeerConnected_PeerLoginExpiration(t *testing.
 	}
 }
 
+func TestDefaultAccountManager_SchedulePeerLoginExpiration_IncludesOfflinePeers(t *testing.T) {
+	manager, updateManager, err := createManager(t)
+	require.NoError(t, err, "unable to create account manager")
+
+	accountID, err := manager.GetAccountIDByUserID(context.Background(), auth.UserAuth{UserId: userID})
+	require.NoError(t, err, "unable to create an account")
+
+	connectedKey, offlineKey := addExpiringPeers(t, manager)
+	_, err = manager.UpdateAccountSettings(context.Background(), accountID, userID, &types.Settings{
+		PeerLoginExpiration:        time.Hour,
+		PeerLoginExpirationEnabled: true,
+		Extra:                      &types.ExtraSettings{},
+	})
+	require.NoError(t, err, "expecting to update account settings successfully but got error")
+	manager.peerLoginExpiry.CancelAll(context.Background())
+
+	// The connected peer logged in just now, so a job computed from connected peers alone
+	// would be armed for an hour. The offline peer's login expires in two seconds; a
+	// reconnect of that peer must not have to wait for the connected peer's tick.
+	now := time.Now().UTC()
+	setPeerLogin(t, manager, accountID, connectedKey, true, now)
+	setPeerLogin(t, manager, accountID, offlineKey, false, now.Add(-time.Hour+2*time.Second))
+
+	offlinePeer, err := manager.Store.GetPeerByPeerPubKey(context.Background(), store.LockingStrengthNone, offlineKey)
+	require.NoError(t, err)
+	updateManager.CreateChannel(context.Background(), offlinePeer.ID)
+
+	manager.peerLoginExpiry = NewDefaultScheduler()
+	t.Cleanup(func() { manager.peerLoginExpiry.CancelAll(context.Background()) })
+	manager.schedulePeerLoginExpiration(context.Background(), accountID)
+
+	// The flag is committed per peer before the disconnect fans out, so wait for both.
+	require.Eventually(t, func() bool {
+		peer, err := manager.Store.GetPeerByPeerPubKey(context.Background(), store.LockingStrengthNone, offlineKey)
+		return err == nil && peer.Status.LoginExpired && !updateManager.HasChannel(offlinePeer.ID)
+	}, 10*time.Second, 100*time.Millisecond, "offline peer should be expired and disconnected at its own deadline")
+
+	connectedPeer, err := manager.Store.GetPeerByPeerPubKey(context.Background(), store.LockingStrengthNone, connectedKey)
+	require.NoError(t, err)
+	assert.False(t, connectedPeer.Status.LoginExpired, "connected peer with a fresh login must not expire")
+}
+
+func TestDefaultAccountManager_SchedulePeerLoginExpiration_DetachesRequestContext(t *testing.T) {
+	manager, _, err := createManager(t)
+	require.NoError(t, err, "unable to create account manager")
+
+	accountID, err := manager.GetAccountIDByUserID(context.Background(), auth.UserAuth{UserId: userID})
+	require.NoError(t, err, "unable to create an account")
+	connectedKey, _ := addExpiringPeers(t, manager)
+	setPeerLogin(t, manager, accountID, connectedKey, true, time.Now().UTC())
+
+	scheduled := make(chan context.Context, 1)
+	manager.peerLoginExpiry = &MockScheduler{
+		IsSchedulerRunningFunc: func(string) bool { return false },
+		ScheduleFunc: func(ctx context.Context, _ time.Duration, _ string, _ func() (time.Duration, bool)) {
+			scheduled <- ctx
+		},
+	}
+
+	requestCtx, cancel := context.WithCancel(context.Background())
+	manager.schedulePeerLoginExpiration(requestCtx, accountID)
+	cancel()
+
+	select {
+	case jobCtx := <-scheduled:
+		assert.NoError(t, jobCtx.Err(), "the expiration job must outlive the request that armed it")
+	case <-time.After(time.Second):
+		t.Fatal("timeout while waiting for the job to be scheduled")
+	}
+}
+
+func TestDefaultAccountManager_ExpireAndUpdatePeers_SkipsPeerThatLoggedInAgain(t *testing.T) {
+	manager, updateManager, err := createManager(t)
+	require.NoError(t, err, "unable to create account manager")
+
+	accountID, err := manager.GetAccountIDByUserID(context.Background(), auth.UserAuth{UserId: userID})
+	require.NoError(t, err, "unable to create an account")
+
+	reloggedKey, staleKey := addExpiringPeers(t, manager)
+	_, err = manager.UpdateAccountSettings(context.Background(), accountID, userID, &types.Settings{
+		PeerLoginExpiration:        time.Hour,
+		PeerLoginExpirationEnabled: true,
+		Extra:                      &types.ExtraSettings{},
+	})
+	require.NoError(t, err, "expecting to update account settings successfully but got error")
+	manager.peerLoginExpiry.CancelAll(context.Background())
+
+	expiredLogin := time.Now().UTC().Add(-2 * time.Hour)
+	setPeerLogin(t, manager, accountID, reloggedKey, true, expiredLogin)
+	setPeerLogin(t, manager, accountID, staleKey, true, expiredLogin)
+
+	expiredPeers, err := manager.getExpiredPeers(context.Background(), accountID)
+	require.NoError(t, err)
+	require.Len(t, expiredPeers, 2, "both peers should be due for expiration")
+
+	// The job holds the candidate list while one peer completes a fresh login, which
+	// moves its deadline into the future and must win over the stale candidate entry.
+	setPeerLogin(t, manager, accountID, reloggedKey, true, time.Now().UTC())
+
+	reloggedPeer, err := manager.Store.GetPeerByPeerPubKey(context.Background(), store.LockingStrengthNone, reloggedKey)
+	require.NoError(t, err)
+	stalePeer, err := manager.Store.GetPeerByPeerPubKey(context.Background(), store.LockingStrengthNone, staleKey)
+	require.NoError(t, err)
+	updateManager.CreateChannel(context.Background(), reloggedPeer.ID)
+	updateManager.CreateChannel(context.Background(), stalePeer.ID)
+
+	err = manager.expireAndUpdatePeers(context.Background(), accountID, expiredPeers, peerExpirationSessionExpired)
+	require.NoError(t, err)
+
+	reloggedPeer, err = manager.Store.GetPeerByPeerPubKey(context.Background(), store.LockingStrengthNone, reloggedKey)
+	require.NoError(t, err)
+	assert.False(t, reloggedPeer.Status.LoginExpired, "a peer that logged in again must not be flagged from the stale candidate list")
+	assert.True(t, reloggedPeer.Status.Connected, "the re-logged peer must keep its connected status")
+	assert.True(t, updateManager.HasChannel(reloggedPeer.ID), "the re-logged peer's update channel must stay open")
+
+	stalePeer, err = manager.Store.GetPeerByPeerPubKey(context.Background(), store.LockingStrengthNone, staleKey)
+	require.NoError(t, err)
+	assert.True(t, stalePeer.Status.LoginExpired, "a peer that is still due must be flagged")
+	assert.False(t, updateManager.HasChannel(stalePeer.ID), "the expired peer's update channel must be closed")
+}
+
+// addExpiringPeers registers two SSO peers with login expiration enabled and returns their public keys.
+func addExpiringPeers(t *testing.T, manager *DefaultAccountManager) (string, string) {
+	t.Helper()
+	keys := make([]string, 0, 2)
+	for _, hostname := range []string{"connected-peer", "offline-peer"} {
+		key, err := wgtypes.GenerateKey()
+		require.NoError(t, err, "unable to generate WireGuard key")
+		_, _, _, _, err = manager.AddPeer(context.Background(), "", "", userID, &nbpeer.Peer{
+			Key:                    key.PublicKey().String(),
+			Meta:                   nbpeer.PeerSystemMeta{Hostname: hostname},
+			LoginExpirationEnabled: true,
+		}, false)
+		require.NoError(t, err, "unable to add peer")
+		keys = append(keys, key.PublicKey().String())
+	}
+	return keys[0], keys[1]
+}
+
+func setPeerLogin(t *testing.T, manager *DefaultAccountManager, accountID, peerKey string, connected bool, lastLogin time.Time) {
+	t.Helper()
+	peer, err := manager.Store.GetPeerByPeerPubKey(context.Background(), store.LockingStrengthNone, peerKey)
+	require.NoError(t, err)
+	peer.Status.Connected = connected
+	peer.LastLogin = &lastLogin
+	require.NoError(t, manager.Store.SavePeer(context.Background(), accountID, peer))
+}
+
 func TestDefaultAccountManager_MarkPeerDisconnected_SchedulesInactivityExpiration(t *testing.T) {
 	manager, _, err := createManager(t)
 	require.NoError(t, err, "unable to create account manager")
@@ -2702,7 +2850,7 @@ func TestAccount_GetNextPeerExpiration(t *testing.T) {
 			expectedNextExpiration: time.Duration(0),
 		},
 		{
-			name: "No connected peers, no expiration",
+			name: "Offline peer with expiration, return expiration",
 			peers: map[string]*nbpeer.Peer{
 				"peer-1": {
 					Status: &nbpeer.PeerStatus{
@@ -2721,8 +2869,33 @@ func TestAccount_GetNextPeerExpiration(t *testing.T) {
 			},
 			expiration:             time.Second,
 			expirationEnabled:      false,
-			expectedNextRun:        false,
+			expectedNextRun:        true,
-			expectedNextExpiration: time.Duration(0),
+			expectedNextExpiration: time.Second,
+		},
+		{
+			name: "Offline peer with the earliest deadline defines the next run",
+			peers: map[string]*nbpeer.Peer{
+				"peer-1": {
+					Status: &nbpeer.PeerStatus{
+						Connected: true,
+					},
+					LoginExpirationEnabled: true,
+					LastLogin:              util.ToPtr(time.Now().UTC()),
+					UserID:                 userID,
+				},
+				"peer-2": {
+					Status: &nbpeer.PeerStatus{
+						Connected: false,
+					},
+					LoginExpirationEnabled: true,
+					LastLogin:              util.ToPtr(time.Now().UTC().Add(-50 * time.Minute)),
+					UserID:                 userID,
+				},
+			},
+			expiration:             time.Hour,
+			expirationEnabled:      true,
+			expectedNextRun:        true,
+			expectedNextExpiration: 10 * time.Minute,
 		},
 		{
 			name: "Connected peers with disabled expiration, no expiration",

management/server/peer.go

@@ -1494,9 +1494,12 @@ func checkAuth(ctx context.Context, loginUserID string, peer *nbpeer.Peer) error
 
 func peerLoginExpired(ctx context.Context, peer *nbpeer.Peer, settings *types.Settings) bool {
 	expired, expiresIn := peer.LoginExpired(settings.PeerLoginExpiration)
-	expired = settings.PeerLoginExpirationEnabled && expired
+	if settings.PeerLoginExpirationEnabled && expired {
-	if expired || peer.Status.LoginExpired {
+		log.WithContext(ctx).Debugf("peer's %s login expired %v ago", peer.ID, -expiresIn)
-		log.WithContext(ctx).Debugf("peer's %s login expired %v ago", peer.ID, expiresIn)
+		return true
+	}
+	if peer.Status.LoginExpired {
+		log.WithContext(ctx).Debugf("peer's %s login is marked as expired", peer.ID)
 		return true
 	}
 	return false
@@ -1643,7 +1646,9 @@ func (am *DefaultAccountManager) UpdateAccountPeer(ctx context.Context, accountI
 
 // getNextPeerExpiration returns the minimum duration in which the next peer of the account will expire if it was found.
 // If there is no peer that expires this function returns false and a duration of 0.
-// This function only considers peers that haven't been expired yet and that are connected.
+// This function only considers peers that haven't been expired yet. Offline peers count too:
+// a running job is never re-armed on connect, so a peer that reconnects with an old login
+// must already be part of the scheduled run.
 func (am *DefaultAccountManager) getNextPeerExpiration(ctx context.Context, accountID string) (time.Duration, bool) {
 	peersWithExpiry, err := am.Store.GetAccountPeersWithExpiration(ctx, store.LockingStrengthNone, accountID)
 	if err != nil {
@@ -1663,8 +1668,7 @@ func (am *DefaultAccountManager) getNextPeerExpiration(ctx context.Context, acco
 
 	var nextExpiry *time.Duration
 	for _, peer := range peersWithExpiry {
-		// consider only connected peers because others will require login on connecting to the management server
+		if peer.Status.LoginExpired {
-		if peer.Status.LoginExpired || !peer.Status.Connected {
 			continue
 		}
 		_, duration := peer.LoginExpired(settings.PeerLoginExpiration)

management/server/scheduler.go

@@ -117,6 +117,7 @@ func (wm *DefaultScheduler) Schedule(ctx context.Context, in time.Duration, ID s
 	}
 
 	ticker := time.NewTicker(in)
+	period := in
 
 	wm.jobs[ID] = cancel
 	log.WithContext(ctx).Debugf("scheduled a job %s to run in %s. There are %d total jobs scheduled.", ID, in.String(), len(wm.jobs))
@@ -136,14 +137,18 @@ func (wm *DefaultScheduler) Schedule(ctx context.Context, in time.Duration, ID s
 				if !reschedule {
 					wm.mu.Lock()
 					defer wm.mu.Unlock()
-					delete(wm.jobs, ID)
+					// A Cancel during job() may have registered a replacement under this ID.
+					if current, ok := wm.jobs[ID]; ok && current == cancel {
+						delete(wm.jobs, ID)
+					}
 					log.WithContext(ctx).Debugf("job %s is not scheduled to run again", ID)
 					ticker.Stop()
 					return
 				}
 				// we need this comparison to avoid resetting the ticker with the same duration and missing the current elapsesed time
-				if runIn != in {
+				if runIn != period {
 					ticker.Reset(runIn)
+					period = runIn
 				}
 			case <-cancel:
 				log.WithContext(ctx).Debugf("job %s was canceled, stopping timer", ID)

management/server/scheduler_test.go

@@ -6,10 +6,12 @@ import (
 	"math/rand"
 	"runtime"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestScheduler_Performance(t *testing.T) {
@@ -150,3 +152,90 @@ func TestScheduler_Schedule(t *testing.T) {
 	scheduler.cancel(context.Background(), jobID)
 
 }
+
+func TestScheduler_Schedule_ResetsTickerAfterReturningInitialInterval(t *testing.T) {
+	jobID := "test-scheduler-job-2"
+	scheduler := NewDefaultScheduler()
+	defer scheduler.Cancel(context.Background(), []string{jobID})
+
+	initial := 30 * time.Millisecond
+	stretched := 400 * time.Millisecond
+	runs := make(chan time.Time, 3)
+	count := 0
+	// The first run stretches the period; the second returns the initial interval again,
+	// which must shrink the period back instead of keeping the stretched one.
+	job := func() (nextRunIn time.Duration, reschedule bool) {
+		count++
+		runs <- time.Now()
+		switch count {
+		case 1:
+			return stretched, true
+		case 2:
+			return initial, true
+		default:
+			return 0, false
+		}
+	}
+	scheduler.Schedule(context.Background(), initial, jobID, job)
+
+	var stamps []time.Time
+	for len(stamps) < 3 {
+		select {
+		case ts := <-runs:
+			stamps = append(stamps, ts)
+		case <-time.After(2 * time.Second):
+			t.Fatalf("timed out after %d runs", len(stamps))
+		}
+	}
+	assert.Less(t, stamps[2].Sub(stamps[1]), stretched/2, "returning the initial interval must reset the stretched ticker")
+}
+
+func TestScheduler_Schedule_StaleCompletionKeepsReplacement(t *testing.T) {
+	jobID := "test-scheduler-job-3"
+	scheduler := NewDefaultScheduler()
+	defer scheduler.Cancel(context.Background(), []string{jobID})
+
+	started := make(chan struct{})
+	release := make(chan struct{})
+	staleJob := func() (nextRunIn time.Duration, reschedule bool) {
+		close(started)
+		<-release
+		return 0, false
+	}
+	scheduler.Schedule(context.Background(), 10*time.Millisecond, jobID, staleJob)
+	select {
+	case <-started:
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for the first job to start")
+	}
+
+	// Cancel the job while it is still executing and register a replacement under the
+	// same ID, as the expiration paths do on a settings change.
+	scheduler.Cancel(context.Background(), []string{jobID})
+	var replacementRuns atomic.Int32
+	scheduler.Schedule(context.Background(), 20*time.Millisecond, jobID, func() (nextRunIn time.Duration, reschedule bool) {
+		replacementRuns.Add(1)
+		return 20 * time.Millisecond, true
+	})
+	require.True(t, scheduler.IsSchedulerRunning(jobID), "replacement must be registered")
+
+	// The stale job now completes without rescheduling; its cleanup must leave the
+	// replacement's entry in place.
+	close(release)
+	assert.Never(t, func() bool { return !scheduler.IsSchedulerRunning(jobID) }, 200*time.Millisecond, 10*time.Millisecond,
+		"stale completion must not drop the replacement job")
+
+	var duplicateRuns atomic.Int32
+	scheduler.Schedule(context.Background(), 10*time.Millisecond, jobID, func() (nextRunIn time.Duration, reschedule bool) {
+		duplicateRuns.Add(1)
+		return 10 * time.Millisecond, true
+	})
+	assert.Never(t, func() bool { return duplicateRuns.Load() > 0 }, 100*time.Millisecond, 10*time.Millisecond,
+		"a duplicate schedule must be refused while the replacement is registered")
+
+	scheduler.Cancel(context.Background(), []string{jobID})
+	assert.False(t, scheduler.IsSchedulerRunning(jobID), "cancel must find and remove the replacement")
+	runsAfterCancel := replacementRuns.Load()
+	assert.Never(t, func() bool { return replacementRuns.Load() > runsAfterCancel+1 }, 150*time.Millisecond, 10*time.Millisecond,
+		"the replacement must stop after cancel")
+}

management/server/store/sql_store.go

@@ -5686,6 +5686,23 @@ func (s *SqlStore) ListCustomDomains(ctx context.Context, accountID string) ([]*
 	return domains, nil
 }
 
+// GetCustomDomainByName returns the custom domain row holding the given name,
+// regardless of which account owns it.
+func (s *SqlStore) GetCustomDomainByName(ctx context.Context, domainName string) (*domain.Domain, error) {
+	customDomain := &domain.Domain{}
+	result := s.db.Take(customDomain, "domain = ?", domainName)
+	if result.Error != nil {
+		if errors.Is(result.Error, gorm.ErrRecordNotFound) {
+			return nil, status.Errorf(status.NotFound, "custom domain %s not found", domainName)
+		}
+
+		log.WithContext(ctx).Errorf("failed to get custom domain by name from store: %v", result.Error)
+		return nil, status.Errorf(status.Internal, "failed to get custom domain from store")
+	}
+
+	return customDomain, nil
+}
+
 func (s *SqlStore) CreateCustomDomain(ctx context.Context, accountID string, domainName string, targetCluster string, validated bool) (*domain.Domain, error) {
 	newDomain := &domain.Domain{
 		ID:            xid.New().String(), // Generate our own ID because gorm doesn't always configure the database to handle this for us.
@@ -5697,6 +5714,18 @@ func (s *SqlStore) CreateCustomDomain(ctx context.Context, accountID string, dom
 	}
 	result := s.db.Create(newDomain)
 	if result.Error != nil {
+		// The unique index is the last guard when two requests clear the
+		// manager's availability check at the same time. The one that loses the
+		// insert is a conflict, not an internal failure.
+		var count int64
+		if err := s.db.Model(&domain.Domain{}).Where("domain = ?", domainName).Count(&count).Error; err == nil && count > 0 {
+			// The insert error is logged even on this path: the name being taken
+			// is what the caller has to act on, but if the insert also failed for
+			// an unrelated reason the operator still needs to see it.
+			log.WithContext(ctx).Warnf("create reverse proxy custom domain %s rejected, name already registered: %v", domainName, result.Error)
+			return nil, status.Errorf(status.AlreadyExists, "domain %s is already registered", domainName)
+		}
+
 		log.WithContext(ctx).Errorf("failed to create reverse proxy custom domain to store: %v", result.Error)
 		return nil, status.Errorf(status.Internal, "failed to create reverse proxy custom domain to store")
 	}

management/server/store/store.go

@@ -302,6 +302,7 @@ type Store interface {
 	GetCustomDomain(ctx context.Context, accountID string, domainID string) (*domain.Domain, error)
 	ListFreeDomains(ctx context.Context, accountID string) ([]string, error)
 	ListCustomDomains(ctx context.Context, accountID string) ([]*domain.Domain, error)
+	GetCustomDomainByName(ctx context.Context, domainName string) (*domain.Domain, error)
 	CreateCustomDomain(ctx context.Context, accountID string, domainName string, targetCluster string, validated bool) (*domain.Domain, error)
 	UpdateCustomDomain(ctx context.Context, accountID string, d *domain.Domain) (*domain.Domain, error)
 	DeleteCustomDomain(ctx context.Context, accountID string, domainID string) error

management/server/store/store_mock.go

@@ -1941,6 +1941,21 @@ func (mr *MockStoreMockRecorder) GetCustomDomain(ctx, accountID, domainID any) *
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetCustomDomain", reflect.TypeOf((*MockStore)(nil).GetCustomDomain), ctx, accountID, domainID)
 }
 
+// GetCustomDomainByName mocks base method.
+func (m *MockStore) GetCustomDomainByName(ctx context.Context, domainName string) (*domain.Domain, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetCustomDomainByName", ctx, domainName)
+	ret0, _ := ret[0].(*domain.Domain)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// GetCustomDomainByName indicates an expected call of GetCustomDomainByName.
+func (mr *MockStoreMockRecorder) GetCustomDomainByName(ctx, domainName any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetCustomDomainByName", reflect.TypeOf((*MockStore)(nil).GetCustomDomainByName), ctx, domainName)
+}
+
 // GetCustomDomainsCounts mocks base method.
 func (m *MockStore) GetCustomDomainsCounts(ctx context.Context) (int64, int64, error) {
 	m.ctrl.T.Helper()

management/server/types/account.go

@@ -404,7 +404,7 @@ func (a *Account) GetExpiredPeers() []*nbpeer.Peer {
 
 // GetNextPeerExpiration returns the minimum duration in which the next peer of the account will expire if it was found.
 // If there is no peer that expires this function returns false and a duration of 0.
-// This function only considers peers that haven't been expired yet and that are connected.
+// This function only considers peers that haven't been expired yet, whether connected or not.
 func (a *Account) GetNextPeerExpiration() (time.Duration, bool) {
 	peersWithExpiry := a.GetPeersWithExpiration()
 	if len(peersWithExpiry) == 0 {
@@ -412,8 +412,7 @@ func (a *Account) GetNextPeerExpiration() (time.Duration, bool) {
 	}
 	var nextExpiry *time.Duration
 	for _, peer := range peersWithExpiry {
-		// consider only connected peers because others will require login on connecting to the management server
+		if peer.Status.LoginExpired {
-		if peer.Status.LoginExpired || !peer.Status.Connected {
 			continue
 		}
 		_, duration := peer.LoginExpired(a.Settings.PeerLoginExpiration)

management/server/user.go

@@ -1177,28 +1177,35 @@ func (am *DefaultAccountManager) expireAndUpdatePeers(ctx context.Context, accou
 	dnsDomain := am.networkMapController.GetDNSDomain(settings)
 
 	var peerIDs []string
-	for _, peer := range peers {
+	defer func() {
+		if len(peerIDs) == 0 {
+			return
+		}
+		// this will trigger peer disconnect from the management service
+		log.Debugf("Expiring %d peers for account %s", len(peerIDs), accountID)
+		am.networkMapController.DisconnectPeers(ctx, accountID, peerIDs)
+	}()
+	for _, candidate := range peers {
 		// nolint:staticcheck
-		ctx = context.WithValue(ctx, nbcontext.PeerIDKey, peer.Key)
+		peerCtx := context.WithValue(ctx, nbcontext.PeerIDKey, candidate.Key)
 
-		if peer.UserID == "" {
+		if candidate.UserID == "" {
 			// we do not want to expire peers that are added via setup key
 			continue
 		}
 
-		if peer.Status.LoginExpired {
+		peer, err := am.expirePeerIfStillDue(peerCtx, accountID, candidate.ID, settings, reason)
+		if err != nil {
+			return err
+		}
+		if peer == nil {
 			continue
 		}
 		peerIDs = append(peerIDs, peer.ID)
-		peer.MarkLoginExpired(true)
-
-		if err := am.Store.SavePeerStatus(ctx, accountID, peer.ID, *peer.Status); err != nil {
-			return err
-		}
 		meta := peer.EventMeta(dnsDomain)
 		meta["reason"] = string(reason)
 		am.StoreEvent(
-			ctx,
+			peerCtx,
 			peer.UserID, peer.ID, accountID,
 			activity.PeerLoginExpired, meta,
 		)
@@ -1215,15 +1222,53 @@ func (am *DefaultAccountManager) expireAndUpdatePeers(ctx context.Context, accou
 	if err != nil {
 		return fmt.Errorf("notify network map controller of peer update: %w", err)
 	}
-
-	if len(peerIDs) != 0 {
-		// this will trigger peer disconnect from the management service
-		log.Debugf("Expiring %d peers for account %s", len(peerIDs), accountID)
-		am.networkMapController.DisconnectPeers(ctx, accountID, peerIDs)
-	}
 	return nil
 }
 
+// expirePeerIfStillDue flags the peer as login-expired and returns its fresh copy, or nil
+// when it no longer qualifies. The candidate list is read without a lock, so a login that
+// landed in between would otherwise be overwritten with a stale expired status.
+func (am *DefaultAccountManager) expirePeerIfStillDue(ctx context.Context, accountID, peerID string, settings *types.Settings, reason peerExpirationReason) (*nbpeer.Peer, error) {
+	var expired *nbpeer.Peer
+	err := am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error {
+		peer, err := transaction.GetPeerByID(ctx, store.LockingStrengthUpdate, accountID, peerID)
+		if err != nil {
+			if s, ok := status.FromError(err); ok && s.Type() == status.NotFound {
+				return nil
+			}
+			return err
+		}
+		if peer.Status.LoginExpired || !peerExpirationDue(peer, settings, reason) {
+			return nil
+		}
+		peer.MarkLoginExpired(true)
+		if err := transaction.SavePeerStatus(ctx, accountID, peer.ID, *peer.Status); err != nil {
+			return err
+		}
+		expired = peer
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	return expired, nil
+}
+
+// peerExpirationDue re-evaluates a time-based expiry against the peer's current state.
+// Administrative reasons expire the peer unconditionally.
+func peerExpirationDue(peer *nbpeer.Peer, settings *types.Settings, reason peerExpirationReason) bool {
+	switch reason {
+	case peerExpirationSessionExpired:
+		expired, _ := peer.LoginExpired(settings.PeerLoginExpiration)
+		return settings.PeerLoginExpirationEnabled && expired
+	case peerExpirationInactivity:
+		expired, _ := peer.SessionExpired(settings.PeerInactivityExpiration)
+		return settings.PeerInactivityExpirationEnabled && expired
+	default:
+		return true
+	}
+}
+
 func (am *DefaultAccountManager) deleteUserFromIDP(ctx context.Context, targetUserID, accountID string) error {
 	if am.userDeleteFromIDPEnabled {
 		log.WithContext(ctx).Debugf("user %s deleted from IdP", targetUserID)

util/serviceurl.go

@@ -0,0 +1,69 @@
+package util
+
+import (
+	"net/url"
+	"strconv"
+	"strings"
+)
+
+// SameServiceURL reports whether two service URLs address the same endpoint.
+// One endpoint can be written several ways, and every spelling below reaches
+// the same server, so none of them is a divergence from another:
+//
+//	an implicit default port    https://mgmt.example.com   :443
+//	a zero-padded port          https://mgmt.example.com:0443
+//	a different host case       https://MGMT.example.com
+//	a trailing slash            https://mgmt.example.com/
+//
+// A path is otherwise part of the identity: https://mgmt.example.com and
+// https://mgmt.example.com/other are two endpoints.
+//
+// It lives here rather than next to any one caller because several of them
+// compare the same kind of URL — an MDM-enforced management URL against a
+// requested one, a stored profile URL against a command-line one — and every
+// copy of these rules that drifts turns an equivalent URL into a refused
+// request.
+func SameServiceURL(a, b *url.URL) bool {
+	if a == nil || b == nil {
+		return a == b
+	}
+
+	return strings.EqualFold(a.Hostname(), b.Hostname()) &&
+		strings.EqualFold(a.Scheme, b.Scheme) &&
+		ServiceURLPort(a) == ServiceURLPort(b) &&
+		strings.TrimSuffix(a.Path, "/") == strings.TrimSuffix(b.Path, "/")
+}
+
+// SameServiceURLStrings is SameServiceURL for unparsed input. Input that does
+// not parse falls back to string equality, which is the strictest thing left
+// to do with it.
+func SameServiceURLStrings(a, b string) bool {
+	ua, errA := url.ParseRequestURI(a)
+	ub, errB := url.ParseRequestURI(b)
+	if errA != nil || errB != nil {
+		return a == b
+	}
+
+	return SameServiceURL(ua, ub)
+}
+
+// ServiceURLPort is the port a URL addresses: the one it carries, normalized
+// numerically so ":0443" and ":443" are one port, or the scheme's default.
+func ServiceURLPort(u *url.URL) string {
+	port := u.Port()
+	if port == "" {
+		switch strings.ToLower(u.Scheme) {
+		case "https":
+			return "443"
+		case "http":
+			return "80"
+		default:
+			return ""
+		}
+	}
+
+	if n, err := strconv.Atoi(port); err == nil {
+		return strconv.Itoa(n)
+	}
+	return port
+}

util/serviceurl_test.go

@@ -0,0 +1,73 @@
+package util
+
+import (
+	"net/url"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestSameServiceURLSpellings(t *testing.T) {
+	tests := []struct {
+		a, b string
+		want bool
+	}{
+		// One endpoint, written several ways.
+		{a: "https://mgmt.example.com", b: "https://mgmt.example.com:443", want: true},
+		{a: "https://mgmt.example.com", b: "https://mgmt.example.com/", want: true},
+		{a: "https://mgmt.example.com/", b: "https://mgmt.example.com:443/", want: true},
+		{a: "https://MGMT.example.com", b: "https://mgmt.example.com", want: true},
+		{a: "https://mgmt.example.com:0443", b: "https://mgmt.example.com:443", want: true},
+		{a: "http://mgmt.example.com", b: "http://mgmt.example.com:80", want: true},
+		{a: "HTTPS://mgmt.example.com", b: "https://mgmt.example.com", want: true},
+
+		// Different endpoints.
+		{a: "https://mgmt.example.com", b: "http://mgmt.example.com", want: false},
+		{a: "https://mgmt.example.com", b: "https://mgmt.example.com:8443", want: false},
+		{a: "https://mgmt.example.com", b: "https://other.example.com", want: false},
+		{a: "https://mgmt.example.com", b: "https://mgmt.example.com/other", want: false},
+
+		// Unparseable input falls back to string equality.
+		{a: "mgmt.example.com", b: "mgmt.example.com", want: true},
+		{a: "mgmt.example.com", b: "https://mgmt.example.com", want: false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.a+" vs "+tt.b, func(t *testing.T) {
+			assert.Equal(t, tt.want, SameServiceURLStrings(tt.a, tt.b))
+			assert.Equal(t, tt.want, SameServiceURLStrings(tt.b, tt.a), "the comparison must be symmetric")
+		})
+	}
+}
+
+// The parsed form is the primitive the string form delegates to, so it must
+// answer the same for a spelling that only the parser can tell apart.
+func TestSameServiceURLParsed(t *testing.T) {
+	parse := func(raw string) *url.URL {
+		t.Helper()
+		u, err := url.ParseRequestURI(raw)
+		require.NoError(t, err)
+		return u
+	}
+
+	assert.True(t, SameServiceURL(parse("https://mgmt.example.com:0443/"), parse("https://MGMT.example.com")))
+	assert.False(t, SameServiceURL(parse("https://mgmt.example.com"), parse("https://mgmt.example.com:8443")))
+
+	assert.True(t, SameServiceURL(nil, nil), "two absent URLs are the same absence")
+	assert.False(t, SameServiceURL(nil, parse("https://mgmt.example.com")))
+}
+
+func TestServiceURLPort(t *testing.T) {
+	parse := func(raw string) *url.URL {
+		t.Helper()
+		u, err := url.ParseRequestURI(raw)
+		require.NoError(t, err)
+		return u
+	}
+
+	assert.Equal(t, "443", ServiceURLPort(parse("https://mgmt.example.com")))
+	assert.Equal(t, "80", ServiceURLPort(parse("http://mgmt.example.com")))
+	assert.Equal(t, "443", ServiceURLPort(parse("https://mgmt.example.com:0443")))
+	assert.Equal(t, "8443", ServiceURLPort(parse("https://mgmt.example.com:8443")))
+}

util/wsproxy/server/proxy.go

@@ -1,12 +1,8 @@
 package server
 
 import (
-	"context"
-	"io"
-	"net"
 	"net/http"
-	"sync"
+	"sync/atomic"
-	"time"
 
 	"github.com/coder/websocket"
 	log "github.com/sirupsen/logrus"
@@ -15,11 +11,6 @@ import (
 	"github.com/netbirdio/netbird/util/wsproxy"
 )
 
-const (
-	bufferSize = 32 * 1024
-	ioTimeout  = 5 * time.Second
-)
-
 // Config contains the configuration for the WebSocket proxy.
 type Config struct {
 	Handler         http.Handler
@@ -53,14 +44,23 @@ func New(handler http.Handler, opts ...Option) *Proxy {
 
 // Handler returns an http.Handler that proxies WebSocket connections to the local gRPC server.
 func (p *Proxy) Handler() http.Handler {
-	return http.HandlerFunc(p.handleWebSocket)
+	return &proxyHandler{
+		metrics: p.config.MetricsRecorder,
+		handler: p.config.Handler,
+	}
 }
 
-func (p *Proxy) handleWebSocket(w http.ResponseWriter, r *http.Request) {
+type proxyHandler struct {
+	metrics MetricsRecorder
+	handler http.Handler
+	conn    atomic.Pointer[wsConnAdapter]
+}
+
+func (ph *proxyHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	ctx := r.Context()
 
-	p.metrics.RecordConnection(ctx)
+	ph.metrics.RecordConnection(ctx)
-	defer p.metrics.RecordDisconnection(ctx)
+	defer ph.metrics.RecordDisconnection(ctx)
 
 	log.Debugf("WebSocket proxy handling connection from %s, forwarding to internal gRPC handler", r.RemoteAddr)
 	acceptOptions := &websocket.AcceptOptions{
@@ -69,121 +69,44 @@ func (p *Proxy) handleWebSocket(w http.ResponseWriter, r *http.Request) {
 
 	wsConn, err := websocket.Accept(w, r, acceptOptions)
 	if err != nil {
-		p.metrics.RecordError(ctx, "websocket_accept_failed")
+		ph.metrics.RecordError(ctx, "websocket_accept_failed")
 		log.Errorf("WebSocket upgrade failed from %s: %v", r.RemoteAddr, err)
 		return
 	}
-	defer func() {
+	serverConn := (&wsConnAdapter{
-		_ = wsConn.Close(websocket.StatusNormalClosure, "")
+		ctx:        ctx,
-	}()
+		conn:       wsConn,
+		metrics:    ph.metrics,
+		clientAddr: r.RemoteAddr,
+	})
 
-	clientConn, serverConn := net.Pipe()
 	defer func() {
-		_ = clientConn.Close()
 		_ = serverConn.Close()
 	}()
 
+	ph.conn.Store(serverConn) // used in tests only
+
 	log.Debugf("WebSocket proxy established: %s -> gRPC handler", r.RemoteAddr)
 
-	go func() {
+	(&http2.Server{
-		(&http2.Server{}).ServeConn(serverConn, &http2.ServeConnOpts{
+		// TODO (dmitri) we should limit the number of concurrent streams per connection (peer)
-			Context: ctx,
+		// and idle timeouts
-			Handler: p.config.Handler,
+		// MaxConcurrentStreams: 20,
-		})
+		// IdleTimeout: 10 * time.Second,
-	}()
+	}).ServeConn(serverConn, &http2.ServeConnOpts{
+		Context:    ctx,
+		Handler:    ph.handler,
+		BaseConfig: &http.Server{
+			// b/c we are wrapping a ws connection, read and write connection deadlines normally set
+			// via ReadTimeout and WriteTimeout http.Server fields aren't available to us. The ws
+			// library doesn't expose connection deadline timer config, and we ignore these calls in "wsConnAdapter".
+			//
+			// Another issue is that Server.ServeConn() call bypasses setting of connection deadlines altogether,
+			// ReadTimeout and Writetimeout set here would only apply to h2 streams, i.e. after a HEADERS frame
+			// arrival and processing, turning ReadTimeout into a request body read deadline, and WriteTimeout into
+			// a response deadline (the latter not useful for streaming requests).
+		},
+	})
 
-	p.proxyData(ctx, wsConn, clientConn, r.RemoteAddr)
+	log.Debugf("WebSocket proxy closing: %s -> gRPC handler", r.RemoteAddr)
-}
-
-func (p *Proxy) proxyData(ctx context.Context, wsConn *websocket.Conn, pipeConn net.Conn, clientAddr string) {
-	proxyCtx, cancel := context.WithCancel(ctx)
-	defer cancel()
-
-	var wg sync.WaitGroup
-	wg.Add(2)
-
-	go p.wsToPipe(proxyCtx, cancel, &wg, wsConn, pipeConn, clientAddr)
-	go p.pipeToWS(proxyCtx, cancel, &wg, wsConn, pipeConn, clientAddr)
-
-	wg.Wait()
-}
-
-func (p *Proxy) wsToPipe(ctx context.Context, cancel context.CancelFunc, wg *sync.WaitGroup, wsConn *websocket.Conn, pipeConn net.Conn, clientAddr string) {
-	defer wg.Done()
-	defer cancel()
-
-	for {
-		msgType, data, err := wsConn.Read(ctx)
-		if err != nil {
-			switch {
-			case ctx.Err() != nil:
-				log.Debugf("WebSocket from %s terminating due to context cancellation", clientAddr)
-			case websocket.CloseStatus(err) != -1:
-				log.Debugf("WebSocket from %s disconnected", clientAddr)
-			default:
-				p.metrics.RecordError(ctx, "websocket_read_error")
-				log.Debugf("WebSocket read error from %s: %v", clientAddr, err)
-			}
-			return
-		}
-
-		if msgType != websocket.MessageBinary {
-			log.Warnf("Unexpected WebSocket message type from %s: %v", clientAddr, msgType)
-			continue
-		}
-
-		if ctx.Err() != nil {
-			log.Tracef("wsToPipe goroutine terminating due to context cancellation before pipe write")
-			return
-		}
-
-		if err := pipeConn.SetWriteDeadline(time.Now().Add(ioTimeout)); err != nil {
-			log.Debugf("Failed to set pipe write deadline: %v", err)
-		}
-
-		n, err := pipeConn.Write(data)
-		if err != nil {
-			p.metrics.RecordError(ctx, "pipe_write_error")
-			log.Warnf("Pipe write error for %s: %v", clientAddr, err)
-			return
-		}
-
-		p.metrics.RecordBytesTransferred(ctx, "ws_to_grpc", int64(n))
-	}
-}
-
-func (p *Proxy) pipeToWS(ctx context.Context, cancel context.CancelFunc, wg *sync.WaitGroup, wsConn *websocket.Conn, pipeConn net.Conn, clientAddr string) {
-	defer wg.Done()
-	defer cancel()
-
-	buf := make([]byte, bufferSize)
-	for {
-		n, err := pipeConn.Read(buf)
-		if err != nil {
-			if ctx.Err() != nil {
-				log.Tracef("pipeToWS goroutine terminating due to context cancellation")
-				return
-			}
-
-			if err != io.EOF {
-				log.Debugf("Pipe read error for %s: %v", clientAddr, err)
-			}
-			return
-		}
-
-		if ctx.Err() != nil {
-			log.Tracef("pipeToWS goroutine terminating due to context cancellation before WebSocket write")
-			return
-		}
-
-		if n > 0 {
-			if err := wsConn.Write(ctx, websocket.MessageBinary, buf[:n]); err != nil {
-				p.metrics.RecordError(ctx, "websocket_write_error")
-				log.Warnf("WebSocket write error for %s: %v", clientAddr, err)
-				return
-			}
-
-			p.metrics.RecordBytesTransferred(ctx, "grpc_to_ws", int64(n))
-		}
-	}
 }

util/wsproxy/server/ws_conn_adapter.go

@@ -0,0 +1,126 @@
+package server
+
+import (
+	"context"
+	"net"
+	"sync/atomic"
+	"time"
+
+	"github.com/coder/websocket"
+	log "github.com/sirupsen/logrus"
+)
+
+type wsConnAdapter struct {
+	prefix       string
+	ctx          context.Context
+	conn         *websocket.Conn
+	metrics      MetricsRecorder
+	clientAddr   string
+	closed       atomic.Bool
+	bufferedRead []byte
+}
+
+var _ net.Conn = &wsConnAdapter{}
+
+type wsAddr struct{ prefix string }
+
+func (wa wsAddr) Network() string { return wa.prefix + "ws-proxy" }
+func (wa wsAddr) String() string  { return wa.prefix + "ws-proxy" }
+
+func (ws *wsConnAdapter) Read(b []byte) (int, error) {
+	if len(ws.bufferedRead) > 0 {
+		return ws.readFromBuffer(b)
+	}
+
+	msgType, data, err := ws.conn.Read(ws.ctx)
+	if err != nil {
+		switch {
+		case ws.ctx.Err() != nil:
+			log.Debugf("WebSocket from %s terminating due to context cancellation", ws.clientAddr)
+		case websocket.CloseStatus(err) != -1:
+			log.Debugf("WebSocket from %s disconnected", ws.clientAddr)
+		default:
+			ws.recordError(ws.ctx, "websocket_read_error")
+			log.Debugf("WebSocket read error from %s: %v", ws.clientAddr, err)
+		}
+		return copy(b, data), err
+	}
+	if msgType != websocket.MessageBinary {
+		log.Warnf("Unexpected WebSocket message type from %s: %v", ws.clientAddr, msgType)
+		return 0, nil
+	}
+
+	ws.bufferedRead = data
+	return ws.readFromBuffer(b)
+}
+
+func (ws *wsConnAdapter) readFromBuffer(b []byte) (int, error) {
+	n := copy(b, ws.bufferedRead)
+
+	ws.recordBytesTransferred(ws.ctx, "ws_to_grpc", n)
+	if n == len(ws.bufferedRead) {
+		ws.bufferedRead = nil
+		return n, nil
+	} else {
+		ws.bufferedRead = ws.bufferedRead[n:]
+	}
+	return n, nil
+}
+
+func (ws *wsConnAdapter) Write(b []byte) (int, error) {
+	maybeErr := ws.ctx.Err()
+
+	n := len(b)
+	if n == 0 {
+		return n, maybeErr
+	}
+	if maybeErr != nil {
+		return 0, maybeErr
+	}
+	if err := ws.conn.Write(ws.ctx, websocket.MessageBinary, b[:n]); err != nil {
+		ws.recordError(ws.ctx, "websocket_write_error")
+		log.Warnf("WebSocket write error for %s: %v", ws.clientAddr, err)
+		return 0, err // we don't know how many bytes have been written
+	}
+
+	ws.recordBytesTransferred(ws.ctx, "grpc_to_ws", n)
+	return n, nil
+}
+
+func (ws *wsConnAdapter) Close() error {
+	ws.closed.Store(true)
+	return ws.conn.Close(websocket.StatusNormalClosure, "")
+}
+
+func (ws *wsConnAdapter) LocalAddr() net.Addr  { return wsAddr{ws.prefix} }
+func (ws *wsConnAdapter) RemoteAddr() net.Addr { return wsAddr{ws.prefix} }
+
+func (ws *wsConnAdapter) SetDeadline(t time.Time) error {
+	return nil
+}
+
+func (ws *wsConnAdapter) SetReadDeadline(t time.Time) error {
+	return nil
+}
+
+func (ws *wsConnAdapter) SetWriteDeadline(t time.Time) error {
+	return nil
+}
+
+func (ws *wsConnAdapter) recordError(ctx context.Context, errorType string) {
+	if ws.metrics == nil {
+		return
+	}
+	ws.metrics.RecordError(ctx, errorType)
+}
+
+func (ws *wsConnAdapter) recordBytesTransferred(ctx context.Context, direction string, bytes int) {
+	if ws.metrics == nil {
+		return
+	}
+	ws.metrics.RecordBytesTransferred(ctx, direction, int64(bytes))
+}
+
+func (ws *wsConnAdapter) IsClosed() bool {
+	return ws.closed.Load()
+}

util/wsproxy/server/ws_conn_adapter_test.go

@@ -0,0 +1,204 @@
+package server
+
+import (
+	"bytes"
+	"context"
+	"crypto/tls"
+	"io"
+	"math/rand/v2"
+	"net"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/coder/websocket"
+	"github.com/stretchr/testify/assert"
+	"golang.org/x/net/http2"
+	"golang.org/x/net/http2/hpack"
+)
+
+func TestAdapterHandlingConnectionClosures(t *testing.T) {
+	var cases = []struct {
+		description string
+		casenum     int
+	}{
+		{"client-side ws connection is closed", 0},
+		{"server-side ws connection is closed", 1},
+		{"client-side context is cancelled", 2},
+		{"server-side context is cancelled", 3},
+	}
+
+	for _, c := range cases {
+		t.Run(c.description, func(t *testing.T) {
+			serversock := filepath.Join(os.TempDir(), "http-server-"+strconv.FormatInt(rand.Int64(), 10)+".sock")
+			t.Cleanup(func() { os.Remove(serversock) })
+
+			l, err := net.Listen("unix", serversock)
+			assert.NoError(t, err)
+
+			proxy := New(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				buf, _ := io.ReadAll(r.Body)
+				defer r.Body.Close()
+				w.Write([]byte("echo: " + string(buf))) //nolint:errcheck
+			}))
+
+			handler, ok := proxy.Handler().(*proxyHandler)
+			assert.True(t, ok)
+
+			protocols := new(http.Protocols)
+			protocols.SetHTTP1(true)
+			protocols.SetUnencryptedHTTP2(true)
+			httpServer := http.Server{
+				Handler: handler,
+			}
+			go httpServer.Serve(l) //nolint:errcheck
+			t.Cleanup(func() { httpServer.Close() })
+
+			clientconn, _, err := websocket.Dial(context.Background(), "http://whatever", //nolint:bodyclose
+				&websocket.DialOptions{HTTPClient: &http.Client{
+					Transport: &http.Transport{
+						DialContext: func(_ context.Context, _, _ string) (net.Conn, error) {
+							return net.Dial("unix", serversock)
+						},
+					}}})
+			assert.NoError(t, err)
+
+			clientCtx, cancel := context.WithCancel(context.Background()) //nolint:govet
+			h2client := &http.Client{
+				Transport: &http2.Transport{
+					AllowHTTP: true,
+					DialTLSContext: func(_ context.Context, _, _ string, _ *tls.Config) (net.Conn, error) {
+						return &wsConnAdapter{
+							prefix: "test-client",
+							ctx:    clientCtx,
+							conn:   clientconn,
+						}, nil
+					},
+				}}
+
+			resp, err := h2client.Post("http://whatever", "text/html", strings.NewReader("g'day"))
+			assert.NoError(t, err)
+
+			body, err := io.ReadAll(resp.Body)
+			defer resp.Body.Close()
+
+			assert.NoError(t, err)
+			assert.Equal(t, "echo: g'day", string(body))
+
+			switch c.casenum {
+			case 0:
+				clientconn.Close(websocket.StatusNormalClosure, "")
+			case 1:
+				handler.conn.Load().Close()
+			case 2:
+				cancel()
+			case 3:
+				resp.Body.Close()
+				h2client.CloseIdleConnections()
+			}
+
+			assert.EventuallyWithT(t, func(c *assert.CollectT) {
+				assert.True(c, handler.conn.Load().IsClosed())
+			}, 3*time.Second, 100*time.Millisecond)
+		}) //nolint:govet
+	}
+}
+
+func TestAdapterHandlingHttpConnection_NoHeadersSent(t *testing.T) {
+	t.Skip("currently disabled as it requires idle timeout to be set")
+
+	serversock := filepath.Join(os.TempDir(), "http-server-"+strconv.FormatInt(rand.Int64(), 10)+".sock")
+	defer os.Remove(serversock)
+
+	l, err := net.Listen("unix", serversock)
+	assert.NoError(t, err)
+
+	proxy := New(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		buf, _ := io.ReadAll(r.Body)
+		defer r.Body.Close()                    //nolint:errcheck
+		w.Write([]byte("echo: " + string(buf))) //nolint:errcheck
+	}))
+
+	handler, ok := proxy.Handler().(*proxyHandler)
+	assert.True(t, ok)
+
+	protocols := new(http.Protocols)
+	protocols.SetHTTP1(true)
+	protocols.SetUnencryptedHTTP2(true)
+	httpServer := http.Server{
+		Handler: handler,
+	}
+	go httpServer.Serve(l) //nolint:errcheck
+
+	clientconn, _, err := websocket.Dial(context.Background(), "http://whatever", //nolint:bodyclose
+		&websocket.DialOptions{HTTPClient: &http.Client{
+			Transport: &http.Transport{
+				DialContext: func(_ context.Context, _, _ string) (net.Conn, error) {
+					return net.Dial("unix", serversock)
+				},
+			}}})
+	assert.NoError(t, err)
+
+	h2client := &http.Client{
+		Transport: &http2.Transport{
+			AllowHTTP: true,
+			DialTLSContext: func(_ context.Context, _, _ string, _ *tls.Config) (net.Conn, error) {
+				return &h2ConnectionSnooper{wrappedConn: &wsConnAdapter{
+					prefix: "test-client",
+					ctx:    context.Background(),
+					conn:   clientconn,
+				}, shouldDropFrame: func(f http2.FrameType) bool { return f == http2.FrameHeaders || f == http2.FrameData }}, nil
+			},
+		}}
+
+	_, err = h2client.Post("http://whatever", "text/html", strings.NewReader("g'day"))
+	assert.Error(t, err)
+
+	assert.EventuallyWithT(t, func(c *assert.CollectT) {
+		assert.True(c, handler.conn.Load().IsClosed())
+	}, 3*time.Second, 100*time.Millisecond)
+}
+
+type h2ConnectionSnooper struct {
+	wrappedConn     net.Conn
+	shouldDropFrame func(f http2.FrameType) bool
+}
+
+func (hs *h2ConnectionSnooper) Read(b []byte) (n int, err error) {
+	return hs.wrappedConn.Read(b)
+}
+
+func (hs *h2ConnectionSnooper) Write(b []byte) (n int, err error) {
+	fr := http2.NewFramer(nil, bytes.NewReader(b))
+	fr.ReadMetaHeaders = hpack.NewDecoder(0, nil)
+	f, err := fr.ReadFrame()
+	if err != nil {
+		return hs.wrappedConn.Write(b)
+	}
+
+	if hs.shouldDropFrame != nil && hs.shouldDropFrame(f.Header().Type) {
+		return len(b), nil
+	}
+
+	return hs.wrappedConn.Write(b)
+}
+
+func (hs *h2ConnectionSnooper) Close() error { return hs.wrappedConn.Close() }
+
+func (hs *h2ConnectionSnooper) LocalAddr() net.Addr { return hs.wrappedConn.LocalAddr() }
+
+func (hs *h2ConnectionSnooper) RemoteAddr() net.Addr { return hs.wrappedConn.RemoteAddr() }
+
+func (hs *h2ConnectionSnooper) SetDeadline(t time.Time) error { return hs.wrappedConn.SetDeadline(t) }
+
+func (hs *h2ConnectionSnooper) SetReadDeadline(t time.Time) error {
+	return hs.wrappedConn.SetReadDeadline(t)
+}
+
+func (hs *h2ConnectionSnooper) SetWriteDeadline(t time.Time) error {
+	return hs.wrappedConn.SetWriteDeadline(t)
+}