mirror of
https://github.com/netbirdio/netbird.git
synced 2026-08-08 02:55:28 -04:00
Compare commits
7 Commits
v0.65.1
...
test/perft
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
920fe73096 | ||
|
|
0146e39714 | ||
|
|
baed6e46ec | ||
|
|
0d1ffba75f | ||
|
|
1024d45698 | ||
|
|
e5d4947d60 | ||
|
|
cb9b39b950 |
@@ -1,10 +1,19 @@
|
||||
package android
|
||||
|
||||
import "github.com/netbirdio/netbird/client/internal/peer"
|
||||
import (
|
||||
"github.com/netbirdio/netbird/client/internal/lazyconn"
|
||||
"github.com/netbirdio/netbird/client/internal/peer"
|
||||
)
|
||||
|
||||
var (
|
||||
// EnvKeyNBForceRelay Exported for Android java client
|
||||
// EnvKeyNBForceRelay Exported for Android java client to force relay connections
|
||||
EnvKeyNBForceRelay = peer.EnvKeyNBForceRelay
|
||||
|
||||
// EnvKeyNBLazyConn Exported for Android java client to configure lazy connection
|
||||
EnvKeyNBLazyConn = lazyconn.EnvEnableLazyConn
|
||||
|
||||
// EnvKeyNBInactivityThreshold Exported for Android java client to configure connection inactivity threshold
|
||||
EnvKeyNBInactivityThreshold = lazyconn.EnvInactivityThreshold
|
||||
)
|
||||
|
||||
// EnvList wraps a Go map for export to Java
|
||||
|
||||
@@ -42,6 +42,8 @@ const (
|
||||
dnsPolicyConfigConfigOptionsKey = "ConfigOptions"
|
||||
dnsPolicyConfigConfigOptionsValue = 0x8
|
||||
|
||||
nrptMaxDomainsPerRule = 50
|
||||
|
||||
interfaceConfigPath = `SYSTEM\CurrentControlSet\Services\Tcpip\Parameters\Interfaces`
|
||||
interfaceConfigNameServerKey = "NameServer"
|
||||
interfaceConfigSearchListKey = "SearchList"
|
||||
@@ -198,10 +200,11 @@ func (r *registryConfigurator) applyDNSConfig(config HostDNSConfig, stateManager
|
||||
|
||||
if len(matchDomains) != 0 {
|
||||
count, err := r.addDNSMatchPolicy(matchDomains, config.ServerIP)
|
||||
// Update count even on error to ensure cleanup covers partially created rules
|
||||
r.nrptEntryCount = count
|
||||
if err != nil {
|
||||
return fmt.Errorf("add dns match policy: %w", err)
|
||||
}
|
||||
r.nrptEntryCount = count
|
||||
} else {
|
||||
r.nrptEntryCount = 0
|
||||
}
|
||||
@@ -239,23 +242,33 @@ func (r *registryConfigurator) addDNSSetupForAll(ip netip.Addr) error {
|
||||
func (r *registryConfigurator) addDNSMatchPolicy(domains []string, ip netip.Addr) (int, error) {
|
||||
// if the gpo key is present, we need to put our DNS settings there, otherwise our config might be ignored
|
||||
// see https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-gpnrpt/8cc31cb9-20cb-4140-9e85-3e08703b4745
|
||||
for i, domain := range domains {
|
||||
localPath := fmt.Sprintf("%s-%d", dnsPolicyConfigMatchPath, i)
|
||||
gpoPath := fmt.Sprintf("%s-%d", gpoDnsPolicyConfigMatchPath, i)
|
||||
|
||||
singleDomain := []string{domain}
|
||||
// We need to batch domains into chunks and create one NRPT rule per batch.
|
||||
ruleIndex := 0
|
||||
for i := 0; i < len(domains); i += nrptMaxDomainsPerRule {
|
||||
end := i + nrptMaxDomainsPerRule
|
||||
if end > len(domains) {
|
||||
end = len(domains)
|
||||
}
|
||||
batchDomains := domains[i:end]
|
||||
|
||||
if err := r.configureDNSPolicy(localPath, singleDomain, ip); err != nil {
|
||||
return i, fmt.Errorf("configure DNS Local policy for domain %s: %w", domain, err)
|
||||
localPath := fmt.Sprintf("%s-%d", dnsPolicyConfigMatchPath, ruleIndex)
|
||||
gpoPath := fmt.Sprintf("%s-%d", gpoDnsPolicyConfigMatchPath, ruleIndex)
|
||||
|
||||
if err := r.configureDNSPolicy(localPath, batchDomains, ip); err != nil {
|
||||
return ruleIndex, fmt.Errorf("configure DNS Local policy for rule %d: %w", ruleIndex, err)
|
||||
}
|
||||
|
||||
// Increment immediately so the caller's cleanup path knows about this rule
|
||||
ruleIndex++
|
||||
|
||||
if r.gpo {
|
||||
if err := r.configureDNSPolicy(gpoPath, singleDomain, ip); err != nil {
|
||||
return i, fmt.Errorf("configure gpo DNS policy: %w", err)
|
||||
if err := r.configureDNSPolicy(gpoPath, batchDomains, ip); err != nil {
|
||||
return ruleIndex, fmt.Errorf("configure gpo DNS policy for rule %d: %w", ruleIndex-1, err)
|
||||
}
|
||||
}
|
||||
|
||||
log.Debugf("added NRPT entry for domain: %s", domain)
|
||||
log.Debugf("added NRPT rule %d with %d domains", ruleIndex-1, len(batchDomains))
|
||||
}
|
||||
|
||||
if r.gpo {
|
||||
@@ -264,8 +277,8 @@ func (r *registryConfigurator) addDNSMatchPolicy(domains []string, ip netip.Addr
|
||||
}
|
||||
}
|
||||
|
||||
log.Infof("added %d separate NRPT entries. Domain list: %s", len(domains), domains)
|
||||
return len(domains), nil
|
||||
log.Infof("added %d NRPT rules for %d domains. Domain list: %v", ruleIndex, len(domains), domains)
|
||||
return ruleIndex, nil
|
||||
}
|
||||
|
||||
func (r *registryConfigurator) configureDNSPolicy(policyPath string, domains []string, ip netip.Addr) error {
|
||||
|
||||
@@ -12,6 +12,7 @@ import (
|
||||
|
||||
// TestNRPTEntriesCleanupOnConfigChange tests that old NRPT entries are properly cleaned up
|
||||
// when the number of match domains decreases between configuration changes.
|
||||
// With batching enabled (50 domains per rule), we need enough domains to create multiple rules.
|
||||
func TestNRPTEntriesCleanupOnConfigChange(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping registry integration test in short mode")
|
||||
@@ -37,51 +38,60 @@ func TestNRPTEntriesCleanupOnConfigChange(t *testing.T) {
|
||||
gpo: false,
|
||||
}
|
||||
|
||||
config5 := HostDNSConfig{
|
||||
ServerIP: testIP,
|
||||
Domains: []DomainConfig{
|
||||
{Domain: "domain1.com", MatchOnly: true},
|
||||
{Domain: "domain2.com", MatchOnly: true},
|
||||
{Domain: "domain3.com", MatchOnly: true},
|
||||
{Domain: "domain4.com", MatchOnly: true},
|
||||
{Domain: "domain5.com", MatchOnly: true},
|
||||
},
|
||||
// Create 125 domains which will result in 3 NRPT rules (50+50+25)
|
||||
domains125 := make([]DomainConfig, 125)
|
||||
for i := 0; i < 125; i++ {
|
||||
domains125[i] = DomainConfig{
|
||||
Domain: fmt.Sprintf("domain%d.com", i+1),
|
||||
MatchOnly: true,
|
||||
}
|
||||
}
|
||||
|
||||
err = cfg.applyDNSConfig(config5, nil)
|
||||
config125 := HostDNSConfig{
|
||||
ServerIP: testIP,
|
||||
Domains: domains125,
|
||||
}
|
||||
|
||||
err = cfg.applyDNSConfig(config125, nil)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Verify all 5 entries exist
|
||||
for i := 0; i < 5; i++ {
|
||||
// Verify 3 NRPT rules exist
|
||||
assert.Equal(t, 3, cfg.nrptEntryCount, "Should create 3 NRPT rules for 125 domains")
|
||||
for i := 0; i < 3; i++ {
|
||||
exists, err := registryKeyExists(fmt.Sprintf("%s-%d", dnsPolicyConfigMatchPath, i))
|
||||
require.NoError(t, err)
|
||||
assert.True(t, exists, "Entry %d should exist after first config", i)
|
||||
assert.True(t, exists, "NRPT rule %d should exist after first config", i)
|
||||
}
|
||||
|
||||
config2 := HostDNSConfig{
|
||||
// Reduce to 75 domains which will result in 2 NRPT rules (50+25)
|
||||
domains75 := make([]DomainConfig, 75)
|
||||
for i := 0; i < 75; i++ {
|
||||
domains75[i] = DomainConfig{
|
||||
Domain: fmt.Sprintf("domain%d.com", i+1),
|
||||
MatchOnly: true,
|
||||
}
|
||||
}
|
||||
|
||||
config75 := HostDNSConfig{
|
||||
ServerIP: testIP,
|
||||
Domains: []DomainConfig{
|
||||
{Domain: "domain1.com", MatchOnly: true},
|
||||
{Domain: "domain2.com", MatchOnly: true},
|
||||
},
|
||||
Domains: domains75,
|
||||
}
|
||||
|
||||
err = cfg.applyDNSConfig(config2, nil)
|
||||
err = cfg.applyDNSConfig(config75, nil)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Verify first 2 entries exist
|
||||
// Verify first 2 NRPT rules exist
|
||||
assert.Equal(t, 2, cfg.nrptEntryCount, "Should create 2 NRPT rules for 75 domains")
|
||||
for i := 0; i < 2; i++ {
|
||||
exists, err := registryKeyExists(fmt.Sprintf("%s-%d", dnsPolicyConfigMatchPath, i))
|
||||
require.NoError(t, err)
|
||||
assert.True(t, exists, "Entry %d should exist after second config", i)
|
||||
assert.True(t, exists, "NRPT rule %d should exist after second config", i)
|
||||
}
|
||||
|
||||
// Verify entries 2-4 are cleaned up
|
||||
for i := 2; i < 5; i++ {
|
||||
exists, err := registryKeyExists(fmt.Sprintf("%s-%d", dnsPolicyConfigMatchPath, i))
|
||||
require.NoError(t, err)
|
||||
assert.False(t, exists, "Entry %d should NOT exist after reducing to 2 domains", i)
|
||||
}
|
||||
// Verify rule 2 is cleaned up
|
||||
exists, err := registryKeyExists(fmt.Sprintf("%s-%d", dnsPolicyConfigMatchPath, 2))
|
||||
require.NoError(t, err)
|
||||
assert.False(t, exists, "NRPT rule 2 should NOT exist after reducing to 75 domains")
|
||||
}
|
||||
|
||||
func registryKeyExists(path string) (bool, error) {
|
||||
@@ -97,6 +107,106 @@ func registryKeyExists(path string) (bool, error) {
|
||||
}
|
||||
|
||||
func cleanupRegistryKeys(*testing.T) {
|
||||
cfg := ®istryConfigurator{nrptEntryCount: 10}
|
||||
// Clean up more entries to account for batching tests with many domains
|
||||
cfg := ®istryConfigurator{nrptEntryCount: 20}
|
||||
_ = cfg.removeDNSMatchPolicies()
|
||||
}
|
||||
|
||||
// TestNRPTDomainBatching verifies that domains are correctly batched into NRPT rules.
|
||||
func TestNRPTDomainBatching(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping registry integration test in short mode")
|
||||
}
|
||||
|
||||
defer cleanupRegistryKeys(t)
|
||||
cleanupRegistryKeys(t)
|
||||
|
||||
testIP := netip.MustParseAddr("100.64.0.1")
|
||||
|
||||
// Create a test interface registry key so updateSearchDomains doesn't fail
|
||||
testGUID := "{12345678-1234-1234-1234-123456789ABC}"
|
||||
interfacePath := `SYSTEM\CurrentControlSet\Services\Tcpip\Parameters\Interfaces\` + testGUID
|
||||
testKey, _, err := registry.CreateKey(registry.LOCAL_MACHINE, interfacePath, registry.SET_VALUE)
|
||||
require.NoError(t, err, "Should create test interface registry key")
|
||||
testKey.Close()
|
||||
defer func() {
|
||||
_ = registry.DeleteKey(registry.LOCAL_MACHINE, interfacePath)
|
||||
}()
|
||||
|
||||
cfg := ®istryConfigurator{
|
||||
guid: testGUID,
|
||||
gpo: false,
|
||||
}
|
||||
|
||||
testCases := []struct {
|
||||
name string
|
||||
domainCount int
|
||||
expectedRuleCount int
|
||||
}{
|
||||
{
|
||||
name: "Less than 50 domains (single rule)",
|
||||
domainCount: 30,
|
||||
expectedRuleCount: 1,
|
||||
},
|
||||
{
|
||||
name: "Exactly 50 domains (single rule)",
|
||||
domainCount: 50,
|
||||
expectedRuleCount: 1,
|
||||
},
|
||||
{
|
||||
name: "51 domains (two rules)",
|
||||
domainCount: 51,
|
||||
expectedRuleCount: 2,
|
||||
},
|
||||
{
|
||||
name: "100 domains (two rules)",
|
||||
domainCount: 100,
|
||||
expectedRuleCount: 2,
|
||||
},
|
||||
{
|
||||
name: "125 domains (three rules: 50+50+25)",
|
||||
domainCount: 125,
|
||||
expectedRuleCount: 3,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range testCases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
// Clean up before each subtest
|
||||
cleanupRegistryKeys(t)
|
||||
|
||||
// Generate domains
|
||||
domains := make([]DomainConfig, tc.domainCount)
|
||||
for i := 0; i < tc.domainCount; i++ {
|
||||
domains[i] = DomainConfig{
|
||||
Domain: fmt.Sprintf("domain%d.com", i+1),
|
||||
MatchOnly: true,
|
||||
}
|
||||
}
|
||||
|
||||
config := HostDNSConfig{
|
||||
ServerIP: testIP,
|
||||
Domains: domains,
|
||||
}
|
||||
|
||||
err := cfg.applyDNSConfig(config, nil)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Verify that exactly expectedRuleCount rules were created
|
||||
assert.Equal(t, tc.expectedRuleCount, cfg.nrptEntryCount,
|
||||
"Should create %d NRPT rules for %d domains", tc.expectedRuleCount, tc.domainCount)
|
||||
|
||||
// Verify all expected rules exist
|
||||
for i := 0; i < tc.expectedRuleCount; i++ {
|
||||
exists, err := registryKeyExists(fmt.Sprintf("%s-%d", dnsPolicyConfigMatchPath, i))
|
||||
require.NoError(t, err)
|
||||
assert.True(t, exists, "NRPT rule %d should exist", i)
|
||||
}
|
||||
|
||||
// Verify no extra rules were created
|
||||
exists, err := registryKeyExists(fmt.Sprintf("%s-%d", dnsPolicyConfigMatchPath, tc.expectedRuleCount))
|
||||
require.NoError(t, err)
|
||||
assert.False(t, exists, "No NRPT rule should exist at index %d", tc.expectedRuleCount)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -84,3 +84,18 @@ func (m *MockServer) UpdateServerConfig(domains dnsconfig.ServerDomains) error {
|
||||
func (m *MockServer) PopulateManagementDomain(mgmtURL *url.URL) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// BeginBatch mock implementation of BeginBatch from Server interface
|
||||
func (m *MockServer) BeginBatch() {
|
||||
// Mock implementation - no-op
|
||||
}
|
||||
|
||||
// EndBatch mock implementation of EndBatch from Server interface
|
||||
func (m *MockServer) EndBatch() {
|
||||
// Mock implementation - no-op
|
||||
}
|
||||
|
||||
// CancelBatch mock implementation of CancelBatch from Server interface
|
||||
func (m *MockServer) CancelBatch() {
|
||||
// Mock implementation - no-op
|
||||
}
|
||||
|
||||
@@ -45,6 +45,9 @@ type IosDnsManager interface {
|
||||
type Server interface {
|
||||
RegisterHandler(domains domain.List, handler dns.Handler, priority int)
|
||||
DeregisterHandler(domains domain.List, priority int)
|
||||
BeginBatch()
|
||||
EndBatch()
|
||||
CancelBatch()
|
||||
Initialize() error
|
||||
Stop()
|
||||
DnsIP() netip.Addr
|
||||
@@ -87,6 +90,7 @@ type DefaultServer struct {
|
||||
currentConfigHash uint64
|
||||
handlerChain *HandlerChain
|
||||
extraDomains map[domain.Domain]int
|
||||
batchMode bool
|
||||
|
||||
mgmtCacheResolver *mgmt.Resolver
|
||||
|
||||
@@ -234,7 +238,9 @@ func (s *DefaultServer) RegisterHandler(domains domain.List, handler dns.Handler
|
||||
// convert to zone with simple ref counter
|
||||
s.extraDomains[toZone(domain)]++
|
||||
}
|
||||
s.applyHostConfig()
|
||||
if !s.batchMode {
|
||||
s.applyHostConfig()
|
||||
}
|
||||
}
|
||||
|
||||
func (s *DefaultServer) registerHandler(domains []string, handler dns.Handler, priority int) {
|
||||
@@ -263,9 +269,41 @@ func (s *DefaultServer) DeregisterHandler(domains domain.List, priority int) {
|
||||
delete(s.extraDomains, zone)
|
||||
}
|
||||
}
|
||||
if !s.batchMode {
|
||||
s.applyHostConfig()
|
||||
}
|
||||
}
|
||||
|
||||
// BeginBatch starts batch mode for DNS handler registration/deregistration.
|
||||
// In batch mode, applyHostConfig() is not called after each handler operation,
|
||||
// allowing multiple handlers to be registered/deregistered efficiently.
|
||||
// Must be followed by EndBatch() to apply the accumulated changes.
|
||||
func (s *DefaultServer) BeginBatch() {
|
||||
s.mux.Lock()
|
||||
defer s.mux.Unlock()
|
||||
log.Debugf("DNS batch mode enabled")
|
||||
s.batchMode = true
|
||||
}
|
||||
|
||||
// EndBatch ends batch mode and applies all accumulated DNS configuration changes.
|
||||
func (s *DefaultServer) EndBatch() {
|
||||
s.mux.Lock()
|
||||
defer s.mux.Unlock()
|
||||
log.Debugf("DNS batch mode disabled, applying accumulated changes")
|
||||
s.batchMode = false
|
||||
s.applyHostConfig()
|
||||
}
|
||||
|
||||
// CancelBatch cancels batch mode without applying accumulated changes.
|
||||
// This is useful when operations fail partway through and you want to
|
||||
// discard partial state rather than applying it.
|
||||
func (s *DefaultServer) CancelBatch() {
|
||||
s.mux.Lock()
|
||||
defer s.mux.Unlock()
|
||||
log.Debugf("DNS batch mode cancelled, discarding accumulated changes")
|
||||
s.batchMode = false
|
||||
}
|
||||
|
||||
func (s *DefaultServer) deregisterHandler(domains []string, priority int) {
|
||||
log.Debugf("deregistering handler with priority %d for %v", priority, domains)
|
||||
|
||||
@@ -523,6 +561,7 @@ func (s *DefaultServer) applyConfiguration(update nbdns.Config) error {
|
||||
s.currentConfig.RouteAll = false
|
||||
}
|
||||
|
||||
// Always apply host config for management updates, regardless of batch mode
|
||||
s.applyHostConfig()
|
||||
|
||||
s.shutdownWg.Add(1)
|
||||
@@ -887,6 +926,7 @@ func (s *DefaultServer) upstreamCallbacks(
|
||||
}
|
||||
}
|
||||
|
||||
// Always apply host config when nameserver goes down, regardless of batch mode
|
||||
s.applyHostConfig()
|
||||
|
||||
go func() {
|
||||
@@ -922,6 +962,7 @@ func (s *DefaultServer) upstreamCallbacks(
|
||||
s.registerHandler([]string{nbdns.RootZone}, handler, priority)
|
||||
}
|
||||
|
||||
// Always apply host config when nameserver reactivates, regardless of batch mode
|
||||
s.applyHostConfig()
|
||||
|
||||
s.updateNSState(nsGroup, nil, true)
|
||||
|
||||
@@ -18,7 +18,12 @@ func TestGetServerDns(t *testing.T) {
|
||||
t.Errorf("invalid dns server instance: %s", err)
|
||||
}
|
||||
|
||||
if srvB != srv {
|
||||
mockSrvB, ok := srvB.(*MockServer)
|
||||
if !ok {
|
||||
t.Errorf("returned server is not a MockServer")
|
||||
}
|
||||
|
||||
if mockSrvB != srv {
|
||||
t.Errorf("mismatch dns instances")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -410,7 +410,7 @@ func (conn *Conn) onICEConnectionIsReady(priority conntype.ConnPriority, iceConn
|
||||
conn.doOnConnected(iceConnInfo.RosenpassPubKey, iceConnInfo.RosenpassAddr)
|
||||
}
|
||||
|
||||
func (conn *Conn) onICEStateDisconnected() {
|
||||
func (conn *Conn) onICEStateDisconnected(sessionChanged bool) {
|
||||
conn.mu.Lock()
|
||||
defer conn.mu.Unlock()
|
||||
|
||||
@@ -430,6 +430,10 @@ func (conn *Conn) onICEStateDisconnected() {
|
||||
if conn.isReadyToUpgrade() {
|
||||
conn.Log.Infof("ICE disconnected, set Relay to active connection")
|
||||
conn.dumpState.SwitchToRelay()
|
||||
if sessionChanged {
|
||||
conn.resetEndpoint()
|
||||
}
|
||||
|
||||
conn.wgProxyRelay.Work()
|
||||
|
||||
presharedKey := conn.presharedKey(conn.rosenpassRemoteKey)
|
||||
@@ -757,6 +761,17 @@ func (conn *Conn) newProxy(remoteConn net.Conn) (wgproxy.Proxy, error) {
|
||||
return wgProxy, nil
|
||||
}
|
||||
|
||||
func (conn *Conn) resetEndpoint() {
|
||||
if !isController(conn.config) {
|
||||
return
|
||||
}
|
||||
conn.Log.Infof("reset wg endpoint")
|
||||
conn.wgWatcher.Reset()
|
||||
if err := conn.endpointUpdater.RemoveEndpointAddress(); err != nil {
|
||||
conn.Log.Warnf("failed to remove endpoint address before update: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func (conn *Conn) isReadyToUpgrade() bool {
|
||||
return conn.wgProxyRelay != nil && conn.currentConnPriority != conntype.Relay
|
||||
}
|
||||
|
||||
@@ -66,6 +66,10 @@ func (e *EndpointUpdater) RemoveWgPeer() error {
|
||||
return e.wgConfig.WgInterface.RemovePeer(e.wgConfig.RemoteKey)
|
||||
}
|
||||
|
||||
func (e *EndpointUpdater) RemoveEndpointAddress() error {
|
||||
return e.wgConfig.WgInterface.RemoveEndpointAddress(e.wgConfig.RemoteKey)
|
||||
}
|
||||
|
||||
func (e *EndpointUpdater) waitForCloseTheDelayedUpdate() {
|
||||
if e.cancelFunc == nil {
|
||||
return
|
||||
|
||||
@@ -32,6 +32,8 @@ type WGWatcher struct {
|
||||
|
||||
enabled bool
|
||||
muEnabled sync.RWMutex
|
||||
|
||||
resetCh chan struct{}
|
||||
}
|
||||
|
||||
func NewWGWatcher(log *log.Entry, wgIfaceStater WGInterfaceStater, peerKey string, stateDump *stateDump) *WGWatcher {
|
||||
@@ -40,6 +42,7 @@ func NewWGWatcher(log *log.Entry, wgIfaceStater WGInterfaceStater, peerKey strin
|
||||
wgIfaceStater: wgIfaceStater,
|
||||
peerKey: peerKey,
|
||||
stateDump: stateDump,
|
||||
resetCh: make(chan struct{}, 1),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -76,6 +79,15 @@ func (w *WGWatcher) IsEnabled() bool {
|
||||
return w.enabled
|
||||
}
|
||||
|
||||
// Reset signals the watcher that the WireGuard peer has been reset and a new
|
||||
// handshake is expected. This restarts the handshake timeout from scratch.
|
||||
func (w *WGWatcher) Reset() {
|
||||
select {
|
||||
case w.resetCh <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
// wgStateCheck help to check the state of the WireGuard handshake and relay connection
|
||||
func (w *WGWatcher) periodicHandshakeCheck(ctx context.Context, onDisconnectedFn func(), enabledTime time.Time, initialHandshake time.Time) {
|
||||
w.log.Infof("WireGuard watcher started")
|
||||
@@ -105,6 +117,12 @@ func (w *WGWatcher) periodicHandshakeCheck(ctx context.Context, onDisconnectedFn
|
||||
w.stateDump.WGcheckSuccess()
|
||||
|
||||
w.log.Debugf("WireGuard watcher reset timer: %v", resetTime)
|
||||
case <-w.resetCh:
|
||||
w.log.Infof("WireGuard watcher received peer reset, restarting handshake timeout")
|
||||
lastHandshake = time.Time{}
|
||||
enabledTime = time.Now()
|
||||
timer.Stop()
|
||||
timer.Reset(wgHandshakeOvertime)
|
||||
case <-ctx.Done():
|
||||
w.log.Infof("WireGuard watcher stopped")
|
||||
return
|
||||
|
||||
@@ -52,8 +52,9 @@ type WorkerICE struct {
|
||||
// increase by one when disconnecting the agent
|
||||
// with it the remote peer can discard the already deprecated offer/answer
|
||||
// Without it the remote peer may recreate a workable ICE connection
|
||||
sessionID ICESessionID
|
||||
muxAgent sync.Mutex
|
||||
sessionID ICESessionID
|
||||
remoteSessionChanged bool
|
||||
muxAgent sync.Mutex
|
||||
|
||||
localUfrag string
|
||||
localPwd string
|
||||
@@ -106,6 +107,7 @@ func (w *WorkerICE) OnNewOffer(remoteOfferAnswer *OfferAnswer) {
|
||||
return
|
||||
}
|
||||
w.log.Debugf("agent already exists, recreate the connection")
|
||||
w.remoteSessionChanged = true
|
||||
w.agentDialerCancel()
|
||||
if w.agent != nil {
|
||||
if err := w.agent.Close(); err != nil {
|
||||
@@ -306,13 +308,17 @@ func (w *WorkerICE) connect(ctx context.Context, agent *icemaker.ThreadSafeAgent
|
||||
w.conn.onICEConnectionIsReady(selectedPriority(pair), ci)
|
||||
}
|
||||
|
||||
func (w *WorkerICE) closeAgent(agent *icemaker.ThreadSafeAgent, cancel context.CancelFunc) {
|
||||
func (w *WorkerICE) closeAgent(agent *icemaker.ThreadSafeAgent, cancel context.CancelFunc) bool {
|
||||
cancel()
|
||||
if err := agent.Close(); err != nil {
|
||||
w.log.Warnf("failed to close ICE agent: %s", err)
|
||||
}
|
||||
|
||||
w.muxAgent.Lock()
|
||||
defer w.muxAgent.Unlock()
|
||||
|
||||
sessionChanged := w.remoteSessionChanged
|
||||
w.remoteSessionChanged = false
|
||||
|
||||
if w.agent == agent {
|
||||
// consider to remove from here and move to the OnNewOffer
|
||||
@@ -325,7 +331,7 @@ func (w *WorkerICE) closeAgent(agent *icemaker.ThreadSafeAgent, cancel context.C
|
||||
w.agentConnecting = false
|
||||
w.remoteSessionID = ""
|
||||
}
|
||||
w.muxAgent.Unlock()
|
||||
return sessionChanged
|
||||
}
|
||||
|
||||
func (w *WorkerICE) punchRemoteWGPort(pair *ice.CandidatePair, remoteWgPort int) {
|
||||
@@ -426,11 +432,11 @@ func (w *WorkerICE) onConnectionStateChange(agent *icemaker.ThreadSafeAgent, dia
|
||||
// ice.ConnectionStateClosed happens when we recreate the agent. For the P2P to TURN switch important to
|
||||
// notify the conn.onICEStateDisconnected changes to update the current used priority
|
||||
|
||||
w.closeAgent(agent, dialerCancel)
|
||||
sessionChanged := w.closeAgent(agent, dialerCancel)
|
||||
|
||||
if w.lastKnownState == ice.ConnectionStateConnected {
|
||||
w.lastKnownState = ice.ConnectionStateDisconnected
|
||||
w.conn.onICEStateDisconnected()
|
||||
w.conn.onICEStateDisconnected(sessionChanged)
|
||||
}
|
||||
default:
|
||||
return
|
||||
|
||||
@@ -346,6 +346,23 @@ func (m *DefaultManager) updateSystemRoutes(newRoutes route.HAMap) error {
|
||||
}
|
||||
|
||||
var merr *multierror.Error
|
||||
|
||||
// Begin batch mode to avoid calling applyHostConfig() after each DNS handler operation
|
||||
batchStarted := false
|
||||
if m.dnsServer != nil {
|
||||
m.dnsServer.BeginBatch()
|
||||
batchStarted = true
|
||||
defer func() {
|
||||
if merr != nil {
|
||||
// On error, cancel batch to discard partial DNS state
|
||||
m.dnsServer.CancelBatch()
|
||||
} else {
|
||||
// On success, apply accumulated DNS changes
|
||||
m.dnsServer.EndBatch()
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
for id, handler := range toRemove {
|
||||
if err := handler.RemoveRoute(); err != nil {
|
||||
merr = multierror.Append(merr, fmt.Errorf("remove route %s: %w", handler.String(), err))
|
||||
@@ -376,6 +393,7 @@ func (m *DefaultManager) updateSystemRoutes(newRoutes route.HAMap) error {
|
||||
m.activeRoutes[id] = handler
|
||||
}
|
||||
|
||||
_ = batchStarted // Mark as used
|
||||
return nberrors.FormatErrorOrNil(merr)
|
||||
}
|
||||
|
||||
|
||||
@@ -2,7 +2,10 @@
|
||||
|
||||
package NetBirdSDK
|
||||
|
||||
import "github.com/netbirdio/netbird/client/internal/peer"
|
||||
import (
|
||||
"github.com/netbirdio/netbird/client/internal/lazyconn"
|
||||
"github.com/netbirdio/netbird/client/internal/peer"
|
||||
)
|
||||
|
||||
// EnvList is an exported struct to be bound by gomobile
|
||||
type EnvList struct {
|
||||
@@ -32,3 +35,13 @@ func (el *EnvList) AllItems() map[string]string {
|
||||
func GetEnvKeyNBForceRelay() string {
|
||||
return peer.EnvKeyNBForceRelay
|
||||
}
|
||||
|
||||
// GetEnvKeyNBLazyConn Exports the environment variable for the iOS client
|
||||
func GetEnvKeyNBLazyConn() string {
|
||||
return lazyconn.EnvEnableLazyConn
|
||||
}
|
||||
|
||||
// GetEnvKeyNBInactivityThreshold Exports the environment variable for the iOS client
|
||||
func GetEnvKeyNBInactivityThreshold() string {
|
||||
return lazyconn.EnvInactivityThreshold
|
||||
}
|
||||
|
||||
463
combined/perftest/main.go
Normal file
463
combined/perftest/main.go
Normal file
@@ -0,0 +1,463 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/binary"
|
||||
"flag"
|
||||
"fmt"
|
||||
"io"
|
||||
"math"
|
||||
"net"
|
||||
"os"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/netbirdio/netbird/client/embed"
|
||||
)
|
||||
|
||||
const (
|
||||
echoPort = 9000
|
||||
connectTimeout = 120 * time.Second
|
||||
startTimeout = 60 * time.Second
|
||||
stopTimeout = 30 * time.Second
|
||||
)
|
||||
|
||||
type peerInfo struct {
|
||||
client *embed.Client
|
||||
tunnelIP string
|
||||
name string
|
||||
}
|
||||
|
||||
type pairStats struct {
|
||||
from string
|
||||
to string
|
||||
sent int64
|
||||
received int64
|
||||
lost int64
|
||||
rtts []time.Duration
|
||||
}
|
||||
|
||||
func (s *pairStats) summary() (avgRTT, minRTT, maxRTT time.Duration, lossPercent float64) {
|
||||
if len(s.rtts) == 0 {
|
||||
return 0, 0, 0, 100
|
||||
}
|
||||
minRTT = s.rtts[0]
|
||||
maxRTT = s.rtts[0]
|
||||
var total time.Duration
|
||||
for _, rtt := range s.rtts {
|
||||
total += rtt
|
||||
if rtt < minRTT {
|
||||
minRTT = rtt
|
||||
}
|
||||
if rtt > maxRTT {
|
||||
maxRTT = rtt
|
||||
}
|
||||
}
|
||||
avgRTT = total / time.Duration(len(s.rtts))
|
||||
if s.sent > 0 {
|
||||
lossPercent = float64(s.lost) / float64(s.sent) * 100
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func main() {
|
||||
managementURL := flag.String("management-url", "", "Management server URL (required)")
|
||||
setupKey := flag.String("setup-key", "", "Reusable setup key (required)")
|
||||
numPeers := flag.Int("peers", 5, "Number of peers to spawn")
|
||||
forceRelay := flag.Bool("force-relay", false, "Force relay connections (NB_FORCE_RELAY=true)")
|
||||
duration := flag.Duration("duration", 30*time.Second, "Traffic test duration")
|
||||
packetSize := flag.Int("packet-size", 512, "UDP packet size in bytes")
|
||||
logLevel := flag.String("log-level", "panic", "Client log level (trace, debug, info, warn, error, panic)")
|
||||
flag.Parse()
|
||||
|
||||
if *managementURL == "" || *setupKey == "" {
|
||||
fmt.Fprintln(os.Stderr, "Error: --management-url and --setup-key are required")
|
||||
flag.Usage()
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
if *numPeers < 2 {
|
||||
fmt.Fprintln(os.Stderr, "Error: --peers must be at least 2")
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
// Minimum packet size: 8 bytes for timestamp + 8 bytes for sequence number
|
||||
if *packetSize < 16 {
|
||||
fmt.Fprintln(os.Stderr, "Error: --packet-size must be at least 16")
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
if *forceRelay {
|
||||
os.Setenv("NB_FORCE_RELAY", "true")
|
||||
}
|
||||
os.Setenv("NB_USE_NETSTACK_MODE", "true")
|
||||
|
||||
fmt.Println("=== NetBird Performance Test ===")
|
||||
fmt.Printf("Management URL: %s\n", *managementURL)
|
||||
fmt.Printf("Peers: %d\n", *numPeers)
|
||||
fmt.Printf("Force relay: %v\n", *forceRelay)
|
||||
fmt.Printf("Duration: %s\n", *duration)
|
||||
fmt.Printf("Packet size: %d bytes\n", *packetSize)
|
||||
fmt.Println()
|
||||
|
||||
// Phase 1: Create peers
|
||||
fmt.Println("--- Phase 1: Creating peers ---")
|
||||
peers := make([]peerInfo, *numPeers)
|
||||
for i := 0; i < *numPeers; i++ {
|
||||
name := fmt.Sprintf("perf-peer-%d", i)
|
||||
port := 0
|
||||
c, err := embed.New(embed.Options{
|
||||
DeviceName: name,
|
||||
SetupKey: *setupKey,
|
||||
ManagementURL: *managementURL,
|
||||
WireguardPort: &port,
|
||||
LogLevel: *logLevel,
|
||||
LogOutput: io.Discard,
|
||||
})
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "Error creating peer %s: %v\n", name, err)
|
||||
os.Exit(1)
|
||||
}
|
||||
peers[i] = peerInfo{client: c, name: name}
|
||||
fmt.Printf(" Created %s\n", name)
|
||||
}
|
||||
|
||||
// Phase 2: Start peers in parallel
|
||||
fmt.Println("\n--- Phase 2: Starting peers ---")
|
||||
startTime := time.Now()
|
||||
var wg sync.WaitGroup
|
||||
startErrors := make([]error, *numPeers)
|
||||
|
||||
for i := range peers {
|
||||
wg.Add(1)
|
||||
go func(idx int) {
|
||||
defer wg.Done()
|
||||
ctx, cancel := context.WithTimeout(context.Background(), startTimeout)
|
||||
defer cancel()
|
||||
t := time.Now()
|
||||
if err := peers[idx].client.Start(ctx); err != nil {
|
||||
startErrors[idx] = err
|
||||
return
|
||||
}
|
||||
fmt.Printf(" %s started in %s\n", peers[idx].name, time.Since(t).Round(time.Millisecond))
|
||||
}(i)
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
// Check for start errors
|
||||
var failed bool
|
||||
for i, err := range startErrors {
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, " Error starting %s: %v\n", peers[i].name, err)
|
||||
failed = true
|
||||
}
|
||||
}
|
||||
if failed {
|
||||
cleanup(peers)
|
||||
os.Exit(1)
|
||||
}
|
||||
fmt.Printf(" All peers started in %s\n", time.Since(startTime).Round(time.Millisecond))
|
||||
|
||||
// Get tunnel IPs
|
||||
for i := range peers {
|
||||
status, err := peers[i].client.Status()
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, " Error getting status for %s: %v\n", peers[i].name, err)
|
||||
cleanup(peers)
|
||||
os.Exit(1)
|
||||
}
|
||||
ip := status.LocalPeerState.IP
|
||||
// Strip CIDR suffix if present (e.g. "100.64.0.1/16" -> "100.64.0.1")
|
||||
if idx := strings.Index(ip, "/"); idx != -1 {
|
||||
ip = ip[:idx]
|
||||
}
|
||||
peers[i].tunnelIP = ip
|
||||
fmt.Printf(" %s -> %s\n", peers[i].name, peers[i].tunnelIP)
|
||||
}
|
||||
|
||||
// Phase 3: Wait for connections
|
||||
fmt.Println("\n--- Phase 3: Waiting for peer connections ---")
|
||||
connStart := time.Now()
|
||||
expectedPeers := *numPeers - 1
|
||||
deadline := time.After(connectTimeout)
|
||||
ticker := time.NewTicker(2 * time.Second)
|
||||
defer ticker.Stop()
|
||||
|
||||
allConnected := false
|
||||
waitLoop:
|
||||
for {
|
||||
select {
|
||||
case <-deadline:
|
||||
fmt.Fprintf(os.Stderr, " Timeout waiting for connections after %s\n", connectTimeout)
|
||||
printConnectionStatus(peers)
|
||||
cleanup(peers)
|
||||
os.Exit(1)
|
||||
case <-ticker.C:
|
||||
allConnected = true
|
||||
for i := range peers {
|
||||
connected := countConnectedPeers(peers[i].client)
|
||||
if connected < expectedPeers {
|
||||
allConnected = false
|
||||
break
|
||||
}
|
||||
}
|
||||
if allConnected {
|
||||
break waitLoop
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fmt.Printf(" All peers connected in %s\n", time.Since(connStart).Round(time.Millisecond))
|
||||
printConnectionStatus(peers)
|
||||
|
||||
// Phase 4: Traffic test
|
||||
fmt.Printf("\n--- Phase 4: Traffic test (%s) ---\n", *duration)
|
||||
|
||||
// Start echo listeners on all peers
|
||||
listeners := make([]net.PacketConn, *numPeers)
|
||||
for i := range peers {
|
||||
conn, err := peers[i].client.ListenUDP(fmt.Sprintf(":%d", echoPort))
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, " Error creating listener on %s: %v\n", peers[i].name, err)
|
||||
cleanup(peers)
|
||||
os.Exit(1)
|
||||
}
|
||||
listeners[i] = conn
|
||||
go echoServer(conn, *packetSize)
|
||||
fmt.Printf(" Echo listener started on %s:%d\n", peers[i].tunnelIP, echoPort)
|
||||
}
|
||||
|
||||
// Run traffic between all pairs (i < j)
|
||||
var statsMu sync.Mutex
|
||||
var allStats []pairStats
|
||||
|
||||
var trafficWg sync.WaitGroup
|
||||
for i := 0; i < *numPeers; i++ {
|
||||
for j := i + 1; j < *numPeers; j++ {
|
||||
trafficWg.Add(1)
|
||||
go func(from, to int) {
|
||||
defer trafficWg.Done()
|
||||
stats := runTraffic(peers[from].client, peers[from].name, peers[to].tunnelIP, peers[to].name, *duration, *packetSize)
|
||||
statsMu.Lock()
|
||||
allStats = append(allStats, stats)
|
||||
statsMu.Unlock()
|
||||
}(i, j)
|
||||
}
|
||||
}
|
||||
trafficWg.Wait()
|
||||
|
||||
// Close listeners
|
||||
for _, l := range listeners {
|
||||
if l != nil {
|
||||
l.Close()
|
||||
}
|
||||
}
|
||||
|
||||
// Phase 5: Report
|
||||
fmt.Println("\n--- Phase 5: Results ---")
|
||||
printReport(allStats)
|
||||
|
||||
// Cleanup
|
||||
fmt.Println("\n--- Cleanup ---")
|
||||
cleanup(peers)
|
||||
fmt.Println("Done.")
|
||||
}
|
||||
|
||||
func countConnectedPeers(c *embed.Client) int {
|
||||
status, err := c.Status()
|
||||
if err != nil {
|
||||
return 0
|
||||
}
|
||||
count := 0
|
||||
for _, p := range status.Peers {
|
||||
if p.ConnStatus == embed.PeerStatusConnected {
|
||||
count++
|
||||
}
|
||||
}
|
||||
return count
|
||||
}
|
||||
|
||||
func printConnectionStatus(peers []peerInfo) {
|
||||
for i := range peers {
|
||||
status, err := peers[i].client.Status()
|
||||
if err != nil {
|
||||
fmt.Printf(" %s: error getting status: %v\n", peers[i].name, err)
|
||||
continue
|
||||
}
|
||||
connected := 0
|
||||
relayed := 0
|
||||
for _, p := range status.Peers {
|
||||
if p.ConnStatus == embed.PeerStatusConnected {
|
||||
connected++
|
||||
if p.Relayed {
|
||||
relayed++
|
||||
}
|
||||
}
|
||||
}
|
||||
connType := "direct"
|
||||
if relayed > 0 {
|
||||
connType = fmt.Sprintf("%d direct, %d relayed", connected-relayed, relayed)
|
||||
}
|
||||
fmt.Printf(" %s: %d/%d connected (%s)\n", peers[i].name, connected, len(status.Peers), connType)
|
||||
}
|
||||
}
|
||||
|
||||
func echoServer(conn net.PacketConn, maxSize int) {
|
||||
buf := make([]byte, maxSize+100)
|
||||
for {
|
||||
n, addr, err := conn.ReadFrom(buf)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
_, _ = conn.WriteTo(buf[:n], addr)
|
||||
}
|
||||
}
|
||||
|
||||
func runTraffic(client *embed.Client, fromName, toIP, toName string, duration time.Duration, packetSize int) pairStats {
|
||||
stats := pairStats{
|
||||
from: fromName,
|
||||
to: toName,
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), duration+10*time.Second)
|
||||
defer cancel()
|
||||
|
||||
conn, err := client.Dial(ctx, "udp", fmt.Sprintf("%s:%d", toIP, echoPort))
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, " Error dialing %s -> %s: %v\n", fromName, toName, err)
|
||||
return stats
|
||||
}
|
||||
defer conn.Close()
|
||||
|
||||
deadline := time.Now().Add(duration)
|
||||
buf := make([]byte, packetSize)
|
||||
recvBuf := make([]byte, packetSize+100)
|
||||
var seq uint64
|
||||
|
||||
for time.Now().Before(deadline) {
|
||||
seq++
|
||||
// Encode timestamp and sequence number
|
||||
binary.BigEndian.PutUint64(buf[0:8], uint64(time.Now().UnixNano()))
|
||||
binary.BigEndian.PutUint64(buf[8:16], seq)
|
||||
|
||||
stats.sent++
|
||||
conn.SetWriteDeadline(time.Now().Add(5 * time.Second))
|
||||
_, err := conn.Write(buf)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, " Error sending packet to %s: %v\n", toName, err)
|
||||
stats.lost++
|
||||
continue
|
||||
}
|
||||
|
||||
conn.SetReadDeadline(time.Now().Add(5 * time.Second))
|
||||
n, err := conn.Read(recvBuf)
|
||||
if err != nil {
|
||||
stats.lost++
|
||||
continue
|
||||
}
|
||||
|
||||
if n >= 8 {
|
||||
sentNano := binary.BigEndian.Uint64(recvBuf[0:8])
|
||||
rtt := time.Since(time.Unix(0, int64(sentNano)))
|
||||
stats.received++
|
||||
stats.rtts = append(stats.rtts, rtt)
|
||||
} else {
|
||||
stats.received++
|
||||
}
|
||||
|
||||
// Small sleep to avoid flooding
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
|
||||
return stats
|
||||
}
|
||||
|
||||
func printReport(allStats []pairStats) {
|
||||
if len(allStats) == 0 {
|
||||
fmt.Println(" No traffic data collected.")
|
||||
return
|
||||
}
|
||||
|
||||
fmt.Printf(" %-30s %8s %8s %8s %8s %10s %10s %10s\n",
|
||||
"Pair", "Sent", "Recv", "Lost", "Loss%", "Avg RTT", "Min RTT", "Max RTT")
|
||||
fmt.Println(" " + strings.Repeat("-", 108))
|
||||
|
||||
var totalSent, totalRecv, totalLost int64
|
||||
var totalRTTs []time.Duration
|
||||
|
||||
for _, s := range allStats {
|
||||
avg, min, max, loss := s.summary()
|
||||
pair := fmt.Sprintf("%s -> %s", s.from, s.to)
|
||||
fmt.Printf(" %-30s %8d %8d %8d %7.1f%% %10s %10s %10s\n",
|
||||
pair, s.sent, s.received, s.lost, loss,
|
||||
avg.Round(time.Microsecond), min.Round(time.Microsecond), max.Round(time.Microsecond))
|
||||
totalSent += s.sent
|
||||
totalRecv += s.received
|
||||
totalLost += s.lost
|
||||
totalRTTs = append(totalRTTs, s.rtts...)
|
||||
}
|
||||
|
||||
fmt.Println(" " + strings.Repeat("-", 108))
|
||||
|
||||
// Overall summary
|
||||
var overallLoss float64
|
||||
if totalSent > 0 {
|
||||
overallLoss = float64(totalLost) / float64(totalSent) * 100
|
||||
}
|
||||
|
||||
var avgRTT, minRTT, maxRTT time.Duration
|
||||
if len(totalRTTs) > 0 {
|
||||
minRTT = totalRTTs[0]
|
||||
maxRTT = totalRTTs[0]
|
||||
var total time.Duration
|
||||
for _, rtt := range totalRTTs {
|
||||
total += rtt
|
||||
if rtt < minRTT {
|
||||
minRTT = rtt
|
||||
}
|
||||
if rtt > maxRTT {
|
||||
maxRTT = rtt
|
||||
}
|
||||
}
|
||||
avgRTT = total / time.Duration(len(totalRTTs))
|
||||
}
|
||||
|
||||
fmt.Printf(" %-30s %8d %8d %8d %7.1f%% %10s %10s %10s\n",
|
||||
"TOTAL", totalSent, totalRecv, totalLost, overallLoss,
|
||||
avgRTT.Round(time.Microsecond), minRTT.Round(time.Microsecond), maxRTT.Round(time.Microsecond))
|
||||
|
||||
// Extra stats
|
||||
if len(totalRTTs) > 0 {
|
||||
fmt.Println()
|
||||
var sumSq float64
|
||||
avgNs := float64(avgRTT.Nanoseconds())
|
||||
for _, rtt := range totalRTTs {
|
||||
diff := float64(rtt.Nanoseconds()) - avgNs
|
||||
sumSq += diff * diff
|
||||
}
|
||||
stddev := time.Duration(math.Sqrt(sumSq / float64(len(totalRTTs))))
|
||||
|
||||
fmt.Printf(" RTT stddev: %s\n", stddev.Round(time.Microsecond))
|
||||
fmt.Printf(" Pairs tested: %d\n", len(allStats))
|
||||
}
|
||||
}
|
||||
|
||||
func cleanup(peers []peerInfo) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), stopTimeout)
|
||||
defer cancel()
|
||||
|
||||
var wg sync.WaitGroup
|
||||
for i := range peers {
|
||||
wg.Add(1)
|
||||
go func(idx int) {
|
||||
defer wg.Done()
|
||||
if err := peers[idx].client.Stop(ctx); err != nil {
|
||||
fmt.Fprintf(os.Stderr, " Error stopping %s: %v\n", peers[idx].name, err)
|
||||
} else {
|
||||
fmt.Printf(" Stopped %s\n", peers[idx].name)
|
||||
}
|
||||
}(i)
|
||||
}
|
||||
wg.Wait()
|
||||
}
|
||||
61
combined/perftest/run.sh
Normal file
61
combined/perftest/run.sh
Normal file
@@ -0,0 +1,61 @@
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
|
||||
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
BINARY="$SCRIPT_DIR/perftest"
|
||||
|
||||
# Defaults
|
||||
MANAGEMENT_URL="${MANAGEMENT_URL:-}"
|
||||
SETUP_KEY="${SETUP_KEY:-}"
|
||||
PEERS="${PEERS:-5}"
|
||||
DURATION="${DURATION:-30s}"
|
||||
PACKET_SIZE="${PACKET_SIZE:-512}"
|
||||
FORCE_RELAY="${FORCE_RELAY:-false}"
|
||||
LOG_LEVEL="${LOG_LEVEL:-panic}"
|
||||
|
||||
usage() {
|
||||
cat <<EOF
|
||||
Usage: MANAGEMENT_URL=... SETUP_KEY=... $0 [options]
|
||||
|
||||
Environment variables (or flags):
|
||||
MANAGEMENT_URL Management server URL (required)
|
||||
SETUP_KEY Reusable setup key (required). Use ephemeral.
|
||||
PEERS Number of peers (default: 5)
|
||||
DURATION Traffic test duration (default: 30s)
|
||||
PACKET_SIZE UDP packet size in bytes (default: 512)
|
||||
FORCE_RELAY Force relay mode (default: false)
|
||||
LOG_LEVEL Client log level (default: panic)
|
||||
|
||||
All extra arguments are passed directly to the binary.
|
||||
EOF
|
||||
exit 1
|
||||
}
|
||||
|
||||
if [[ -z "$MANAGEMENT_URL" || -z "$SETUP_KEY" ]]; then
|
||||
echo "Error: MANAGEMENT_URL and SETUP_KEY must be set"
|
||||
echo
|
||||
usage
|
||||
fi
|
||||
|
||||
# Build
|
||||
echo "Building perftest..."
|
||||
cd "$SCRIPT_DIR"
|
||||
go build -o "$BINARY" .
|
||||
echo "Build OK: $BINARY"
|
||||
echo
|
||||
|
||||
# Run
|
||||
ARGS=(
|
||||
--management-url "$MANAGEMENT_URL"
|
||||
--setup-key "$SETUP_KEY"
|
||||
--peers "$PEERS"
|
||||
--duration "$DURATION"
|
||||
--packet-size "$PACKET_SIZE"
|
||||
--log-level "$LOG_LEVEL"
|
||||
)
|
||||
|
||||
if [[ "$FORCE_RELAY" == "true" ]]; then
|
||||
ARGS+=(--force-relay)
|
||||
fi
|
||||
|
||||
exec "$BINARY" "${ARGS[@]}" "$@"
|
||||
1
go.mod
1
go.mod
@@ -83,6 +83,7 @@ require (
|
||||
github.com/pion/stun/v3 v3.1.0
|
||||
github.com/pion/transport/v3 v3.1.1
|
||||
github.com/pion/turn/v3 v3.0.1
|
||||
github.com/pires/go-proxyproto v0.11.0
|
||||
github.com/pkg/sftp v1.13.9
|
||||
github.com/prometheus/client_golang v1.23.2
|
||||
github.com/quic-go/quic-go v0.55.0
|
||||
|
||||
2
go.sum
2
go.sum
@@ -474,6 +474,8 @@ github.com/pion/turn/v3 v3.0.1 h1:wLi7BTQr6/Q20R0vt/lHbjv6y4GChFtC33nkYbasoT8=
|
||||
github.com/pion/turn/v3 v3.0.1/go.mod h1:MrJDKgqryDyWy1/4NT9TWfXWGMC7UHT6pJIv1+gMeNE=
|
||||
github.com/pion/turn/v4 v4.1.1 h1:9UnY2HB99tpDyz3cVVZguSxcqkJ1DsTSZ+8TGruh4fc=
|
||||
github.com/pion/turn/v4 v4.1.1/go.mod h1:2123tHk1O++vmjI5VSD0awT50NywDAq5A2NNNU4Jjs8=
|
||||
github.com/pires/go-proxyproto v0.11.0 h1:gUQpS85X/VJMdUsYyEgyn59uLJvGqPhJV5YvG68wXH4=
|
||||
github.com/pires/go-proxyproto v0.11.0/go.mod h1:ZKAAyp3cgy5Y5Mo4n9AlScrkCZwUy0g3Jf+slqQVcuU=
|
||||
github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
|
||||
github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
|
||||
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
|
||||
|
||||
@@ -183,14 +183,14 @@ read_enable_proxy() {
|
||||
}
|
||||
|
||||
read_proxy_domain() {
|
||||
local suggested_proxy="proxy.${NETBIRD_DOMAIN}"
|
||||
local suggested_proxy="proxy.${BASE_DOMAIN}"
|
||||
|
||||
echo "" > /dev/stderr
|
||||
echo "NOTE: The proxy domain must be different from the management domain ($NETBIRD_DOMAIN)" > /dev/stderr
|
||||
echo "to avoid TLS certificate conflicts." > /dev/stderr
|
||||
echo "" > /dev/stderr
|
||||
echo "You also need to add a wildcard DNS record for the proxy domain," > /dev/stderr
|
||||
echo "e.g. *.${suggested_proxy} pointing to the same server IP as $NETBIRD_DOMAIN." > /dev/stderr
|
||||
echo "e.g. *.${suggested_proxy} pointing to the same server domain as $NETBIRD_DOMAIN with a CNAME record." > /dev/stderr
|
||||
echo "" > /dev/stderr
|
||||
echo -n "Enter the domain for the NetBird Proxy (e.g. ${suggested_proxy}): " > /dev/stderr
|
||||
read -r READ_PROXY_DOMAIN < /dev/tty
|
||||
@@ -202,13 +202,16 @@ read_proxy_domain() {
|
||||
fi
|
||||
|
||||
if [[ "$READ_PROXY_DOMAIN" == "$NETBIRD_DOMAIN" ]]; then
|
||||
echo "The proxy domain cannot be the same as the management domain ($NETBIRD_DOMAIN)." > /dev/stderr
|
||||
echo "" > /dev/stderr
|
||||
echo "WARNING: The proxy domain cannot be the same as the management domain ($NETBIRD_DOMAIN)." > /dev/stderr
|
||||
read_proxy_domain
|
||||
return
|
||||
fi
|
||||
|
||||
if [[ "$READ_PROXY_DOMAIN" == *".${NETBIRD_DOMAIN}" ]]; then
|
||||
echo "The proxy domain cannot be a subdomain of the management domain ($NETBIRD_DOMAIN)." > /dev/stderr
|
||||
echo ${READ_PROXY_DOMAIN} | grep ${NETBIRD_DOMAIN} > /dev/null
|
||||
if [[ $? -eq 0 ]]; then
|
||||
echo "" > /dev/stderr
|
||||
echo "WARNING: The proxy domain cannot be a subdomain of the management domain ($NETBIRD_DOMAIN)." > /dev/stderr
|
||||
read_proxy_domain
|
||||
return
|
||||
fi
|
||||
@@ -326,6 +329,9 @@ initialize_default_values() {
|
||||
BIND_LOCALHOST_ONLY="true"
|
||||
EXTERNAL_PROXY_NETWORK=""
|
||||
|
||||
# Traefik static IP within the internal bridge network
|
||||
TRAEFIK_IP="172.30.0.10"
|
||||
|
||||
# NetBird Proxy configuration
|
||||
ENABLE_PROXY="false"
|
||||
PROXY_DOMAIN=""
|
||||
@@ -340,10 +346,12 @@ configure_domain() {
|
||||
|
||||
if [[ "$NETBIRD_DOMAIN" == "use-ip" ]]; then
|
||||
NETBIRD_DOMAIN=$(get_main_ip_address)
|
||||
BASE_DOMAIN=$NETBIRD_DOMAIN
|
||||
else
|
||||
NETBIRD_PORT=443
|
||||
NETBIRD_HTTP_PROTOCOL="https"
|
||||
NETBIRD_RELAY_PROTO="rels"
|
||||
BASE_DOMAIN=$(echo $NETBIRD_DOMAIN | sed -E 's/^[^.]+\.//')
|
||||
fi
|
||||
return 0
|
||||
}
|
||||
@@ -388,7 +396,7 @@ check_existing_installation() {
|
||||
echo "Generated files already exist, if you want to reinitialize the environment, please remove them first."
|
||||
echo "You can use the following commands:"
|
||||
echo " $DOCKER_COMPOSE_COMMAND down --volumes # to remove all containers and volumes"
|
||||
echo " rm -f docker-compose.yml dashboard.env config.yaml proxy.env nginx-netbird.conf caddyfile-netbird.txt npm-advanced-config.txt"
|
||||
echo " rm -f docker-compose.yml dashboard.env config.yaml proxy.env traefik-dynamic.yaml nginx-netbird.conf caddyfile-netbird.txt npm-advanced-config.txt"
|
||||
echo "Be aware that this will remove all data from the database, and you will have to reconfigure the dashboard."
|
||||
exit 1
|
||||
fi
|
||||
@@ -407,6 +415,8 @@ generate_configuration_files() {
|
||||
# This will be overwritten with the actual token after netbird-server starts
|
||||
echo "# Placeholder - will be updated with token after netbird-server starts" > proxy.env
|
||||
echo "NB_PROXY_TOKEN=placeholder" >> proxy.env
|
||||
# TCP ServersTransport for PROXY protocol v2 to the proxy backend
|
||||
render_traefik_dynamic > traefik-dynamic.yaml
|
||||
fi
|
||||
;;
|
||||
1)
|
||||
@@ -554,10 +564,14 @@ init_environment() {
|
||||
############################################
|
||||
|
||||
render_docker_compose_traefik_builtin() {
|
||||
# Generate proxy service section if enabled
|
||||
# Generate proxy service section and Traefik dynamic config if enabled
|
||||
local proxy_service=""
|
||||
local proxy_volumes=""
|
||||
local traefik_file_provider=""
|
||||
local traefik_dynamic_volume=""
|
||||
if [[ "$ENABLE_PROXY" == "true" ]]; then
|
||||
traefik_file_provider=' - "--providers.file.filename=/etc/traefik/dynamic.yaml"'
|
||||
traefik_dynamic_volume=" - ./traefik-dynamic.yaml:/etc/traefik/dynamic.yaml:ro"
|
||||
proxy_service="
|
||||
# NetBird Proxy - exposes internal resources to the internet
|
||||
proxy:
|
||||
@@ -565,7 +579,9 @@ render_docker_compose_traefik_builtin() {
|
||||
container_name: netbird-proxy
|
||||
# Hairpin NAT fix: route domain back to traefik's static IP within Docker
|
||||
extra_hosts:
|
||||
- \"$NETBIRD_DOMAIN:172.30.0.10\"
|
||||
- \"$NETBIRD_DOMAIN:$TRAEFIK_IP\"
|
||||
ports:
|
||||
- 51820:51820/udp
|
||||
restart: unless-stopped
|
||||
networks: [netbird]
|
||||
depends_on:
|
||||
@@ -583,6 +599,7 @@ render_docker_compose_traefik_builtin() {
|
||||
- traefik.tcp.routers.proxy-passthrough.service=proxy-tls
|
||||
- traefik.tcp.routers.proxy-passthrough.priority=1
|
||||
- traefik.tcp.services.proxy-tls.loadbalancer.server.port=8443
|
||||
- traefik.tcp.services.proxy-tls.loadbalancer.serverstransport=pp-v2@file
|
||||
logging:
|
||||
driver: \"json-file\"
|
||||
options:
|
||||
@@ -602,7 +619,7 @@ services:
|
||||
restart: unless-stopped
|
||||
networks:
|
||||
netbird:
|
||||
ipv4_address: 172.30.0.10
|
||||
ipv4_address: $TRAEFIK_IP
|
||||
command:
|
||||
# Logging
|
||||
- "--log.level=INFO"
|
||||
@@ -629,12 +646,14 @@ services:
|
||||
# gRPC transport settings
|
||||
- "--serverstransport.forwardingtimeouts.responseheadertimeout=0s"
|
||||
- "--serverstransport.forwardingtimeouts.idleconntimeout=0s"
|
||||
$traefik_file_provider
|
||||
ports:
|
||||
- '443:443'
|
||||
- '80:80'
|
||||
volumes:
|
||||
- /var/run/docker.sock:/var/run/docker.sock:ro
|
||||
- netbird_traefik_letsencrypt:/letsencrypt
|
||||
$traefik_dynamic_volume
|
||||
logging:
|
||||
driver: "json-file"
|
||||
options:
|
||||
@@ -744,6 +763,10 @@ server:
|
||||
cliRedirectURIs:
|
||||
- "http://localhost:53000/"
|
||||
|
||||
reverseProxy:
|
||||
trustedHTTPProxies:
|
||||
- "$TRAEFIK_IP/32"
|
||||
|
||||
store:
|
||||
engine: "sqlite"
|
||||
encryptionKey: "$DATASTORE_ENCRYPTION_KEY"
|
||||
@@ -773,6 +796,17 @@ EOF
|
||||
return 0
|
||||
}
|
||||
|
||||
render_traefik_dynamic() {
|
||||
cat <<'EOF'
|
||||
tcp:
|
||||
serversTransports:
|
||||
pp-v2:
|
||||
proxyProtocol:
|
||||
version: 2
|
||||
EOF
|
||||
return 0
|
||||
}
|
||||
|
||||
render_proxy_env() {
|
||||
cat <<EOF
|
||||
# NetBird Proxy Configuration
|
||||
@@ -792,6 +826,10 @@ NB_PROXY_OIDC_CLIENT_ID=netbird-proxy
|
||||
NB_PROXY_OIDC_ENDPOINT=$NETBIRD_HTTP_PROTOCOL://$NETBIRD_DOMAIN/oauth2
|
||||
NB_PROXY_OIDC_SCOPES=openid,profile,email
|
||||
NB_PROXY_FORWARDED_PROTO=https
|
||||
# Enable PROXY protocol to preserve client IPs through L4 proxies (Traefik TCP passthrough)
|
||||
NB_PROXY_PROXY_PROTOCOL=true
|
||||
# Trust Traefik's IP for PROXY protocol headers
|
||||
NB_PROXY_TRUSTED_PROXIES=$TRAEFIK_IP
|
||||
EOF
|
||||
return 0
|
||||
}
|
||||
@@ -1150,23 +1188,30 @@ print_builtin_traefik_instructions() {
|
||||
echo " NETBIRD SETUP COMPLETE"
|
||||
echo "$MSG_SEPARATOR"
|
||||
echo ""
|
||||
echo "You can access the NetBird dashboard at $NETBIRD_HTTP_PROTOCOL://$NETBIRD_DOMAIN"
|
||||
echo "You can access the NetBird dashboard at:"
|
||||
echo " $NETBIRD_HTTP_PROTOCOL://$NETBIRD_DOMAIN"
|
||||
echo ""
|
||||
echo "Follow the onboarding steps to set up your NetBird instance."
|
||||
echo ""
|
||||
echo "Traefik is handling TLS certificates automatically via Let's Encrypt."
|
||||
echo "If you see certificate warnings, wait a moment for certificate issuance to complete."
|
||||
echo ""
|
||||
echo "Open ports:"
|
||||
echo " - 443/tcp (HTTPS - all NetBird services)"
|
||||
echo " - 80/tcp (HTTP - redirects to HTTPS)"
|
||||
echo " - $NETBIRD_STUN_PORT/udp (STUN - required for NAT traversal)"
|
||||
echo " - 443/tcp (HTTPS - all NetBird services)"
|
||||
echo " - 80/tcp (HTTP - redirects to HTTPS)"
|
||||
echo " - $NETBIRD_STUN_PORT/udp (STUN - required for NAT traversal)"
|
||||
if [[ "$ENABLE_PROXY" == "true" ]]; then
|
||||
echo " - 51820/udp (WIREGUARD - (optional) for P2P proxy connections)"
|
||||
echo ""
|
||||
echo "NetBird Proxy:"
|
||||
echo " The proxy service is enabled and running."
|
||||
echo " Any domain NOT matching $NETBIRD_DOMAIN will be passed through to the proxy."
|
||||
echo " The proxy handles its own TLS certificates via ACME TLS-ALPN-01 challenge."
|
||||
echo " Point your proxy domains (CNAMEs) to this server's IP address."
|
||||
echo " Point your proxy domain to this server's domain address like in the examples below:"
|
||||
echo ""
|
||||
echo " $PROXY_DOMAIN CNAME $NETBIRD_DOMAIN"
|
||||
echo " *.$PROXY_DOMAIN CNAME $NETBIRD_DOMAIN"
|
||||
echo ""
|
||||
fi
|
||||
return 0
|
||||
}
|
||||
|
||||
@@ -56,6 +56,7 @@ var (
|
||||
certKeyFile string
|
||||
certLockMethod string
|
||||
wgPort int
|
||||
proxyProtocol bool
|
||||
)
|
||||
|
||||
var rootCmd = &cobra.Command{
|
||||
@@ -90,6 +91,7 @@ func init() {
|
||||
rootCmd.Flags().StringVar(&certKeyFile, "cert-key-file", envStringOrDefault("NB_PROXY_CERTIFICATE_KEY_FILE", "tls.key"), "TLS certificate key filename within the certificate directory")
|
||||
rootCmd.Flags().StringVar(&certLockMethod, "cert-lock-method", envStringOrDefault("NB_PROXY_CERT_LOCK_METHOD", "auto"), "Certificate lock method for cross-replica coordination: auto, flock, or k8s-lease")
|
||||
rootCmd.Flags().IntVar(&wgPort, "wg-port", envIntOrDefault("NB_PROXY_WG_PORT", 0), "WireGuard listen port (0 = random). Fixed port only works with single-account deployments")
|
||||
rootCmd.Flags().BoolVar(&proxyProtocol, "proxy-protocol", envBoolOrDefault("NB_PROXY_PROXY_PROTOCOL", false), "Enable PROXY protocol on TCP listeners to preserve client IPs behind L4 proxies")
|
||||
}
|
||||
|
||||
// Execute runs the root command.
|
||||
@@ -165,6 +167,7 @@ func runServer(cmd *cobra.Command, args []string) error {
|
||||
TrustedProxies: parsedTrustedProxies,
|
||||
CertLockMethod: nbacme.CertLockMethod(certLockMethod),
|
||||
WireguardPort: wgPort,
|
||||
ProxyProtocol: proxyProtocol,
|
||||
}
|
||||
|
||||
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
|
||||
|
||||
106
proxy/proxyprotocol_test.go
Normal file
106
proxy/proxyprotocol_test.go
Normal file
@@ -0,0 +1,106 @@
|
||||
package proxy
|
||||
|
||||
import (
|
||||
"net"
|
||||
"net/netip"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
proxyproto "github.com/pires/go-proxyproto"
|
||||
log "github.com/sirupsen/logrus"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestWrapProxyProtocol_OverridesRemoteAddr(t *testing.T) {
|
||||
srv := &Server{
|
||||
Logger: log.StandardLogger(),
|
||||
TrustedProxies: []netip.Prefix{netip.MustParsePrefix("127.0.0.1/32")},
|
||||
ProxyProtocol: true,
|
||||
}
|
||||
|
||||
raw, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
require.NoError(t, err)
|
||||
defer raw.Close()
|
||||
|
||||
ln := srv.wrapProxyProtocol(raw)
|
||||
|
||||
realClientIP := "203.0.113.50"
|
||||
realClientPort := uint16(54321)
|
||||
|
||||
accepted := make(chan net.Conn, 1)
|
||||
go func() {
|
||||
conn, err := ln.Accept()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
accepted <- conn
|
||||
}()
|
||||
|
||||
// Connect and send a PROXY v2 header.
|
||||
conn, err := net.Dial("tcp", ln.Addr().String())
|
||||
require.NoError(t, err)
|
||||
defer conn.Close()
|
||||
|
||||
header := &proxyproto.Header{
|
||||
Version: 2,
|
||||
Command: proxyproto.PROXY,
|
||||
TransportProtocol: proxyproto.TCPv4,
|
||||
SourceAddr: &net.TCPAddr{IP: net.ParseIP(realClientIP), Port: int(realClientPort)},
|
||||
DestinationAddr: &net.TCPAddr{IP: net.ParseIP("10.0.0.1"), Port: 443},
|
||||
}
|
||||
_, err = header.WriteTo(conn)
|
||||
require.NoError(t, err)
|
||||
|
||||
select {
|
||||
case accepted := <-accepted:
|
||||
defer accepted.Close()
|
||||
host, _, err := net.SplitHostPort(accepted.RemoteAddr().String())
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, realClientIP, host, "RemoteAddr should reflect the PROXY header source IP")
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("timed out waiting for connection")
|
||||
}
|
||||
}
|
||||
|
||||
func TestProxyProtocolPolicy_TrustedRequires(t *testing.T) {
|
||||
srv := &Server{
|
||||
Logger: log.StandardLogger(),
|
||||
TrustedProxies: []netip.Prefix{netip.MustParsePrefix("10.0.0.0/8")},
|
||||
}
|
||||
|
||||
opts := proxyproto.ConnPolicyOptions{
|
||||
Upstream: &net.TCPAddr{IP: net.ParseIP("10.0.0.1"), Port: 1234},
|
||||
}
|
||||
policy, err := srv.proxyProtocolPolicy(opts)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, proxyproto.REQUIRE, policy, "trusted source should require PROXY header")
|
||||
}
|
||||
|
||||
func TestProxyProtocolPolicy_UntrustedIgnores(t *testing.T) {
|
||||
srv := &Server{
|
||||
Logger: log.StandardLogger(),
|
||||
TrustedProxies: []netip.Prefix{netip.MustParsePrefix("10.0.0.0/8")},
|
||||
}
|
||||
|
||||
opts := proxyproto.ConnPolicyOptions{
|
||||
Upstream: &net.TCPAddr{IP: net.ParseIP("203.0.113.50"), Port: 1234},
|
||||
}
|
||||
policy, err := srv.proxyProtocolPolicy(opts)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, proxyproto.IGNORE, policy, "untrusted source should have PROXY header ignored")
|
||||
}
|
||||
|
||||
func TestProxyProtocolPolicy_InvalidIPRejects(t *testing.T) {
|
||||
srv := &Server{
|
||||
Logger: log.StandardLogger(),
|
||||
TrustedProxies: []netip.Prefix{netip.MustParsePrefix("10.0.0.0/8")},
|
||||
}
|
||||
|
||||
opts := proxyproto.ConnPolicyOptions{
|
||||
Upstream: &net.UnixAddr{Name: "/tmp/test.sock", Net: "unix"},
|
||||
}
|
||||
policy, err := srv.proxyProtocolPolicy(opts)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, proxyproto.REJECT, policy, "unparsable address should be rejected")
|
||||
}
|
||||
180
proxy/server.go
180
proxy/server.go
@@ -23,6 +23,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/cenkalti/backoff/v4"
|
||||
proxyproto "github.com/pires/go-proxyproto"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"github.com/prometheus/client_golang/prometheus/promhttp"
|
||||
log "github.com/sirupsen/logrus"
|
||||
@@ -92,7 +93,7 @@ type Server struct {
|
||||
DebugEndpointEnabled bool
|
||||
// DebugEndpointAddress is the address for the debug HTTP endpoint (default: ":8444").
|
||||
DebugEndpointAddress string
|
||||
// HealthAddress is the address for the health probe endpoint (default: "localhost:8080").
|
||||
// HealthAddress is the address for the health probe endpoint.
|
||||
HealthAddress string
|
||||
// ProxyToken is the access token for authenticating with the management server.
|
||||
ProxyToken string
|
||||
@@ -107,6 +108,10 @@ type Server struct {
|
||||
// random OS-assigned port. A fixed port only works with single-account
|
||||
// deployments; multiple accounts will fail to bind the same port.
|
||||
WireguardPort int
|
||||
// ProxyProtocol enables PROXY protocol (v1/v2) on TCP listeners.
|
||||
// When enabled, the real client IP is extracted from the PROXY header
|
||||
// sent by upstream L4 proxies that support PROXY protocol.
|
||||
ProxyProtocol bool
|
||||
}
|
||||
|
||||
// NotifyStatus sends a status update to management about tunnel connectivity
|
||||
@@ -137,23 +142,8 @@ func (s *Server) NotifyCertificateIssued(ctx context.Context, accountID, service
|
||||
}
|
||||
|
||||
func (s *Server) ListenAndServe(ctx context.Context, addr string) (err error) {
|
||||
s.startTime = time.Now()
|
||||
s.initDefaults()
|
||||
|
||||
// If no ID is set then one can be generated.
|
||||
if s.ID == "" {
|
||||
s.ID = "netbird-proxy-" + s.startTime.Format("20060102150405")
|
||||
}
|
||||
// Fallback version option in case it is not set.
|
||||
if s.Version == "" {
|
||||
s.Version = "dev"
|
||||
}
|
||||
|
||||
// If no logger is specified fallback to the standard logger.
|
||||
if s.Logger == nil {
|
||||
s.Logger = log.StandardLogger()
|
||||
}
|
||||
|
||||
// Start up metrics gathering
|
||||
reg := prometheus.NewRegistry()
|
||||
s.meter = metrics.New(reg)
|
||||
|
||||
@@ -189,40 +179,11 @@ func (s *Server) ListenAndServe(ctx context.Context, addr string) (err error) {
|
||||
|
||||
s.healthChecker = health.NewChecker(s.Logger, s.netbird)
|
||||
|
||||
if s.DebugEndpointEnabled {
|
||||
debugAddr := debugEndpointAddr(s.DebugEndpointAddress)
|
||||
debugHandler := debug.NewHandler(s.netbird, s.healthChecker, s.Logger)
|
||||
if s.acme != nil {
|
||||
debugHandler.SetCertStatus(s.acme)
|
||||
}
|
||||
s.debug = &http.Server{
|
||||
Addr: debugAddr,
|
||||
Handler: debugHandler,
|
||||
ErrorLog: newHTTPServerLogger(s.Logger, logtagValueDebug),
|
||||
}
|
||||
go func() {
|
||||
s.Logger.Infof("starting debug endpoint on %s", debugAddr)
|
||||
if err := s.debug.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
|
||||
s.Logger.Errorf("debug endpoint error: %v", err)
|
||||
}
|
||||
}()
|
||||
}
|
||||
s.startDebugEndpoint()
|
||||
|
||||
// Start health probe server.
|
||||
healthAddr := s.HealthAddress
|
||||
if healthAddr == "" {
|
||||
healthAddr = "localhost:8080"
|
||||
if err := s.startHealthServer(reg); err != nil {
|
||||
return err
|
||||
}
|
||||
s.healthServer = health.NewServer(healthAddr, s.healthChecker, s.Logger, promhttp.HandlerFor(reg, promhttp.HandlerOpts{}))
|
||||
healthListener, err := net.Listen("tcp", healthAddr)
|
||||
if err != nil {
|
||||
return fmt.Errorf("health probe server listen on %s: %w", healthAddr, err)
|
||||
}
|
||||
go func() {
|
||||
if err := s.healthServer.Serve(healthListener); err != nil && !errors.Is(err, http.ErrServerClosed) {
|
||||
s.Logger.Errorf("health probe server: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
// Start the reverse proxy HTTPS server.
|
||||
s.https = &http.Server{
|
||||
@@ -232,10 +193,19 @@ func (s *Server) ListenAndServe(ctx context.Context, addr string) (err error) {
|
||||
ErrorLog: newHTTPServerLogger(s.Logger, logtagValueHTTPS),
|
||||
}
|
||||
|
||||
lc := net.ListenConfig{}
|
||||
ln, err := lc.Listen(ctx, "tcp", addr)
|
||||
if err != nil {
|
||||
return fmt.Errorf("listen on %s: %w", addr, err)
|
||||
}
|
||||
if s.ProxyProtocol {
|
||||
ln = s.wrapProxyProtocol(ln)
|
||||
}
|
||||
|
||||
httpsErr := make(chan error, 1)
|
||||
go func() {
|
||||
s.Logger.Debugf("starting reverse proxy server on %s", addr)
|
||||
httpsErr <- s.https.ListenAndServeTLS("", "")
|
||||
httpsErr <- s.https.ServeTLS(ln, "", "")
|
||||
}()
|
||||
|
||||
select {
|
||||
@@ -251,7 +221,115 @@ func (s *Server) ListenAndServe(ctx context.Context, addr string) (err error) {
|
||||
}
|
||||
}
|
||||
|
||||
// initDefaults sets fallback values for optional Server fields.
|
||||
func (s *Server) initDefaults() {
|
||||
s.startTime = time.Now()
|
||||
|
||||
// If no ID is set then one can be generated.
|
||||
if s.ID == "" {
|
||||
s.ID = "netbird-proxy-" + s.startTime.Format("20060102150405")
|
||||
}
|
||||
// Fallback version option in case it is not set.
|
||||
if s.Version == "" {
|
||||
s.Version = "dev"
|
||||
}
|
||||
|
||||
// If no logger is specified fallback to the standard logger.
|
||||
if s.Logger == nil {
|
||||
s.Logger = log.StandardLogger()
|
||||
}
|
||||
}
|
||||
|
||||
// startDebugEndpoint launches the debug HTTP server if enabled.
|
||||
func (s *Server) startDebugEndpoint() {
|
||||
if !s.DebugEndpointEnabled {
|
||||
return
|
||||
}
|
||||
debugAddr := debugEndpointAddr(s.DebugEndpointAddress)
|
||||
debugHandler := debug.NewHandler(s.netbird, s.healthChecker, s.Logger)
|
||||
if s.acme != nil {
|
||||
debugHandler.SetCertStatus(s.acme)
|
||||
}
|
||||
s.debug = &http.Server{
|
||||
Addr: debugAddr,
|
||||
Handler: debugHandler,
|
||||
ErrorLog: newHTTPServerLogger(s.Logger, logtagValueDebug),
|
||||
}
|
||||
go func() {
|
||||
s.Logger.Infof("starting debug endpoint on %s", debugAddr)
|
||||
if err := s.debug.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
|
||||
s.Logger.Errorf("debug endpoint error: %v", err)
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// startHealthServer launches the health probe and metrics server.
|
||||
func (s *Server) startHealthServer(reg *prometheus.Registry) error {
|
||||
healthAddr := s.HealthAddress
|
||||
if healthAddr == "" {
|
||||
healthAddr = defaultHealthAddr
|
||||
}
|
||||
s.healthServer = health.NewServer(healthAddr, s.healthChecker, s.Logger, promhttp.HandlerFor(reg, promhttp.HandlerOpts{}))
|
||||
healthListener, err := net.Listen("tcp", healthAddr)
|
||||
if err != nil {
|
||||
return fmt.Errorf("health probe server listen on %s: %w", healthAddr, err)
|
||||
}
|
||||
go func() {
|
||||
if err := s.healthServer.Serve(healthListener); err != nil && !errors.Is(err, http.ErrServerClosed) {
|
||||
s.Logger.Errorf("health probe server: %v", err)
|
||||
}
|
||||
}()
|
||||
return nil
|
||||
}
|
||||
|
||||
// wrapProxyProtocol wraps a listener with PROXY protocol support.
|
||||
// When TrustedProxies is configured, only those sources may send PROXY headers;
|
||||
// connections from untrusted sources have any PROXY header ignored.
|
||||
func (s *Server) wrapProxyProtocol(ln net.Listener) net.Listener {
|
||||
ppListener := &proxyproto.Listener{
|
||||
Listener: ln,
|
||||
ReadHeaderTimeout: proxyProtoHeaderTimeout,
|
||||
}
|
||||
if len(s.TrustedProxies) > 0 {
|
||||
ppListener.ConnPolicy = s.proxyProtocolPolicy
|
||||
} else {
|
||||
s.Logger.Warn("PROXY protocol enabled without trusted proxies; any source may send PROXY headers")
|
||||
}
|
||||
s.Logger.Info("PROXY protocol enabled on listener")
|
||||
return ppListener
|
||||
}
|
||||
|
||||
// proxyProtocolPolicy returns whether to require, skip, or reject the PROXY
|
||||
// header based on whether the connection source is in TrustedProxies.
|
||||
func (s *Server) proxyProtocolPolicy(opts proxyproto.ConnPolicyOptions) (proxyproto.Policy, error) {
|
||||
// No logging on reject to prevent abuse
|
||||
tcpAddr, ok := opts.Upstream.(*net.TCPAddr)
|
||||
if !ok {
|
||||
return proxyproto.REJECT, nil
|
||||
}
|
||||
addr, ok := netip.AddrFromSlice(tcpAddr.IP)
|
||||
if !ok {
|
||||
return proxyproto.REJECT, nil
|
||||
}
|
||||
addr = addr.Unmap()
|
||||
|
||||
// called per accept
|
||||
for _, prefix := range s.TrustedProxies {
|
||||
if prefix.Contains(addr) {
|
||||
return proxyproto.REQUIRE, nil
|
||||
}
|
||||
}
|
||||
return proxyproto.IGNORE, nil
|
||||
}
|
||||
|
||||
const (
|
||||
defaultHealthAddr = "localhost:8080"
|
||||
defaultDebugAddr = "localhost:8444"
|
||||
|
||||
// proxyProtoHeaderTimeout is the deadline for reading the PROXY protocol
|
||||
// header after accepting a connection.
|
||||
proxyProtoHeaderTimeout = 5 * time.Second
|
||||
|
||||
// shutdownPreStopDelay is the time to wait after receiving a shutdown signal
|
||||
// before draining connections. This allows the load balancer to propagate
|
||||
// the endpoint removal.
|
||||
@@ -647,7 +725,7 @@ func (s *Server) protoToMapping(mapping *proto.ProxyMapping) proxy.Mapping {
|
||||
// If addr is empty, it defaults to localhost:8444 for security.
|
||||
func debugEndpointAddr(addr string) string {
|
||||
if addr == "" {
|
||||
return "localhost:8444"
|
||||
return defaultDebugAddr
|
||||
}
|
||||
return addr
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user