diff --git a/cmd/cli/dns_intercept_darwin.go b/cmd/cli/dns_intercept_darwin.go index 7b72df9..364a29a 100644 --- a/cmd/cli/dns_intercept_darwin.go +++ b/cmd/cli/dns_intercept_darwin.go @@ -41,6 +41,11 @@ const ( // 2s re-check misses. pfAnchorRecheckDelayLong = 4 * time.Second + // pfExecFailureBackoff suppresses repeated pfctl/scutil checks briefly after + // macOS reports local resource exhaustion. Without this, network-change storms + // can turn a pf ruleset race into a fork/file-descriptor exhaustion loop. + pfExecFailureBackoff = 5 * time.Second + // pfVPNInterfacePrefixes lists interface name prefixes that indicate VPN/tunnel // interfaces on macOS. Used to add interface-specific DNS intercept rules so that // VPN software with "pass out quick on " rules cannot bypass our intercept. @@ -1303,6 +1308,15 @@ func (p *prog) ensurePFAnchorActive() bool { if p.dnsInterceptState == nil { return false } + if !p.pfEnsureRunning.CompareAndSwap(false, true) { + mainLog.Load().Debug().Msg("DNS intercept watchdog: check already running, skipping duplicate") + return false + } + defer p.pfEnsureRunning.Store(false) + + if p.pfExecBackoffActive() { + return false + } // While stabilizing (VPN connecting), suppress all restores. // The stabilization loop will restore once pf settles. @@ -1336,6 +1350,9 @@ func (p *prog) ensurePFAnchorActive() bool { // Check 1: anchor references in the main ruleset. natOut, err := exec.Command("pfctl", "-sn").CombinedOutput() if err != nil { + if p.pfBackoffResourceExhaustion(err, natOut, "dump NAT rules") { + return false + } mainLog.Load().Warn().Err(err).Msg("DNS intercept watchdog: could not dump NAT rules") return false } @@ -1348,6 +1365,9 @@ func (p *prog) ensurePFAnchorActive() bool { if !needsRestore { filterOut, err := exec.Command("pfctl", "-sr").CombinedOutput() if err != nil { + if p.pfBackoffResourceExhaustion(err, filterOut, "dump filter rules") { + return false + } mainLog.Load().Warn().Err(err).Msg("DNS intercept watchdog: could not dump filter rules") return false } @@ -1365,6 +1385,9 @@ func (p *prog) ensurePFAnchorActive() bool { if !needsRestore { anchorFilter, err := exec.Command("pfctl", "-a", pfAnchorName, "-sr").CombinedOutput() if err != nil || len(strings.TrimSpace(string(anchorFilter))) == 0 { + if p.pfBackoffResourceExhaustion(err, anchorFilter, "dump anchor filter rules") { + return false + } mainLog.Load().Warn().Msg("DNS intercept watchdog: anchor has no filter rules — content was flushed") needsRestore = true } @@ -1372,6 +1395,9 @@ func (p *prog) ensurePFAnchorActive() bool { if !needsRestore { anchorNat, err := exec.Command("pfctl", "-a", pfAnchorName, "-sn").CombinedOutput() if err != nil || len(strings.TrimSpace(string(anchorNat))) == 0 { + if p.pfBackoffResourceExhaustion(err, anchorNat, "dump anchor NAT rules") { + return false + } mainLog.Load().Warn().Msg("DNS intercept watchdog: anchor has no rdr rules — translation was flushed (will cause packet loop on lo0)") needsRestore = true } @@ -1405,6 +1431,7 @@ func (p *prog) ensurePFAnchorActive() bool { // Restore: re-inject anchor references into the main ruleset. mainLog.Load().Info().Msg("DNS intercept watchdog: restoring pf anchor references") if err := p.ensurePFAnchorReference(); err != nil { + p.pfBackoffResourceExhaustion(err, nil, "restore anchor references") mainLog.Load().Error().Err(err).Msg("DNS intercept watchdog: failed to restore anchor references") return true } @@ -1421,6 +1448,7 @@ func (p *prog) ensurePFAnchorActive() bool { if err := os.WriteFile(pfAnchorFile, []byte(rulesStr), 0644); err != nil { mainLog.Load().Error().Err(err).Msg("DNS intercept watchdog: failed to write anchor file") } else if out, err := exec.Command("pfctl", "-a", pfAnchorName, "-f", pfAnchorFile).CombinedOutput(); err != nil { + p.pfBackoffResourceExhaustion(err, out, "load rebuilt anchor") mainLog.Load().Error().Err(err).Msgf("DNS intercept watchdog: failed to load rebuilt anchor (output: %s)", strings.TrimSpace(string(out))) } else { flushPFStates() @@ -1458,6 +1486,49 @@ func (p *prog) scheduleDNSAfterVPNSettleRefresh(reason string, delay time.Durati }) } +func (p *prog) pfExecBackoffActive() bool { + until := p.pfExecBackoffUntil.Load() + if until == 0 { + return false + } + remaining := time.Until(time.UnixMilli(until)) + if remaining <= 0 { + p.pfExecBackoffUntil.CompareAndSwap(until, 0) + return false + } + mainLog.Load().Debug().Dur("remaining", remaining).Msg("DNS intercept watchdog: suppressed during pf exec backoff") + return true +} + +func (p *prog) pfBackoffResourceExhaustion(err error, output []byte, operation string) bool { + if !isResourceExhaustion(err, output) { + return false + } + until := time.Now().Add(pfExecFailureBackoff) + p.pfExecBackoffUntil.Store(until.UnixMilli()) + mainLog.Load().Warn().Err(err).Dur("backoff", pfExecFailureBackoff).Str("operation", operation). + Msg("DNS intercept watchdog: backing off after local exec resource exhaustion") + return true +} + +func isResourceExhaustion(err error, output []byte) bool { + if err == nil && len(output) == 0 { + return false + } + var msg string + if err != nil { + msg = err.Error() + } + if len(output) > 0 { + msg += "\n" + string(output) + } + msg = strings.ToLower(msg) + return strings.Contains(msg, "resource temporarily unavailable") || + strings.Contains(msg, "too many open files") || + strings.Contains(msg, "too many processes") || + strings.Contains(msg, "cannot allocate memory") +} + // pfWatchdog periodically checks that our pf anchor is still active. // Other programs (e.g., Windscribe desktop app, macOS configd) can replace // scheduleDelayedRechecks schedules delayed re-checks after a network change event. @@ -1470,8 +1541,19 @@ func (p *prog) scheduleDNSAfterVPNSettleRefresh(reason string, delay time.Durati // // Two delays (2s and 4s) cover both fast and slow VPN teardowns. func (p *prog) scheduleDelayedRechecks() { + p.pfDelayedRecheckMu.Lock() + defer p.pfDelayedRecheckMu.Unlock() + + for _, timer := range p.pfDelayedRecheckTimers { + if timer != nil { + timer.Stop() + } + } + p.pfDelayedRecheckTimers = p.pfDelayedRecheckTimers[:0] + for _, delay := range []time.Duration{pfAnchorRecheckDelay, pfAnchorRecheckDelayLong} { - time.AfterFunc(delay, func() { + delay := delay + timer := time.AfterFunc(delay, func() { if p.dnsInterceptState == nil || p.pfStabilizing.Load() { return } @@ -1485,6 +1567,7 @@ func (p *prog) scheduleDelayedRechecks() { p.vpnDNS.Refresh(true) } }) + p.pfDelayedRecheckTimers = append(p.pfDelayedRecheckTimers, timer) } } diff --git a/cmd/cli/dns_intercept_darwin_test.go b/cmd/cli/dns_intercept_darwin_test.go index d0834d7..c8d90b0 100644 --- a/cmd/cli/dns_intercept_darwin_test.go +++ b/cmd/cli/dns_intercept_darwin_test.go @@ -3,6 +3,7 @@ package cli import ( + "errors" "strings" "testing" @@ -141,3 +142,47 @@ func TestPFAddressFamily(t *testing.T) { } } } + +func TestIsResourceExhaustion(t *testing.T) { + tests := []struct { + name string + err error + output []byte + want bool + }{ + { + name: "exec start failure", + err: errors.New("fork/exec /sbin/pfctl: resource temporarily unavailable"), + want: true, + }, + { + name: "fd exhaustion from stderr output", + err: errors.New("exit status 1"), + output: []byte("pfctl: Pipe: Too many open files"), + want: true, + }, + { + name: "process exhaustion from wrapped restore error", + err: errors.New("failed to dump running filter rules: exit status 1 (output: too many processes)"), + want: true, + }, + { + name: "ordinary pf syntax failure", + err: errors.New("exit status 1"), + output: []byte("pfctl: syntax error"), + want: false, + }, + { + name: "nil error and empty output", + want: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := isResourceExhaustion(tt.err, tt.output); got != tt.want { + t.Fatalf("isResourceExhaustion() = %v, want %v", got, tt.want) + } + }) + } +} diff --git a/cmd/cli/dns_proxy.go b/cmd/cli/dns_proxy.go index fe37f60..e8e238c 100644 --- a/cmd/cli/dns_proxy.go +++ b/cmd/cli/dns_proxy.go @@ -1937,6 +1937,9 @@ func (p *prog) handleRecovery(reason RecoveryReason) { // waitForUpstreamRecovery checks the provided upstreams concurrently until one recovers. // It returns the name of the recovered upstream or an error if the check times out. func (p *prog) waitForUpstreamRecovery(ctx context.Context, upstreams map[string]*ctrld.UpstreamConfig) (string, error) { + recoveryCtx, cancel := context.WithCancel(ctx) + defer cancel() + recoveredCh := make(chan string, 1) var wg sync.WaitGroup @@ -1950,7 +1953,7 @@ func (p *prog) waitForUpstreamRecovery(ctx context.Context, upstreams map[string attempts := 0 for { select { - case <-ctx.Done(): + case <-recoveryCtx.Done(): mainLog.Load().Debug().Msgf("Context canceled for upstream %s", name) return default: @@ -1962,13 +1965,16 @@ func (p *prog) waitForUpstreamRecovery(ctx context.Context, upstreams map[string select { case recoveredCh <- name: mainLog.Load().Debug().Msgf("Sent recovery notification for upstream %s", name) + cancel() default: mainLog.Load().Debug().Msg("Recovery channel full, another upstream already recovered") } return } mainLog.Load().Debug().Msgf("Upstream %s check failed, sleeping before retry", name) - time.Sleep(checkUpstreamBackoffSleep) + if !sleepWithContext(recoveryCtx, checkUpstreamBackoffSleep) { + return + } // if this is the upstreamOS and it's the 3rd attempt (or multiple of 3), // we should try to reinit the OS resolver to ensure we can recover @@ -1996,6 +2002,17 @@ func (p *prog) waitForUpstreamRecovery(ctx context.Context, upstreams map[string return recovered, nil } +func sleepWithContext(ctx context.Context, d time.Duration) bool { + timer := time.NewTimer(d) + defer timer.Stop() + select { + case <-timer.C: + return true + case <-ctx.Done(): + return false + } +} + // buildRecoveryUpstreams constructs the map of upstream configurations to test. // For OS failures we supply the manual OS resolver upstream configuration. // For network change or regular failure we use the upstreams defined in p.cfg (ignoring OS). diff --git a/cmd/cli/prog.go b/cmd/cli/prog.go index 5c5b275..1bd1680 100644 --- a/cmd/cli/prog.go +++ b/cmd/cli/prog.go @@ -188,6 +188,21 @@ type prog struct { // interception with exponential backoff and auto-heals if broken. pfMonitorRunning atomic.Bool //lint:ignore U1000 used on darwin + // pfEnsureRunning ensures only one pf anchor validation/restoration runs at a time. + // Network-change callbacks, delayed rechecks, and the periodic watchdog can all + // converge during macOS interface churn; concurrent pfctl/scutil exec storms can + // exhaust process/file limits and make the outage worse. + pfEnsureRunning atomic.Bool //lint:ignore U1000 used on darwin + + // pfExecBackoffUntil suppresses pf anchor validation after pfctl/scutil execs + // fail due host resource exhaustion (fork unavailable, too many open files). + pfExecBackoffUntil atomic.Int64 //lint:ignore U1000 used on darwin + + // pfDelayedRecheckTimers coalesces delayed DNS-intercept rechecks after noisy + // network changes. Protected by pfDelayedRecheckMu. + pfDelayedRecheckMu sync.Mutex //lint:ignore U1000 used on darwin + pfDelayedRecheckTimers []*time.Timer //lint:ignore U1000 used on darwin + // pfProbeExpected holds the domain name of a pending pf interception probe. // When non-empty, the DNS handler checks incoming queries against this value // and signals pfProbeCh if matched. The probe verifies that pf's rdr rules @@ -1355,6 +1370,7 @@ func errNetworkError(err error) bool { case errors.Is(opErr.Err, syscall.ECONNREFUSED), errors.Is(opErr.Err, syscall.EINVAL), errors.Is(opErr.Err, syscall.ENETUNREACH), + errors.Is(opErr.Err, syscall.EHOSTUNREACH), errors.Is(opErr.Err, windowsENETUNREACH), errors.Is(opErr.Err, windowsEINVAL), errors.Is(opErr.Err, windowsECONNREFUSED), diff --git a/cmd/cli/prog_test.go b/cmd/cli/prog_test.go index 6831f5a..c8c17a8 100644 --- a/cmd/cli/prog_test.go +++ b/cmd/cli/prog_test.go @@ -1,7 +1,11 @@ package cli import ( + "context" + "net" + "net/url" "runtime" + "syscall" "testing" "time" @@ -12,6 +16,23 @@ import ( "github.com/Control-D-Inc/ctrld" ) +func TestErrNetworkErrorTreatsNoRouteAsNetworkError(t *testing.T) { + err := &net.OpError{Op: "dial", Net: "tcp", Err: syscall.EHOSTUNREACH} + assert.True(t, errNetworkError(err)) + assert.True(t, errUrlNetworkError(&url.Error{Op: "Get", URL: "https://dns.controld.com", Err: err})) +} + +func TestSleepWithContext(t *testing.T) { + assert.True(t, sleepWithContext(context.Background(), time.Millisecond)) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + start := time.Now() + assert.False(t, sleepWithContext(ctx, time.Minute)) + assert.Less(t, time.Since(start), 100*time.Millisecond) +} + func Test_prog_dnsWatchdogEnabled(t *testing.T) { p := &prog{cfg: &ctrld.Config{}} diff --git a/cmd/cli/vpn_dns.go b/cmd/cli/vpn_dns.go index f6f8c20..115ed80 100644 --- a/cmd/cli/vpn_dns.go +++ b/cmd/cli/vpn_dns.go @@ -6,6 +6,7 @@ import ( "runtime" "strings" "sync" + "sync/atomic" "github.com/rs/zerolog" "tailscale.com/net/netmon" @@ -50,6 +51,9 @@ type vpnDNSManager struct { // discoverVPNDNS is injected for tests so Refresh does not depend on the // runner host's real VPN/virtual adapter state. discoverVPNDNS func(context.Context) []ctrld.VPNDNSConfig + // refreshRunning keeps noisy network-change storms from running overlapping + // scutil/networksetup VPN DNS discovery work. + refreshRunning atomic.Bool // Called when VPN DNS server list changes, to update intercept exemptions. onServersChanged vpnDNSExemptFunc } @@ -69,6 +73,11 @@ func newVPNDNSManager(exemptFunc vpnDNSExemptFunc) *vpnDNSManager { // Called on network change events. func (m *vpnDNSManager) Refresh(guardAgainstNoNameservers bool) { logger := mainLog.Load() + if !m.refreshRunning.CompareAndSwap(false, true) { + logger.Debug().Msg("VPN DNS refresh already running, skipping duplicate") + return + } + defer m.refreshRunning.Store(false) logger.Debug().Msg("Refreshing VPN DNS configurations") discoverVPNDNS := m.discoverVPNDNS diff --git a/cmd/cli/vpn_dns_test.go b/cmd/cli/vpn_dns_test.go index 6a6d520..846dc8e 100644 --- a/cmd/cli/vpn_dns_test.go +++ b/cmd/cli/vpn_dns_test.go @@ -2,6 +2,8 @@ package cli import ( "context" + "sync" + "sync/atomic" "testing" "github.com/Control-D-Inc/ctrld" @@ -14,6 +16,36 @@ func withVPNDNSSettlingEnabled(t *testing.T) { t.Cleanup(func() { vpnDNSSettlingEnabled = old }) } +func TestVPNDNSRefreshSkipsConcurrentDuplicate(t *testing.T) { + m := newVPNDNSManager(nil) + started := make(chan struct{}) + release := make(chan struct{}) + done := make(chan struct{}) + var once sync.Once + var calls atomic.Int32 + + m.discoverVPNDNS = func(context.Context) []ctrld.VPNDNSConfig { + calls.Add(1) + once.Do(func() { close(started) }) + <-release + return nil + } + + go func() { + defer close(done) + m.Refresh(true) + }() + + <-started + m.Refresh(true) + close(release) + <-done + + if calls.Load() != 1 { + t.Fatalf("expected overlapping refresh to be skipped, got %d discovery calls", calls.Load()) + } +} + func TestVPNDNSRefreshRetainsStateForOneGuardedEmptyDiscovery(t *testing.T) { withVPNDNSSettlingEnabled(t) var gotExemptions []vpnDNSExemption diff --git a/resolver.go b/resolver.go index fbd2ad6..2284d04 100644 --- a/resolver.go +++ b/resolver.go @@ -45,8 +45,12 @@ const ( const controldPublicDns = "76.76.2.0" +const maxConcurrentOSResolverExchanges = 128 + var controldPublicDnsWithPort = net.JoinHostPort(controldPublicDns, "53") +var osResolverExchangeSem = make(chan struct{}, maxConcurrentOSResolverExchanges) + var localResolver Resolver func init() { @@ -136,14 +140,15 @@ func availableNameservers() []string { // It's the caller's responsibility to ensure the system DNS is in a clean state before // calling this function. func InitializeOsResolver(guardAgainstNoNameservers bool) []string { + resolverMutex.Lock() + defer resolverMutex.Unlock() + nameservers := availableNameservers() // if no nameservers, return empty slice so we dont remove all nameservers if len(nameservers) == 0 && guardAgainstNoNameservers { return []string{} } ns := initializeOsResolver(nameservers) - resolverMutex.Lock() - defer resolverMutex.Unlock() or = newResolverWithNameserver(ns) return ns } @@ -466,6 +471,13 @@ func (o *osResolver) resolve(ctx context.Context, msg *dns.Msg) (*dns.Msg, error for _, server := range servers { go func(server string) { defer wg.Done() + release, ok := acquireOSResolverExchangeSlot(ctx) + if !ok { + ch <- &osResolverResult{err: ctx.Err(), server: server, lan: isLan} + return + } + defer release() + var answer *dns.Msg var err error var localOSResolverIP net.IP @@ -576,6 +588,15 @@ func (o *osResolver) resolve(ctx context.Context, msg *dns.Msg) (*dns.Msg, error return nil, errors.Join(errs...) } +func acquireOSResolverExchangeSlot(ctx context.Context) (func(), bool) { + select { + case osResolverExchangeSem <- struct{}{}: + return func() { <-osResolverExchangeSem }, true + case <-ctx.Done(): + return nil, false + } +} + func (o *osResolver) removeCache(key string) { o.cache.Delete(key) }