diff --git a/cmd/cli/awake_clock_others.go b/cmd/cli/awake_clock_others.go new file mode 100644 index 0000000..ce1ab23 --- /dev/null +++ b/cmd/cli/awake_clock_others.go @@ -0,0 +1,13 @@ +//go:build !windows + +package cli + +import "time" + +var awakeClockBase = time.Now() + +// awakeClock returns how long the host has been awake since ctrld started. +// Go's monotonic clock stops while Linux and macOS sleep. +func awakeClock() time.Duration { + return time.Since(awakeClockBase) +} diff --git a/cmd/cli/awake_clock_test.go b/cmd/cli/awake_clock_test.go new file mode 100644 index 0000000..1926992 --- /dev/null +++ b/cmd/cli/awake_clock_test.go @@ -0,0 +1,22 @@ +package cli + +import ( + "testing" + "time" +) + +// While the host is awake, the awake clock runs with the wall clock, so the +// real clocks find no sleep. +func TestAwakeClockRunsWhileAwake(t *testing.T) { + before := awakeClock() + time.Sleep(50 * time.Millisecond) + if d := awakeClock() - before; d < 40*time.Millisecond || d > 5*time.Second { + t.Fatalf("awake clock moved %v over a 50ms wait", d) + } + var w sleepWatch + w.slept() + time.Sleep(10 * time.Millisecond) + if w.slept() { + t.Fatal("the real clocks found a sleep while awake") + } +} diff --git a/cmd/cli/awake_clock_windows.go b/cmd/cli/awake_clock_windows.go new file mode 100644 index 0000000..5029f66 --- /dev/null +++ b/cmd/cli/awake_clock_windows.go @@ -0,0 +1,35 @@ +package cli + +import ( + "time" + "unsafe" + + "golang.org/x/sys/windows" +) + +var procQueryUnbiasedInterruptTime = windows.NewLazySystemDLL("kernel32.dll").NewProc("QueryUnbiasedInterruptTime") + +var awakeClockBase = time.Now() + +// awakeClock returns how long the host has been awake, from the unbiased +// interrupt time, which stops while Windows sleeps. Go's monotonic clock reads +// the interrupt time, which does not (golang/go#35482). If the call fails, it +// falls back to that clock, and the sleep check then finds no sleep. +func awakeClock() time.Duration { + if d, ok := unbiasedInterruptTime(); ok { + return d + } + return time.Since(awakeClockBase) +} + +// unbiasedInterruptTime returns the time Windows has been awake since boot. +func unbiasedInterruptTime() (time.Duration, bool) { + var t uint64 // 100 ns units + if err := procQueryUnbiasedInterruptTime.Find(); err != nil { + return 0, false + } + if r, _, _ := procQueryUnbiasedInterruptTime.Call(uintptr(unsafe.Pointer(&t))); r == 0 { + return 0, false + } + return time.Duration(t) * 100, true +} diff --git a/cmd/cli/awake_clock_windows_test.go b/cmd/cli/awake_clock_windows_test.go new file mode 100644 index 0000000..d18a623 --- /dev/null +++ b/cmd/cli/awake_clock_windows_test.go @@ -0,0 +1,11 @@ +package cli + +import "testing" + +// The sleep check needs the unbiased interrupt time: the fallback clock keeps +// running while Windows sleeps, so the check would never find a sleep. +func TestUnbiasedInterruptTimeAvailable(t *testing.T) { + if _, ok := unbiasedInterruptTime(); !ok { + t.Fatal("QueryUnbiasedInterruptTime failed") + } +} diff --git a/cmd/cli/dns_proxy.go b/cmd/cli/dns_proxy.go index a5657ac..26e5535 100644 --- a/cmd/cli/dns_proxy.go +++ b/cmd/cli/dns_proxy.go @@ -930,113 +930,134 @@ func (p *prog) proxy(ctx context.Context, req *proxyRequest) *proxyResponse { // answers, so a name nobody resolves gets a real negative answer. internalFallback := p.internalDomainFallbackUpstreams(upstreams) var internalNegative *dns.Msg - for n, upstreamConfig := range upstreamConfigs { - if upstreamConfig == nil { - continue - } - logger := mainLog.Load().Debug(). - Str("upstream", upstreamConfig.String()). - Str("query", req.msg.Question[0].Name). - Bool("is_ad_query", p.isAdDomainQuery(req.msg)). - Bool("is_lan_query", isLanOrPtrQuery) - - if p.isLoop(upstreamConfig) { - ctrld.Log(ctx, logger, "DNS loop detected") - continue - } - // A fallback-mode Internal Domain resolver that the monitor reports as - // down is not waited on: off the organization network every query - // would otherwise pay its timeout before the network fallback runs. - // Nothing else marks a generated resolver up again, so skipping it - // starts a background re-check that does. - if internalFallback && p.um.isDown(upstreams[n]) { - ctrld.Log(ctx, mainLog.Load().Debug(), "internal domain resolver %s is down, skipping it", upstreams[n]) - p.internalDomainProbes.probe(p.um, upstreams[n], upstreamConfig, req.msg) - continue - } - answer := resolve(upstreams[n], upstreamConfig, req.msg) - if answer == nil { - if serveStaleCache && staleAnswer != nil { - ctrld.Log(ctx, mainLog.Load().Debug(), "serving stale cached response") - p.health.countCacheHit() - now := time.Now() - setCachedAnswerTTL(staleAnswer, now, now.Add(staleTTL)) - res.answer = staleAnswer - res.cached = true - return res + // tryUpstreams returns nil when no upstream produced an answer to serve. + tryUpstreams := func() *proxyResponse { + for n, upstreamConfig := range upstreamConfigs { + if upstreamConfig == nil { + continue } - continue - } - // Reject an answer whose question does not match the request before it - // can be served or cached. A mismatched question means the upstream - // answered a different name/type than asked; caching it would poison - // the shared cache with wrong-domain records for the requested name. - // See github.com/Control-D-Inc/ctrld/issues/322. - if !sameQuestion(req.msg, answer) { - ctrld.Log(ctx, mainLog.Load().Debug(), - "discarding answer from %s: question mismatch (asked %q, got %q)", - upstreams[n], questionString(req.msg), questionString(answer)) - continue - } - // We are doing LAN/PTR lookup using private resolver, so always process next one. - // Except for the last, we want to send response instead of saying all upstream failed. - if answer.Rcode != dns.RcodeSuccess && isLanOrPtrQuery && n != len(upstreamConfigs)-1 { - ctrld.Log(ctx, mainLog.Load().Debug(), "no response from %s, process to next upstream", upstreams[n]) - continue - } - if answer.Rcode != dns.RcodeSuccess && len(upstreamConfigs) > 1 && containRcode(req.failoverRcodes, answer.Rcode) { - ctrld.Log(ctx, mainLog.Load().Debug(), "failover rcode matched, process to next upstream") - continue - } - if internalFallback && internalDomainFallbackRcode(answer.Rcode) { - ctrld.Log(ctx, mainLog.Load().Debug(), "internal domain resolver %s answered %s, trying next resolver", - upstreams[n], dns.RcodeToString[answer.Rcode]) - internalNegative = betterInternalDomainNegative(internalNegative, answer) - continue - } + logger := mainLog.Load().Debug(). + Str("upstream", upstreamConfig.String()). + Str("query", req.msg.Question[0].Name). + Bool("is_ad_query", p.isAdDomainQuery(req.msg)). + Bool("is_lan_query", isLanOrPtrQuery) - // set compression, as it is not set by default when unpacking - answer.Compress = true - - if cacheable { - p.addCachedResponse(dnscache.NewKey(req.msg, upstreams[n]), answer) - ctrld.Log(ctx, mainLog.Load().Debug(), "add cached response") - } - hostname := "" - if req.ci != nil { - hostname = req.ci.Hostname - } - // DNS64 synthesis for IPv6-only networks without CLAT: applied to the - // policy-approved answer only, using the same upstream for the companion - // A resolution. No-op unless the network state requires it. - var synthesizedPrefix netip.Prefix - answer, synthesizedPrefix = p.maybeDNS64(ctx, req.msg, answer, func(aReq *dns.Msg) *dns.Msg { - key := dnscache.NewKey(aReq, upstreams[n]) - if cacheable { - if cachedValue := p.cache.Get(key); cachedValue != nil { + if p.isLoop(upstreamConfig) { + ctrld.Log(ctx, logger, "DNS loop detected") + continue + } + // A fallback-mode Internal Domain resolver that the monitor reports as + // down is not waited on: off the organization network every query + // would otherwise pay its timeout before the network fallback runs. + // Nothing else marks a generated resolver up again, so skipping it + // starts a background re-check that does. + if internalFallback && p.um.isDown(upstreams[n]) { + ctrld.Log(ctx, mainLog.Load().Debug(), "internal domain resolver %s is down, skipping it", upstreams[n]) + p.internalDomainProbes.probe(p.um, upstreams[n], upstreamConfig, req.msg) + continue + } + answer := resolve(upstreams[n], upstreamConfig, req.msg) + if answer == nil { + if serveStaleCache && staleAnswer != nil { + ctrld.Log(ctx, mainLog.Load().Debug(), "serving stale cached response") + p.health.countCacheHit() now := time.Now() - if cachedValue.Expire.After(now) { - cached := cachedValue.Msg.Copy() - ctrld.SetCacheReply(cached, aReq, cached.Rcode) - setCachedAnswerTTL(cached, now, cachedValue.Expire) - return cached + setCachedAnswerTTL(staleAnswer, now, now.Add(staleTTL)) + res.answer = staleAnswer + res.cached = true + return res + } + continue + } + // Reject an answer whose question does not match the request before it + // can be served or cached. A mismatched question means the upstream + // answered a different name/type than asked; caching it would poison + // the shared cache with wrong-domain records for the requested name. + // See github.com/Control-D-Inc/ctrld/issues/322. + if !sameQuestion(req.msg, answer) { + ctrld.Log(ctx, mainLog.Load().Debug(), + "discarding answer from %s: question mismatch (asked %q, got %q)", + upstreams[n], questionString(req.msg), questionString(answer)) + continue + } + // We are doing LAN/PTR lookup using private resolver, so always process next one. + // Except for the last, we want to send response instead of saying all upstream failed. + if answer.Rcode != dns.RcodeSuccess && isLanOrPtrQuery && n != len(upstreamConfigs)-1 { + ctrld.Log(ctx, mainLog.Load().Debug(), "no response from %s, process to next upstream", upstreams[n]) + continue + } + if answer.Rcode != dns.RcodeSuccess && len(upstreamConfigs) > 1 && containRcode(req.failoverRcodes, answer.Rcode) { + ctrld.Log(ctx, mainLog.Load().Debug(), "failover rcode matched, process to next upstream") + continue + } + if internalFallback && internalDomainFallbackRcode(answer.Rcode) { + ctrld.Log(ctx, mainLog.Load().Debug(), "internal domain resolver %s answered %s, trying next resolver", + upstreams[n], dns.RcodeToString[answer.Rcode]) + internalNegative = betterInternalDomainNegative(internalNegative, answer) + continue + } + + // set compression, as it is not set by default when unpacking + answer.Compress = true + + if cacheable { + p.addCachedResponse(dnscache.NewKey(req.msg, upstreams[n]), answer) + ctrld.Log(ctx, mainLog.Load().Debug(), "add cached response") + } + hostname := "" + if req.ci != nil { + hostname = req.ci.Hostname + } + // DNS64 synthesis for IPv6-only networks without CLAT: applied to the + // policy-approved answer only, using the same upstream for the companion + // A resolution. No-op unless the network state requires it. + var synthesizedPrefix netip.Prefix + answer, synthesizedPrefix = p.maybeDNS64(ctx, req.msg, answer, func(aReq *dns.Msg) *dns.Msg { + key := dnscache.NewKey(aReq, upstreams[n]) + if cacheable { + if cachedValue := p.cache.Get(key); cachedValue != nil { + now := time.Now() + if cachedValue.Expire.After(now) { + cached := cachedValue.Msg.Copy() + ctrld.SetCacheReply(cached, aReq, cached.Rcode) + setCachedAnswerTTL(cached, now, cachedValue.Expire) + return cached + } } } + resolved := resolve(upstreams[n], upstreamConfig, aReq) + if cacheable && resolved != nil && sameQuestion(aReq, resolved) { + p.addCachedResponse(key, resolved) + } + return resolved + }) + if cacheable && synthesizedPrefix.IsValid() { + p.addCachedResponse(dns64CacheKey(req.msg, upstreams[n], synthesizedPrefix), answer) + ctrld.Log(ctx, mainLog.Load().Debug(), "dns64: add cached response variant") } - resolved := resolve(upstreams[n], upstreamConfig, aReq) - if cacheable && resolved != nil && sameQuestion(aReq, resolved) { - p.addCachedResponse(key, resolved) - } - return resolved - }) - if cacheable && synthesizedPrefix.IsValid() { - p.addCachedResponse(dns64CacheKey(req.msg, upstreams[n], synthesizedPrefix), answer) - ctrld.Log(ctx, mainLog.Load().Debug(), "dns64: add cached response variant") + ctrld.Log(ctx, mainLog.Load().Info(), "REPLY: %s -> %s (%s): %s", upstreams[n], req.ufr.srcAddr, hostname, dns.RcodeToString[answer.Rcode]) + res.answer = answer + res.upstream = upstreamConfig.Endpoint + return res + } + return nil + } + // See fresh_connection.go: after a sleep, renew the connections before + // the query rides them. + p.renewConnectionsIfSlept(ctx) + connEpoch := p.networkEpoch.Load() + if r := tryUpstreams(); r != nil { + return r + } + // A query that ran on connections the network left behind gets one more + // pass. It can take as long as the first, timeouts included, before the + // fallback: that is the price of not sending the query in plaintext, and + // it is paid only by the queries already in flight when the network + // changed or the host slept. + if p.retryOnFreshConnection(ctx, upstreams, upstreamConfigs, connEpoch) { + if r := tryUpstreams(); r != nil { + return r } - ctrld.Log(ctx, mainLog.Load().Info(), "REPLY: %s -> %s (%s): %s", upstreams[n], req.ufr.srcAddr, hostname, dns.RcodeToString[answer.Rcode]) - res.answer = answer - res.upstream = upstreamConfig.Endpoint - return res } if internalFallback { ctrld.Log(ctx, mainLog.Load().Debug(), "internal domain resolvers did not resolve the query; trying network resolvers") @@ -2046,9 +2067,7 @@ func (p *prog) handleNetworkChange(delta *netmon.ChangeDelta, isMajorChange bool } mainLog.Load().Debug().Msg("Link state changed, re-bootstrapping") - for _, uc := range p.cfg.Upstream { - uc.ReBootstrap() - } + p.renewUpstreamConnections() // Get IPs from default route interface in new state selfIP := networkChangeDefaultRouteIPFn() diff --git a/cmd/cli/fresh_connection.go b/cmd/cli/fresh_connection.go new file mode 100644 index 0000000..f676640 --- /dev/null +++ b/cmd/cli/fresh_connection.go @@ -0,0 +1,128 @@ +package cli + +import ( + "context" + "sync" + "time" + + "github.com/Control-D-Inc/ctrld" +) + +// A query that an encrypted upstream fails falls back to the OS resolver, +// which sends it in plaintext on port 53. After a sleep or a network change, +// that failure is often not the upstream's: the query rode a connection opened +// on the previous network, or before the sleep, and timed out on it. A new +// connection answers at once (#631). So: +// +// - every query first checks whether the host slept since the previous +// query, and if so renews the upstream connections before it is sent. A +// sleep shows on the clocks before netmon or a wake detector reports it. +// - before the fallback, a failed query is retried once on fresh +// connections when the network epoch moved while it ran: a network change +// or a wake was handled, or a sleep was found then, and the renewal gave +// the retry new transports. + +// hostSleepSlack is how far the wall clock may move ahead of the awake clock +// between two checks before the host counts as having slept. The awake +// clock's rate also differs from the NTP-disciplined wall clock, by far less +// than 1%, which the check allows on top. +const hostSleepSlack = 2 * time.Second + +// Clocks of the sleep check. Tests replace them. +var ( + sleepWatchWallFn = func() time.Time { return time.Now().Round(0) } + sleepWatchAwakeFn = awakeClock +) + +// sleepWatch finds a sleep of the host between two of its checks: the wall +// clock keeps running while the host sleeps, and the awake clock does not. +type sleepWatch struct { + mu sync.Mutex + seen bool + wall time.Time + awake time.Duration +} + +// slept reports whether the host slept since the previous call. A wall clock +// step, as from NTP, reads the same, and costs one renewal of the connections. +func (w *sleepWatch) slept() bool { + wall, awake := sleepWatchWallFn(), sleepWatchAwakeFn() + w.mu.Lock() + defer w.mu.Unlock() + seen, prevWall, prevAwake := w.seen, w.wall, w.awake + w.seen, w.wall, w.awake = true, wall, awake + if !seen { + return false + } + awakeFor := awake - prevAwake + return wall.Sub(prevWall)-awakeFor > hostSleepSlack+awakeFor/100 +} + +// renewConnectionsIfSlept renews the upstream connections if the host slept +// since the previous check, and reports whether it did. +func (p *prog) renewConnectionsIfSlept(ctx context.Context) bool { + if !p.sleepWatch.slept() { + return false + } + ctrld.Log(ctx, mainLog.Load().Info(), "host slept since the previous query; renewing upstream connections") + p.renewUpstreamConnections() + return true +} + +// renewUpstreamConnections makes every upstream open new connections for its +// next query, and moves the network epoch, so queries running on the old +// connections retry on the new ones. The re-bootstrap is lazy: the next query +// builds the transports. +func (p *prog) renewUpstreamConnections() { + p.connRenewMu.Lock() + defer p.connRenewMu.Unlock() + p.renewUpstreamConnectionsLocked() +} + +func (p *prog) renewUpstreamConnectionsLocked() { + if p.cfg != nil { + for _, uc := range p.cfg.Upstream { + uc.ReBootstrap() + } + } + // After the re-bootstrap, so a query that sees the new epoch retries on + // the new transports. + p.networkEpoch.Add(1) +} + +// retryOnFreshConnection reports whether a query whose upstreams all failed +// gets one more try on fresh connections before any fallback. epoch is the +// network epoch before its first try. +func (p *prog) retryOnFreshConnection(ctx context.Context, upstreams []string, configs []*ctrld.UpstreamConfig, epoch uint64) bool { + // Internal Domain resolvers have their own fallback, and the network + // resolvers they fall back to are not the plaintext catch-all. + if p.internalDomainExplicitUpstreams(upstreams) || p.internalDomainFallbackUpstreams(upstreams) { + return false + } + if !hasConnectionUpstream(configs) { + return false + } + // A sleep that no query has seen yet renews the connections, and moves + // the epoch. + p.renewConnectionsIfSlept(ctx) + if p.networkEpoch.Load() == epoch { + return false + } + ctrld.Log(ctx, mainLog.Load().Debug(), "network changed or host slept during the query; retrying upstreams on fresh connections") + return true +} + +// hasConnectionUpstream reports whether any upstream keeps connections that a +// network change can leave dead: DoH, DoH3, DoT and DoQ. +func hasConnectionUpstream(configs []*ctrld.UpstreamConfig) bool { + for _, uc := range configs { + if uc == nil { + continue + } + switch uc.Type { + case ctrld.ResolverTypeDOH, ctrld.ResolverTypeDOH3, ctrld.ResolverTypeDOT, ctrld.ResolverTypeDOQ: + return true + } + } + return false +} diff --git a/cmd/cli/fresh_connection_test.go b/cmd/cli/fresh_connection_test.go new file mode 100644 index 0000000..1772af2 --- /dev/null +++ b/cmd/cli/fresh_connection_test.go @@ -0,0 +1,236 @@ +package cli + +import ( + "context" + "errors" + "sync" + "testing" + "time" + + "github.com/miekg/dns" + + "github.com/Control-D-Inc/ctrld" +) + +// fakeSleepClocks drives the sleep check: sleep moves the wall clock only, as +// a sleeping host does, and awake moves both. +type fakeSleepClocks struct { + mu sync.Mutex + wall time.Time + awake time.Duration +} + +func stubSleepClocks(t *testing.T) *fakeSleepClocks { + t.Helper() + c := &fakeSleepClocks{wall: time.Date(2026, 10, 14, 21, 38, 0, 0, time.UTC)} + oldWall, oldAwake := sleepWatchWallFn, sleepWatchAwakeFn + t.Cleanup(func() { sleepWatchWallFn, sleepWatchAwakeFn = oldWall, oldAwake }) + sleepWatchWallFn = func() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.wall + } + sleepWatchAwakeFn = func() time.Duration { + c.mu.Lock() + defer c.mu.Unlock() + return c.awake + } + return c +} + +func (c *fakeSleepClocks) sleep(d time.Duration) { + c.mu.Lock() + defer c.mu.Unlock() + c.wall = c.wall.Add(d) +} + +func (c *fakeSleepClocks) run(d time.Duration) { + c.mu.Lock() + defer c.mu.Unlock() + c.wall = c.wall.Add(d) + c.awake += d +} + +// The wall clock moving ahead of the awake clock is a sleep. Both moving +// together is not, however long, and neither is the awake clock's rate error. +func TestSleepWatch(t *testing.T) { + for _, tc := range []struct { + name string + pass func(*fakeSleepClocks) + slept bool + }{ + {"awake", func(c *fakeSleepClocks) { c.run(time.Hour) }, false}, + {"slept", func(c *fakeSleepClocks) { c.run(time.Second); c.sleep(time.Minute) }, true}, + {"short nap", func(c *fakeSleepClocks) { c.sleep(time.Second) }, false}, + {"drift", func(c *fakeSleepClocks) { c.run(24 * time.Hour); c.sleep(10 * time.Minute) }, false}, + {"slept after a long idle", func(c *fakeSleepClocks) { c.run(24 * time.Hour); c.sleep(time.Hour) }, true}, + } { + t.Run(tc.name, func(t *testing.T) { + c := stubSleepClocks(t) + var w sleepWatch + if w.slept() { + t.Fatal("the first check found a sleep") + } + tc.pass(c) + if got := w.slept(); got != tc.slept { + t.Fatalf("slept = %v, want %v", got, tc.slept) + } + if w.slept() { + t.Fatal("the same sleep was found twice") + } + }) + } +} + +// freshConnResolver models upstream connections opened in network epoch 0 as +// dead: the upstream fails until the connections are renewed. onFirst runs +// during the upstream's first query. The OS resolver records its use and +// answers. +type freshConnResolver struct { + p *prog + os bool + state *freshConnState +} + +type freshConnState struct { + upstreamCalls int + osCalls int + onFirst func() +} + +func (r freshConnResolver) Resolve(_ context.Context, msg *dns.Msg) (*dns.Msg, error) { + if r.os { + r.state.osCalls++ + } else { + r.state.upstreamCalls++ + dead := r.p.networkEpoch.Load() == 0 + if r.state.upstreamCalls == 1 && r.state.onFirst != nil { + r.state.onFirst() + } + if dead { + return nil, errors.New("context deadline exceeded") + } + } + answer := new(dns.Msg) + answer.SetReply(msg) + return answer, nil +} + +func newFreshConnProg(t *testing.T, typ string) (*prog, *freshConnState, *fakeSleepClocks) { + t.Helper() + captureDebugMainLog(t) + clocks := stubSleepClocks(t) + leak := true + cfg := &ctrld.Config{Upstream: map[string]*ctrld.UpstreamConfig{ + "0": {Name: "control d", Type: typ, Endpoint: "https://192.0.2.1/dns-query", Timeout: 100}, + }} + cfg.Service.LeakOnUpstreamFailure = &leak + p := &prog{cfg: cfg, um: newUpstreamMonitor(cfg)} + p.um.after = func(time.Duration, func()) {} + p.lanLoopGuard = newLoopGuard() + p.ptrLoopGuard = newLoopGuard() + t.Cleanup(func() { p.querySampler.closeExpired(time.Now().Add(querySampleWindow)) }) + state := &freshConnState{} + oldResolver, oldRecovery := newResolverFn, queryRecoveryFn + t.Cleanup(func() { newResolverFn, queryRecoveryFn = oldResolver, oldRecovery }) + newResolverFn = func(uc *ctrld.UpstreamConfig) (ctrld.Resolver, error) { + return freshConnResolver{p: p, os: uc.Name == osUpstreamConfig.Name, state: state}, nil + } + queryRecoveryFn = func(*prog, RecoveryReason) {} + // The watch has seen the host awake before the test. + p.sleepWatch.slept() + return p, state, clocks +} + +func freshConnQuery(p *prog) *proxyResponse { + msg := new(dns.Msg) + msg.SetQuestion("time.apple.com.", dns.TypeA) + return p.proxy(context.Background(), &proxyRequest{msg: msg, ufr: &upstreamForResult{ + upstreams: []string{upstreamPrefix + "0"}, srcAddr: "127.0.0.1:5353", + }}) +} + +// A query does not ride connections that predate the network to a plaintext +// fallback (#631). A sleep found before the query renews the connections +// before it is sent. A network change, or a sleep found when the query fails, +// retries it once on fresh connections. Without such evidence a failure goes +// to the OS resolver at once, as before. +func TestFreshConnectionRetryBeforeOSFallback(t *testing.T) { + for _, tc := range []struct { + name string + typ string + before func(*prog, *fakeSleepClocks) + during func(*prog, *fakeSleepClocks) + wantUpstream int + wantOS int + }{ + {name: "host slept before the query", typ: ctrld.ResolverTypeDOH, + before: func(_ *prog, c *fakeSleepClocks) { c.sleep(time.Hour) }, + wantUpstream: 1}, + {name: "host slept during the query", typ: ctrld.ResolverTypeDOH, + during: func(_ *prog, c *fakeSleepClocks) { c.sleep(time.Hour) }, + wantUpstream: 2}, + {name: "network change during the query", typ: ctrld.ResolverTypeDOH, + during: func(p *prog, _ *fakeSleepClocks) { p.renewUpstreamConnections() }, + wantUpstream: 2}, + {name: "DoT, network change during the query", typ: ctrld.ResolverTypeDOT, + during: func(p *prog, _ *fakeSleepClocks) { p.renewUpstreamConnections() }, + wantUpstream: 2}, + {name: "stable network", typ: ctrld.ResolverTypeDOH, + during: func(_ *prog, c *fakeSleepClocks) { c.run(time.Second) }, + wantUpstream: 1, wantOS: 1}, + {name: "plain DNS upstream", typ: ctrld.ResolverTypeLegacy, + during: func(p *prog, _ *fakeSleepClocks) { p.renewUpstreamConnections() }, + wantUpstream: 1, wantOS: 1}, + } { + t.Run(tc.name, func(t *testing.T) { + p, state, clocks := newFreshConnProg(t, tc.typ) + if tc.before != nil { + tc.before(p, clocks) + } + if tc.during != nil { + state.onFirst = func() { tc.during(p, clocks) } + } + + res := freshConnQuery(p) + + if res.answer == nil || res.answer.Rcode != dns.RcodeSuccess { + t.Fatalf("answer = %v, want an answer", res.answer) + } + if state.upstreamCalls != tc.wantUpstream || state.osCalls != tc.wantOS { + t.Fatalf("upstream calls = %d, OS resolver calls = %d; want %d and %d", + state.upstreamCalls, state.osCalls, tc.wantUpstream, tc.wantOS) + } + }) + } +} + +// Of the queries that run after a sleep, one renews the connections. +func TestFreshConnectionSleepRenewsOnce(t *testing.T) { + p, _, clocks := newFreshConnProg(t, ctrld.ResolverTypeDOH) + clocks.sleep(time.Hour) + var wg sync.WaitGroup + for range 8 { + wg.Go(func() { p.renewConnectionsIfSlept(context.Background()) }) + } + wg.Wait() + if got := p.networkEpoch.Load(); got != 1 { + t.Fatalf("network epoch = %d, want 1", got) + } +} + +// A wake and an accepted network change both renew the upstream connections. +func TestWakeAndNetworkChangeRenewConnections(t *testing.T) { + h := newNetworkEventsHarness(t, "en0", "en7") + h.prog.noteHostWoke("detector", time.Minute, nil) + if got := h.prog.networkEpoch.Load(); got != 1 { + t.Fatalf("network epoch after the wake = %d, want 1", got) + } + before := orderTestState("", "", map[string][]string{"en0": nil}, map[string]bool{"en0": true}) + before.HaveV4 = false + after := orderTestState("en7", "", map[string][]string{"en0": nil, "en7": {"192.0.2.30/24"}}, map[string]bool{"en0": true}) + h.handle(before, after, false) + if got := h.prog.networkEpoch.Load(); got != 2 { + t.Fatalf("network epoch after the change = %d, want 2", got) + } +} diff --git a/cmd/cli/network_events.go b/cmd/cli/network_events.go index 959fb29..9116075 100644 --- a/cmd/cli/network_events.go +++ b/cmd/cli/network_events.go @@ -42,6 +42,9 @@ func (p *prog) noteHostWoke(source string, gap time.Duration, state *netmon.Stat if p.dnsConfig != nil { p.dnsConfig.noteActivity(now) } + // Connections opened before the sleep are likely dead, on the same + // network too, once a NAT or the server dropped them. + p.renewUpstreamConnections() p.wake.note(now, wakeReport{source: source, gap: gap, state: state}, logHostWoke) } diff --git a/cmd/cli/prog.go b/cmd/cli/prog.go index de56eca..c4dfc3f 100644 --- a/cmd/cli/prog.go +++ b/cmd/cli/prog.go @@ -215,6 +215,14 @@ type prog struct { // See os_resolver_stale.go. osResolverStaleUntil atomic.Int64 + // networkEpoch counts the events after which the upstream connections + // may predate the network: accepted network changes and wakes. See + // fresh_connection.go. + networkEpoch atomic.Uint64 + connRenewMu sync.Mutex + // sleepWatch finds a sleep of the host between two queries. + sleepWatch sleepWatch + // dns64 tracks DNS64/NAT64 synthesis state for IPv6-only networks // without client-side 464XLAT. See cmd/cli/dns64.go. dns64 dns64State diff --git a/doh_fresh_connection_test.go b/doh_fresh_connection_test.go new file mode 100644 index 0000000..f886e37 --- /dev/null +++ b/doh_fresh_connection_test.go @@ -0,0 +1,163 @@ +package ctrld + +import ( + "context" + "crypto/tls" + "encoding/base64" + "net" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "github.com/miekg/dns" +) + +// staleConnProxy forwards TCP connections to target. Connections it accepted +// before staleExisting stop relaying, as connections opened before a sleep or +// on the previous network do; later connections work. +type staleConnProxy struct { + ln net.Listener + target string + mu sync.Mutex + conns map[net.Conn]bool // accepted connection -> stale +} + +func newStaleConnProxy(t *testing.T, target string) *staleConnProxy { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + p := &staleConnProxy{ln: ln, target: target, conns: map[net.Conn]bool{}} + t.Cleanup(func() { + ln.Close() + p.mu.Lock() + defer p.mu.Unlock() + for c := range p.conns { + c.Close() + } + }) + go func() { + for { + c, err := ln.Accept() + if err != nil { + return + } + p.mu.Lock() + p.conns[c] = false + p.mu.Unlock() + go p.relay(c) + } + }() + return p +} + +func (p *staleConnProxy) staleExisting() { + p.mu.Lock() + defer p.mu.Unlock() + for c := range p.conns { + p.conns[c] = true + } +} + +func (p *staleConnProxy) stale(c net.Conn) bool { + p.mu.Lock() + defer p.mu.Unlock() + return p.conns[c] +} + +func (p *staleConnProxy) relay(client net.Conn) { + defer client.Close() + server, err := net.Dial("tcp", p.target) + if err != nil { + return + } + defer server.Close() + pipe := func(dst, src net.Conn) { + buf := make([]byte, 32*1024) + for { + n, err := src.Read(buf) + if err != nil { + return + } + if p.stale(client) { + continue + } + if _, err := dst.Write(buf[:n]); err != nil { + return + } + } + } + go pipe(client, server) + pipe(server, client) +} + +// dnsAnsweringHandler answers every DoH GET with an empty NOERROR reply. +func dnsAnsweringHandler() http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + data, err := base64.RawURLEncoding.DecodeString(r.URL.Query().Get("dns")) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + msg := new(dns.Msg) + if err := msg.Unpack(data); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + answer := new(dns.Msg) + answer.SetReply(msg) + out, _ := answer.Pack() + w.Header().Set("Content-Type", "application/dns-message") + _, _ = w.Write(out) + }) +} + +// A query on a connection that went stale times out, and after ReBootstrap the +// next query opens a new connection and is answered: the retry on fresh +// connections that ctrld makes after a sleep or a network change (#631) +// depends on it. +func TestDoHReBootstrapReplacesStaleConnection(t *testing.T) { + srv := httptest.NewUnstartedServer(dnsAnsweringHandler()) + srv.TLS = &tls.Config{ + Certificates: []tls.Certificate{generateTestCertificate(t).tlsCert}, + NextProtos: []string{"h2", "http/1.1"}, + MinVersion: tls.VersionTLS12, + } + srv.EnableHTTP2 = true + srv.StartTLS() + t.Cleanup(srv.Close) + proxy := newStaleConnProxy(t, srv.Listener.Addr().String()) + _, port, _ := net.SplitHostPort(proxy.ln.Addr().String()) + uc := dohUpstreamForTLSServer(t, srv) + uc.Endpoint = "https://127.0.0.1:" + port + "/dns-query" + uc.Init() + uc.BootstrapIP = "127.0.0.1" + + resolve := func() error { + r, err := NewResolver(uc) + if err != nil { + return err + } + ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond) + defer cancel() + msg := new(dns.Msg) + msg.SetQuestion("time.apple.com.", dns.TypeA) + _, err = r.Resolve(ctx, msg) + return err + } + + if err := resolve(); err != nil { + t.Fatalf("first query: %v", err) + } + proxy.staleExisting() + if err := resolve(); err == nil { + t.Fatal("query on the stale connection succeeded; the test does not reproduce a dead connection") + } + uc.ReBootstrap() + if err := resolve(); err != nil { + t.Fatalf("query after ReBootstrap: %v", err) + } +}