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 { Close() error 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 } return networkSourceState(interfaces, (*net.Interface).Addrs) } // networkSourceState builds the state of the given interfaces. The address // reader is a parameter, so a test feeds addresses without a host network. func networkSourceState(interfaces []net.Interface, addressesOf func(*net.Interface) ([]net.Addr, error)) (*netmon.State, error) { state := &netmon.State{Interface: map[string]netmon.Interface{}, InterfaceIPs: map[string][]netip.Prefix{}} for _, iface := range interfaces { addresses, err := addressesOf(&iface) if err != nil { return nil, err } state.Interface[iface.Name] = netmon.Interface{Interface: &iface} for _, address := range addresses { prefix, err := interfaceAddressPrefix(address) if err != nil { return nil, err } state.InterfaceIPs[iface.Name] = append(state.InterfaceIPs[iface.Name], prefix) } } return state, nil } // interfaceAddressPrefix turns one interface address into a prefix. An address // with a subnet mask keeps the mask and its host bits, so a fresh state shows // the same /24 or /64 as the network monitor. An address without a mask is a // host prefix. func interfaceAddressPrefix(address net.Addr) (netip.Prefix, error) { var ip net.IP var mask net.IPMask switch address := address.(type) { case *net.IPNet: ip, mask = address.IP, address.Mask case *net.IPAddr: ip = address.IP default: return netip.Prefix{}, errors.New("unsupported interface address type") } addr, ok := netip.AddrFromSlice(ip) if !ok { return netip.Prefix{}, errors.New("invalid interface address") } addr = addr.Unmap() return netip.PrefixFrom(addr, prefixBits(addr, mask)), nil } // prefixBits reads the length of a mask for an unmapped address. The mask of // an IPv4 address is 4 or 16 bytes long, and a 16-byte mask counts the 96 // bits of the IPv6 mapping too. A missing or irregular mask means a host // prefix. func prefixBits(addr netip.Addr, mask net.IPMask) int { bits, total := mask.Size() if total == 0 { return addr.BitLen() } if addr.Is4() && total == 128 { bits -= 96 } if bits < 0 || bits > addr.BitLen() { return addr.BitLen() } return bits } // 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 // The flow fills these while it runs. The end event reports them, so one // journal line tells what the recovery found and what it changed. recoveredUpstream string dhcpServers []string interceptTargetAction string bypassActive 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") }) } // The outcomes that one recovery pass ends with. const ( recoveryOutcomeCompleted = "completed" recoveryOutcomeCanceled = "canceled" recoveryOutcomeSuperseded = "superseded" ) // journalRecoveredUpstream bounds the upstream name of the end event. An empty // name means that no upstream recovered, so it stays empty. func journalRecoveredUpstream(upstream string) string { if upstream == "" { return "" } return journalUpstreamName(upstream) } func (d *recoveryDiagnostic) end(outcome string) { e := mainLog.Load().Info() if outcome == recoveryOutcomeCanceled || d.failed.Load() { e = mainLog.Load().Warn() } journal(d.event(e)).Str("outcome", outcome).Bool("had_failure", d.failed.Load()). Dur("duration_ms", time.Since(d.started)). Str("recovered_upstream", journalRecoveredUpstream(d.recoveredUpstream)). Strs("dhcp_servers", d.dhcpServers). Int("dhcp_server_count", len(d.dhcpServers)). Str("intercept_target_action", d.interceptTargetAction). Bool("bypass_active", d.bypassActive). Msg("Recovery end") }