diff --git a/cmd/cli/dns_intercept_darwin.go b/cmd/cli/dns_intercept_darwin.go index eb9eafc..ab72492 100644 --- a/cmd/cli/dns_intercept_darwin.go +++ b/cmd/cli/dns_intercept_darwin.go @@ -7,6 +7,7 @@ import ( "crypto/sha256" "fmt" "net" + "net/netip" "os" "os/exec" "path/filepath" @@ -77,7 +78,8 @@ const ( // The full probe domain is "_pf-probe-.". // These queries are sent by a subprocess WITHOUT the _ctrld group GID, // so pf should intercept them and redirect to ctrld. If ctrld receives - // the query, pf interception is working. If not (timeout), rdr is broken. + // the query, interception is working. A sent query without receipt permits + // repair. An unsent or unavailable probe is indeterminate. // No trailing dot — canonicalName() in the DNS handler strips trailing dots. pfProbeDomain = "pf-probe.ctrld.test" @@ -1477,10 +1479,19 @@ func (p *prog) finishPFStabilization(stableRequired time.Duration) { } // probePFInterceptFn and forceReloadPFInterceptFn are the functional verification seams -// shared by both probers. +// shared by all four probe callers. var ( probePFInterceptFn = (*prog).probePFIntercept forceReloadPFInterceptFn = (*prog).forceReloadPFMainRuleset + pfProbeNameservers = ctrld.OsResolverNameservers + newPFProbeCommand = func(ctx context.Context, host, packet string) *exec.Cmd { + cmd := exec.CommandContext(ctx, os.Args[0], "pf-probe-send", host, packet) + // wheel, not _ctrld: the probe must traverse interception. + cmd.SysProcAttr = &syscall.SysProcAttr{ + Credential: &syscall.Credential{Uid: 0, Gid: 0}, + } + return cmd + } ) // pfFunctionalProbeOwnerWait bounds how long the post-stabilization verifier waits for @@ -1551,17 +1562,29 @@ func (p *prog) verifyInterceptAfterStabilization() { return } - if probePFInterceptFn(p) { + result := probePFInterceptFn(p) + if result.result == pfProbeIndeterminate { + mainLog.Load().Debug().Msg("DNS intercept: post-stabilization probe indeterminate; repair deferred") + return + } + if result.result == pfProbeIntercepted { mainLog.Load().Debug().Msg("DNS intercept: post-stabilization probe passed — interception is translating") return } mainLog.Load().Warn().Msg("DNS intercept: post-stabilization rules are intact but the probe FAILED — forcing one reload") - if !forceReloadPFInterceptFn(p) { + if !p.repairPFIntercept(result, "post_stabilization") { mainLog.Load().Error().Msg("DNS intercept: post-stabilization forced reload did not run — leaving recovery to the watchdog") return } - if probePFInterceptFn(p) { + cause := result + result = probePFInterceptFn(p) + logPFRepair(cause, "post_stabilization", result.result.String(), true, result) + if result.result == pfProbeIndeterminate { + mainLog.Load().Warn().Msg("DNS intercept: post-stabilization reload completed; verification indeterminate") + return + } + if result.result == pfProbeIntercepted { mainLog.Load().Info().Msg("DNS intercept: interception restored by the post-stabilization reload") return } @@ -1600,8 +1623,8 @@ var pfWakeProbeOwnerWait = 250 * time.Millisecond // 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. +// first usable IPv4 LAN target, or the original usable first public target. +// No target is indeterminate. Refreshing avoids probing a stale VPN resolver. func (p *prog) verifyInterceptAfterWake(gap time.Duration) { if p.dnsInterceptState == nil { return @@ -1653,7 +1676,7 @@ func (p *prog) verifyInterceptAfterWake(gap time.Duration) { 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", + mainLog.Load().Error().Msgf("DNS intercept: interception not verified after resume (%d probe attempts, %d forced reloads) - the watchdog will retry", probed, repairs) } @@ -1680,7 +1703,12 @@ func (p *prog) waitBeforeWakeProbe(delay time.Duration) bool { // The caller holds functional-probe ownership for the call. func (p *prog) runWakeProbeAttempt(attempt, repairs int) (restored, repaired bool) { initializeOsResolver(true) - if probePFInterceptFn(p) { + result := probePFInterceptFn(p) + if result.result == pfProbeIndeterminate { + mainLog.Load().Debug().Msg("DNS intercept: post-wake probe indeterminate; repair deferred") + return false, false + } + if result.result == pfProbeIntercepted { 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 { @@ -1692,16 +1720,19 @@ func (p *prog) runWakeProbeAttempt(attempt, repairs int) (restored, repaired boo 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) + logPFRepair(result, "post_wake", "budget_exhausted", false, pfProbeObservation{}) 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) { + if !p.repairPFIntercept(result, "post_wake") { mainLog.Load().Error().Msg("DNS intercept: post-wake forced reload did not run") return false, false } - if probePFInterceptFn(p) { + confirmation := probePFInterceptFn(p) + logPFRepair(result, "post_wake", confirmation.result.String(), true, confirmation) + if confirmation.result == pfProbeIntercepted { mainLog.Load().Info().Msg("DNS intercept: interception restored by the post-wake reload") return true, true } @@ -2011,18 +2042,7 @@ func (p *prog) pfWatchdog() { ensureInterceptDNSTargetFn(p, []string{}) result := p.ensurePFAnchorActive() - if result == pfAnchorCheckIntact { - // Only an authoritative intact result may trigger the functional probe. - // Skipped/backoff/failed checks must not be treated as healthy text state. - if !p.pfMonitorRunning.Load() && !p.probePFIntercept() { - mainLog.Load().Warn().Msg("DNS intercept watchdog: rules intact but probe FAILED — forcing full reload") - if p.forceReloadPFMainRuleset() { - result = pfAnchorCheckRestored - } else { - result = pfAnchorCheckFailed - } - } - } + result = p.checkPFWatchdogProbe(result) if result == pfAnchorCheckIntact { // Check if backoff should be reset only after an authoritative healthy check. @@ -2070,6 +2090,27 @@ func (p *prog) pfWatchdog() { } } +// A successful forced reload following a probe miss is not evidence that an +// anchor was missing. Keep that repair out of the missing-anchor counter. +func (p *prog) checkPFWatchdogProbe(result pfAnchorCheckResult) pfAnchorCheckResult { + if result != pfAnchorCheckIntact || p.pfMonitorRunning.Load() { + return result + } + probe := probePFInterceptFn(p) + switch probe.result { + case pfProbeIntercepted: + return result + case pfProbeNotIntercepted: + mainLog.Load().Warn().Msg("DNS intercept watchdog: sent probe was not intercepted; forcing full reload") + if !p.repairPFIntercept(probe, "watchdog") { + mainLog.Load().Warn().Msg("DNS intercept watchdog: probe repair did not complete; will retry") + } + case pfProbeIndeterminate: + mainLog.Load().Debug().Msg("DNS intercept watchdog: probe indeterminate; repair deferred") + } + return pfAnchorCheckDeferred +} + // exemptVPNDNSServers rebuilds the pf anchor rules to exclude VPN DNS interfaces // and server IPs from interception. VPN DNS handlers using Network Extensions // (e.g., Tailscale MagicDNS) need DNS traffic to flow without any pf interference. @@ -2144,67 +2185,91 @@ func (p *prog) exemptVPNDNSServers(exemptions []vpnDNSExemption) error { return nil } -// probePFIntercept tests whether pf's rdr translation is actually working by -// sending a DNS query through the interception path from a subprocess that does -// NOT have the _ctrld group GID. If pf interception is working, the query gets -// redirected to 127.0.0.1:53 (ctrld), and the DNS handler signals us. If broken -// (rdr rules present but not evaluating), the query goes to the real DNS server -// and we time out. -// -// Returns true if interception is working, false if broken or indeterminate. -func (p *prog) probePFIntercept() bool { +// Keep warning coalescing available throughout intercept startup and recovery. +var pfProbeLogs pfProbeLogState + +// probePFIntercept distinguishes a meaningful interception failure from a probe +// that could not send. Only pfProbeNotIntercepted permits destructive repair. +func (p *prog) probePFIntercept() pfProbeObservation { + observed := pfProbeObservation{ + probeID: fmt.Sprintf("_pf-probe-%x.%s", time.Now().UnixNano(), pfProbeDomain), + recoveryGeneration: p.recoveryGen.Load(), + stage: "target", code: "unavailable", + } if p.dnsInterceptState == nil { - return true + return observed } - - nsIPs := ctrld.OsResolverNameservers() - if len(nsIPs) == 0 { - mainLog.Load().Debug().Msg("DNS intercept probe: no OS resolver nameservers available") - return true // can't probe without a target + started := time.Now() + servers := pfProbeNameservers() + host := pfProbeTarget(servers) + family := "none" + if len(servers) > 0 { + if first, _, err := net.SplitHostPort(servers[0]); err == nil { + if addr, err := netip.ParseAddr(first); err == nil { + observed.target = net.JoinHostPort(addr.String(), "53") + family = "ipv6" + if addr.Is4() { + family = "ipv4" + } + } + } } - host, _, _ := net.SplitHostPort(nsIPs[0]) - if host == "" || host == "127.0.0.1" || host == "::1" { - mainLog.Load().Debug().Msg("DNS intercept probe: OS resolver is localhost, skipping probe") - return true // can't probe through localhost + if host != "" { + observed.target = net.JoinHostPort(host, "53") + family = "ipv4" + probeCh, deregister := p.registerInterceptProbe(observed.probeID) + defer deregister() + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + go func() { + select { + case <-p.stopCh: + cancel() + case <-ctx.Done(): + } + }() + cmd := newPFProbeCommand(ctx, host, fmt.Sprintf("%x", buildDNSQueryPacket(observed.probeID))) + result := runPFProbe(ctx, cmd, probeCh, pfProbeTimeout) + observed.result, observed.stage, observed.code = result.result, result.stage, result.code } - - // Generate unique probe domain - probeID := fmt.Sprintf("_pf-probe-%x.%s", time.Now().UnixNano()&0xFFFFFFFF, pfProbeDomain) - - // Register this attempt's own domain: overlapping probes must not cancel each other. - probeCh, deregister := p.registerInterceptProbe(probeID) - defer deregister() - - // Build a minimal DNS query packet for the probe domain. - // We use exec.Command to send from a subprocess with GID=0 (wheel), - // so pf's _ctrld group exemption does NOT apply and the query gets intercepted. - dnsPacket := buildDNSQueryPacket(probeID) - - // Send via a helper subprocess that drops the _ctrld group - cmd := exec.Command(os.Args[0], "pf-probe-send", host, fmt.Sprintf("%x", dnsPacket)) - cmd.SysProcAttr = &syscall.SysProcAttr{ - Credential: &syscall.Credential{ - Uid: 0, - Gid: 0, // wheel group — NOT _ctrld, so pf intercepts it - }, + // Cancellation is a shutdown diagnostic, not a change in PF health. + retain, repeats := false, uint64(0) + if observed.code != "canceled" { + retain, repeats = pfProbeLogs.change(observed) } - - if err := cmd.Start(); err != nil { - mainLog.Load().Debug().Err(err).Msg("DNS intercept probe: failed to start probe subprocess") - return true // can't probe, assume OK + event := mainLog.Load().Debug() + if retain { + event = mainLog.Load().Warn() } + event.Str("probe_id", observed.probeID).Uint64("recovery_generation", observed.recoveryGeneration). + Str("resolver_target", observed.target).Str("resolver_family", family). + Str("stage", observed.stage).Str("error_code", observed.code). + Str("outcome", observed.result.String()).Uint64("repeated_results", repeats). + Bool("repair_eligible", observed.result == pfProbeNotIntercepted). + Dur("elapsed_ms", time.Since(started)).Msg("DNS intercept probe result") + return observed +} - // Don't leak the subprocess - go func() { - _ = cmd.Wait() - }() - - select { - case <-probeCh: - return true - case <-time.After(pfProbeTimeout): - return false +// Carry the cause by value. A concurrent probe cannot replace its correlation fields. +func (p *prog) repairPFIntercept(cause pfProbeObservation, caller string) bool { + logPFRepair(cause, caller, "started", false, pfProbeObservation{}) + performed := forceReloadPFInterceptFn(p) + outcome := "not_run" + if performed { + outcome = "reload_completed" } + logPFRepair(cause, caller, outcome, performed, pfProbeObservation{}) + return performed +} + +func logPFRepair(cause pfProbeObservation, caller, outcome string, performed bool, confirmation pfProbeObservation) { + mainLog.Load().Warn().Str("probe_id", cause.probeID). + Uint64("recovery_generation", cause.recoveryGeneration). + Str("caller", caller).Str("cause", cause.result.String()). + Str("stage", cause.stage).Str("error_code", cause.code). + Str("resolver_target", cause.target).Str("outcome", outcome). + Bool("reload_performed", performed).Str("confirmation_probe_id", confirmation.probeID). + Msg("PF repair") } // buildDNSQueryPacket constructs a minimal DNS query packet (wire format) for the given domain. @@ -2235,6 +2300,8 @@ func buildDNSQueryPacket(domain string) []byte { return append(header, question...) } +var pfInterceptMonitorDelays = []time.Duration{0, 500 * time.Millisecond, time.Second, 2 * time.Second, 4 * time.Second} + // pfInterceptMonitor runs asynchronously after interface changes are detected. // It probes pf interception with exponential backoff and forces a full pf reload // if the probe fails. Only one instance runs at a time (singleton via atomic.Bool). @@ -2259,7 +2326,7 @@ func (p *prog) pfInterceptMonitor() { // Backoff schedule: probe quickly first, then space out. // Total monitoring window: ~0 + 0.5 + 1 + 2 + 4 = ~7.5s - delays := []time.Duration{0, 500 * time.Millisecond, time.Second, 2 * time.Second, 4 * time.Second} + delays := pfInterceptMonitorDelays for i, delay := range delays { if delay > 0 { @@ -2270,22 +2337,33 @@ func (p *prog) pfInterceptMonitor() { return } - if probePFInterceptFn(p) { + result := probePFInterceptFn(p) + if result.result == pfProbeIndeterminate { + mainLog.Load().Debug().Msg("DNS intercept monitor: probe indeterminate; repair deferred") + continue + } + if result.result == pfProbeIntercepted { mainLog.Load().Debug().Msgf("DNS intercept monitor: probe %d/%d passed", i+1, len(delays)) continue // working now — keep monitoring in case it breaks later in the window } // Probe failed — pf translation is broken. Force full reload. mainLog.Load().Warn().Msgf("DNS intercept monitor: probe %d/%d FAILED — pf translation broken, forcing full ruleset reload", i+1, len(delays)) - forceReloadPFInterceptFn(p) + if !p.repairPFIntercept(result, "monitor") { + continue + } // Verify the reload fixed it time.Sleep(200 * time.Millisecond) - if probePFInterceptFn(p) { - mainLog.Load().Info().Msg("DNS intercept monitor: probe passed after reload — interception restored") - // Continue monitoring in case the hypervisor reloads pf again - } else { - mainLog.Load().Error().Msg("DNS intercept monitor: probe still failing after reload — pf may need manual intervention") + confirmation := probePFInterceptFn(p) + logPFRepair(result, "monitor", confirmation.result.String(), true, confirmation) + switch confirmation.result { + case pfProbeIntercepted: + mainLog.Load().Info().Msg("DNS intercept monitor: probe passed after reload - interception restored") + case pfProbeNotIntercepted: + mainLog.Load().Warn().Msg("DNS intercept monitor: probe still failing after reload - pf may need manual intervention") + case pfProbeIndeterminate: + mainLog.Load().Warn().Msg("DNS intercept monitor: reload completed; verification indeterminate") } } diff --git a/cmd/cli/dns_intercept_darwin_test.go b/cmd/cli/dns_intercept_darwin_test.go index aad97a9..6c8b24e 100644 --- a/cmd/cli/dns_intercept_darwin_test.go +++ b/cmd/cli/dns_intercept_darwin_test.go @@ -838,13 +838,16 @@ func stubStabilizationProbe(t *testing.T, probeResults []bool, reloadOK bool) (p probePFInterceptFn, forceReloadPFInterceptFn = originalProbe, originalReload }) probeCalls, reloadCalls := 0, 0 - probePFInterceptFn = func(*prog) bool { + probePFInterceptFn = func(*prog) pfProbeObservation { result := false if probeCalls < len(probeResults) { result = probeResults[probeCalls] } probeCalls++ - return result + if result { + return pfProbeObservation{result: pfProbeIntercepted} + } + return pfProbeObservation{result: pfProbeNotIntercepted} } forceReloadPFInterceptFn = func(*prog) bool { reloadCalls++ diff --git a/cmd/cli/dns_proxy.go b/cmd/cli/dns_proxy.go index f1318d1..dc8b844 100644 --- a/cmd/cli/dns_proxy.go +++ b/cmd/cli/dns_proxy.go @@ -1655,7 +1655,7 @@ func FlushDNSCache() error { // monitorNetworkChanges starts monitoring for network interface changes func (p *prog) monitorNetworkChanges() error { - mon, err := netmon.New(func(format string, args ...any) { + mon, err := newNetworkChangeMonitorFn(func(format string, args ...any) { // Always fetch the latest logger (and inject the prefix) mainLog.Load().Printf("netmon: "+format, args...) }) @@ -1664,183 +1664,7 @@ func (p *prog) monitorNetworkChanges() error { } mon.RegisterChangeCallback(func(delta *netmon.ChangeDelta) { - isMajorChange := mon.IsMajorChangeFrom(delta.Old, delta.New) - p.handleDNS64NetworkChange(delta, isMajorChange) - - // Get map of valid interfaces - validIfaces := validInterfacesMap() - - mainLog.Load().Debug(). - Interface("old_state", delta.Old). - Interface("new_state", delta.New). - Bool("is_major_change", isMajorChange). - Msg("Network change detected") - - changed := false - activeInterfaceExists := false - var changeIPs []netip.Prefix - // Check each valid interface for changes - for ifaceName := range validIfaces { - oldIface, oldExists := delta.Old.Interface[ifaceName] - newIface, newExists := delta.New.Interface[ifaceName] - if !newExists { - continue - } - - oldIPs := delta.Old.InterfaceIPs[ifaceName] - newIPs := delta.New.InterfaceIPs[ifaceName] - - // if a valid interface did not exist in old - // check that its up and has usable IPs - if !oldExists { - // The interface is new (was not present in the old state). - usableNewIPs := filterUsableIPs(newIPs) - if newIface.IsUp() && len(usableNewIPs) > 0 { - changed = true - changeIPs = usableNewIPs - mainLog.Load().Debug(). - Str("interface", ifaceName). - Interface("new_ips", usableNewIPs). - Msg("Interface newly appeared (was not present in old state)") - break - } - continue - } - - // Filter new IPs to only those that are usable. - usableNewIPs := filterUsableIPs(newIPs) - - // Check if interface is up and has usable IPs. - if newIface.IsUp() && len(usableNewIPs) > 0 { - activeInterfaceExists = true - } - - // Compare interface states and IPs (interfaceIPsEqual will itself filter the IPs). - if !interfaceStatesEqual(&oldIface, &newIface) || !interfaceIPsEqual(oldIPs, newIPs) { - if newIface.IsUp() && len(usableNewIPs) > 0 { - changed = true - changeIPs = usableNewIPs - mainLog.Load().Debug(). - Str("interface", ifaceName). - Interface("old_ips", oldIPs). - Interface("new_ips", usableNewIPs). - Msg("Interface state or IPs changed") - break - } - } - } - - // if the default route changed, set changed to true - if delta.New.DefaultRouteInterface != delta.Old.DefaultRouteInterface { - changed = true - mainLog.Load().Debug().Msgf("Default route changed from %s to %s", delta.Old.DefaultRouteInterface, delta.New.DefaultRouteInterface) - } - - if !changed { - mainLog.Load().Debug().Msg("Ignoring interface change - no valid interfaces affected") - // check if the default IPs are still on an interface that is up - ValidateDefaultLocalIPsFromDelta(delta.New) - // Minor interface changes can still accompany pf/WFP or VPN DNS changes. - // On macOS, bound the immediate full reconciliation so link-local-only - // notification storms do not run pfctl/scutil work for every event. - // Windows keeps the existing immediate behavior. Tunnel changes always - // bypass the macOS limit, and delayed checks provide a trailing refresh. - if dnsIntercept && p.dnsInterceptState != nil { - p.handleDNSInterceptIgnoredNetworkChange(delta, time.Now()) - } - return - } - - if !activeInterfaceExists { - mainLog.Load().Debug().Msg("No active interfaces found, skipping reinitialization") - return - } - - mainLog.Load().Debug().Msg("Link state changed, re-bootstrapping") - for _, uc := range p.cfg.Upstream { - uc.ReBootstrap() - } - - // Get IPs from default route interface in new state - selfIP := defaultRouteIP() - - // Ensure that selfIP is an IPv4 address. - // If defaultRouteIP mistakenly returns an IPv6 (such as a ULA), clear it - if ip := net.ParseIP(selfIP); ip != nil && ip.To4() == nil { - mainLog.Load().Debug().Msgf("defaultRouteIP returned a non-IPv4 address: %s, ignoring it", selfIP) - selfIP = "" - } - var ipv6 string - - if delta.New.DefaultRouteInterface != "" { - mainLog.Load().Debug().Msgf("default route interface: %s, IPs: %v", delta.New.DefaultRouteInterface, delta.New.InterfaceIPs[delta.New.DefaultRouteInterface]) - for _, ip := range delta.New.InterfaceIPs[delta.New.DefaultRouteInterface] { - ipAddr, _ := netip.ParsePrefix(ip.String()) - addr := ipAddr.Addr() - if selfIP == "" && addr.Is4() { - mainLog.Load().Debug().Msgf("checking IP: %s", addr.String()) - if !addr.IsLoopback() && !addr.IsLinkLocalUnicast() { - selfIP = addr.String() - } - } - if addr.Is6() && !addr.IsLoopback() && !addr.IsLinkLocalUnicast() { - ipv6 = addr.String() - } - } - } else { - // If no default route interface is set yet, use the changed IPs - mainLog.Load().Debug().Msgf("no default route interface found, using changed IPs: %v", changeIPs) - for _, ip := range changeIPs { - ipAddr, _ := netip.ParsePrefix(ip.String()) - addr := ipAddr.Addr() - if selfIP == "" && addr.Is4() { - mainLog.Load().Debug().Msgf("checking IP: %s", addr.String()) - if !addr.IsLoopback() && !addr.IsLinkLocalUnicast() { - selfIP = addr.String() - } - } - if addr.Is6() && !addr.IsLoopback() && !addr.IsLinkLocalUnicast() { - ipv6 = addr.String() - } - } - } - - // Only set the IPv4 default if selfIP is a valid IPv4 address. - if ip := net.ParseIP(selfIP); ip != nil && ip.To4() != nil { - ctrld.SetDefaultLocalIPv4(ip) - if !isMobile() && p.ciTable != nil { - p.ciTable.SetSelfIP(selfIP) - } - } - if ip := net.ParseIP(ipv6); ip != nil { - ctrld.SetDefaultLocalIPv6(ip) - } - mainLog.Load().Debug().Msgf("Set default local IPv4: %s, IPv6: %s", selfIP, ipv6) - - // we only trigger recovery flow for network changes on non router devices - if router.Name() == "" { - p.debounceRecovery() - } - - // After network changes, verify our pf anchor is still active and - // refresh VPN DNS state. Order matters: tunnel checks first (may rebuild - // anchor), then VPN DNS refresh (updates exemptions in anchor), then - // delayed re-checks for async VPN teardown. - if dnsIntercept && p.dnsInterceptState != nil { - if !p.pfStabilizing.Load() { - p.ensurePFAnchorActive() - } - // Check tunnel interfaces unconditionally — it decides internally - // whether to enter stabilization or rebuild immediately. - p.checkTunnelInterfaceChanges() - // Refresh VPN DNS routes — runs after tunnel checks so the anchor - // rebuild includes current VPN DNS exemptions. - if p.vpnDNS != nil { - p.vpnDNS.Refresh(true) - } - // Schedule delayed re-checks to catch async VPN teardown changes. - p.scheduleDelayedRechecks() - } + p.handleNetworkChange(delta, mon.IsMajorChangeFrom(delta.Old, delta.New)) }) mon.Start() @@ -1848,6 +1672,269 @@ func (p *prog) monitorNetworkChanges() error { return nil } +// handleNetworkChange is the network monitor callback, kept separate for synthetic deltas. +func (p *prog) handleNetworkChange(delta *netmon.ChangeDelta, isMajorChange bool) { + // netmon v1.74.0 dispatches callbacks concurrently and caches only major + // snapshots. Minor events refer to that cache through Old. The synthetic + // delta.Major flag after a time jump does not change the cache. + p.networkSourceMu.Lock() + currentState := networkChangeCurrentStateFn(delta) + if !networkSnapshotCurrent(delta, isMajorChange, currentState) { + p.networkSourceMu.Unlock() + mainLog.Load().Debug().Msg("Network transition skipped: snapshot superseded") + return + } + transitionID := p.networkTransitionGen.Add(1) + sourceState := delta.New + // A late major callback and reordered minor callbacks share one cache + // epoch. For these cases, source validity comes from the OS, not ordering. + if delta.Monitor != nil && (!isMajorChange || (p.networkSourceEpoch == currentState && p.networkSourceState != currentState)) { + fresh, err := readNetworkSourceStateFn() + if err != nil || fresh == nil { + if !p.networkSourceReadFailed { + mainLog.Load().Warn().Uint64("transition_id", transitionID). + Msg("Network source snapshot unavailable; source writes deferred") + } + p.networkSourceReadFailed = true + sourceState = nil + } else { + snapshot := *fresh + snapshot.DefaultRouteInterface = currentState.DefaultRouteInterface + sourceState = &snapshot + if p.networkSourceReadFailed { + mainLog.Load().Warn().Uint64("transition_id", transitionID).Msg("Network source snapshot available again") + } + p.networkSourceReadFailed = false + } + } + p.networkSourceState, p.networkSourceEpoch = sourceState, currentState + sourceSnapshotAvailable := sourceState != nil + before4, before6 := ctrld.GetDefaultLocalIPv4(), ctrld.GetDefaultLocalIPv6() + validateDefaultLocalIPsFromDelta(sourceState, transitionID) + after4, after6 := ctrld.GetDefaultLocalIPv4(), ctrld.GetDefaultLocalIPv6() + p.networkSourceMu.Unlock() + + p.handleDNS64NetworkChange(delta, isMajorChange) + validIfaces := networkChangeValidInterfacesFn() + + changed := false + changedInterface := "" + outcome := "ignored" + defer func() { + mainLog.Load().Debug().Uint64("transition_id", transitionID). + Bool("is_major_change", isMajorChange).Bool("changed", changed). + Str("interface", changedInterface). + Str("default_route_before", delta.Old.DefaultRouteInterface). + Str("default_route_after", delta.New.DefaultRouteInterface). + Str("source_ipv4_before", before4.String()).Str("source_ipv6_before", before6.String()). + Str("source_ipv4_after", after4.String()). + Str("source_ipv6_after", after6.String()). + Bool("source_snapshot_available", sourceSnapshotAvailable).Str("outcome", outcome).Msg("Network transition") + }() + activeInterfaceExists := false + var changeIPs []netip.Prefix + // Check each valid interface for changes + for ifaceName := range validIfaces { + oldIface, oldExists := delta.Old.Interface[ifaceName] + newIface, newExists := delta.New.Interface[ifaceName] + if !newExists { + continue + } + + oldIPs := delta.Old.InterfaceIPs[ifaceName] + newIPs := delta.New.InterfaceIPs[ifaceName] + + // if a valid interface did not exist in old + // check that its up and has usable IPs + if !oldExists { + // The interface is new (was not present in the old state). + usableNewIPs := filterUsableIPs(newIPs) + if newIface.IsUp() && len(usableNewIPs) > 0 { + changed = true + changeIPs = usableNewIPs + changedInterface = ifaceName + mainLog.Load().Debug(). + Str("interface", ifaceName). + Interface("new_ips", usableNewIPs). + Msg("Interface newly appeared (was not present in old state)") + break + } + continue + } + + // Filter new IPs to only those that are usable. + usableNewIPs := filterUsableIPs(newIPs) + + // Check if interface is up and has usable IPs. + if newIface.IsUp() && len(usableNewIPs) > 0 { + activeInterfaceExists = true + } + + // Compare interface states and IPs (interfaceIPsEqual will itself filter the IPs). + if !interfaceStatesEqual(&oldIface, &newIface) || !interfaceIPsEqual(oldIPs, newIPs) { + if newIface.IsUp() && len(usableNewIPs) > 0 { + changed = true + changeIPs = usableNewIPs + changedInterface = ifaceName + mainLog.Load().Debug(). + Str("interface", ifaceName). + Interface("old_ips", oldIPs). + Interface("new_ips", usableNewIPs). + Msg("Interface state or IPs changed") + break + } + } + } + + // if the default route changed, set changed to true + if delta.New.DefaultRouteInterface != delta.Old.DefaultRouteInterface { + changed = true + mainLog.Load().Debug().Msgf("Default route changed from %s to %s", delta.Old.DefaultRouteInterface, delta.New.DefaultRouteInterface) + } + + if !changed { + mainLog.Load().Debug().Msg("Ignoring interface change - no valid interfaces affected") + // Minor interface changes can still accompany pf/WFP or VPN DNS changes. + // On macOS, bound the immediate full reconciliation so link-local-only + // notification storms do not run pfctl/scutil work for every event. + // Windows keeps the existing immediate behavior. Tunnel changes always + // bypass the macOS limit, and delayed checks provide a trailing refresh. + if dnsIntercept && p.dnsInterceptState != nil { + networkChangeIgnoredInterceptFn(p, delta, time.Now()) + } + return + } + + if !activeInterfaceExists { + outcome = "no_active_interface" + mainLog.Load().Debug().Msg("No active interfaces found, skipping reinitialization") + return + } + + // An ignored event must not cancel this accepted transition. Only a newer + // accepted transition can replace its source commit or debounced recovery. + p.networkSourceMu.Lock() + if p.networkAcceptedGen.Load() > transitionID { + p.networkSourceMu.Unlock() + outcome = "superseded" + return + } + p.networkAcceptedGen.Store(transitionID) + p.networkSourceMu.Unlock() + + mainLog.Load().Debug().Msg("Link state changed, re-bootstrapping") + for _, uc := range p.cfg.Upstream { + uc.ReBootstrap() + } + + // Get IPs from default route interface in new state + selfIP := networkChangeDefaultRouteIPFn() + + // Ensure that selfIP is an IPv4 address. + // If defaultRouteIP mistakenly returns an IPv6 (such as a ULA), clear it + if ip := net.ParseIP(selfIP); ip != nil && ip.To4() == nil { + mainLog.Load().Debug().Msgf("defaultRouteIP returned a non-IPv4 address: %s, ignoring it", selfIP) + selfIP = "" + } + // Route discovery can lag the delta; do not reintroduce an invalid source. + if ip := net.ParseIP(selfIP); ip != nil && sourceInvalidReason(delta.New, ip) != "" { + selfIP = "" + } + var ipv6 string + + if delta.New.DefaultRouteInterface != "" { + mainLog.Load().Debug().Msgf("default route interface: %s, IPs: %v", delta.New.DefaultRouteInterface, delta.New.InterfaceIPs[delta.New.DefaultRouteInterface]) + for _, ip := range delta.New.InterfaceIPs[delta.New.DefaultRouteInterface] { + if sourceInvalidReason(delta.New, net.ParseIP(ip.Addr().String())) != "" { + continue + } + ipAddr, _ := netip.ParsePrefix(ip.String()) + addr := ipAddr.Addr() + if selfIP == "" && addr.Is4() { + mainLog.Load().Debug().Msgf("checking IP: %s", addr.String()) + if !addr.IsLoopback() && !addr.IsLinkLocalUnicast() { + selfIP = addr.String() + } + } + if addr.Is6() && !addr.IsLoopback() && !addr.IsLinkLocalUnicast() { + ipv6 = addr.String() + } + } + } else { + // If no default route interface is set yet, use the changed IPs + mainLog.Load().Debug().Msgf("no default route interface found, using changed IPs: %v", changeIPs) + for _, ip := range changeIPs { + ipAddr, _ := netip.ParsePrefix(ip.String()) + addr := ipAddr.Addr() + if selfIP == "" && addr.Is4() { + mainLog.Load().Debug().Msgf("checking IP: %s", addr.String()) + if !addr.IsLoopback() && !addr.IsLinkLocalUnicast() { + selfIP = addr.String() + } + } + if addr.Is6() && !addr.IsLoopback() && !addr.IsLinkLocalUnicast() { + ipv6 = addr.String() + } + } + } + + // An ignored event can change source validity without replacing recovery. + // Recheck addresses against the newest state after blocking route lookup. + p.networkSourceMu.Lock() + if p.networkAcceptedGen.Load() != transitionID { + p.networkSourceMu.Unlock() + outcome = "superseded" + return + } + latestState := p.sourceCommitState(delta) + sourceSnapshotAvailable = latestState != nil + validateDefaultLocalIPsFromDelta(latestState, transitionID) + // Only keep candidate addresses that still exist on an up interface. + if ip := net.ParseIP(selfIP); ip != nil && ip.To4() != nil && latestState != nil && sourceInvalidReason(latestState, ip) == "" { + ctrld.SetDefaultLocalIPv4(ip) + if !isMobile() && p.ciTable != nil { + p.ciTable.SetSelfIP(selfIP) + } + } + if ip := net.ParseIP(ipv6); ip != nil && latestState != nil && sourceInvalidReason(latestState, ip) == "" { + ctrld.SetDefaultLocalIPv6(ip) + } + after4, after6 = ctrld.GetDefaultLocalIPv4(), ctrld.GetDefaultLocalIPv6() + p.networkSourceMu.Unlock() + mainLog.Load().Debug().Msgf("Set default local IPv4: %s, IPv6: %s", selfIP, ipv6) + + outcome = "accepted" + networkChangeReconcileFn(p, transitionID) +} + +// reconcileNetworkChange is the first DNS/PF mutation boundary after source updates. +func (p *prog) reconcileNetworkChange(transitionID uint64) { + // we only trigger recovery flow for network changes on non router devices + if router.Name() == "" { + p.debounceRecovery(transitionID) + } + + // After network changes, verify our pf anchor is still active and + // refresh VPN DNS state. Order matters: tunnel checks first (may rebuild + // anchor), then VPN DNS refresh (updates exemptions in anchor), then + // delayed re-checks for async VPN teardown. + if dnsIntercept && p.dnsInterceptState != nil { + if !p.pfStabilizing.Load() { + p.ensurePFAnchorActive() + } + // Check tunnel interfaces unconditionally — it decides internally + // whether to enter stabilization or rebuild immediately. + p.checkTunnelInterfaceChanges() + // Refresh VPN DNS routes — runs after tunnel checks so the anchor + // rebuild includes current VPN DNS exemptions. + if p.vpnDNS != nil { + p.vpnDNS.Refresh(true) + } + // Schedule delayed re-checks to catch async VPN teardown changes. + p.scheduleDelayedRechecks() + } +} + // handleDNSInterceptIgnoredNetworkChange runs the DNS-intercept work for a // network delta that did not affect a usable interface. Keeping this path in a // method lets tests exercise the callback wiring with synthetic deltas. @@ -2035,7 +2122,12 @@ const recoveryDebounceWindow = 500 * time.Millisecond // recovery runs once with the final network state. All other state updates (IP, // pf anchor, VPN DNS, tunnel checks) run immediately — only the recovery flow // with its upstream probing and DHCP bypass logic is debounced. -func (p *prog) debounceRecovery() { +func (p *prog) debounceRecovery(transitionID uint64) { + p.networkSourceMu.Lock() + defer p.networkSourceMu.Unlock() + if p.networkAcceptedGen.Load() != transitionID { + return + } p.recoveryDebounceMu.Lock() defer p.recoveryDebounceMu.Unlock() @@ -2044,10 +2136,16 @@ func (p *prog) debounceRecovery() { mainLog.Load().Debug().Msg("Recovery debounce: resetting timer (rapid network change)") } p.recoveryDebounceTimer = time.AfterFunc(recoveryDebounceWindow, func() { + p.networkSourceMu.Lock() + if p.networkAcceptedGen.Load() != transitionID { + p.networkSourceMu.Unlock() + return + } p.recoveryDebounceMu.Lock() p.recoveryDebounceTimer = nil p.recoveryDebounceMu.Unlock() - p.handleRecovery(RecoveryReasonNetworkChange) + p.networkSourceMu.Unlock() + handleRecoveryForTransitionFn(p, RecoveryReasonNetworkChange, transitionID) }) mainLog.Load().Debug().Msg("Recovery debounce: scheduled (500ms window)") } @@ -2057,13 +2155,28 @@ func (p *prog) debounceRecovery() { // upstream failure recoveries, waiting for recovery to complete (using a cancellable context without timeout), // and then re-applying the DNS settings. func (p *prog) handleRecovery(reason RecoveryReason) { - mainLog.Load().Debug().Msg("Starting recovery process: removing DNS settings") + p.handleRecoveryForTransition(reason, 0) +} - recoveryCtx, gen, interceptRecovery, ok := p.beginRecovery(reason) - if !ok { - mainLog.Load().Debug().Msg("Upstream recovery already in progress; skipping duplicate trigger") +func (p *prog) handleRecoveryForTransition(reason RecoveryReason, transitionID uint64) { + p.networkSourceMu.Lock() + if transitionID != 0 && p.networkAcceptedGen.Load() != transitionID { + p.networkSourceMu.Unlock() + mainLog.Load().Debug().Uint64("transition_id", transitionID).Msg("Recovery skipped: transition superseded") return } + recoveryCtx, gen, interceptRecovery, ok := p.beginRecovery(reason) + p.networkSourceMu.Unlock() + if !ok { + mainLog.Load().Debug().Uint64("transition_id", transitionID). + Uint64("recovery_generation", p.recoveryGen.Load()). + Msg("Upstream recovery already in progress; skipping duplicate trigger") + return + } + diagnostic := recoveryDiagnostic{transitionID: transitionID, generation: gen, reason: reason, started: time.Now()} + diagnostic.event(mainLog.Load().Debug()).Bool("intercept", interceptRecovery).Msg("Recovery begin") + outcome := "superseded" + defer func() { diagnostic.end(outcome) }() if reason == RecoveryReasonNetworkChange { mainLog.Load().Debug().Msg("Network change recovery now owns shared recovery state") } @@ -2125,7 +2238,7 @@ func (p *prog) handleRecovery(reason RecoveryReason) { } } else { // Traditional flow: remove DNS settings to expose DHCP nameservers - p.resetDNS(false, false) + recoveryResetDNSFn(p, false, false) // For an OS failure, reinitialize OS resolver nameservers immediately. if reason == RecoveryReasonOSFailure { @@ -2143,9 +2256,11 @@ func (p *prog) handleRecovery(reason RecoveryReason) { upstreams := p.buildRecoveryUpstreams(reason) // Wait indefinitely until one of the upstreams recovers. - recovered, err := p.waitForUpstreamRecovery(recoveryCtx, upstreams) + recovered, err := p.waitForUpstreamRecovery(recoveryCtx, upstreams, &diagnostic) if err != nil { - mainLog.Load().Error().Err(err).Msg("Recovery canceled; DNS settings remain removed") + if p.recoveryOwnsState(gen) { + outcome = "canceled" + } p.recoveryCanceledCleanup(gen) return } @@ -2200,6 +2315,7 @@ func (p *prog) handleRecovery(reason RecoveryReason) { mainLog.Load().Debug().Msgf("Recovery generation %d was superseded during completion; preserving successor state", gen) return } + outcome = "completed" if interceptRecovery { mainLog.Load().Info().Msg("DNS intercept recovery complete: disabling DHCP bypass, resuming normal flow") } @@ -2207,7 +2323,7 @@ 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) { +func (p *prog) waitForUpstreamRecovery(ctx context.Context, upstreams map[string]*ctrld.UpstreamConfig, diagnostic *recoveryDiagnostic) (string, error) { recoveryCtx, cancel := context.WithCancel(ctx) defer cancel() @@ -2243,6 +2359,7 @@ func (p *prog) waitForUpstreamRecovery(ctx context.Context, upstreams map[string } return } + diagnostic.failure(err) // Back off the retry cadence for an unroutable endpoint so a // host with IPv6 up but no route to the IPv6 DoH endpoint does // not re-bootstrap/re-check every checkUpstreamBackoffSleep and @@ -2330,27 +2447,8 @@ func (p *prog) buildRecoveryUpstreams(reason RecoveryReason) map[string]*ctrld.U // are still present in the new network state (provided by delta.New). // If a stored default IP is no longer active, it resets that default (sets it to nil) // so that it won't be used in subsequent custom dialer contexts. +// Deprecated: retained for callers of the existing exported cli API. +// The network monitor uses transition-aware validation instead. func ValidateDefaultLocalIPsFromDelta(newState *netmon.State) { - currentIPv4 := ctrld.GetDefaultLocalIPv4() - currentIPv6 := ctrld.GetDefaultLocalIPv6() - - // Build a map of active IP addresses from the new state. - activeIPs := make(map[string]bool) - for _, prefixes := range newState.InterfaceIPs { - for _, prefix := range prefixes { - activeIPs[prefix.Addr().String()] = true - } - } - - // Check if the default IPv4 is still active. - if currentIPv4 != nil && !activeIPs[currentIPv4.String()] { - mainLog.Load().Debug().Msgf("DefaultLocalIPv4 %s is no longer active in the new state. Resetting.", currentIPv4) - ctrld.SetDefaultLocalIPv4(nil) - } - - // Check if the default IPv6 is still active. - if currentIPv6 != nil && !activeIPs[currentIPv6.String()] { - mainLog.Load().Debug().Msgf("DefaultLocalIPv6 %s is no longer active in the new state. Resetting.", currentIPv6) - ctrld.SetDefaultLocalIPv6(nil) - } + validateDefaultLocalIPsFromDelta(newState, 0) } diff --git a/cmd/cli/main.go b/cmd/cli/main.go index 7e6dd8d..d372e8d 100644 --- a/cmd/cli/main.go +++ b/cmd/cli/main.go @@ -1,7 +1,6 @@ package cli import ( - "encoding/hex" "io" "net" "os" @@ -214,18 +213,7 @@ func initCache() { // // Usage: ctrld pf-probe-send func pfProbeSend(host, hexPacket string) { - packet, err := hex.DecodeString(hexPacket) - if err != nil { + if err := sendPFProbe(host, hexPacket, net.DialTimeout, os.Stdout); err != nil { os.Exit(1) } - conn, err := net.DialTimeout("udp", net.JoinHostPort(host, "53"), time.Second) - if err != nil { - os.Exit(1) - } - defer conn.Close() - conn.SetDeadline(time.Now().Add(time.Second)) - _, _ = conn.Write(packet) - // Read response (don't care about result, just need the send to happen) - buf := make([]byte, 512) - _, _ = conn.Read(buf) } diff --git a/cmd/cli/network_monitor_test.go b/cmd/cli/network_monitor_test.go new file mode 100644 index 0000000..33dba8d --- /dev/null +++ b/cmd/cli/network_monitor_test.go @@ -0,0 +1,52 @@ +package cli + +import ( + "net" + "testing" + + ctrld "github.com/Control-D-Inc/ctrld" + "tailscale.com/net/netmon" +) + +type sourceTestMonitor struct { + callback netmon.ChangeFunc + started bool +} + +func (m *sourceTestMonitor) RegisterChangeCallback(fn netmon.ChangeFunc) func() { + m.callback = fn + return func() {} +} +func (*sourceTestMonitor) IsMajorChangeFrom(*netmon.State, *netmon.State) bool { return true } +func (m *sourceTestMonitor) Start() { m.started = true } + +func TestNetworkMonitorRegistersSourceCleanup(t *testing.T) { + sourceTestGlobals(t) + old := newNetworkChangeMonitorFn + t.Cleanup(func() { newNetworkChangeMonitorFn = old }) + mon := &sourceTestMonitor{} + newNetworkChangeMonitorFn = func(func(string, ...any)) (networkChangeMonitor, error) { return mon, nil } + ctrld.SetDefaultLocalIPv4(net.ParseIP("192.0.2.10")) + ctrld.SetDefaultLocalIPv6(net.ParseIP("2001:db8::10")) + calls := 0 + networkChangeReconcileFn = func(*prog, uint64) { + calls++ + if ctrld.GetDefaultLocalIPv4() != nil { + t.Fatal("departed IPv4 survived to reconciliation") + } + } + p := &prog{cfg: &ctrld.Config{}, dnsInterceptState: &interceptStateStub{}} + if err := p.monitorNetworkChanges(); err != nil { + t.Fatal(err) + } + if !mon.started || mon.callback == nil { + t.Fatal("monitor did not register/start") + } + mon.callback(&netmon.ChangeDelta{ + Old: sourceTestState("en0", true, "192.0.2.10/24", "2001:db8::10/64"), + New: sourceTestState("en0", true, "2001:db8::10/64"), + }) + if calls != 1 { + t.Fatalf("registered callback never reached source reconciliation: %d", calls) + } +} diff --git a/cmd/cli/network_source_error_test.go b/cmd/cli/network_source_error_test.go new file mode 100644 index 0000000..54450fc --- /dev/null +++ b/cmd/cli/network_source_error_test.go @@ -0,0 +1,32 @@ +package cli + +import ( + "errors" + "net" + "testing" + + ctrld "github.com/Control-D-Inc/ctrld" + "tailscale.com/net/netmon" +) + +func TestSourceReadFailureDefersAcceptedWrites(t *testing.T) { + sourceTestGlobals(t) + dnsIntercept = false + oldCurrent, oldRead := networkChangeCurrentStateFn, readNetworkSourceStateFn + t.Cleanup(func() { networkChangeCurrentStateFn, readNetworkSourceStateFn = oldCurrent, oldRead }) + major := sourceTestState("en0", true, "192.0.2.10/24") + networkChangeCurrentStateFn = func(*netmon.ChangeDelta) *netmon.State { return major } + readNetworkSourceStateFn = func() (*netmon.State, error) { return nil, errors.New("read failed") } + ctrld.SetDefaultLocalIPv4(net.ParseIP("192.0.2.1")) + ctrld.SetDefaultLocalIPv6(nil) + p := &prog{cfg: &ctrld.Config{}, networkSourceEpoch: major} + calls := 0 + networkChangeReconcileFn = func(*prog, uint64) { calls++ } + p.handleNetworkChange(&netmon.ChangeDelta{ + Monitor: netmon.NewStatic(), + Old: sourceTestState("en0", true, "192.0.2.1/24"), New: major, + }, true) + if calls != 1 || !ctrld.GetDefaultLocalIPv4().Equal(net.ParseIP("192.0.2.1")) { + t.Fatal("unknown source validity changed a binding or skipped accepted recovery") + } +} diff --git a/cmd/cli/network_transition.go b/cmd/cli/network_transition.go new file mode 100644 index 0000000..5638e2c --- /dev/null +++ b/cmd/cli/network_transition.go @@ -0,0 +1,200 @@ +package cli + +import ( + "context" + "errors" + "net" + "net/netip" + "sync" + "sync/atomic" + "syscall" + "time" + + "github.com/rs/zerolog" + "github.com/Control-D-Inc/ctrld" + ctrldnet "github.com/Control-D-Inc/ctrld/internal/net" + "tailscale.com/net/netmon" +) + +type networkChangeMonitor interface { + RegisterChangeCallback(netmon.ChangeFunc) func() + IsMajorChangeFrom(*netmon.State, *netmon.State) bool + Start() +} + +// Tests replace these functions to avoid changes to host DNS and PF state. +var ( + newNetworkChangeMonitorFn = func(logf func(string, ...any)) (networkChangeMonitor, error) { + return netmon.New(logf) + } + networkChangeCurrentStateFn = func(delta *netmon.ChangeDelta) *netmon.State { + if delta.Monitor != nil { + return delta.Monitor.InterfaceState() + } + return delta.New + } + readNetworkSourceStateFn = readNetworkSourceState + networkChangeValidInterfacesFn = validInterfacesMap + networkChangeDefaultRouteIPFn = defaultRouteIP + networkChangeReconcileFn = (*prog).reconcileNetworkChange + networkChangeIgnoredInterceptFn = (*prog).handleDNSInterceptIgnoredNetworkChange + handleRecoveryForTransitionFn = (*prog).handleRecoveryForTransition + recoveryResetDNSFn = (*prog).resetDNS +) + +// Minor callbacks have no observation sequence in netmon v1.74.0. Read only +// current interface addresses and flags rather than trusting their arrival order. +func readNetworkSourceState() (*netmon.State, error) { + interfaces, err := net.Interfaces() + if err != nil { + return nil, err + } + state := &netmon.State{Interface: map[string]netmon.Interface{}, InterfaceIPs: map[string][]netip.Prefix{}} + for _, iface := range interfaces { + addresses, err := iface.Addrs() + if err != nil { + return nil, err + } + state.Interface[iface.Name] = netmon.Interface{Interface: &iface} + for _, address := range addresses { + var ip net.IP + switch address := address.(type) { + case *net.IPNet: + ip = address.IP + case *net.IPAddr: + ip = address.IP + default: + return nil, errors.New("unsupported interface address type") + } + addr, ok := netip.AddrFromSlice(ip) + if !ok { + return nil, errors.New("invalid interface address") + } + addr = addr.Unmap() + state.InterfaceIPs[iface.Name] = append(state.InterfaceIPs[iface.Name], netip.PrefixFrom(addr, addr.BitLen())) + } + } + return state, nil +} + +// netmon v1.74.0 caches only major snapshots. Minor and time-jump events +// belong to their Old snapshot, not their distinct New snapshot. +func networkSnapshotCurrent(delta *netmon.ChangeDelta, major bool, current *netmon.State) bool { + if major || delta.Monitor == nil { + return current == delta.New + } + return current == delta.Old +} + +// Callers hold networkSourceMu. Preserve minor changes within the current +// major epoch, but use a new major snapshot even before its callback runs. +func (p *prog) sourceCommitState(delta *netmon.ChangeDelta) *netmon.State { + if delta.Monitor != nil { + if current := networkChangeCurrentStateFn(delta); current != p.networkSourceEpoch { + return current + } + } + return p.networkSourceState +} + +// sourceInvalidReason considers all up interfaces, not just hardware ports: a +// preferred source remains valid if it moved to another active interface. +func sourceInvalidReason(state *netmon.State, ip net.IP) string { + if ip == nil || state == nil { + return "" + } + reason := "address_removed" + for name, prefixes := range state.InterfaceIPs { + for _, prefix := range prefixes { + if !ip.Equal(net.ParseIP(prefix.Addr().String())) { + continue + } + iface, exists := state.Interface[name] + if exists && iface.IsUp() { + return "" + } + if exists { + reason = "interface_down" + } else if reason != "interface_down" { + reason = "interface_removed" + } + } + } + return reason +} + +func validateDefaultLocalIPsFromDelta(state *netmon.State, transitionID uint64) { + before4, before6 := ctrld.GetDefaultLocalIPv4(), ctrld.GetDefaultLocalIPv6() + reason4, reason6 := sourceInvalidReason(state, before4), sourceInvalidReason(state, before6) + if reason4 != "" { + ctrld.SetDefaultLocalIPv4(nil) + } + if reason6 != "" { + ctrld.SetDefaultLocalIPv6(nil) + } + if reason4 == "" && reason6 == "" { + return + } + // One warning per actual invalidation, not per notification or query. The + // normal warning buffer retains this summary when debug traffic rotates. + mainLog.Load().Warn().Uint64("transition_id", transitionID). + Str("default_route", state.DefaultRouteInterface). + Str("source_ipv4_before", before4.String()).Str("source_ipv6_before", before6.String()). + Str("source_ipv4_after", ctrld.GetDefaultLocalIPv4().String()). + Str("source_ipv6_after", ctrld.GetDefaultLocalIPv6().String()). + Str("ipv4_clear_reason", reason4).Str("ipv6_clear_reason", reason6). + Msg("Removed stale resolver source") +} + +type recoveryDiagnostic struct { + transitionID uint64 + generation uint64 + reason RecoveryReason + started time.Time + firstFailure sync.Once + failed atomic.Bool +} + +func recoveryReasonName(reason RecoveryReason) string { + switch reason { + case RecoveryReasonNetworkChange: + return "network_change" + case RecoveryReasonRegularFailure: + return "upstream_failure" + case RecoveryReasonOSFailure: + return "os_resolver_failure" + default: + return "unknown" + } +} + +func (d *recoveryDiagnostic) event(e *zerolog.Event) *zerolog.Event { + return e.Uint64("transition_id", d.transitionID).Uint64("recovery_generation", d.generation). + Str("recovery_reason", recoveryReasonName(d.reason)) +} + +func (d *recoveryDiagnostic) failure(err error) { + d.firstFailure.Do(func() { + d.failed.Store(true) + // Classify only; resolver errors can contain queried names or endpoints. + failure := "upstream_error" + var networkError net.Error + if errors.Is(err, syscall.EADDRNOTAVAIL) { + failure = "source_unavailable" + } else if ctrldnet.IsUnreachable(err) { + failure = "network_unreachable" + } else if errors.Is(err, context.DeadlineExceeded) || (errors.As(err, &networkError) && networkError.Timeout()) { + failure = "timeout" + } + d.event(mainLog.Load().Warn()).Str("failure", failure). + Msg("Recovery waiting for upstream; first failure") + }) +} + +func (d *recoveryDiagnostic) end(outcome string) { + e := mainLog.Load().Debug() + if outcome == "canceled" || d.failed.Load() { + e = mainLog.Load().Warn() + } + d.event(e).Str("outcome", outcome).Bool("had_failure", d.failed.Load()).Dur("duration_ms", time.Since(d.started)).Msg("Recovery end") +} diff --git a/cmd/cli/network_transition_concurrency_test.go b/cmd/cli/network_transition_concurrency_test.go new file mode 100644 index 0000000..1b9e1bf --- /dev/null +++ b/cmd/cli/network_transition_concurrency_test.go @@ -0,0 +1,87 @@ +package cli + +import ( + "net" + "strings" + "testing" + "time" + + ctrld "github.com/Control-D-Inc/ctrld" + "tailscale.com/net/netmon" +) + +func TestNetworkChangeOverlappingCallbackCannotRestoreRemovedSource(t *testing.T) { + sourceTestGlobals(t) + logs := captureTransitionLogs(t) + ctrld.SetDefaultLocalIPv4(net.ParseIP("192.0.2.10")) + ctrld.SetDefaultLocalIPv6(nil) + entered, release, done := make(chan struct{}), make(chan struct{}), make(chan struct{}) + networkChangeDefaultRouteIPFn = func() string { + close(entered) + <-release + return "192.0.2.10" + } + reconciliations := 0 + networkChangeReconcileFn = func(*prog, uint64) { reconciliations++ } + networkChangeIgnoredInterceptFn = func(*prog, *netmon.ChangeDelta, time.Time) {} + p := &prog{cfg: &ctrld.Config{}, dnsInterceptState: &interceptStateStub{}} + stateA := sourceTestState("en0", true, "192.0.2.10/24") + t.Cleanup(func() { + select { + case <-release: + default: + close(release) + } + <-done + }) + go func() { + defer close(done) + p.handleNetworkChange(&netmon.ChangeDelta{Old: sourceTestState("en0", true, "192.0.2.20/24"), New: stateA}, true) + }() + select { + case <-entered: + case <-time.After(3 * time.Second): + t.Fatal("first callback did not reach route discovery") + } + // The newer callback completes while A is paused. Its ignored/address-loss + // path clears the source and deliberately schedules no replacement recovery. + p.handleNetworkChange(&netmon.ChangeDelta{Old: stateA, New: sourceTestState("en0", true)}, true) + close(release) + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("superseded callback did not finish") + } + if ctrld.GetDefaultLocalIPv4() != nil { + t.Fatal("older callback restored the removed source") + } + if reconciliations != 1 || !strings.Contains(logs.String(), `"outcome":"accepted"`) { + t.Fatal("an ignored address-loss delta must preserve the accepted recovery") + } +} + +func TestNetworkChangeLateOldSnapshotIsRejected(t *testing.T) { + sourceTestGlobals(t) + old := networkChangeCurrentStateFn + t.Cleanup(func() { networkChangeCurrentStateFn = old }) + latest := sourceTestState("en0", true) + networkChangeCurrentStateFn = func(*netmon.ChangeDelta) *netmon.State { return latest } + ctrld.SetDefaultLocalIPv4(nil) + ctrld.SetDefaultLocalIPv6(nil) + networkChangeReconcileFn = func(*prog, uint64) { t.Fatal("old snapshot reconciled") } + networkChangeIgnoredInterceptFn = func(*prog, *netmon.ChangeDelta, time.Time) { t.Fatal("old snapshot reconciled") } + p := &prog{cfg: &ctrld.Config{}} + p.handleNetworkChange(&netmon.ChangeDelta{Old: latest, New: sourceTestState("en0", true, "192.0.2.10/24")}, true) + if p.networkTransitionGen.Load() != 0 || ctrld.GetDefaultLocalIPv4() != nil { + t.Fatal("old snapshot became a new transition/source") + } +} + +func TestRecoveryRejectsSupersededTransition(t *testing.T) { + p := &prog{cfg: &ctrld.Config{}} + p.networkAcceptedGen.Store(2) + p.handleRecoveryForTransition(RecoveryReasonNetworkChange, 1) + if p.recoveryGen.Load() != 0 || p.recoveryRunning.Load() { + t.Fatal("superseded transition started recovery") + } +} diff --git a/cmd/cli/network_transition_reordered_test.go b/cmd/cli/network_transition_reordered_test.go new file mode 100644 index 0000000..2946af2 --- /dev/null +++ b/cmd/cli/network_transition_reordered_test.go @@ -0,0 +1,94 @@ +package cli + +import ( + "errors" + "net" + "net/netip" + "testing" + "time" + + ctrld "github.com/Control-D-Inc/ctrld" + "tailscale.com/net/netmon" +) + +func TestLateSameEpochCallbacksCannotRestoreRemovedSource(t *testing.T) { + for _, initial := range []string{"192.0.2.1", "192.0.2.10"} { + t.Run(initial, func(t *testing.T) { + sourceTestGlobals(t) + dnsIntercept = false + oldCurrent, oldRead := networkChangeCurrentStateFn, readNetworkSourceStateFn + t.Cleanup(func() { networkChangeCurrentStateFn, readNetworkSourceStateFn = oldCurrent, oldRead }) + mon := netmon.NewStatic() + before := sourceTestState("en0", true, "192.0.2.1/24") + major := sourceTestState("ipsec0", true, "192.0.2.1/24") + major.Interface["ipsec0"] = netmon.Interface{Interface: &net.Interface{Name: "ipsec0", Flags: net.FlagUp | net.FlagPointToPoint}} + major.InterfaceIPs["ipsec0"] = []netip.Prefix{netip.MustParsePrefix("192.0.2.10/24")} + latest := sourceTestState("ipsec0", true, "192.0.2.1/24") + latest.Interface["ipsec0"] = major.Interface["ipsec0"] + // netmon ignores ipsec address changes on Darwin. Its major cache + // stays at A while B removes the source. These are OS-read seams. + networkChangeCurrentStateFn = func(*netmon.ChangeDelta) *netmon.State { return major } + reads := 0 + readNetworkSourceStateFn = func() (*netmon.State, error) { reads++; return latest, nil } + ctrld.SetDefaultLocalIPv4(net.ParseIP(initial)) + ctrld.SetDefaultLocalIPv6(nil) + p := &prog{cfg: &ctrld.Config{}} + reconciled := 0 + networkChangeReconcileFn = func(*prog, uint64) { reconciled++ } + // B finishes before A's major callback even enters. A second, older + // minor callback also arrives late in that same cache epoch. + p.handleNetworkChange(&netmon.ChangeDelta{Monitor: mon, Old: major, New: latest}, false) + olderMinor := *major + p.handleNetworkChange(&netmon.ChangeDelta{Monitor: mon, Old: major, New: &olderMinor}, false) + if len(p.networkSourceState.InterfaceIPs["ipsec0"]) != 0 { + t.Fatal("late minor callback replaced current source validity") + } + p.handleNetworkChange(&netmon.ChangeDelta{Monitor: mon, Old: before, New: major}, true) + want := net.IP(nil) + if initial == "192.0.2.1" { + want = net.ParseIP(initial) + } + if !ctrld.GetDefaultLocalIPv4().Equal(want) { + t.Fatalf("late major restored an obsolete source: got %v want %v", ctrld.GetDefaultLocalIPv4(), want) + } + if reads != 3 || reconciled != 1 { + t.Fatalf("fresh reads=%d reconciliation=%d", reads, reconciled) + } + }) + } +} + +func TestMinorSourceReadFailureDoesNotSkipReconciliation(t *testing.T) { + sourceTestGlobals(t) + oldRead := readNetworkSourceStateFn + t.Cleanup(func() { readNetworkSourceStateFn = oldRead }) + readNetworkSourceStateFn = func() (*netmon.State, error) { return nil, errors.New("read failed") } + mon := netmon.NewStatic() + old := mon.InterfaceState() + next := *old + networkChangeValidInterfacesFn = func() map[string]struct{} { return map[string]struct{}{} } + calls := 0 + networkChangeIgnoredInterceptFn = func(*prog, *netmon.ChangeDelta, time.Time) { calls++ } + p := &prog{cfg: &ctrld.Config{}, dnsInterceptState: &interceptStateStub{}} + p.handleNetworkChange(&netmon.ChangeDelta{Monitor: mon, Old: old, New: &next}, false) + if calls != 1 || p.networkSourceState != nil || !p.networkSourceReadFailed { + t.Fatal("source-read failure skipped reconciliation or treated unknown data as current") + } +} + +func TestReadNetworkSourceState(t *testing.T) { + state, err := readNetworkSourceState() + if err != nil { + t.Fatal(err) + } + for name, iface := range state.Interface { + if iface.Interface == nil || iface.Name != name { + t.Fatal("interface snapshots share the wrong object") + } + for _, prefix := range state.InterfaceIPs[name] { + if !prefix.IsValid() { + t.Fatalf("invalid source prefix on %s", name) + } + } + } +} diff --git a/cmd/cli/network_transition_review_test.go b/cmd/cli/network_transition_review_test.go new file mode 100644 index 0000000..5d91488 --- /dev/null +++ b/cmd/cli/network_transition_review_test.go @@ -0,0 +1,128 @@ +package cli + +import ( + "net" + "sync/atomic" + "testing" + "time" + + ctrld "github.com/Control-D-Inc/ctrld" + "tailscale.com/net/netmon" +) + +func TestNetmonMinorAndWakeSnapshotReconciliation(t *testing.T) { + for _, timeJump := range []bool{false, true} { + t.Run(map[bool]string{false: "minor", true: "time_jump"}[timeJump], func(t *testing.T) { + sourceTestGlobals(t) + mon := netmon.NewStatic() + // NewStatic has no monitoring resources to close. + old := mon.InterfaceState() + next := *old + // An identical snapshot is non-major; NewStatic has no OS monitor for IsMajorChangeFrom. + major := false + networkChangeValidInterfacesFn = func() map[string]struct{} { return map[string]struct{}{} } + calls := 0 + networkChangeIgnoredInterceptFn = func(*prog, *netmon.ChangeDelta, time.Time) { calls++ } + p := &prog{cfg: &ctrld.Config{}, dnsInterceptState: &interceptStateStub{}} + delta := &netmon.ChangeDelta{Monitor: mon, Old: old, New: &next, TimeJumped: timeJump} + //lint:ignore SA1019 This fixture reproduces netmon v1.74.0's legacy wake flag. + delta.Major = timeJump + p.handleNetworkChange(delta, major) + if calls != 1 { + t.Fatalf("real netmon non-major delta discarded: ignored-path calls=%d transition ID=%d", calls, p.networkTransitionGen.Load()) + } + }) + } +} + +func TestIgnoredDeltaPreservesAcceptedRecovery(t *testing.T) { + sourceTestGlobals(t) + dnsIntercept = false + oldRecovery := handleRecoveryForTransitionFn + t.Cleanup(func() { handleRecoveryForTransitionFn = oldRecovery }) + calls := make(chan uint64, 2) + handleRecoveryForTransitionFn = func(_ *prog, _ RecoveryReason, id uint64) { calls <- id } + networkChangeReconcileFn = (*prog).reconcileNetworkChange // Real accepted-path wiring, with recovery stubbed below. + p := &prog{cfg: &ctrld.Config{}} + a := sourceTestState("en0", true, "192.0.2.20/24") + p.handleNetworkChange(&netmon.ChangeDelta{Old: sourceTestState("en0", true, "192.0.2.10/24"), New: a}, true) + b := sourceTestState("en0", true, "192.0.2.20/24") + p.handleNetworkChange(&netmon.ChangeDelta{Old: a, New: b}, false) + select { + case <-calls: + case <-time.After(1200 * time.Millisecond): + t.Fatal("ignored delta canceled accepted recovery; no replacement was scheduled") + } +} + +func TestIgnoredDeltaPreservesValidPendingSource(t *testing.T) { + sourceTestGlobals(t) + dnsIntercept = false + ctrld.SetDefaultLocalIPv4(nil) + ctrld.SetDefaultLocalIPv6(nil) + entered, release, done := make(chan struct{}), make(chan struct{}), make(chan struct{}) + networkChangeDefaultRouteIPFn = func() string { close(entered); <-release; return "192.0.2.20" } + reconciled := make(chan uint64, 2) + networkChangeReconcileFn = func(_ *prog, id uint64) { reconciled <- id } + p := &prog{cfg: &ctrld.Config{}} + a := sourceTestState("en0", true, "192.0.2.20/24") + go func() { + defer close(done) + p.handleNetworkChange(&netmon.ChangeDelta{Old: sourceTestState("en0", true, "192.0.2.10/24"), New: a}, true) + }() + <-entered + p.handleNetworkChange(&netmon.ChangeDelta{Old: a, New: sourceTestState("en0", true, "192.0.2.20/24")}, false) + close(release) + <-done + if !ctrld.GetDefaultLocalIPv4().Equal(net.ParseIP("192.0.2.20")) || len(reconciled) != 1 { + t.Fatalf("valid source discarded after ignored delta: source=%v reconciliations=%d", ctrld.GetDefaultLocalIPv4(), len(reconciled)) + } +} + +func TestNewAcceptedDeltaSupersedesPendingAcceptedSource(t *testing.T) { + sourceTestGlobals(t) + dnsIntercept = false + entered, release, done := make(chan struct{}), make(chan struct{}), make(chan struct{}) + var lookups atomic.Int32 + networkChangeDefaultRouteIPFn = func() string { + if lookups.Add(1) == 1 { + close(entered) + <-release + return "192.0.2.20" + } + return "" + } + reconciled := make(chan uint64, 2) + networkChangeReconcileFn = func(_ *prog, id uint64) { reconciled <- id } + p := &prog{cfg: &ctrld.Config{}} + a := sourceTestState("en0", true, "192.0.2.20/24") + go func() { + defer close(done) + p.handleNetworkChange(&netmon.ChangeDelta{Old: sourceTestState("en0", true, "192.0.2.10/24"), New: a}, true) + }() + <-entered + p.handleNetworkChange(&netmon.ChangeDelta{Old: a, New: sourceTestState("en0", true, "192.0.2.30/24")}, true) + close(release) + <-done + if !ctrld.GetDefaultLocalIPv4().Equal(net.ParseIP("192.0.2.30")) || len(reconciled) != 1 || <-reconciled != 2 { + t.Fatal("older accepted delta replaced the newer source or recovery") + } +} + +func TestSourceCommitUsesCurrentMajorEpoch(t *testing.T) { + mon := netmon.NewStatic() + latest := mon.InterfaceState() + minor := *latest + p := &prog{networkSourceState: &minor, networkSourceEpoch: latest} + delta := &netmon.ChangeDelta{Monitor: mon, Old: latest, New: &minor} + if p.sourceCommitState(delta) != &minor { + t.Fatal("discarded the observed minor update within the current epoch") + } + p.networkSourceEpoch = &minor + if p.sourceCommitState(delta) != latest { + t.Fatal("did not refresh from a newer major epoch") + } + if networkSnapshotCurrent(&netmon.ChangeDelta{Monitor: mon, Old: &minor, New: &minor}, false, latest) { + t.Fatal("accepted a minor callback from an obsolete major epoch") + } +} diff --git a/cmd/cli/network_transition_test.go b/cmd/cli/network_transition_test.go new file mode 100644 index 0000000..5db1779 --- /dev/null +++ b/cmd/cli/network_transition_test.go @@ -0,0 +1,308 @@ +package cli + +import ( + "context" + "encoding/json" + "errors" + "net" + "net/netip" + "strings" + "sync" + "testing" + "time" + + "github.com/rs/zerolog" + "github.com/Control-D-Inc/ctrld" + "tailscale.com/net/netmon" +) + +func sourceTestState(route string, up bool, ips ...string) *netmon.State { + flags := net.Flags(0) + if up { + flags = net.FlagUp + } + s := &netmon.State{ + DefaultRouteInterface: route, + Interface: map[string]netmon.Interface{"en0": {Interface: &net.Interface{Name: "en0", Flags: flags}}}, + InterfaceIPs: map[string][]netip.Prefix{}, + } + for _, ip := range ips { + s.InterfaceIPs["en0"] = append(s.InterfaceIPs["en0"], netip.MustParsePrefix(ip)) + } + return s +} + +func captureTransitionLogs(t *testing.T) *syncBuffer { + t.Helper() + old := mainLog.Load() + buf := &syncBuffer{} + logger := zerolog.New(buf) + mainLog.Store(&logger) + t.Cleanup(func() { mainLog.Store(old) }) + return buf +} + +func sourceTestGlobals(t *testing.T) { + t.Helper() + v4, v6 := ctrld.GetDefaultLocalIPv4(), ctrld.GetDefaultLocalIPv6() + valid, route := networkChangeValidInterfacesFn, networkChangeDefaultRouteIPFn + reconcile, ignored := networkChangeReconcileFn, networkChangeIgnoredInterceptFn + intercept := dnsIntercept + dnsIntercept = true + networkChangeValidInterfacesFn = func() map[string]struct{} { return map[string]struct{}{"en0": {}} } + networkChangeDefaultRouteIPFn = func() string { return "" } + t.Cleanup(func() { + ctrld.SetDefaultLocalIPv4(v4) + ctrld.SetDefaultLocalIPv6(v6) + networkChangeValidInterfacesFn, networkChangeDefaultRouteIPFn = valid, route + networkChangeReconcileFn, networkChangeIgnoredInterceptFn = reconcile, ignored + dnsIntercept = intercept + }) +} + +func TestNetworkChangeSourceLifecycle(t *testing.T) { + const v4, v6 = "192.0.2.10", "2001:db8::10" + for _, tt := range []struct { + name string + newState *netmon.State + preferred string + want4, want6 string + outcome, reason4, reason6 string + }{ + {"dhcp_lease_replacement_keeps_ipv6", sourceTestState("en0", true, "192.0.2.20/24", v6+"/64"), "", "192.0.2.20", v6, "accepted", "address_removed", ""}, + {"dhcp_lease_replacement", sourceTestState("en0", true, "192.0.2.20/24"), "", "192.0.2.20", "", "accepted", "address_removed", "address_removed"}, + {"accepted_ipv4_loss", sourceTestState("en0", true, v6+"/64"), "", "", v6, "accepted", "address_removed", ""}, + {"accepted_ipv6_loss", sourceTestState("en0", true, v4+"/24"), v4, v4, "", "accepted", "", "address_removed"}, + {"stale_route_discovery", sourceTestState("en0", true, v6+"/64"), v4, "", v6, "accepted", "address_removed", ""}, + {"ignored_down_retains_addresses", sourceTestState("en0", false, v4+"/24", v6+"/64"), "", "", "", "ignored", "interface_down", "interface_down"}, + {"ignored_address_loss", sourceTestState("en0", true), "", "", "", "ignored", "address_removed", "address_removed"}, + {"skipped_no_active_interface", sourceTestState("", false, v4+"/24", v6+"/64"), "", "", "", "no_active_interface", "interface_down", "interface_down"}, + {"ignored_valid_sources", sourceTestState("en0", true, v4+"/24", v6+"/64"), "", v4, v6, "ignored", "", ""}, + {"accepted_valid_preferred_source", sourceTestState("en0", true, "192.0.2.20/24", v4+"/24", v6+"/64"), v4, v4, v6, "accepted", "", ""}, + } { + t.Run(tt.name, func(t *testing.T) { + sourceTestGlobals(t) + logs := captureTransitionLogs(t) + ctrld.SetDefaultLocalIPv4(net.ParseIP(v4)) + ctrld.SetDefaultLocalIPv6(net.ParseIP(v6)) + networkChangeDefaultRouteIPFn = func() string { return tt.preferred } + p := &prog{cfg: &ctrld.Config{}, dnsInterceptState: &interceptStateStub{}} + accepted, ignored := 0, 0 + assertSources := func() { + t.Helper() + if got := ctrld.GetDefaultLocalIPv4(); !got.Equal(net.ParseIP(tt.want4)) { + t.Errorf("IPv4 = %v, want %s", got, tt.want4) + } + if got := ctrld.GetDefaultLocalIPv6(); !got.Equal(net.ParseIP(tt.want6)) { + t.Errorf("IPv6 = %v, want %s", got, tt.want6) + } + } + // These seams precede all DNS/PF writes and recovery timers, on every OS. + networkChangeReconcileFn = func(_ *prog, id uint64) { + accepted++ + if id != 1 { + t.Errorf("transition ID = %d", id) + } + assertSources() + } + networkChangeIgnoredInterceptFn = func(_ *prog, _ *netmon.ChangeDelta, _ time.Time) { ignored++; assertSources() } + p.handleNetworkChange(&netmon.ChangeDelta{Old: sourceTestState("en0", true, v4+"/24", v6+"/64"), New: tt.newState}, true) + assertSources() + wantAccepted, wantIgnored := 0, 0 + if tt.outcome == "accepted" { + wantAccepted = 1 + } + if tt.outcome == "ignored" { + wantIgnored = 1 + } + if accepted != wantAccepted || ignored != wantIgnored { + t.Fatalf("mutation boundaries: accepted=%d ignored=%d; want %d/%d", accepted, ignored, wantAccepted, wantIgnored) + } + var transition, warning map[string]any + for _, line := range strings.Split(strings.TrimSpace(logs.String()), "\n") { + var event map[string]any + if err := json.Unmarshal([]byte(line), &event); err != nil { + t.Fatal(err) + } + switch event["message"] { + case "Network transition": + transition = event + case "Removed stale resolver source": + warning = event + } + } + if transition["outcome"] != tt.outcome || transition["transition_id"] != float64(1) { + t.Fatalf("transition: %v", transition) + } + wantWarning := tt.reason4 != "" || tt.reason6 != "" + if (warning != nil) != wantWarning { + t.Fatalf("warning = %v, want warning %v", warning, wantWarning) + } + if wantWarning && (warning["ipv4_clear_reason"] != tt.reason4 || warning["ipv6_clear_reason"] != tt.reason6 || warning["level"] != "warn") { + t.Fatalf("warning reasons/level: %v", warning) + } + }) + } +} + +func TestNetworkChangeDoesNotRestoreDownDefaultRouteSources(t *testing.T) { + sourceTestGlobals(t) + const v4, v6 = "192.0.2.10", "2001:db8::10" + ctrld.SetDefaultLocalIPv4(net.ParseIP(v4)) + ctrld.SetDefaultLocalIPv6(net.ParseIP(v6)) + networkChangeDefaultRouteIPFn = func() string { return v4 } + old := sourceTestState("en1", true, "2001:db8:1::10/64") + old.Interface["en1"] = netmon.Interface{Interface: &net.Interface{Name: "en1", Flags: net.FlagUp}} + old.InterfaceIPs["en1"] = []netip.Prefix{netip.MustParsePrefix(v4 + "/24"), netip.MustParsePrefix(v6 + "/64")} + newState := sourceTestState("en1", true, "2001:db8:1::20/64") + newState.Interface["en1"] = netmon.Interface{Interface: &net.Interface{Name: "en1"}} + newState.InterfaceIPs["en1"] = old.InterfaceIPs["en1"] + called := 0 + networkChangeReconcileFn = func(_ *prog, _ uint64) { + called++ + if ctrld.GetDefaultLocalIPv4() != nil || ctrld.GetDefaultLocalIPv6() != nil { + t.Error("down-interface sources reintroduced before recovery") + } + } + networkChangeIgnoredInterceptFn = func(_ *prog, _ *netmon.ChangeDelta, _ time.Time) { t.Fatal("accepted delta was ignored") } + (&prog{cfg: &ctrld.Config{}, dnsInterceptState: &interceptStateStub{}}).handleNetworkChange(&netmon.ChangeDelta{Old: old, New: newState}, true) + if called != 1 { + t.Fatalf("reconciliation called %d times", called) + } +} + +func TestNetworkChangeSourceMovedToUpInterface(t *testing.T) { + sourceTestGlobals(t) + ctrld.SetDefaultLocalIPv4(net.ParseIP("192.0.2.10")) + ctrld.SetDefaultLocalIPv6(nil) + state := sourceTestState("en0", false, "192.0.2.10/24") + state.Interface["utun0"] = netmon.Interface{Interface: &net.Interface{Name: "utun0", Flags: net.FlagUp}} + state.InterfaceIPs["utun0"] = state.InterfaceIPs["en0"] + called := 0 + networkChangeReconcileFn = func(_ *prog, _ uint64) { t.Fatal("unexpected recovery") } + networkChangeIgnoredInterceptFn = func(_ *prog, _ *netmon.ChangeDelta, _ time.Time) { called++ } + (&prog{cfg: &ctrld.Config{}, dnsInterceptState: &interceptStateStub{}}).handleNetworkChange(&netmon.ChangeDelta{Old: state, New: state}, false) + if called != 1 || !ctrld.GetDefaultLocalIPv4().Equal(net.ParseIP("192.0.2.10")) { + t.Fatal("valid source on another up interface was cleared") + } +} + +func TestRecoveryTransitionDiagnostics(t *testing.T) { + logs := captureTransitionLogs(t) + original, intercept := recoveryResetDNSFn, dnsIntercept + dnsIntercept = false + t.Cleanup(func() { recoveryResetDNSFn = original; dnsIntercept = intercept }) + for _, superseded := range []bool{false, true} { + p := &prog{cfg: &ctrld.Config{}} + calls := 0 + recoveryResetDNSFn = func(p *prog, start, restore bool) { + calls++ + if start || restore { + t.Error("changed reset flags") + } + if superseded { + p.beginRecovery(RecoveryReasonNetworkChange) + } else { + p.recoveryCancel() + } + } + p.networkAcceptedGen.Store(42) + p.handleRecoveryForTransition(RecoveryReasonNetworkChange, 42) + if calls != 1 { + t.Fatalf("resetDNS seam called %d times", calls) + } + if p.recoveryCancel != nil { + p.recoveryCancel() + } + } + for _, expected := range []string{`"transition_id":42`, `"recovery_generation":1`, `"message":"Recovery begin"`, `"outcome":"canceled"`, `"outcome":"superseded"`, `"message":"Recovery end"`} { + if !strings.Contains(logs.String(), expected) { + t.Errorf("missing %s in %s", expected, logs.String()) + } + } +} + +func TestRecoveryDebouncePreservesTransitionID(t *testing.T) { + original := handleRecoveryForTransitionFn + t.Cleanup(func() { handleRecoveryForTransitionFn = original }) + ids := make(chan uint64, 2) + handleRecoveryForTransitionFn = func(_ *prog, reason RecoveryReason, id uint64) { + if reason != RecoveryReasonNetworkChange { + t.Error("wrong recovery reason") + } + ids <- id + } + p := &prog{cfg: &ctrld.Config{}} + p.networkAcceptedGen.Store(11) + p.debounceRecovery(11) + p.networkAcceptedGen.Store(12) + p.debounceRecovery(12) + p.debounceRecovery(11) // a delayed older callback must not replace the new timer + select { + case id := <-ids: + if id != 12 { + t.Fatalf("recovery transition = %d, want 12", id) + } + case <-time.After(3 * time.Second): + t.Fatal("debounced recovery did not run") + } +} + +func TestTransitionWarningsSurviveDebugRotation(t *testing.T) { + sourceTestGlobals(t) + old := mainLog.Load() + debug, warnings := newLogWriterWithSize(2048), newSmallLogWriter() + // Runtime summaries occur after initialization, outside the preserved prefix. + debug.Write([]byte(logWriterInitEndMarker)) + logger := zerolog.New(zerolog.MultiLevelWriter(debug, &zerolog.FilteredLevelWriter{Writer: zerolog.LevelWriterAdapter{Writer: warnings}, Level: zerolog.WarnLevel})) + mainLog.Store(&logger) + t.Cleanup(func() { mainLog.Store(old) }) + ctrld.SetDefaultLocalIPv4(net.ParseIP("192.0.2.10")) + ctrld.SetDefaultLocalIPv6(nil) + state := sourceTestState("en0", false, "192.0.2.10/24") + for range 100 { + validateDefaultLocalIPsFromDelta(state, 7) + } + d := &recoveryDiagnostic{transitionID: 7, generation: 9, started: time.Now()} + var wg sync.WaitGroup + for range 100 { + wg.Add(1) + go func() { defer wg.Done(); d.failure(errors.New("private.customer.example secret payload")) }() + } + wg.Wait() + d.end("completed") + (&recoveryDiagnostic{transitionID: 8, generation: 10, started: time.Now()}).end("completed") + for range 100 { + logger.Debug().Msg(strings.Repeat("noise", 100)) + } + text := warnings.buf.String() + for _, msg := range []string{"Removed stale resolver source", "Recovery waiting for upstream; first failure"} { + if strings.Count(text, msg) != 1 { + t.Fatalf("expected one retained %q: %s", msg, text) + } + if strings.Contains(debug.buf.String(), msg) { + t.Fatalf("debug buffer did not rotate %q", msg) + } + } + if strings.Count(text, "Recovery end") != 1 || !strings.Contains(text, `"outcome":"completed"`) { + t.Fatal("recovery completion must survive rotation beside its retained failure") + } + if strings.Contains(text, "private.customer") || strings.Contains(text, "secret payload") { + t.Fatalf("raw payload or normal debug elevated: %s", text) + } +} + +func TestRecoveryWaitLogsFirstFailure(t *testing.T) { + logs := captureTransitionLogs(t) + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + d := &recoveryDiagnostic{transitionID: 8, generation: 2, started: time.Now()} + // Invalid resolver type fails construction before any network access. + _, err := (&prog{}).waitForUpstreamRecovery(ctx, map[string]*ctrld.UpstreamConfig{"invalid": {Type: "invalid"}}, d) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("wait result = %v", err) + } + if strings.Count(logs.String(), "Recovery waiting for upstream; first failure") != 1 { + t.Fatalf("missing first-failure summary: %s", logs.String()) + } +} diff --git a/cmd/cli/pf_probe.go b/cmd/cli/pf_probe.go new file mode 100644 index 0000000..83d93a7 --- /dev/null +++ b/cmd/cli/pf_probe.go @@ -0,0 +1,233 @@ +package cli + +import ( + "bufio" + "context" + "encoding/hex" + "errors" + "fmt" + "io" + "net" + "net/netip" + "os/exec" + "strings" + "sync" + "syscall" + "time" + + "tailscale.com/net/tsaddr" +) + +type pfProbeResult uint8 + +const ( + pfProbeIndeterminate pfProbeResult = iota + pfProbeIntercepted + pfProbeNotIntercepted +) + +func (r pfProbeResult) String() string { + switch r { + case pfProbeIntercepted: + return "intercepted" + case pfProbeNotIntercepted: + return "not_intercepted" + default: + return "indeterminate" + } +} + +type pfProbeObservation struct { + result pfProbeResult + stage string + code string + probeID string + recoveryGeneration uint64 + target string +} + +// Conditions exclude attempt IDs and generations so repeated failures stay bounded. +type pfProbeCondition struct { + result pfProbeResult + stage, code, target string +} + +type pfProbeLogState struct { + mu sync.Mutex + previous pfProbeCondition + initialized bool + repeats uint64 +} + +// change returns whether to retain a transition and the preceding repeat count. +// A restored receipt is retained too, so an old warning does not imply an outage. +func (s *pfProbeLogState) change(o pfProbeObservation) (retain bool, repeats uint64) { + s.mu.Lock() + defer s.mu.Unlock() + condition := pfProbeCondition{o.result, o.stage, o.code, o.target} + if s.initialized && condition == s.previous { + s.repeats++ + return false, s.repeats + } + retain = o.result != pfProbeIntercepted || (s.initialized && s.previous.result != pfProbeIntercepted) + repeats = s.repeats + s.previous, s.initialized, s.repeats = condition, true, 0 + return retain, repeats +} + +// OsResolverNameservers groups LAN entries before public entries. Preserve a usable +// first target, including public DNS. Otherwise scan only the LAN group. Stop at +// the public boundary, not at a particular IP: the public group can be synthetic. +func pfProbeTarget(servers []string) string { + for i, server := range servers { + host, _, err := net.SplitHostPort(server) + if err != nil { + return "" + } + ip, err := netip.ParseAddr(host) + if err != nil { + return "" + } + lan := ip.IsPrivate() || ip.IsLoopback() || ip.IsLinkLocalUnicast() || tsaddr.CGNATRange().Contains(ip) + usable := ip.Is4() && !ip.IsLoopback() && !ip.IsUnspecified() && !ip.IsMulticast() + if usable && (i == 0 || lan) { + return host + } + if !lan { + return "" + } + } + return "" +} + +// Keep the helper protocol bounded and free of resolver IDs, queries and raw OS errors. +func pfProbeErrorCode(err error) string { + switch { + case errors.Is(err, syscall.ENETUNREACH), errors.Is(err, syscall.EHOSTUNREACH): + return "unreachable" + case errors.Is(err, syscall.EADDRNOTAVAIL): + return "source_unavailable" + case errors.Is(err, syscall.EACCES), errors.Is(err, syscall.EPERM): + return "permission" + default: + var ne net.Error + if errors.As(err, &ne) && ne.Timeout() { + return "timeout" + } + return "io" + } +} + +func sendPFProbe(host, hexPacket string, dial func(string, string, time.Duration) (net.Conn, error), status io.Writer) error { + fail := func(stage string, err error) error { + _, _ = fmt.Fprintf(status, "%s:%s\n", stage, pfProbeErrorCode(err)) + return err + } + packet, err := hex.DecodeString(hexPacket) + if err != nil { + _, _ = io.WriteString(status, "decode:invalid_argument\n") + return err + } + conn, err := dial("udp4", net.JoinHostPort(host, "53"), time.Second) + if err != nil { + return fail("dial", err) + } + defer conn.Close() + if err := conn.SetDeadline(time.Now().Add(time.Second)); err != nil { + return fail("deadline", err) + } + n, err := conn.Write(packet) + if err == nil && n != len(packet) { + err = io.ErrShortWrite + } + if err != nil { + return fail("write", err) + } + if _, err := io.WriteString(status, "sent\n"); err != nil { + return err + } + // Keep the socket alive for the local handler's response. A read failure does + // not undo the successful send acknowledged above. + _, _ = conn.Read(make([]byte, 512)) + return nil +} + +func parsePFProbeStatus(status string) pfProbeObservation { + if status == "sent\n" { + return pfProbeObservation{stage: "sent"} + } + stage, code, ok := strings.Cut(strings.TrimSuffix(status, "\n"), ":") + if ok && (stage == "decode" || stage == "dial" || stage == "deadline" || stage == "write") { + switch code { + case "unreachable", "source_unavailable", "permission", "timeout", "io", "invalid_argument": + return pfProbeObservation{stage: stage, code: code} + } + } + return pfProbeObservation{stage: "helper", code: "invalid_status"} +} + +func pfProbeContextCode(ctx context.Context) string { + if errors.Is(ctx.Err(), context.DeadlineExceeded) { + return "timeout" + } + return "canceled" +} + +// Receipt by the local handler is conclusive. A missing receipt is a failed +// interception test only after the child confirms a successful packet send. +func runPFProbe(ctx context.Context, cmd *exec.Cmd, received <-chan struct{}, deliveryTimeout time.Duration) pfProbeObservation { + stdout, err := cmd.StdoutPipe() + if err != nil { + return pfProbeObservation{stage: "start", code: pfProbeErrorCode(err)} + } + if err := cmd.Start(); err != nil { + _ = stdout.Close() + return pfProbeObservation{stage: "start", code: pfProbeErrorCode(err)} + } + statusCh := make(chan pfProbeObservation, 1) + done := make(chan struct{}) + go func() { + line, _ := bufio.NewReader(io.LimitReader(stdout, 128)).ReadString('\n') + statusCh <- parsePFProbeStatus(line) + _ = cmd.Wait() + close(done) + }() + defer func() { + _ = cmd.Process.Kill() + <-done + }() + + var sent pfProbeObservation + select { + case <-received: + return pfProbeObservation{result: pfProbeIntercepted, stage: "received"} + case <-ctx.Done(): + return pfProbeObservation{stage: "helper", code: pfProbeContextCode(ctx)} + case sent = <-statusCh: + } + if sent.stage != "sent" { + select { + case <-received: + return pfProbeObservation{result: pfProbeIntercepted, stage: "received"} + default: + return sent + } + } + + timer := time.NewTimer(deliveryTimeout) + defer timer.Stop() + select { + case <-received: + return pfProbeObservation{result: pfProbeIntercepted, stage: "received"} + case <-ctx.Done(): + return pfProbeObservation{stage: "delivery", code: pfProbeContextCode(ctx)} + case <-timer.C: + // Prefer receipt when notification and deadline became ready together. + select { + case <-received: + return pfProbeObservation{result: pfProbeIntercepted, stage: "received"} + default: + return pfProbeObservation{result: pfProbeNotIntercepted, stage: "delivery", code: "timeout"} + } + } +} diff --git a/cmd/cli/pf_probe_darwin_test.go b/cmd/cli/pf_probe_darwin_test.go new file mode 100644 index 0000000..87f4d72 --- /dev/null +++ b/cmd/cli/pf_probe_darwin_test.go @@ -0,0 +1,213 @@ +//go:build darwin + +package cli + +import ( + "context" + "encoding/json" + "os" + "os/exec" + "reflect" + "strings" + "testing" + "time" +) + +func TestPFProbeDefaultCommand(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + cmd := newPFProbeCommand(ctx, "192.0.2.1", "0102") + if !reflect.DeepEqual(cmd.Args, []string{os.Args[0], "pf-probe-send", "192.0.2.1", "0102"}) { + t.Fatalf("args=%q", cmd.Args) + } + if cmd.SysProcAttr == nil || cmd.SysProcAttr.Credential == nil { + t.Fatal("missing credentials") + } + cred := cmd.SysProcAttr.Credential + if cred.Uid != 0 || cred.Gid != 0 || cred.NoSetGroups || len(cred.Groups) != 0 { + t.Fatalf("unexpected credentials: %+v", cred) + } + if cmd.Cancel == nil { + t.Fatal("command is not context-bound") + } + // Do not start this command: credential changes require native privileged QA. +} + +func TestPFProbeMonitorIndeterminateContinues(t *testing.T) { + oldProbe, oldReload, oldDelays := probePFInterceptFn, forceReloadPFInterceptFn, pfInterceptMonitorDelays + t.Cleanup(func() { + probePFInterceptFn, forceReloadPFInterceptFn, pfInterceptMonitorDelays = oldProbe, oldReload, oldDelays + }) + pfInterceptMonitorDelays = []time.Duration{0, 0, 0} + probes, reloads := 0, 0 + probePFInterceptFn = func(*prog) pfProbeObservation { + probes++ + if probes == 2 { + return pfProbeObservation{result: pfProbeIntercepted} + } + return pfProbeObservation{stage: "dial", code: "unreachable"} + } + forceReloadPFInterceptFn = func(*prog) bool { reloads++; return true } + p := &prog{dnsInterceptState: &pfState{}} + p.pfInterceptMonitor() + if probes != 3 || reloads != 0 || p.pfMonitorRunning.Load() { + t.Fatalf("probes=%d reloads=%d owner=%v", probes, reloads, p.pfMonitorRunning.Load()) + } +} + +func TestPFProbeRepairCorrelation(t *testing.T) { + for _, caller := range []string{"post_stabilization", "post_wake", "watchdog", "monitor"} { + for _, performed := range []bool{false, true} { + t.Run(caller+"/"+map[bool]string{false: "not_run", true: "completed"}[performed], func(t *testing.T) { + oldProbe, oldReload, oldDelays := probePFInterceptFn, forceReloadPFInterceptFn, pfInterceptMonitorDelays + t.Cleanup(func() { + probePFInterceptFn, forceReloadPFInterceptFn, pfInterceptMonitorDelays = oldProbe, oldReload, oldDelays + }) + stubWakeProbeSchedule(t, 1) + pfInterceptMonitorDelays = []time.Duration{0} + p := &prog{dnsInterceptState: &pfState{}, stopCh: make(chan struct{})} + probes, reloads := 0, 0 + probePFInterceptFn = func(*prog) pfProbeObservation { + probes++ + if probes == 1 { + return pfProbeObservation{result: pfProbeNotIntercepted, probeID: "cause-probe", recoveryGeneration: 7, stage: "delivery", code: "timeout", target: "192.0.2.1:53"} + } + return pfProbeObservation{result: pfProbeIndeterminate, probeID: "confirmation-probe", recoveryGeneration: 8, stage: "dial", code: "unreachable"} + } + forceReloadPFInterceptFn = func(*prog) bool { reloads++; p.recoveryGen.Store(99); return performed } + before := len(logOutput.String()) + switch caller { + case "post_stabilization": + p.verifyInterceptAfterStabilization() + case "post_wake": + p.runWakeProbeAttempt(0, 0) + case "watchdog": + p.checkPFWatchdogProbe(pfAnchorCheckIntact) + case "monitor": + p.pfInterceptMonitor() + } + expected := "not_run" + if performed { + expected = "reload_completed" + } + found, confirmation := false, false + for _, line := range strings.Split(logOutput.String()[before:], "\n") { + var event map[string]interface{} + if json.Unmarshal([]byte(line), &event) != nil || event["message"] != "PF repair" { + continue + } + if event["probe_id"] != "cause-probe" || event["recovery_generation"] != float64(7) || event["caller"] != caller { + t.Fatalf("lost cause: %v", event) + } + if event["outcome"] == expected { + found = true + } + if event["outcome"] == "indeterminate" && event["confirmation_probe_id"] == "confirmation-probe" { + confirmation = true + } + } + if reloads != 1 || !found { + t.Fatalf("reloads=%d outcome found=%v", reloads, found) + } + if performed && caller != "watchdog" && !confirmation { + t.Fatal("missing confirmation outcome") + } + }) + } + } +} + +func TestPFProbeMissingTargetLogTransitions(t *testing.T) { + old := pfProbeNameservers + t.Cleanup(func() { pfProbeNameservers = old; pfProbeLogs = pfProbeLogState{} }) + pfProbeLogs = pfProbeLogState{} + first := "[fe80::1%en0]:53" + pfProbeNameservers = func() []string { return []string{first, "76.76.2.0:53"} } + p := &prog{dnsInterceptState: &pfState{}} + before := len(logOutput.String()) + p.probePFIntercept() + p.probePFIntercept() + first = "127.0.0.1:53" + p.probePFIntercept() + var levels []string + for _, line := range strings.Split(logOutput.String()[before:], "\n") { + var event map[string]interface{} + if json.Unmarshal([]byte(line), &event) != nil || event["message"] != "DNS intercept probe result" { + continue + } + levels = append(levels, event["level"].(string)) + if len(levels) == 3 && event["repeated_results"] != float64(1) { + t.Fatal("missing suppressed count") + } + } + if !reflect.DeepEqual(levels, []string{"warn", "debug", "warn"}) { + t.Fatalf("levels=%v", levels) + } +} + +func TestPFProbeUnsentDoesNotReload(t *testing.T) { + oldServers, oldCommand, oldReload := pfProbeNameservers, newPFProbeCommand, forceReloadPFInterceptFn + t.Cleanup(func() { + pfProbeNameservers, newPFProbeCommand, forceReloadPFInterceptFn = oldServers, oldCommand, oldReload + }) + pfProbeNameservers = func() []string { return []string{"192.0.2.1:53"} } + newPFProbeCommand = func(ctx context.Context, host, packet string) *exec.Cmd { return pfProbeTestCommand(ctx, "dial") } + reloads := 0 + forceReloadPFInterceptFn = func(*prog) bool { reloads++; return true } + p := &prog{dnsInterceptState: &pfState{}, stopCh: make(chan struct{})} + before := len(logOutput.String()) + if got := p.checkPFWatchdogProbe(pfAnchorCheckIntact); got != pfAnchorCheckDeferred { + t.Fatalf("watchdog result=%v", got) + } + p.verifyInterceptAfterStabilization() + stubWakeProbeSchedule(t, 1) + if restored, repaired := p.runWakeProbeAttempt(0, 0); restored || repaired { + t.Fatalf("unsent wake probe: restored=%v repaired=%v", restored, repaired) + } + if reloads != 0 { + t.Fatalf("unsent probes caused %d reloads", reloads) + } + logs := logOutput.String()[before:] + for _, field := range []string{`"probe_id":`, `"recovery_generation":`, `"resolver_target":"192.0.2.1:53"`, `"stage":"dial"`, `"error_code":"unreachable"`, `"outcome":"indeterminate"`, `"repair_eligible":false`} { + if !strings.Contains(logs, field) { + t.Errorf("missing %s in logs", field) + } + } + if strings.Contains(logs, "interception is translating") || strings.Contains(logs, "anchor was missing") { + t.Fatal("unknown probe claimed PF health or missing rules") + } +} + +func TestPFProbeWatchdogMeaningfulFailureRepairsWithoutMissingAnchorClaim(t *testing.T) { + _, reloads := stubStabilizationProbe(t, []bool{false}, true) + p := &prog{dnsInterceptState: &pfState{}} + if got := p.checkPFWatchdogProbe(pfAnchorCheckIntact); got != pfAnchorCheckDeferred { + t.Fatalf("probe repair must not be counted as missing-anchor restoration: %v", got) + } + if *reloads != 1 { + t.Fatalf("reloads=%d", *reloads) + } +} + +func TestPFProbeWatchdogSkipsIndeterminateAnchor(t *testing.T) { + probes, reloads := stubStabilizationProbe(t, nil, true) + p := &prog{dnsInterceptState: &pfState{}} + for _, r := range []pfAnchorCheckResult{pfAnchorCheckDeferred, pfAnchorCheckSkipped, pfAnchorCheckFailed, pfAnchorCheckRestored} { + if got := p.checkPFWatchdogProbe(r); got != r { + t.Fatalf("changed outcome %v -> %v", r, got) + } + } + if *probes != 0 || *reloads != 0 { + t.Fatal("probed without intact rule evidence") + } +} + +func TestPFProbeNoIPv4TargetIsIndeterminate(t *testing.T) { + old := pfProbeNameservers + t.Cleanup(func() { pfProbeNameservers = old }) + pfProbeNameservers = func() []string { return []string{"[fe80::1%en0]:53", "76.76.2.0:53"} } + p := &prog{dnsInterceptState: &pfState{}} + if got := p.probePFIntercept(); got.result != pfProbeIndeterminate { + t.Fatalf("got %v", got) + } +} diff --git a/cmd/cli/pf_probe_test.go b/cmd/cli/pf_probe_test.go new file mode 100644 index 0000000..6afae2a --- /dev/null +++ b/cmd/cli/pf_probe_test.go @@ -0,0 +1,292 @@ +package cli + +import ( + "bytes" + "context" + "errors" + "io" + "net" + "os" + "os/exec" + "strings" + "sync" + "syscall" + "testing" + "time" +) + +func TestPFProbeProcess(t *testing.T) { + mode := os.Getenv("CTRLD_PF_PROBE_TEST") + if mode == "" { + return + } + switch mode { + case "main_dispatch": + os.Args = []string{os.Args[0], "pf-probe-send", "127.0.0.1", "zz"} + Main() + os.Exit(99) // Production dispatch must exit on invalid input. + case "sent": + _, _ = os.Stdout.WriteString("sent\n") + case "dial": + _, _ = os.Stdout.WriteString("dial:unreachable\n") + os.Exit(1) + case "write": + _, _ = os.Stdout.WriteString("write:permission\n") + os.Exit(1) + case "malformed": + _, _ = os.Stdout.WriteString(strings.Repeat("private-payload", 1000)) + case "hang": + time.Sleep(time.Minute) + case "silent": + os.Exit(1) + } + os.Exit(0) +} + +func pfProbeTestCommand(ctx context.Context, mode string) *exec.Cmd { + cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestPFProbeProcess$") + cmd.Env = append(os.Environ(), "CTRLD_PF_PROBE_TEST="+mode) + return cmd +} + +func TestPFProbeProcessClassification(t *testing.T) { + for _, tc := range []struct { + mode string + want pfProbeResult + stage string + }{ + {"dial", pfProbeIndeterminate, "dial"}, + {"write", pfProbeIndeterminate, "write"}, + {"silent", pfProbeIndeterminate, "helper"}, + {"malformed", pfProbeIndeterminate, "helper"}, + {"sent", pfProbeNotIntercepted, "delivery"}, + } { + t.Run(tc.mode, func(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + got := runPFProbe(ctx, pfProbeTestCommand(ctx, tc.mode), make(chan struct{}), 10*time.Millisecond) + if got.result != tc.want || got.stage != tc.stage { + t.Fatalf("got %+v, want %v/%s", got, tc.want, tc.stage) + } + if strings.Contains(got.code, "private") { + t.Fatal("helper output escaped into diagnostic") + } + }) + } +} + +func TestPFProbeStartFailureAndCancellation(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + got := runPFProbe(ctx, exec.Command(t.TempDir()+"/missing-helper"), make(chan struct{}), time.Second) + if got.result != pfProbeIndeterminate || got.stage != "start" { + t.Fatalf("start failure: %+v", got) + } + ctx, cancel = context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + got = runPFProbe(ctx, pfProbeTestCommand(ctx, "hang"), make(chan struct{}), time.Second) + if got.result != pfProbeIndeterminate { + t.Fatalf("canceled helper must not request repair: %+v", got) + } +} + +func TestPFProbeReceiptWins(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + received := make(chan struct{}) + close(received) + got := runPFProbe(ctx, pfProbeTestCommand(ctx, "sent"), received, time.Millisecond) + if got.result != pfProbeIntercepted { + t.Fatalf("local receipt is conclusive: %+v", got) + } +} + +type pfProbeErrorConn struct { + net.Conn + writeErr error + short bool +} + +func (c pfProbeErrorConn) SetDeadline(time.Time) error { return nil } +func (c pfProbeErrorConn) Close() error { return nil } +func (c pfProbeErrorConn) Write(b []byte) (int, error) { + if c.short { + return len(b) - 1, nil + } + return 0, c.writeErr +} + +func TestPFProbeSendFailures(t *testing.T) { + for _, tc := range []struct { + name, packet, want string + dial func(string, string, time.Duration) (net.Conn, error) + }{ + {"decode", "zz", "decode:invalid_argument\n", nil}, + {"dial", "0102", "dial:unreachable\n", func(string, string, time.Duration) (net.Conn, error) { + return nil, &net.OpError{Op: "dial", Err: syscall.ENETUNREACH} + }}, + {"write", "0102", "write:permission\n", func(string, string, time.Duration) (net.Conn, error) { + return pfProbeErrorConn{writeErr: syscall.EACCES}, nil + }}, + {"short", "0102", "write:io\n", func(string, string, time.Duration) (net.Conn, error) { return pfProbeErrorConn{short: true}, nil }}, + } { + t.Run(tc.name, func(t *testing.T) { + var out bytes.Buffer + err := sendPFProbe("192.0.2.1", tc.packet, tc.dial, &out) + if err == nil || out.String() != tc.want { + t.Fatalf("status=%q err=%v", out.String(), err) + } + if strings.Contains(out.String(), "sent") { + t.Fatal("failed write claimed success") + } + }) + } +} + +func TestPFProbeSendRealUDP(t *testing.T) { + listener, err := net.ListenPacket("udp4", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer listener.Close() + received := make(chan []byte, 1) + go func() { + b := make([]byte, 100) + n, addr, e := listener.ReadFrom(b) + if e != nil { + return + } + received <- append([]byte(nil), b[:n]...) + _, _ = listener.WriteTo([]byte{0}, addr) + }() + var out bytes.Buffer + err = sendPFProbe("127.0.0.1", "010203", func(network, address string, timeout time.Duration) (net.Conn, error) { + if network != "udp4" || address != "127.0.0.1:53" { + t.Errorf("unexpected target %s %s", network, address) + } + return net.DialTimeout(network, listener.LocalAddr().String(), timeout) + }, &out) + if err != nil || out.String() != "sent\n" { + t.Fatalf("status=%q error=%v", out.String(), err) + } + select { + case b := <-received: + if !bytes.Equal(b, []byte{1, 2, 3}) { + t.Fatalf("wrong packet %v", b) + } + case <-time.After(time.Second): + t.Fatal("helper reported sent without UDP receipt") + } +} + +func TestPFProbeTarget(t *testing.T) { + for _, first := range []string{"[fe80::1%en0]:53", "[2001:db8::1]:53", "127.0.0.53:53", "0.0.0.0:53", "224.0.0.1:53", "bad"} { + // Resolver construction can append this public fallback. No later entry + // may become a new probe recipient when the original first target fails. + if got := pfProbeTarget([]string{first, "76.76.2.0:53"}); got != "" { + t.Fatalf("first=%s selected new recipient=%s", first, got) + } + } + for _, first := range []string{"192.0.2.10", "76.76.2.0"} { + if got := pfProbeTarget([]string{first + ":53", "192.0.2.11:53"}); got != first { + t.Fatalf("changed original first target %s to %s", first, got) + } + } + if got := pfProbeTarget(nil); got != "" { + t.Fatalf("empty list selected %s", got) + } +} + +func TestPFProbeMainDispatch(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + out, err := pfProbeTestCommand(ctx, "main_dispatch").Output() + var exit *exec.ExitError + if !errors.As(err, &exit) || exit.ExitCode() != 1 || string(out) != "decode:invalid_argument\n" { + t.Fatalf("production Main dispatch: output=%q error=%v", out, err) + } +} + +func TestPFProbeTargetLANScan(t *testing.T) { + for _, tc := range []struct { + servers []string + want string + }{ + {[]string{"[fe80::1%en0]:53", "192.168.1.1:53", "76.76.2.0:53"}, "192.168.1.1"}, + {[]string{"127.0.0.1:53", "100.64.0.1:53", "76.76.2.0:53"}, "100.64.0.1"}, + {[]string{"[fd00::1]:53", "169.254.1.1:53", "76.76.2.0:53"}, "169.254.1.1"}, + {[]string{"[fe80::1%en0]:53", "76.76.2.0:53"}, ""}, + {[]string{"[fe80::1%en0]:53", "8.8.8.8:53", "192.168.1.1:53"}, ""}, + {[]string{"[2001:db8::1]:53", "192.168.1.1:53"}, ""}, + {[]string{"76.76.2.0:53"}, "76.76.2.0"}, + {[]string{"8.8.8.8:53", "192.168.1.1:53"}, "8.8.8.8"}, + } { + if got := pfProbeTarget(tc.servers); got != tc.want { + t.Errorf("%v: got %q want %q", tc.servers, got, tc.want) + } + } +} + +func TestPFProbeLogConcurrentRepeats(t *testing.T) { + var state pfProbeLogState + observed := pfProbeObservation{stage: "dial", code: "unreachable"} + state.change(observed) + var wg sync.WaitGroup + for range 20 { + wg.Add(1) + go func() { defer wg.Done(); state.change(observed) }() + } + wg.Wait() + observed.result, observed.stage, observed.code = pfProbeIntercepted, "received", "" + if retained, repeats := state.change(observed); !retained || repeats != 20 { + t.Fatalf("retained=%v repeats=%d", retained, repeats) + } +} + +func TestPFProbeLogConditionChanges(t *testing.T) { + var state pfProbeLogState + missing := pfProbeObservation{stage: "target", code: "unavailable", target: "[fe80::1%en0]:53"} + if retain, repeats := state.change(missing); !retain || repeats != 0 { + t.Fatal("first missing target not retained") + } + missing.probeID, missing.recoveryGeneration = "new-attempt", 2 + if retain, repeats := state.change(missing); retain || repeats != 1 { + t.Fatal("attempt IDs defeat dedupe") + } + for _, change := range []func(*pfProbeObservation){ + func(o *pfProbeObservation) { o.target = "127.0.0.1:53" }, + func(o *pfProbeObservation) { o.stage = "dial" }, + func(o *pfProbeObservation) { o.code = "unreachable" }, + func(o *pfProbeObservation) { o.result = pfProbeNotIntercepted }, + } { + change(&missing) + if retain, repeats := state.change(missing); !retain || repeats != 1 { + t.Fatalf("change not retained with prior repeats: %v %d", retain, repeats) + } + if retain, repeats := state.change(missing); retain || repeats != 1 { + t.Fatal("repeat not coalesced") + } + } + healthy := pfProbeObservation{result: pfProbeIntercepted, stage: "received", target: missing.target} + if retain, repeats := state.change(healthy); !retain || repeats != 1 { + t.Fatal("restoration not retained") + } + if retain, _ := state.change(healthy); retain { + t.Fatal("normal receipt retained") + } +} + +func TestPFProbeStatusIsBounded(t *testing.T) { + if got := parsePFProbeStatus("decode:invalid_argument\n"); got.stage != "decode" || got.code != "invalid_argument" || got.result != pfProbeIndeterminate { + t.Fatalf("invalid argument classification: %+v", got) + } + for _, s := range []string{"dial:secret-token\n", "secret:io\n", strings.Repeat("private", 1000)} { + got := parsePFProbeStatus(s) + if got.result != pfProbeIndeterminate || got.stage != "helper" || got.code != "invalid_status" { + t.Fatalf("untrusted status leaked: %+v", got) + } + } + if pfProbeErrorCode(errors.New("secret")) != "io" || pfProbeErrorCode(io.ErrShortWrite) != "io" { + t.Fatal("unbounded diagnostic") + } +} diff --git a/cmd/cli/prog.go b/cmd/cli/prog.go index 73f4819..60913dc 100644 --- a/cmd/cli/prog.go +++ b/cmd/cli/prog.go @@ -161,6 +161,14 @@ type prog struct { // only touches shared recovery state if it is still the newest (#597). recoveryGen atomic.Uint64 + // All deltas get an ID. Only accepted deltas replace recovery ownership. + networkTransitionGen atomic.Uint64 + networkAcceptedGen atomic.Uint64 + networkSourceMu sync.Mutex + networkSourceState *netmon.State + networkSourceEpoch *netmon.State + networkSourceReadFailed bool + // recoveryDebounceTimer coalesces rapid NetworkChange recovery triggers // into a single handleRecovery call. Only handleRecovery is debounced — // all other state updates (IP, pf anchor, VPN DNS) run immediately. diff --git a/docs/dns-intercept-mode.md b/docs/dns-intercept-mode.md index 508952e..e1724a0 100644 --- a/docs/dns-intercept-mode.md +++ b/docs/dns-intercept-mode.md @@ -312,12 +312,14 @@ Programs like Parallels Desktop reload `/etc/pf.conf` when creating/destroying v ctrld detects interface appearance/disappearance and spawns an async probe monitor: -1. **Probe mechanism:** A subprocess runs with GID=0 (wheel, not `_ctrld`) and sends a DNS query to the OS resolver. If pf interception is working, the query gets redirected to ctrld (127.0.0.1:53) and is detected in the DNS handler. If broken, it times out after 1s. +1. **Probe mechanism:** A subprocess runs with GID=0 (wheel, not `_ctrld`) and sends a DNS query to the OS resolver. If pf interception is working, the query gets redirected to ctrld (127.0.0.1:53) and is detected in the DNS handler. A confirmed send without local receipt within 1s gives `not_intercepted`. 2. **Backoff schedule:** Probes at 0, 0.5, 1, 2, 4 seconds (~8s window) to win the race against async pf reloads by the hypervisor. Only one monitor runs at a time (atomic singleton). -3. **Auto-heal:** On probe failure, `forceReloadPFMainRuleset()` dumps the running ruleset and pipes it back through `pfctl -f -`, resetting pf's translation engine. VPN-safe because it reassembles from the current running state. +3. **Auto-heal:** Only after `not_intercepted`, `forceReloadPFMainRuleset()` dumps the running ruleset and pipes it back through `pfctl -f -`, resetting pf's translation engine. VPN-safe because it reassembles from the current running state. 4. **Watchdog integration:** The 30s watchdog also runs the probe when rule text checks pass, as a safety net for unknown corruption causes. -This approach detects **actual broken DNS** rather than guessing from trigger events, making it robust against future unknown corruption scenarios. +The probe uses IPv4 only. Missing targets and helper failures give `indeterminate`, which does not cause a reload. +See [target selection and diagnostics](network-recovery-diagnostics.md#pf-probes-macos). +A successful probe proves local interception delivery, not remote DNS health. ### 5. Proactive DoH Connection Pool Reset diff --git a/docs/network-recovery-diagnostics.md b/docs/network-recovery-diagnostics.md new file mode 100644 index 0000000..e7c18c6 --- /dev/null +++ b/docs/network-recovery-diagnostics.md @@ -0,0 +1,88 @@ +# Network-recovery diagnostics + +Use continuous `ctrld log tail` capture during a reproduction. Internal logs rotate by size. +A later `log send` can contain recent debug events and older warnings. +Match timestamps and IDs. The last retained error does not necessarily describe the current network. +Issue #604 tracks upload-time device and version headers separately. + +## Transitions and recovery + +`Network transition` debug events include a `transition_id`, affected interface, default route, and stored source addresses before and after the callback. +The outcome identifies an accepted, ignored, superseded, or unusable change. +Newer accepted changes supersede pending recovery. Ignored changes do not cancel accepted recovery. +Source commits validate addresses against the current network state. +Netmon does not order minor callbacks within one major cache epoch. +For minor callbacks and late major callbacks in that epoch, ctrld reads current interface addresses and flags. +This read does not change host DNS. A failed read defers source writes, but reconciliation continues. + +`Removed stale resolver source` is a warning only when ctrld removes an invalid address. +The warning gives the reason and survives debug rotation. +`Recovery begin` and `Recovery end` connect the transition to a `recovery_generation`, reason, outcome, and elapsed time. +Only the first upstream failure in each recovery produces a bounded summary. +If a recovery encounters a failure, its final outcome also enters the warning stream. +Normal successful recovery stays at debug level. + +These summaries exclude upstream URLs and queried names. +Each PF probe records the recovery generation at its start. +A later recovery cannot receive credit for an earlier probe. + +### Cancellation event migration + +`Recovery end` with `outcome=canceled` replaces `Recovery canceled; DNS settings remain removed`. +Update support runbooks and log searches to use the structured event. +The cancellation event alone does not prove that interface DNS remains removed. + +## PF probes (macOS) + +`DNS intercept probe result` includes these fields: + +- `probe_id` and `recovery_generation` +- Numeric `resolver_target` and `resolver_family` +- `stage`, bounded `error_code`, and `outcome` +- `repair_eligible`, `repeated_results`, and elapsed milliseconds + +A changed unsuccessful condition produces a warning. The condition consists of the stage, error code, outcome, and target. +Equal results stay at debug level. The next changed condition includes the preceding repeat count. +The first restored receipt also produces a warning-stream event. This event prevents an older failure from appearing current. +Normal receipts and shutdown cancellation stay at debug level. +The warning buffer is finite. These events do not guarantee indefinite retention. + +The outcomes have these meanings: + +- `intercepted`: the registered query reached the local handler. This proves interception delivery, not remote DNS health. +- `not_intercepted`: the helper confirmed a successful UDP write, but the handler did not receive the query before the deadline. + This result permits bounded repair. It does not identify an external PF writer or explain the delivery failure. +- `indeterminate`: the probe has no suitable target, cannot send, receives an invalid helper acknowledgement, or stops before a conclusive result. + This result alone does not cause a PF reload or transport reset. The existing monitor or watchdog retries later. + Independent evidence of missing rules still permits ordinary repair. + +The literal stages are `target`, `start`, `decode`, `dial`, `deadline`, `write`, `sent`, `helper`, `delivery`, and `received`. +`sent` is the helper acknowledgement before the final delivery result. +The error codes are `unavailable`, `invalid_argument`, `unreachable`, `source_unavailable`, `permission`, `timeout`, `io`, `invalid_status`, and `canceled`. +Successful results have an empty error code. No raw subprocess output enters these fields. + +The sender runs outside the `_ctrld` group exemption. +ctrld blocks outgoing IPv6 DNS with its own PF rule and does not redirect it. +The selector preserves a usable first IPv4 target, including an original public resolver. +Otherwise, it selects the first usable IPv4 address in the leading LAN group. +It stops at the public group, which can contain a synthetic fallback. It does not blacklist a particular public IP. +Loopback, unspecified, and multicast addresses are not usable targets. + +`PF repair` events carry the causal `probe_id` and `recovery_generation` for all four callers. +They record `started`, `not_run`, or `reload_completed`. Completed means that the reload ran, not that interception recovered. +A confirming probe adds its own ID and its interception outcome to the original repair correlation. +The watchdog leaves confirmation to its next scheduled probe. +`budget_exhausted` records a wake repair that the existing budget prevents. +A probe-triggered reload does not count as a missing-anchor repair. + +See [PF DNS interception](pf-dns-intercept.md) and [DNS intercept mode](dns-intercept-mode.md) for the rule lifecycle. + +## Reproduction boundaries + +Compare the affected release with the exact candidate artifact on an owned host. +Keep the configuration unchanged. Capture ordinary DNS and a direct query to the actual listener and port. +Measure recovery after a usable route and resolver return. Exclude unavoidable offline time. +Record source-address changes, PF observations, and upstream results. Success at one layer does not prove success at another. + +Use the recovery mode, ownership, outcome, and effective network-service DNS configuration to assess cancellation. +Never flush global PF state to collect diagnostics. diff --git a/docs/pf-dns-intercept.md b/docs/pf-dns-intercept.md index 1055460..488f56e 100644 --- a/docs/pf-dns-intercept.md +++ b/docs/pf-dns-intercept.md @@ -278,10 +278,13 @@ change handler and spawns an asynchronous interception probe monitor: 1. A subprocess sends a DNS query WITHOUT the `_ctrld` group GID, so pf intercept rules apply to it 2. If ctrld receives the query → pf interception is working -3. If the query times out (1s) → pf translation is broken -4. On failure: `forceReloadPFMainRuleset()` does `pfctl -f -` with the current +3. A successful UDP write without local receipt within 1s gives `not_intercepted`. +4. Only this meaningful failure permits probe-triggered repair: `forceReloadPFMainRuleset()` does `pfctl -f -` with the current running ruleset, resetting pf's translation engine +A missing target or helper failure gives `indeterminate` and does not cause a reload. +The probe uses IPv4 only. See [target selection and diagnostics](network-recovery-diagnostics.md#pf-probes-macos). + The monitor probes with exponential backoff (0, 0.5, 1, 2, 4s) to win the race against async pf reloads. Only one monitor runs at a time (singleton). The watchdog also runs the probe every 30s as a safety net. @@ -300,8 +303,8 @@ Some pf.conf files include `set skip on lo0` which tells pf to skip ALL processi **Mitigation:** the interception probe. `probePFIntercept()` sends a real query from outside the `_ctrld` group and confirms the listener received the redirect, which cannot -succeed while pf is bypassing loopback — so a skip on `lo0` shows up as a probe failure -and triggers a full reload. +succeed while pf bypasses loopback. A confirmed send without receipt permits a full reload. +An unsent or unavailable probe is indeterminate and does not trigger repair. **Not implemented, contrary to earlier versions of this document:** ctrld does *not* strip `lo0` from `set skip on` directives, and the watchdog does *not* inspect skip