From 246a54020de578ca84fdc489a4baa5f9e12b2dc3 Mon Sep 17 00:00:00 2001 From: Cuong Manh Le Date: Wed, 19 Aug 2026 16:59:28 +0700 Subject: [PATCH] fix(darwin): probe interception after the host wakes Rule text survives a suspend unchanged, so the watchdog reports the pf anchor intact while nothing translates, and the probe monitor is only armed by interface changes with a schedule frozen through the sleep. Detect the resume from the wall-clock gap and probe from there, with a bounded repair: at most two forced reloads across the retry schedule. Refresh the OS resolver before every probe - the probe aims at the first OS nameserver and passes when it has none, so a pre-sleep list makes it meaningless. Stand down for stabilization, which verifies on completion. --- cmd/cli/dns_intercept_darwin.go | 273 ++++++++++++++--- cmd/cli/dns_intercept_darwin_test.go | 422 +++++++++++++++++++++++++++ cmd/cli/prog.go | 6 + cmd/cli/suspend_detector.go | 109 +++++++ cmd/cli/suspend_detector_test.go | 208 +++++++++++++ 5 files changed, 981 insertions(+), 37 deletions(-) create mode 100644 cmd/cli/suspend_detector.go create mode 100644 cmd/cli/suspend_detector_test.go diff --git a/cmd/cli/dns_intercept_darwin.go b/cmd/cli/dns_intercept_darwin.go index f0c8759..7af3878 100644 --- a/cmd/cli/dns_intercept_darwin.go +++ b/cmd/cli/dns_intercept_darwin.go @@ -251,6 +251,44 @@ func setCtrldGroupID(gid int) error { func (p *prog) startDNSIntercept() error { mainLog.Load().Info().Msg("DNS intercept: initializing macOS packet filter (pf) redirect") + if err := installPFInterceptFn(p); err != nil { + return err + } + + p.dnsInterceptState = &pfState{ + anchorFile: pfAnchorFile, + anchorName: pfAnchorName, + } + + // Store the initial set of tunnel interfaces so we can detect changes later. + p.mu.Lock() + p.lastTunnelIfaces = discoverTunnelInterfacesForReconcile() + p.pendingTunnelIfaces = nil + p.hasPendingTunnelIfaces = false + p.mu.Unlock() + + lc := p.cfg.FirstListener() + if lc != nil { + mainLog.Load().Info().Msgf("DNS intercept: pf redirect active - all outbound DNS (port 53) redirected to %s:%d via anchor %q", lc.IP, lc.Port, pfAnchorName) + } else { + mainLog.Load().Info().Msgf("DNS intercept: pf redirect active - all outbound DNS (port 53) redirected via anchor %q", pfAnchorName) + } + + p.startInterceptBackgroundWork() + + return nil +} + +// installPFInterceptFn is the privileged-install seam. Everything behind it needs root, +// a live pf and a writable /etc/pf.anchors - which is what kept startDNSIntercept, and +// with it the handoff to the background watchers, out of reach of every test. +var installPFInterceptFn = (*prog).installPFIntercept + +// installPFIntercept performs the privileged part of starting interception: validation, +// _ctrld group setup, anchor file write, pfctl load, enable, and post-load verification. +// It returns nil only once pf is authoritatively active, so the caller may publish +// intercept mode; on failure it has already rolled back whatever it applied. +func (p *prog) installPFIntercept() error { if err := p.validateDNSIntercept(); err != nil { return err } @@ -360,32 +398,33 @@ func (p *prog) startDNSIntercept() error { return fmt.Errorf("dns intercept: post-load PF verification failed") } - p.dnsInterceptState = &pfState{ - anchorFile: pfAnchorFile, - anchorName: pfAnchorName, - } - - // Store the initial set of tunnel interfaces so we can detect changes later. - p.mu.Lock() - p.lastTunnelIfaces = discoverTunnelInterfacesForReconcile() - p.pendingTunnelIfaces = nil - p.hasPendingTunnelIfaces = false - p.mu.Unlock() - - lc := p.cfg.FirstListener() - if lc != nil { - mainLog.Load().Info().Msgf("DNS intercept: pf redirect active — all outbound DNS (port 53) redirected to %s:%d via anchor %q", lc.IP, lc.Port, pfAnchorName) - } else { - mainLog.Load().Info().Msgf("DNS intercept: pf redirect active — all outbound DNS (port 53) redirected via anchor %q", pfAnchorName) - } - - // Start the pf watchdog to detect and restore rules if another program - // (e.g., Windscribe desktop, macOS configd) replaces the pf ruleset. - go p.pfWatchdog() - return nil } +// pfWatchdogFn and runSuspendWatcherFn are the background-work seams. Tests use them to +// observe what startInterceptBackgroundWork registers without running either loop, which +// would otherwise reach pfctl. +var ( + pfWatchdogFn = (*prog).pfWatchdog + runSuspendWatcherFn = runSuspendWatcher +) + +// startInterceptBackgroundWork launches the goroutines that guard an active intercept. +// +// Split out of startDNSIntercept so the registration is reachable from a test: +// startDNSIntercept needs root, a live pf and a writable /etc/pf.anchors, so nothing +// covers this wiring if it lives inline there. +func (p *prog) startInterceptBackgroundWork() { + // Detect and restore rules if another program (e.g., Windscribe desktop, macOS + // configd) replaces the pf ruleset. + go pfWatchdogFn(p) + + // A host sleep freezes every timer above it, and rule text survives the suspend + // unchanged, so no existing path notices that pf stopped translating while the host + // was asleep. Watch for the resume itself and probe from there. + go runSuspendWatcherFn(p.stopCh, p.verifyInterceptAfterWake) +} + func (p *prog) rollbackDNSInterceptStart() { if out, err := exec.Command("pfctl", "-a", pfAnchorName, "-F", "all").CombinedOutput(); err != nil { mainLog.Load().Warn().Err(err).Msgf("DNS intercept: startup rollback could not flush anchor (output: %s)", strings.TrimSpace(string(out))) @@ -562,6 +601,10 @@ func (p *prog) stopDNSIntercept() error { } } p.pfDelayedRecheckTimers = nil + if p.pfSettleFollowupTimer != nil { + p.pfSettleFollowupTimer.Stop() + p.pfSettleFollowupTimer = nil + } p.pfDelayedRecheckMu.Unlock() for !p.pfEnsureRunning.CompareAndSwap(false, true) { @@ -1553,6 +1596,146 @@ func (p *prog) verifyInterceptAfterStabilization() { mainLog.Load().Error().Msg("DNS intercept: interception still not translating after the post-stabilization reload — the watchdog will retry") } +// pfWakeProbeDelays is the bounded retry schedule for the post-resume functional probe. +// Each value is the wait before its own attempt, not an offset from the resume, so the +// attempts land at +0s, +2s, +6s and +14s after the suspend is noticed. A resumed host +// brings its link, default route, resolver list and pf translation state back over +// several seconds, so a single probe at +0 decides too early. A var so tests can shorten +// it. +var pfWakeProbeDelays = []time.Duration{0, 2 * time.Second, 4 * time.Second, 8 * time.Second} + +// pfWakeMaxRepairs bounds forced reloads across the whole schedule. QA on the post-wake +// outage saw one forced reload fail to restore translation and a later one succeed, so a +// single attempt is not enough - but this stays a bounded repair rather than a reload +// loop, because every reload flushes pf state and kills in-flight DoH connections. +const pfWakeMaxRepairs = 2 + +// pfWakeProbeOwnerWait bounds how long one attempt waits for another prober to release +// functional-probe ownership. Short on purpose: a holder that is genuinely probing is +// doing this path's job, and there are further attempts left in the schedule. +var pfWakeProbeOwnerWait = 250 * time.Millisecond + +// verifyInterceptAfterWake proves pf is actually translating after the host resumes from +// sleep, and repairs it a bounded number of times when it is not. +// +// Nothing else covers this window. The periodic watchdog compares rule text, which +// survives a suspend unchanged, so it reports the anchor intact while every query through +// the system resolver times out. The interception probe monitor is only armed by +// interface changes and its backoff schedule is frozen through the suspend, so whether it +// probes after a resume depends on where its timers happened to be when the host slept. +// QA measured 18s of no public DNS on a host whose link and default route were back +// within the first second, recovered only because an unrelated VPN reconnect happened to +// drive the post-stabilization verifier; without that reconnect nothing would have probed. +// +// The OS resolver list is refreshed before every probe. probePFIntercept aims at the +// first OS nameserver and reports success when it has no target, so a list still holding +// a pre-sleep VPN resolver turns this into a meaningless pass or a false failure. +func (p *prog) verifyInterceptAfterWake(gap time.Duration) { + if p.dnsInterceptState == nil { + return + } + mainLog.Load().Info().Msgf("DNS intercept: host resumed after a %s gap - verifying interception is still translating", gap.Round(time.Second)) + + repairs := 0 + probed := 0 + for attempt, delay := range pfWakeProbeDelays { + if !p.waitBeforeWakeProbe(delay) { + return + } + if p.dnsInterceptState == nil { + return + } + // Stabilization owns pf while a VPN's ruleset settles, and runs its own + // functional verification when it finishes. Probing or reloading underneath it + // is the mutual overwriting stabilization exists to prevent. + if p.pfStabilizing.Load() { + mainLog.Load().Debug().Msg("DNS intercept: post-wake probe standing down - stabilization owns pf and verifies on completion") + return + } + if p.pfExecBackoffActive() { + mainLog.Load().Debug().Msg("DNS intercept: post-wake probe deferred this attempt - pf exec backoff active") + continue + } + // Eligibility first, ownership second, and release before the next delay: a + // prober that holds the flag while sleeping starves the post-stabilization + // verifier, which reads a held flag as "somebody is probing". + if !p.claimFunctionalProbeOwner(pfWakeProbeOwnerWait) { + mainLog.Load().Debug().Msg("DNS intercept: post-wake probe skipped this attempt - another prober holds ownership") + continue + } + restored, repaired := p.runWakeProbeAttempt(attempt, repairs) + probed++ + p.pfMonitorRunning.Store(false) + if repaired { + repairs++ + } + if restored { + return + } + } + // Report what actually ran: an attempt deferred by pf exec backoff or skipped for + // want of probe ownership never probes, so the schedule length would claim + // verification that did not happen - and with every attempt deferred, nothing was + // measured at all. + if probed == 0 { + mainLog.Load().Warn().Msg("DNS intercept: post-wake verification never probed - every attempt was deferred or skipped - the watchdog will retry") + return + } + mainLog.Load().Error().Msgf("DNS intercept: interception still not translating after resume (%d probe attempts, %d forced reloads) - the watchdog will retry", + probed, repairs) +} + +// waitBeforeWakeProbe waits out one attempt's delay, reporting false when the service +// stopped while waiting. +func (p *prog) waitBeforeWakeProbe(delay time.Duration) bool { + if delay <= 0 { + return true + } + timer := time.NewTimer(delay) + defer timer.Stop() + select { + case <-p.stopCh: + return false + case <-timer.C: + return true + } +} + +// runWakeProbeAttempt runs one probe of the post-resume schedule, forcing a reload when +// the probe fails and repairs so far leave budget for one. It reports whether +// interception is translating and whether this attempt spent a repair. A reload that +// could not run does not count: nothing was changed, so the next attempt may try again. +// The caller holds functional-probe ownership for the call. +func (p *prog) runWakeProbeAttempt(attempt, repairs int) (restored, repaired bool) { + initializeOsResolver(ctrld.LoggerCtx(context.Background(), mainLog.Load()), true) + if probePFInterceptFn(p) { + if attempt > 0 || repairs > 0 { + mainLog.Load().Info().Msgf("DNS intercept: post-wake probe %d/%d passed - interception is translating", attempt+1, len(pfWakeProbeDelays)) + } else { + mainLog.Load().Debug().Msg("DNS intercept: post-wake probe passed - interception is translating") + } + return true, false + } + + if repairs >= pfWakeMaxRepairs { + mainLog.Load().Warn().Msgf("DNS intercept: post-wake probe %d/%d FAILED - repair budget of %d forced reloads is spent, not reloading again", + attempt+1, len(pfWakeProbeDelays), pfWakeMaxRepairs) + return false, false + } + + mainLog.Load().Warn().Msgf("DNS intercept: post-wake probe %d/%d FAILED - rules survived the suspend but pf is not translating, forcing a reload", + attempt+1, len(pfWakeProbeDelays)) + if !forceReloadPFInterceptFn(p) { + mainLog.Load().Error().Msg("DNS intercept: post-wake forced reload did not run") + return false, false + } + if probePFInterceptFn(p) { + mainLog.Load().Info().Msg("DNS intercept: interception restored by the post-wake reload") + return true, true + } + return false, true +} + var runPFAnchorCheckCommand = func(args ...string) ([]byte, error) { return exec.Command("pfctl", args...).CombinedOutput() } @@ -1714,6 +1897,36 @@ func (p *prog) dnsInterceptIgnoredChangeReconcileDue(now time.Time) bool { } } +// scheduleDNSAfterVPNSettleRefresh queues one follow-up VPN DNS refresh. +// +// The timer is tracked rather than fired and forgotten. Teardown has to be able to +// cancel it, and VPN churn can finish stabilization several times in a row, where +// untracked timers stack up into repeats of the same scutil/VPN-DNS work. +func (p *prog) scheduleDNSAfterVPNSettleRefresh(reason string, delay time.Duration) { + p.pfDelayedRecheckMu.Lock() + defer p.pfDelayedRecheckMu.Unlock() + if p.pfSettleFollowupTimer != nil { + p.pfSettleFollowupTimer.Stop() + } + p.pfSettleFollowupTimer = time.AfterFunc(delay, func() { + if p.dnsInterceptState == nil { + return + } + p.refreshDNSAfterVPNSettle(reason) + }) +} + +// stopPFSettleFollowup cancels a pending post-settle refresh. A callback that already +// fired still checks dnsInterceptState, so either teardown order is safe. +func (p *prog) stopPFSettleFollowup() { + p.pfDelayedRecheckMu.Lock() + defer p.pfDelayedRecheckMu.Unlock() + if p.pfSettleFollowupTimer != nil { + p.pfSettleFollowupTimer.Stop() + p.pfSettleFollowupTimer = nil + } +} + func (p *prog) pfExecBackoffActive() bool { until := p.pfExecBackoffUntil.Load() if until == 0 { @@ -1756,20 +1969,6 @@ func isResourceExhaustion(err error, output []byte) bool { strings.Contains(msg, "cannot allocate memory") } -func (p *prog) scheduleDNSAfterVPNSettleRefresh(reason string, delay time.Duration) { - timer := time.AfterFunc(delay, func() { - if p.dnsInterceptState == nil { - return - } - p.refreshDNSAfterVPNSettle(reason) - }) - // Track the timer like the other delayed rechecks, so intercept teardown - // (and test cleanup) can stop it instead of letting it fire afterwards. - p.pfDelayedRecheckMu.Lock() - p.pfDelayedRecheckTimers = append(p.pfDelayedRecheckTimers, timer) - p.pfDelayedRecheckMu.Unlock() -} - // 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. diff --git a/cmd/cli/dns_intercept_darwin_test.go b/cmd/cli/dns_intercept_darwin_test.go index 79e052a..992ec26 100644 --- a/cmd/cli/dns_intercept_darwin_test.go +++ b/cmd/cli/dns_intercept_darwin_test.go @@ -1000,6 +1000,16 @@ func TestFinishPFStabilizationRunsFunctionalVerification(t *testing.T) { t.Errorf("forced reloads = %d, want 1", *reloads) } + stopPFTestTimers(p) +} + +// stopPFTestTimers cancels every timer a stabilization pass can leave pending. +// +// The whole test binary shares one unsynchronised log sink, so a timer that outlives its +// test logs concurrently with whichever test is running next and trips the race detector +// there - reported against the innocent test. The post-settle follow-up is a 4s timer, so +// the collision lands far from its origin. +func stopPFTestTimers(p *prog) { p.pfDelayedRecheckMu.Lock() timers := append([]*time.Timer(nil), p.pfDelayedRecheckTimers...) p.pfDelayedRecheckTimers = nil @@ -1007,4 +1017,416 @@ func TestFinishPFStabilizationRunsFunctionalVerification(t *testing.T) { for _, timer := range timers { timer.Stop() } + p.stopPFSettleFollowup() +} + +// TestScheduleDNSAfterVPNSettleRefreshIsCancellable pins the tracking itself. An +// untracked follow-up cannot be cancelled by teardown or by this helper, which is how a +// 4s timer escaped its test and raced the log sink of a later one. +func TestScheduleDNSAfterVPNSettleRefreshIsCancellable(t *testing.T) { + p := &prog{dnsInterceptState: &pfState{}} + p.scheduleDNSAfterVPNSettleRefresh("test", time.Hour) + + p.pfDelayedRecheckMu.Lock() + tracked := p.pfSettleFollowupTimer != nil + p.pfDelayedRecheckMu.Unlock() + if !tracked { + t.Fatal("the follow-up refresh was scheduled without being tracked; nothing can cancel it") + } + + p.stopPFSettleFollowup() + p.pfDelayedRecheckMu.Lock() + cleared := p.pfSettleFollowupTimer == nil + p.pfDelayedRecheckMu.Unlock() + if !cleared { + t.Error("stopPFSettleFollowup left the timer in place") + } +} + +// TestScheduleDNSAfterVPNSettleRefreshReplacesPending keeps VPN churn from stacking +// repeats of the same scutil/VPN-DNS refresh. +func TestScheduleDNSAfterVPNSettleRefreshReplacesPending(t *testing.T) { + p := &prog{dnsInterceptState: &pfState{}} + p.scheduleDNSAfterVPNSettleRefresh("first", time.Hour) + p.pfDelayedRecheckMu.Lock() + first := p.pfSettleFollowupTimer + p.pfDelayedRecheckMu.Unlock() + + p.scheduleDNSAfterVPNSettleRefresh("second", time.Hour) + p.pfDelayedRecheckMu.Lock() + second := p.pfSettleFollowupTimer + p.pfDelayedRecheckMu.Unlock() + defer p.stopPFSettleFollowup() + + if first == second { + t.Fatal("the second schedule reused the first timer") + } + if first.Stop() { + t.Error("the superseded timer was still armed; stabilization churn would stack refreshes") + } +} + +// ============================================================================= +// post-wake functional verification tests +// ============================================================================= + +// stubWakeProbeSchedule shortens the post-resume retry schedule and neutralises the OS +// resolver refresh, so the bounded-repair contract can be tested without real delays or +// a real scutil call. It returns the number of resolver refreshes performed. +func stubWakeProbeSchedule(t *testing.T, attempts int) (refreshes *int) { + t.Helper() + originalDelays, originalWait := pfWakeProbeDelays, pfWakeProbeOwnerWait + originalResolver := initializeOsResolver + t.Cleanup(func() { + pfWakeProbeDelays, pfWakeProbeOwnerWait = originalDelays, originalWait + initializeOsResolver = originalResolver + }) + + pfWakeProbeDelays = make([]time.Duration, attempts) + for i := range pfWakeProbeDelays { + pfWakeProbeDelays[i] = time.Millisecond + } + pfWakeProbeOwnerWait = 10 * time.Millisecond + + calls := 0 + initializeOsResolver = func(context.Context, bool) []string { + calls++ + return []string{"10.0.0.1:53"} + } + return &calls +} + +// TestWakeProbeDetectsInterceptionBrokenBySuspend is the hole this closes. Rule text +// survives a suspend unchanged, so the watchdog reports the anchor intact, and the probe +// monitor is only armed by interface changes with a schedule frozen through the sleep. +// Nothing probed after the resume, and QA measured 18s of no public DNS on a host whose +// link and default route were already back. +func TestWakeProbeDetectsInterceptionBrokenBySuspend(t *testing.T) { + stubWakeProbeSchedule(t, 4) + probes, reloads := stubStabilizationProbe(t, []bool{false, true}, true) + + p := &prog{dnsInterceptState: &pfState{}} + p.verifyInterceptAfterWake(18 * time.Second) + + if *probes != 2 { + t.Errorf("probe calls = %d, want 2: one to detect and one to confirm the repair", *probes) + } + if *reloads != 1 { + t.Errorf("forced reloads = %d, want 1", *reloads) + } +} + +// TestWakeProbePassStopsImmediately keeps a healthy resume free of pf work: a reload +// flushes pf state and kills in-flight DoH connections for nothing. +func TestWakeProbePassStopsImmediately(t *testing.T) { + stubWakeProbeSchedule(t, 4) + probes, reloads := stubStabilizationProbe(t, []bool{true}, true) + + p := &prog{dnsInterceptState: &pfState{}} + p.verifyInterceptAfterWake(18 * time.Second) + + if *probes != 1 { + t.Errorf("probe calls = %d, want 1: a passing probe must end the schedule", *probes) + } + if *reloads != 0 { + t.Errorf("forced reloads = %d, want 0 on a healthy resume", *reloads) + } +} + +// TestWakeProbeRetriesAcrossTheSchedule covers the observed shape directly: the first +// forced reload did not restore translation and a later one did, so a single attempt is +// not enough. The second repair must still happen, and the confirming pass must end it. +func TestWakeProbeRetriesAcrossTheSchedule(t *testing.T) { + stubWakeProbeSchedule(t, 4) + // attempt 1: probe fails, reload, confirm fails. + // attempt 2: probe fails, reload, confirm passes. + probes, reloads := stubStabilizationProbe(t, []bool{false, false, false, true}, true) + + p := &prog{dnsInterceptState: &pfState{}} + p.verifyInterceptAfterWake(18 * time.Second) + + if *reloads != 2 { + t.Errorf("forced reloads = %d, want 2: one failed repair must not end the schedule", *reloads) + } + if *probes != 4 { + t.Errorf("probe calls = %d, want 4", *probes) + } +} + +// TestWakeProbeRepairsAreBounded pins the bound. A probe that never passes must exhaust +// the repair budget and then keep probing without reloading, so this can never become the +// unbounded reload loop the issue rules out. +func TestWakeProbeRepairsAreBounded(t *testing.T) { + stubWakeProbeSchedule(t, 6) + probes, reloads := stubStabilizationProbe(t, nil, true) // every probe fails + + p := &prog{dnsInterceptState: &pfState{}} + p.verifyInterceptAfterWake(18 * time.Second) + + if *reloads != pfWakeMaxRepairs { + t.Errorf("forced reloads = %d, want the budget of %d", *reloads, pfWakeMaxRepairs) + } + // Two repaired attempts probe twice each; the remaining four probe once each. + if want := 2*pfWakeMaxRepairs + (6 - pfWakeMaxRepairs); *probes != want { + t.Errorf("probe calls = %d, want %d", *probes, want) + } + if p.pfMonitorRunning.Load() { + t.Error("functional-probe ownership was left held after the schedule finished") + } +} + +// TestWakeProbeRefreshesResolverBeforeEveryProbe is why a stale list matters: the probe +// aims at the first OS nameserver and reports success when it has none, so a list still +// holding a pre-sleep VPN resolver turns the probe into a meaningless pass or a false +// failure. The captured run showed the resolver set changing after the resume. +func TestWakeProbeRefreshesResolverBeforeEveryProbe(t *testing.T) { + refreshes := stubWakeProbeSchedule(t, 3) + stubStabilizationProbe(t, nil, false) // probes fail, reload refuses to run + + p := &prog{dnsInterceptState: &pfState{}} + p.verifyInterceptAfterWake(18 * time.Second) + + if *refreshes != 3 { + t.Errorf("OS resolver refreshes = %d, want one per attempt (3)", *refreshes) + } +} + +// TestWakeProbeStandsDownForStabilization keeps the two repair paths from fighting over +// pf. Stabilization owns the ruleset while a VPN settles and runs its own functional +// verification when it finishes; reloading underneath it is the mutual overwriting +// stabilization exists to prevent. +func TestWakeProbeStandsDownForStabilization(t *testing.T) { + stubWakeProbeSchedule(t, 4) + probes, reloads := stubStabilizationProbe(t, []bool{false, true}, true) + + p := &prog{dnsInterceptState: &pfState{}} + p.pfStabilizing.Store(true) + p.verifyInterceptAfterWake(18 * time.Second) + + if *probes != 0 || *reloads != 0 { + t.Errorf("probe calls = %d, forced reloads = %d, want 0/0 while stabilization owns pf", *probes, *reloads) + } + if !p.claimFunctionalProbeOwner(0) { + t.Error("standing down took functional-probe ownership; the post-stabilization verifier would skip") + } + p.pfMonitorRunning.Store(false) +} + +// TestWakeProbeYieldsToAnotherProberThenRetries is the ownership handoff. A probe monitor +// that is genuinely working owns this attempt, but the schedule must not be spent waiting +// on it: the next attempt still has to probe once ownership is free. +func TestWakeProbeYieldsToAnotherProberThenRetries(t *testing.T) { + stubWakeProbeSchedule(t, 3) + probes, reloads := stubStabilizationProbe(t, []bool{false, true}, true) + + p := &prog{dnsInterceptState: &pfState{}} + p.pfMonitorRunning.Store(true) + go func() { + time.Sleep(30 * time.Millisecond) + p.pfMonitorRunning.Store(false) + }() + + p.verifyInterceptAfterWake(18 * time.Second) + + if *probes == 0 { + t.Fatal("the schedule gave up entirely because another prober held ownership at first") + } + if *reloads != 1 { + t.Errorf("forced reloads = %d, want 1", *reloads) + } +} + +// TestWakeProbeSkippedDuringExecBackoff protects the resource-exhaustion guard: each +// probe forks a helper, which is exactly what backoff exists to stop. +func TestWakeProbeSkippedDuringExecBackoff(t *testing.T) { + stubWakeProbeSchedule(t, 3) + probes, reloads := stubStabilizationProbe(t, []bool{false, true}, true) + + p := &prog{dnsInterceptState: &pfState{}} + p.pfExecBackoffUntil.Store(time.Now().Add(time.Minute).UnixMilli()) + p.verifyInterceptAfterWake(18 * time.Second) + + if *probes != 0 || *reloads != 0 { + t.Errorf("probe calls = %d, forced reloads = %d, want 0/0 during pf exec backoff", *probes, *reloads) + } +} + +// TestWakeProbeSkippedWhenInterceptDisabled keeps the watcher harmless when intercept +// mode is off: the detector runs for the process, the repair is intercept-only. +func TestWakeProbeSkippedWhenInterceptDisabled(t *testing.T) { + stubWakeProbeSchedule(t, 3) + probes, reloads := stubStabilizationProbe(t, []bool{false}, true) + + p := &prog{} + p.verifyInterceptAfterWake(18 * time.Second) + + if *probes != 0 || *reloads != 0 { + t.Errorf("probe calls = %d, forced reloads = %d, want 0/0 with intercept disabled", *probes, *reloads) + } +} + +// TestWakeProbeStopsWhenServiceStops keeps a resume during shutdown from holding the +// schedule open past the stop request. +func TestWakeProbeStopsWhenServiceStops(t *testing.T) { + originalDelays := pfWakeProbeDelays + pfWakeProbeDelays = []time.Duration{0, time.Hour} + t.Cleanup(func() { pfWakeProbeDelays = originalDelays }) + originalResolver := initializeOsResolver + initializeOsResolver = func(context.Context, bool) []string { return []string{"10.0.0.1:53"} } + t.Cleanup(func() { initializeOsResolver = originalResolver }) + + probes, _ := stubStabilizationProbe(t, nil, false) // first probe fails, reload refuses + + stopCh := make(chan struct{}) + close(stopCh) + p := &prog{dnsInterceptState: &pfState{}, stopCh: stopCh} + + done := make(chan struct{}) + go func() { + p.verifyInterceptAfterWake(18 * time.Second) + close(done) + }() + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("verifyInterceptAfterWake ignored the stop request and waited out its schedule") + } + if *probes != 1 { + t.Errorf("probe calls = %d, want 1 before the stop was observed", *probes) + } +} + +// ============================================================================= +// post-wake watcher registration tests +// ============================================================================= + +// TestStartInterceptBackgroundWorkRegistersTheWakeVerifier covers the caller wiring. The +// detector and the wake-repair path can both be correct while nothing connects them, and +// then a host sleep is never noticed in production. +func TestStartInterceptBackgroundWorkRegistersTheWakeVerifier(t *testing.T) { + type registration struct { + stopCh <-chan struct{} + onResume func(time.Duration) + } + registered := make(chan registration, 1) + watchdogStarted := make(chan struct{}, 1) + originalWatcher, originalWatchdog := runSuspendWatcherFn, pfWatchdogFn + runSuspendWatcherFn = func(stopCh <-chan struct{}, onResume func(time.Duration)) { + registered <- registration{stopCh: stopCh, onResume: onResume} + } + // Both loops are stubbed rather than run: each would reach pfctl, and stopping the + // real watchdog means closing stopCh, which the wake verifier reads as "service + // stopping" and would abandon its probe schedule before the first probe. + pfWatchdogFn = func(*prog) { watchdogStarted <- struct{}{} } + t.Cleanup(func() { runSuspendWatcherFn, pfWatchdogFn = originalWatcher, originalWatchdog }) + + stubWakeProbeSchedule(t, 4) + probes, _ := stubStabilizationProbe(t, []bool{false, true}, true) + + // Open for the whole test: an active intercept is not stopping. + stopCh := make(chan struct{}) + p := &prog{dnsInterceptState: &pfState{}, stopCh: stopCh} + p.startInterceptBackgroundWork() + + select { + case <-watchdogStarted: + case <-time.After(2 * time.Second): + t.Error("no pf watchdog was started; a replaced pf ruleset would never be restored") + } + + var reg registration + select { + case reg = <-registered: + case <-time.After(2 * time.Second): + t.Fatal("no suspend watcher was started; a host sleep would go unnoticed") + } + if reg.stopCh != stopCh { + t.Error("the watcher was not wired to the service stop channel; it would outlive the intercept") + } + + // Prove the callback is the post-wake verifier and not some other function: a + // reported resume has to probe interception. How many probes and repairs that + // schedule is worth belongs to TestWakeProbeDetectsInterceptionBrokenBySuspend; + // pinning it again here would only couple this test to that tuning. + reg.onResume(18 * time.Second) + if *probes == 0 { + t.Error("the resume callback probed nothing; it is not wired to verifyInterceptAfterWake") + } +} + +// TestStartDNSInterceptStartsBackgroundWork closes the caller-wiring gap. Everything the +// privileged install does sits behind installPFInterceptFn, so this drives the real +// startDNSIntercept and checks what a successful start hands off: without the handoff, an +// active intercept runs with no pf watchdog and no post-wake verification, and every +// detector, recovery and registration test still passes. +func TestStartDNSInterceptStartsBackgroundWork(t *testing.T) { + installed := 0 + originalInstall := installPFInterceptFn + installPFInterceptFn = func(*prog) error { installed++; return nil } + originalDiscover := discoverTunnelInterfacesForReconcile + discoverTunnelInterfacesForReconcile = func() []string { return nil } + + watchdogStarted := make(chan struct{}, 1) + watcherStarted := make(chan struct{}, 1) + originalWatchdog, originalWatcher := pfWatchdogFn, runSuspendWatcherFn + pfWatchdogFn = func(*prog) { watchdogStarted <- struct{}{} } + runSuspendWatcherFn = func(<-chan struct{}, func(time.Duration)) { watcherStarted <- struct{}{} } + t.Cleanup(func() { + installPFInterceptFn = originalInstall + discoverTunnelInterfacesForReconcile = originalDiscover + pfWatchdogFn, runSuspendWatcherFn = originalWatchdog, originalWatcher + }) + + p := &prog{ + cfg: &ctrld.Config{Listener: map[string]*ctrld.ListenerConfig{"0": {IP: "127.0.0.1", Port: 53}}}, + stopCh: make(chan struct{}), + } + if err := p.startDNSIntercept(); err != nil { + t.Fatalf("startDNSIntercept() = %v, want nil", err) + } + if installed != 1 { + t.Errorf("privileged install ran %d times, want 1", installed) + } + if p.dnsInterceptState == nil { + t.Fatal("a successful start did not publish intercept state") + } + + select { + case <-watchdogStarted: + case <-time.After(2 * time.Second): + t.Error("startDNSIntercept did not start the pf watchdog; a replaced pf ruleset would never be restored") + } + select { + case <-watcherStarted: + case <-time.After(2 * time.Second): + t.Error("startDNSIntercept did not start the suspend watcher; a host sleep would go unnoticed") + } +} + +// TestStartDNSInterceptSkipsBackgroundWorkWhenInstallFails keeps the handoff conditional: +// watchers over an intercept that was never installed would repair state nothing owns. +func TestStartDNSInterceptSkipsBackgroundWorkWhenInstallFails(t *testing.T) { + originalInstall := installPFInterceptFn + installPFInterceptFn = func(*prog) error { return errors.New("pf unavailable") } + originalWatchdog, originalWatcher := pfWatchdogFn, runSuspendWatcherFn + pfWatchdogFn = func(*prog) { t.Error("the pf watchdog started after a failed install") } + runSuspendWatcherFn = func(<-chan struct{}, func(time.Duration)) { + t.Error("the suspend watcher started after a failed install") + } + t.Cleanup(func() { + installPFInterceptFn = originalInstall + pfWatchdogFn, runSuspendWatcherFn = originalWatchdog, originalWatcher + }) + + p := &prog{stopCh: make(chan struct{})} + if err := p.startDNSIntercept(); err == nil { + t.Fatal("startDNSIntercept() = nil after the privileged install failed") + } + if p.dnsInterceptState != nil { + t.Error("a failed start published intercept state; the caller would skip its DNS fallback") + } + // The goroutines above are started before this returns if at all, but give a failing + // wiring a moment to report rather than racing the test's end. + time.Sleep(50 * time.Millisecond) } diff --git a/cmd/cli/prog.go b/cmd/cli/prog.go index af47a69..0d32ce1 100644 --- a/cmd/cli/prog.go +++ b/cmd/cli/prog.go @@ -249,6 +249,12 @@ type prog struct { pfDelayedRecheckMu sync.Mutex //lint:ignore U1000 used on darwin pfDelayedRecheckTimers []*time.Timer //lint:ignore U1000 used on darwin + // pfSettleFollowupTimer is the pending post-settle VPN DNS refresh, if any. + // Tracked so teardown can cancel it and a later stabilization replaces it + // instead of stacking another copy of the same refresh. Protected by + // pfDelayedRecheckMu. + pfSettleFollowupTimer *time.Timer //lint:ignore U1000 used on darwin + // pfIgnoredChangeLastReconcile bounds immediate pf/VPN-DNS work for noisy // ignored macOS network deltas. Tunnel changes bypass this limit, and the // existing delayed checks provide a trailing reconciliation after churn. diff --git a/cmd/cli/suspend_detector.go b/cmd/cli/suspend_detector.go new file mode 100644 index 0000000..b3e3f81 --- /dev/null +++ b/cmd/cli/suspend_detector.go @@ -0,0 +1,109 @@ +package cli + +import "time" + +// Suspend/resume detection without an OS notification API. +// +// A host sleep suspends the whole process: timers stop advancing and resume where they +// left off, so a tick scheduled two seconds out can arrive a minute later. That +// overshoot is the only in-process evidence that the host slept, and nothing else tells +// the DNS-intercept repair paths that the state they proved before sleep - rule text, +// resolver list, upstream connections - may no longer hold. + +const ( + // suspendProbeInterval is how often the detector measures elapsed wall-clock time. + // Short enough that a resume is noticed within a couple of seconds, which is what + // the post-wake continuity contract measures from. + suspendProbeInterval = 2 * time.Second + + // suspendGapThreshold is the elapsed wall clock over one interval that counts as a + // suspend. A pfctl storm or a loaded host can stretch a tick, so this sits at + // several times the interval; a real sleep overshoots by far more. Erring low is + // cheap on purpose: a false positive costs one probe, and a stall this long is + // worth a probe either way. + suspendGapThreshold = 8 * time.Second +) + +// wallNanoNow reads the wall clock in nanoseconds. +// +// The detector works in wall-clock nanoseconds rather than time.Time on purpose. +// time.Time.Sub prefers the monotonic reading whenever both operands carry one, and +// Darwin's monotonic clock stops while the host is asleep - so a monotonic difference +// across a suspend is just the tick interval and hides the very gap this looks for. +// Taking int64 wall nanos makes that mistake unrepresentable rather than a comment. +func wallNanoNow() int64 { return time.Now().UnixNano() } + +// suspendDetector turns observations of a fixed-interval tick into suspend/resume +// events. It measures elapsed time rather than counting ticks, because a suspended +// process gets one late tick, not the ticks it missed. +type suspendDetector struct { + threshold time.Duration + wallNano func() int64 + lastNano int64 +} + +// newSuspendDetector returns a detector that reports gaps of at least threshold. +// wallNano is a seam for tests; nil means the real wall clock. +func newSuspendDetector(threshold time.Duration, wallNano func() int64) *suspendDetector { + if wallNano == nil { + wallNano = wallNanoNow + } + return &suspendDetector{threshold: threshold, wallNano: wallNano, lastNano: wallNano()} +} + +// observe records a tick and reports the elapsed time when it reaches the threshold, +// which means the process was not running for most of that span. +func (d *suspendDetector) observe() (time.Duration, bool) { + current := d.wallNano() + gap := time.Duration(current - d.lastNano) + d.lastNano = current + if gap < d.threshold { + return 0, false + } + return gap, true +} + +// reset re-baselines the detector to the current wall clock. Time that has already +// passed is deliberately discarded rather than measured: use it after doing work +// outside the sampling loop, which is not a suspend even though it looks like one. +func (d *suspendDetector) reset() { d.lastNano = d.wallNano() } + +// watchForResume calls onResume once per detected suspend/resume gap, until stopCh +// closes or tick is closed. tick is a parameter so tests can drive it directly. +// +// onResume runs on this goroutine, so recovery is serialized against detection - one +// resume cannot start a second recovery on top of the first. The cost is that no +// sampling happens while it runs, which the re-baseline below accounts for. +func watchForResume(stopCh <-chan struct{}, tick <-chan time.Time, d *suspendDetector, onResume func(gap time.Duration)) { + for { + select { + case <-stopCh: + return + case _, ok := <-tick: + if !ok { + return + } + if gap, resumed := d.observe(); resumed { + onResume(gap) + // Re-baseline: the recovery just ran, and its own runtime exceeds the + // threshold on its own (the post-wake probe schedule waits for seconds + // before a single probe timeout or reload). The ticker keeps firing and + // buffers a tick while it works, so without this the next observe() + // measures the callback and reports it as a fresh suspend - which starts + // another recovery, which is again long enough to look like a suspend. + // Under a persistent pf failure that is an unbounded reload loop, and + // each pass flushes pf state and kills in-flight DoH connections. + d.reset() + } + } + } +} + +// runSuspendWatcher watches for suspend/resume against a real ticker. +// +//lint:ignore U1000 used on darwin (post-wake interception verification) +func runSuspendWatcher(stopCh <-chan struct{}, onResume func(gap time.Duration)) { + ticker := time.NewTicker(suspendProbeInterval) + defer ticker.Stop() + watchForResume(stopCh, ticker.C, newSuspendDetector(suspendGapThreshold, nil), onResume) +} diff --git a/cmd/cli/suspend_detector_test.go b/cmd/cli/suspend_detector_test.go new file mode 100644 index 0000000..4f31a31 --- /dev/null +++ b/cmd/cli/suspend_detector_test.go @@ -0,0 +1,208 @@ +package cli + +import ( + "testing" + "time" +) + +// fakeWallClock returns the queued offsets from a fixed base, then repeats the last one. +type fakeWallClock struct { + nanos []int64 + calls int +} + +func (c *fakeWallClock) now() int64 { + if c.calls < len(c.nanos) { + n := c.nanos[c.calls] + c.calls++ + return n + } + return c.nanos[len(c.nanos)-1] +} + +// wallClockAt builds a clock whose readings are the given offsets from an arbitrary +// epoch. The first reading is consumed by newSuspendDetector; the rest are ticks. +func wallClockAt(offsets ...time.Duration) *fakeWallClock { + const base = int64(1787081656_000000000) // 2026-08-18T14:34:16-05:00 + nanos := make([]int64, 0, len(offsets)) + for _, off := range offsets { + nanos = append(nanos, base+int64(off)) + } + return &fakeWallClock{nanos: nanos} +} + +func TestSuspendDetectorIgnoresOrdinaryTicks(t *testing.T) { + // Construction, then three ticks two seconds apart, one stretched to 3s by a + // loaded host. None of these is a suspend. + clock := wallClockAt(0, 2*time.Second, 4*time.Second, 7*time.Second) + d := newSuspendDetector(suspendGapThreshold, clock.now) + + for i := 0; i < 3; i++ { + if gap, resumed := d.observe(); resumed { + t.Errorf("tick %d reported a suspend of %s; ordinary jitter must not count", i, gap) + } + } +} + +// TestSuspendDetectorReportsTheGapOnce replays the captured incident: the last pre-sleep +// sample at 14:34:47 and the first post-resume one at 14:35:05. +func TestSuspendDetectorReportsTheGapOnce(t *testing.T) { + clock := wallClockAt( + 31*time.Second, // construction, 14:34:47 + 33*time.Second, // ordinary tick + 49*time.Second, // 14:35:05 - a 16s gap over a 2s interval + 51*time.Second, // ordinary again + ) + d := newSuspendDetector(suspendGapThreshold, clock.now) + + if _, resumed := d.observe(); resumed { + t.Fatal("the tick before the suspend reported a resume") + } + gap, resumed := d.observe() + if !resumed { + t.Fatal("a 16s gap over a 2s interval was not reported as a suspend") + } + if gap != 16*time.Second { + t.Errorf("gap = %s, want 16s: the reported gap must be the measured wall clock", gap) + } + if _, resumed := d.observe(); resumed { + t.Error("the resume was reported twice; the baseline must advance to the late tick") + } +} + +// TestSuspendDetectorThresholdCoversShortNaps guards the lower bound. A nap only a few +// seconds longer than the interval still breaks pf translation, so the threshold must not +// drift up to a value that only catches long sleeps. +func TestSuspendDetectorThresholdCoversShortNaps(t *testing.T) { + if suspendGapThreshold > 10*time.Second { + t.Errorf("suspendGapThreshold = %s; a short nap would go unnoticed", suspendGapThreshold) + } + if suspendGapThreshold <= 2*suspendProbeInterval { + t.Errorf("suspendGapThreshold = %s is too close to the %s interval; ordinary jitter would fire it", + suspendGapThreshold, suspendProbeInterval) + } + + clock := wallClockAt(0, suspendGapThreshold) + d := newSuspendDetector(suspendGapThreshold, clock.now) + if _, resumed := d.observe(); !resumed { + t.Error("a gap exactly at the threshold was not reported") + } +} + +func TestSuspendDetectorIgnoresBackwardClockSteps(t *testing.T) { + // An NTP step backwards must not read as a resume, and must not leave the baseline + // in the future either: the next ordinary tick still has to look ordinary. + clock := wallClockAt(0, -30*time.Second, -28*time.Second) + d := newSuspendDetector(suspendGapThreshold, clock.now) + + if gap, resumed := d.observe(); resumed { + t.Errorf("a backward clock step reported a suspend of %s", gap) + } + if gap, resumed := d.observe(); resumed { + t.Errorf("the tick after a backward step reported a suspend of %s", gap) + } +} + +func TestWatchForResumeCallsBackPerGap(t *testing.T) { + clock := wallClockAt( + 0, + 2*time.Second, // ordinary + 40*time.Second, // resume: 38s + 42*time.Second, // ordinary + 99*time.Second, // second resume: 57s + ) + d := newSuspendDetector(suspendGapThreshold, clock.now) + + tick := make(chan time.Time, 4) + for i := 0; i < 4; i++ { + tick <- time.Time{} + } + close(tick) + + var gaps []time.Duration + watchForResume(nil, tick, d, func(gap time.Duration) { gaps = append(gaps, gap) }) + + if len(gaps) != 2 { + t.Fatalf("onResume called %d times (%v), want 2", len(gaps), gaps) + } + if gaps[0] != 38*time.Second || gaps[1] != 57*time.Second { + t.Errorf("gaps = %v, want [38s 57s]", gaps) + } +} + +// advancingClock is a wall clock the test moves by hand, so a callback's runtime can be +// simulated without sleeping for it. +type advancingClock struct{ nano int64 } + +func (c *advancingClock) now() int64 { return c.nano } +func (c *advancingClock) advance(d time.Duration) { c.nano += int64(d) } + +// TestWatchForResumeDoesNotRetriggerOnItsOwnCallback pins the loop this closes. onResume +// runs on the watcher goroutine, and the real callback - verifyInterceptAfterWake - takes +// longer than the gap threshold by design: its configured waits alone total more than the +// threshold, before any probe timeout or forced reload. The ticker buffers a tick while it +// works, so a detector that keeps its pre-callback baseline measures the callback itself as +// the next gap and starts another recovery, with another repair budget, indefinitely under +// a persistent pf failure. +func TestWatchForResumeDoesNotRetriggerOnItsOwnCallback(t *testing.T) { + clock := &advancingClock{} + d := newSuspendDetector(suspendGapThreshold, clock.now) + + // Capacity 1, like time.Ticker's channel: exactly one tick can queue up behind a + // callback in production. + tick := make(chan time.Time, 1) + calls := 0 + done := make(chan struct{}) + go func() { + defer close(done) + watchForResume(nil, tick, d, func(time.Duration) { + calls++ + if calls > 1 { + // A retrigger - the failure this test reports. There is nothing left to + // simulate, and tick is already closed. + return + } + // The recovery takes longer than the threshold, and a tick lands while it + // runs. Closing here ends the watcher after it drains that queued tick, + // which is the observation under test. + clock.advance(2 * suspendGapThreshold) + tick <- time.Time{} + close(tick) + }) + }() + + clock.advance(2 * suspendGapThreshold) // the genuine suspend + tick <- time.Time{} + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("watchForResume did not finish; the queued tick was never consumed") + } + + if calls != 1 { + t.Errorf("onResume called %d times, want 1: the callback's own runtime was measured as a second suspend", calls) + } +} + +func TestWatchForResumeStopsOnStopCh(t *testing.T) { + stopCh := make(chan struct{}) + tick := make(chan time.Time) // never fires + clock := wallClockAt(0, time.Hour) + d := newSuspendDetector(suspendGapThreshold, clock.now) + + done := make(chan struct{}) + go func() { + watchForResume(stopCh, tick, d, func(time.Duration) { + t.Error("onResume ran without a tick") + }) + close(done) + }() + + close(stopCh) + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("watchForResume did not return once stopCh was closed") + } +}