diff --git a/cmd/ateom-gvisor/hosted.go b/cmd/ateom-gvisor/hosted.go index 8789a6727..43c5edb95 100644 --- a/cmd/ateom-gvisor/hosted.go +++ b/cmd/ateom-gvisor/hosted.go @@ -80,7 +80,6 @@ func (s *AteomService) hostActor(ctx context.Context, attribution resources.Acto ActorUID: uid, Veth: true, EgressPort: s.tunnel.EgressPort, - DNSPort: atunnel.DNSPort, }, s.tunnel.Egress, s.tunnel.DNSRelay) if err != nil { s.actorsMu.Lock() diff --git a/cmd/ateom-microvm/hosted.go b/cmd/ateom-microvm/hosted.go index 0a5e23ae5..c531af9c7 100644 --- a/cmd/ateom-microvm/hosted.go +++ b/cmd/ateom-microvm/hosted.go @@ -86,7 +86,6 @@ func (s *AteomService) hostActor(ctx context.Context, attribution resources.Acto session, err := ateomnet.ServeSandbox(ctx, ateomnet.SandboxNetworkConfig{ ActorUID: uid, EgressPort: s.tunnel.EgressPort, - DNSPort: atunnel.DNSPort, }, s.tunnel.Egress, s.tunnel.DNSRelay) if err != nil { s.actorsMu.Lock() diff --git a/cmd/ateom-microvm/net_linux_test.go b/cmd/ateom-microvm/net_linux_test.go index f25e894b8..1a8b41f3e 100644 --- a/cmd/ateom-microvm/net_linux_test.go +++ b/cmd/ateom-microvm/net_linux_test.go @@ -40,12 +40,8 @@ func TestHostActorReplacesSameActor(t *testing.T) { if err != nil { t.Fatal(err) } - dns, err := atunnel.NewDNSRelay([]string{"127.0.0.1:53"}) - if err != nil { - t.Fatal(err) - } service := &AteomService{ - tunnel: &ateomtunnel.Tunnel{Egress: egress, EgressPort: 15001, DNSRelay: dns}, + tunnel: &ateomtunnel.Tunnel{Egress: egress, EgressPort: 15001}, actors: map[string]*hostedActor{}, maxActors: 1, } diff --git a/internal/atunnel/dns.go b/internal/ateomnet/dns/relay.go similarity index 61% rename from internal/atunnel/dns.go rename to internal/ateomnet/dns/relay.go index 24d6541b3..b636f72a4 100644 --- a/internal/atunnel/dns.go +++ b/internal/ateomnet/dns/relay.go @@ -12,25 +12,25 @@ // See the License for the specific language governing permissions and // limitations under the License. -package atunnel +package dns import ( - "bufio" "context" "errors" "fmt" "io" "log/slog" "net" - "os" - "strings" + "strconv" "sync" "time" + + "github.com/agent-substrate/substrate/internal/ateomnet/netns" ) const ( - // DNSPort is the relay port on the sandbox's gateway. - DNSPort = 53 + // dnsPort is the relay port on the sandbox's gateway. + dnsPort = 53 // Read complete UDP datagrams without truncating EDNS responses. maxDNSDatagram = 65535 @@ -48,10 +48,10 @@ const ( dnsExchangeTimeout = 5 * time.Second ) -// DNSRelay forwards UDP and TCP DNS unchanged to the worker pod's resolvers. +// Relay forwards UDP and TCP DNS unchanged to the worker pod's resolvers. // It listens in the sandbox's gateway namespace and dials from the worker's. // DNS bypasses the egress tunnel and is not checked against egress policy. -type DNSRelay struct { +type Relay struct { upstreams []string // dialer reaches upstream resolvers from the worker namespace. dialer *net.Dialer @@ -61,17 +61,29 @@ type DNSRelay struct { connections chan struct{} } -// NewDNSRelay forwards to upstreams, each "host:port". -func NewDNSRelay(upstreams []string) (*DNSRelay, error) { +// NewRelay reads nameservers from resolvConfPath and forwards to each on port 53. +func NewRelay(resolvConfPath string) (*Relay, error) { + upstreams, err := resolvConfNameservers(resolvConfPath) + if err != nil { + return nil, err + } + return NewRelayForUpstreams(upstreams) +} + +// NewRelayForUpstreams forwards to upstreams, each "host:port", tried in order. +func NewRelayForUpstreams(upstreams []string) (*Relay, error) { if len(upstreams) == 0 { - return nil, fmt.Errorf("atunnel: at least one upstream resolver is required") + return nil, fmt.Errorf("dns: at least one upstream resolver is required") } for _, u := range upstreams { if _, _, err := net.SplitHostPort(u); err != nil { - return nil, fmt.Errorf("atunnel: invalid upstream resolver %q: %w", u, err) + return nil, fmt.Errorf("dns: invalid upstream resolver %q: %w", u, err) } } - return &DNSRelay{ + + slog.Info("DNS relay configured", slog.Any("upstreams", upstreams)) + + return &Relay{ upstreams: upstreams, dialer: &net.Dialer{Timeout: dnsExchangeTimeout}, inFlight: make(chan struct{}, maxInFlightDNS), @@ -79,42 +91,61 @@ func NewDNSRelay(upstreams []string) (*DNSRelay, error) { }, nil } -// ResolvConfNameservers reads nameservers from resolv.conf as "host:53". -func ResolvConfNameservers(path string) ([]string, error) { - f, err := os.Open(path) - if err != nil { - return nil, fmt.Errorf("atunnel: reading resolv.conf: %w", err) - } - defer f.Close() +// Serve serves UDP and TCP DNS in the sandbox's local gateway namespace. +func (r *Relay) Serve(ctx context.Context, ns netns.Handle) ([]io.Closer, []func(), error) { + // Bind the wildcard because the microVM tap's gateway address is added later. + address := net.JoinHostPort("0.0.0.0", strconv.Itoa(dnsPort)) - var out []string - scanner := bufio.NewScanner(f) - for scanner.Scan() { - line := strings.TrimSpace(scanner.Text()) - if i := strings.IndexAny(line, "#;"); i >= 0 { - line = strings.TrimSpace(line[:i]) + var packet net.PacketConn + var stream net.Listener + if err := netns.Do(ctx, ns, func(context.Context) error { + pc, err := net.ListenPacket("udp", address) + if err != nil { + return fmt.Errorf("while opening the actor DNS socket: %w", err) } - rest, ok := strings.CutPrefix(line, "nameserver") - if !ok { - continue + packet = pc + l, err := net.Listen("tcp", address) + if err != nil { + _ = pc.Close() + return fmt.Errorf("while opening the actor DNS listener: %w", err) } - address := strings.TrimSpace(rest) - if address == "" || net.ParseIP(address) == nil { - continue - } - out = append(out, net.JoinHostPort(address, "53")) + stream = l + return nil + }); err != nil { + return nil, nil, err } - if err := scanner.Err(); err != nil { - return nil, fmt.Errorf("atunnel: reading resolv.conf: %w", err) - } - if len(out) == 0 { - return nil, fmt.Errorf("atunnel: %s names no usable nameserver", path) - } - return out, nil + + closers, serve := r.ServeOn(ctx, packet, stream) + return closers, serve, nil } -// ServePacket answers UDP queries until ctx is canceled or the socket fails. -func (r *DNSRelay) ServePacket(ctx context.Context, pc net.PacketConn) error { +// ServeOn serves UDP DNS on packet and TCP DNS on stream, which the caller has +// already bound. It returns the functions that serve, each to be run on its own +// goroutine, and the closers that stop them and release the sockets. +func (r *Relay) ServeOn(ctx context.Context, packet net.PacketConn, stream net.Listener) ([]io.Closer, []func()) { + // Detached from the activation RPC's context but cancelable: the relay's + // capacity is the worker's, so teardown must drop queries still in flight. + serveCtx, stopServing := context.WithCancel(context.WithoutCancel(ctx)) + serve := []func(){ + func() { + if err := r.servePacket(serveCtx, packet); err != nil { + slog.WarnContext(ctx, "Actor DNS socket stopped", slog.Any("err", err)) + } + }, + func() { + if err := r.serveTCP(serveCtx, stream); err != nil { + slog.WarnContext(ctx, "Actor DNS listener stopped", slog.Any("err", err)) + } + }, + } + // Cancel first: closing the sockets alone leaves the queries already being + // resolved holding the relay. + closers := []io.Closer{closerFunc(func() error { stopServing(); return nil }), packet, stream} + return closers, serve +} + +// servePacket answers UDP queries until ctx is canceled or the socket fails. +func (r *Relay) servePacket(ctx context.Context, pc net.PacketConn) error { done := make(chan struct{}) go func() { select { @@ -135,7 +166,7 @@ func (r *DNSRelay) ServePacket(ctx context.Context, pc net.PacketConn) error { if ctx.Err() != nil || errors.Is(err, net.ErrClosed) { return nil } - return fmt.Errorf("atunnel: reading actor DNS query: %w", err) + return fmt.Errorf("dns: reading actor DNS query: %w", err) } // Copied: the buffer is reused by the next read. query := make([]byte, n) @@ -144,7 +175,7 @@ func (r *DNSRelay) ServePacket(ctx context.Context, pc net.PacketConn) error { select { case r.inFlight <- struct{}{}: default: - slog.DebugContext(ctx, "atunnel dropped a DNS query; too many in flight") + slog.DebugContext(ctx, "dns relay dropped a DNS query; too many in flight") continue } wg.Add(1) @@ -153,18 +184,18 @@ func (r *DNSRelay) ServePacket(ctx context.Context, pc net.PacketConn) error { defer func() { <-r.inFlight }() answer, err := r.exchangeUDP(ctx, query) if err != nil { - slog.WarnContext(ctx, "atunnel could not resolve an actor DNS query", slog.Any("err", err)) + slog.WarnContext(ctx, "dns relay could not resolve an actor DNS query", slog.Any("err", err)) return } if _, err := pc.WriteTo(answer, from); err != nil && ctx.Err() == nil { - slog.WarnContext(ctx, "atunnel could not return a DNS answer", slog.Any("err", err)) + slog.WarnContext(ctx, "dns relay could not return a DNS answer", slog.Any("err", err)) } }() } } -// Serve relays TCP DNS connections until ctx is canceled or the listener closes. -func (r *DNSRelay) Serve(ctx context.Context, listener net.Listener) error { +// serveTCP relays TCP DNS connections until ctx is canceled or the listener closes. +func (r *Relay) serveTCP(ctx context.Context, listener net.Listener) error { done := make(chan struct{}) go func() { select { @@ -186,12 +217,12 @@ func (r *DNSRelay) Serve(ctx context.Context, listener net.Listener) error { if ctx.Err() != nil || errors.Is(err, net.ErrClosed) { return nil } - return fmt.Errorf("atunnel: accepting actor DNS connection: %w", err) + return fmt.Errorf("dns: accepting actor DNS connection: %w", err) } select { case r.connections <- struct{}{}: default: - slog.DebugContext(ctx, "atunnel refused a DNS connection; too many open") + slog.DebugContext(ctx, "dns relay refused a DNS connection; too many open") _ = conn.Close() continue } @@ -204,7 +235,7 @@ func (r *DNSRelay) Serve(ctx context.Context, listener net.Listener) error { } } -func (r *DNSRelay) exchangeUDP(ctx context.Context, query []byte) ([]byte, error) { +func (r *Relay) exchangeUDP(ctx context.Context, query []byte) ([]byte, error) { var errs error // deferred holds a server-failure answer to fall back on, see below. var deferred []byte @@ -251,45 +282,11 @@ func (r *DNSRelay) exchangeUDP(ctx context.Context, query []byte) ([]byte, error if deferred != nil { return deferred, nil } - return nil, fmt.Errorf("atunnel: no upstream resolver answered: %w", errs) -} - -// Response codes that say the resolver failed rather than answered. NXDOMAIN -// and NOERROR are answers and are passed back as they are. -const ( - rcodeServFail = 2 - rcodeNotImp = 4 - rcodeRefused = 5 -) - -// failoverRcode reports the response code when the relay should try the next -// upstream. Reads the 12-byte header only; anything shorter is passed through. -func failoverRcode(msg []byte) (byte, bool) { - if len(msg) < 12 { - return 0, false - } - rcode := msg[3] & 0x0f - switch rcode { - case rcodeServFail, rcodeNotImp, rcodeRefused: - return rcode, true - } - return 0, false -} - -func rcodeError(rcode byte) error { - switch rcode { - case rcodeServFail: - return errors.New("answered SERVFAIL") - case rcodeNotImp: - return errors.New("answered NOTIMP") - case rcodeRefused: - return errors.New("answered REFUSED") - } - return fmt.Errorf("answered rcode %d", rcode) + return nil, fmt.Errorf("dns: no upstream resolver answered: %w", errs) } // relayTCP copies a DNS stream without parsing its length-prefixed messages. -func (r *DNSRelay) relayTCP(ctx context.Context, downstream net.Conn) { +func (r *Relay) relayTCP(ctx context.Context, downstream net.Conn) { defer downstream.Close() var upstream net.Conn @@ -304,7 +301,7 @@ func (r *DNSRelay) relayTCP(ctx context.Context, downstream net.Conn) { break } if upstream == nil { - slog.WarnContext(ctx, "atunnel could not reach any resolver for an actor DNS connection", slog.Any("err", errs)) + slog.WarnContext(ctx, "dns relay could not reach any resolver for an actor DNS connection", slog.Any("err", errs)) return } defer upstream.Close() @@ -346,3 +343,43 @@ func (r *DNSRelay) relayTCP(ctx context.Context, downstream net.Conn) { }() wg.Wait() } + +// Response codes that say the resolver failed rather than answered. NXDOMAIN +// and NOERROR are answers and are passed back as they are. +const ( + rcodeServFail = 2 + rcodeNotImp = 4 + rcodeRefused = 5 +) + +// failoverRcode reports the response code when the relay should try the next +// upstream. Reads the 12-byte header only; anything shorter is passed through. +func failoverRcode(msg []byte) (byte, bool) { + if len(msg) < 12 { + return 0, false + } + rcode := msg[3] & 0x0f + switch rcode { + case rcodeServFail, rcodeNotImp, rcodeRefused: + return rcode, true + } + return 0, false +} + +func rcodeError(rcode byte) error { + switch rcode { + case rcodeServFail: + return errors.New("answered SERVFAIL") + case rcodeNotImp: + return errors.New("answered NOTIMP") + case rcodeRefused: + return errors.New("answered REFUSED") + } + return fmt.Errorf("answered rcode %d", rcode) +} + +// closerFunc adapts a cancel function to io.Closer, so a caller takes a +// sandbox's sockets and the work behind them down as one list. +type closerFunc func() error + +func (f closerFunc) Close() error { return f() } diff --git a/internal/atunnel/dns_test.go b/internal/ateomnet/dns/relay_test.go similarity index 79% rename from internal/atunnel/dns_test.go rename to internal/ateomnet/dns/relay_test.go index 67cc3deb9..b6528af18 100644 --- a/internal/atunnel/dns_test.go +++ b/internal/ateomnet/dns/relay_test.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package atunnel +package dns import ( "bytes" @@ -30,54 +30,7 @@ import ( "github.com/google/go-cmp/cmp" ) -func TestResolvConfNameservers(t *testing.T) { - for _, tc := range []struct { - name string - content string - want []string - wantErr bool - }{ - { - name: "cluster resolv.conf", - content: "search ate-system.svc.cluster.local svc.cluster.local\nnameserver 10.96.0.10\noptions ndots:5\n", - want: []string{"10.96.0.10:53"}, - }, - { - name: "several, in order", - content: "nameserver 10.96.0.10\nnameserver 8.8.8.8\n", - want: []string{"10.96.0.10:53", "8.8.8.8:53"}, - }, - { - name: "comments and blanks", - content: "# generated\n\n nameserver 10.96.0.10 # cluster\n;nameserver 1.1.1.1\n", - want: []string{"10.96.0.10:53"}, - }, - {name: "no nameserver", content: "search cluster.local\n", wantErr: true}, - {name: "unparsable address", content: "nameserver not-an-ip\n", wantErr: true}, - } { - t.Run(tc.name, func(t *testing.T) { - path := filepath.Join(t.TempDir(), "resolv.conf") - if err := os.WriteFile(path, []byte(tc.content), 0o644); err != nil { - t.Fatal(err) - } - got, err := ResolvConfNameservers(path) - if tc.wantErr { - if err == nil { - t.Fatalf("ResolvConfNameservers() = %v, want an error", got) - } - return - } - if err != nil { - t.Fatalf("ResolvConfNameservers: %v", err) - } - if diff := cmp.Diff(tc.want, got); diff != "" { - t.Errorf("nameservers mismatch (-want +got):\n%s", diff) - } - }) - } -} - -func TestDNSRelayCancelsUDPExchange(t *testing.T) { +func TestRelayCancelsUDPExchange(t *testing.T) { // A resolver that receives the query and never answers, so the exchange is // blocked on the read when the context is canceled. silent, err := net.ListenPacket("udp", "127.0.0.1:0") @@ -101,7 +54,7 @@ func TestDNSRelayCancelsUDPExchange(t *testing.T) { // Two upstreams: a canceled exchange must not move on to the second. second := newFakeResolver(t, func(query []byte) []byte { return query }) - relay, err := NewDNSRelay([]string{silent.LocalAddr().String(), second}) + relay, err := NewRelayForUpstreams([]string{silent.LocalAddr().String(), second}) if err != nil { t.Fatal(err) } @@ -129,12 +82,12 @@ func TestDNSRelayCancelsUDPExchange(t *testing.T) { } } -func TestDNSRelayForwardsUDPVerbatim(t *testing.T) { +func TestRelayForwardsUDPVerbatim(t *testing.T) { upstream := newFakeResolver(t, func(query []byte) []byte { return append([]byte{0xff}, query...) }) - relay, err := NewDNSRelay([]string{upstream}) + relay, err := NewRelayForUpstreams([]string{upstream}) if err != nil { t.Fatal(err) } @@ -150,7 +103,7 @@ func TestDNSRelayForwardsUDPVerbatim(t *testing.T) { } } -func TestDNSRelayFallsBackToTheNextResolver(t *testing.T) { +func TestRelayFallsBackToTheNextResolver(t *testing.T) { dead, err := net.ListenPacket("udp", "127.0.0.1:0") if err != nil { t.Fatal(err) @@ -160,7 +113,7 @@ func TestDNSRelayFallsBackToTheNextResolver(t *testing.T) { dead.Close() live := newFakeResolver(t, func(query []byte) []byte { return []byte("answered") }) - relay, err := NewDNSRelay([]string{deadAddress, live}) + relay, err := NewRelayForUpstreams([]string{deadAddress, live}) if err != nil { t.Fatal(err) } @@ -174,12 +127,19 @@ func TestDNSRelayFallsBackToTheNextResolver(t *testing.T) { } } -func TestNewDNSRelayRejects(t *testing.T) { - if _, err := NewDNSRelay(nil); err == nil { - t.Error("NewDNSRelay(nil) succeeded; a relay with no upstream can answer nothing") +func TestNewRelayRejects(t *testing.T) { + empty := filepath.Join(t.TempDir(), "resolv.conf") + if err := os.WriteFile(empty, nil, 0o644); err != nil { + t.Fatal(err) } - if _, err := NewDNSRelay([]string{"10.96.0.10"}); err == nil { - t.Error("NewDNSRelay accepted an address with no port") + if _, err := NewRelay(empty); err == nil { + t.Error("NewRelay succeeded with an empty resolv.conf") + } + if _, err := NewRelayForUpstreams(nil); err == nil { + t.Error("NewRelayForUpstreams(nil) succeeded; a relay with no upstream can answer nothing") + } + if _, err := NewRelayForUpstreams([]string{"10.96.0.10"}); err == nil { + t.Error("NewRelayForUpstreams accepted an address with no port") } } @@ -207,7 +167,7 @@ func newFakeResolver(t *testing.T, respond func([]byte) []byte) string { } // serveRelayUDP runs the relay on a loopback socket and returns a connection to it. -func serveRelayUDP(t *testing.T, relay *DNSRelay) net.Conn { +func serveRelayUDP(t *testing.T, relay *Relay) net.Conn { t.Helper() pc, err := net.ListenPacket("udp", "127.0.0.1:0") if err != nil { @@ -215,7 +175,7 @@ func serveRelayUDP(t *testing.T, relay *DNSRelay) net.Conn { } ctx, cancel := context.WithCancel(context.Background()) t.Cleanup(cancel) - go func() { _ = relay.ServePacket(ctx, pc) }() + go func() { _ = relay.servePacket(ctx, pc) }() client, err := net.Dial("udp", pc.LocalAddr().String()) if err != nil { @@ -238,7 +198,7 @@ func readWithin(t *testing.T, conn net.Conn) []byte { return buf[:n] } -func TestDNSRelayForwardsAnswersLargerThanTheCommonBuffer(t *testing.T) { +func TestRelayForwardsAnswersLargerThanTheCommonBuffer(t *testing.T) { const size = 9000 answer := make([]byte, size) for i := range answer { @@ -246,7 +206,7 @@ func TestDNSRelayForwardsAnswersLargerThanTheCommonBuffer(t *testing.T) { } upstream := newFakeResolver(t, func([]byte) []byte { return answer }) - relay, err := NewDNSRelay([]string{upstream}) + relay, err := NewRelayForUpstreams([]string{upstream}) if err != nil { t.Fatal(err) } @@ -264,7 +224,7 @@ func TestDNSRelayForwardsAnswersLargerThanTheCommonBuffer(t *testing.T) { } } -func TestDNSRelayDropsQueriesBeyondItsInFlightLimit(t *testing.T) { +func TestRelayDropsQueriesBeyondItsInFlightLimit(t *testing.T) { // Hold concurrent queries to exercise the relay's limit. pc, err := net.ListenPacket("udp", "127.0.0.1:0") if err != nil { @@ -291,7 +251,7 @@ func TestDNSRelayDropsQueriesBeyondItsInFlightLimit(t *testing.T) { } }() - relay, err := NewDNSRelay([]string{pc.LocalAddr().String()}) + relay, err := NewRelayForUpstreams([]string{pc.LocalAddr().String()}) if err != nil { t.Fatal(err) } @@ -354,14 +314,14 @@ func newHeldTCPResolver(t *testing.T) (address string, accepted *atomic.Int64) { } // serveRelayTCP runs the relay's TCP side on a loopback listener. -func serveRelayTCP(t *testing.T, relay *DNSRelay, ctx context.Context) net.Addr { +func serveRelayTCP(t *testing.T, relay *Relay, ctx context.Context) net.Addr { t.Helper() lis, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { t.Fatal(err) } t.Cleanup(func() { lis.Close() }) - go func() { _ = relay.Serve(ctx, lis) }() + go func() { _ = relay.serveTCP(ctx, lis) }() return lis.Addr() } @@ -378,9 +338,9 @@ func waitFor(t *testing.T, what string, cond func() bool) { t.Fatalf("timed out waiting for %s", what) } -func TestDNSRelayRefusesTCPConnectionsBeyondItsLimit(t *testing.T) { +func TestRelayRefusesTCPConnectionsBeyondItsLimit(t *testing.T) { upstream, held := newHeldTCPResolver(t) - relay, err := NewDNSRelay([]string{upstream}) + relay, err := NewRelayForUpstreams([]string{upstream}) if err != nil { t.Fatal(err) } @@ -414,9 +374,9 @@ func TestDNSRelayRefusesTCPConnectionsBeyondItsLimit(t *testing.T) { } } -func TestDNSRelayClosesTCPConnectionsWhenServingEnds(t *testing.T) { +func TestRelayClosesTCPConnectionsWhenServingEnds(t *testing.T) { upstream, held := newHeldTCPResolver(t) - relay, err := NewDNSRelay([]string{upstream}) + relay, err := NewRelayForUpstreams([]string{upstream}) if err != nil { t.Fatal(err) } @@ -464,7 +424,7 @@ func dnsAnswer(query []byte, rcode byte) []byte { // A resolver that answers SERVFAIL has not answered the question, and the // sandbox can no longer consult the pod's other resolvers itself: it is given // the gateway as its only nameserver. The relay has to fail over for it. -func TestDNSRelayFailsOverOnServerFailure(t *testing.T) { +func TestRelayFailsOverOnServerFailure(t *testing.T) { var sickCalls, healthyCalls atomic.Int32 sick := newFakeResolver(t, func(query []byte) []byte { sickCalls.Add(1) @@ -475,7 +435,7 @@ func TestDNSRelayFailsOverOnServerFailure(t *testing.T) { return dnsAnswer(query, 0) }) - relay, err := NewDNSRelay([]string{sick, healthy}) + relay, err := NewRelayForUpstreams([]string{sick, healthy}) if err != nil { t.Fatal(err) } @@ -493,11 +453,11 @@ func TestDNSRelayFailsOverOnServerFailure(t *testing.T) { // With every resolver failing there is nothing better to return, and a real // SERVFAIL beats a timeout: the sandbox's resolver can act on it. -func TestDNSRelayReturnsServerFailureWhenAllFail(t *testing.T) { +func TestRelayReturnsServerFailureWhenAllFail(t *testing.T) { first := newFakeResolver(t, func(query []byte) []byte { return dnsAnswer(query, rcodeServFail) }) second := newFakeResolver(t, func(query []byte) []byte { return dnsAnswer(query, rcodeRefused) }) - relay, err := NewDNSRelay([]string{first, second}) + relay, err := NewRelayForUpstreams([]string{first, second}) if err != nil { t.Fatal(err) } @@ -512,7 +472,7 @@ func TestDNSRelayReturnsServerFailureWhenAllFail(t *testing.T) { // NXDOMAIN is an answer, not a failure: failing over would ask every resolver // about a name that does not exist. -func TestDNSRelayReturnsNXDomainWithoutFailover(t *testing.T) { +func TestRelayReturnsNXDomainWithoutFailover(t *testing.T) { const rcodeNXDomain = 3 var secondCalls atomic.Int32 first := newFakeResolver(t, func(query []byte) []byte { return dnsAnswer(query, rcodeNXDomain) }) @@ -521,7 +481,7 @@ func TestDNSRelayReturnsNXDomainWithoutFailover(t *testing.T) { return dnsAnswer(query, 0) }) - relay, err := NewDNSRelay([]string{first, second}) + relay, err := NewRelayForUpstreams([]string{first, second}) if err != nil { t.Fatal(err) } diff --git a/internal/ateomnet/dns/resolvconf.go b/internal/ateomnet/dns/resolvconf.go index f33b68bbf..26f6fd805 100644 --- a/internal/ateomnet/dns/resolvconf.go +++ b/internal/ateomnet/dns/resolvconf.go @@ -17,14 +17,50 @@ package dns import ( + "bufio" "errors" "fmt" "io/fs" + "net" "os" "path/filepath" "strings" ) +// resolvConfNameservers reads nameservers from resolv.conf as "host:53". +func resolvConfNameservers(path string) ([]string, error) { + f, err := os.Open(path) + if err != nil { + return nil, fmt.Errorf("dns: reading resolv.conf: %w", err) + } + defer f.Close() + + var out []string + scanner := bufio.NewScanner(f) + for scanner.Scan() { + line := strings.TrimSpace(scanner.Text()) + if i := strings.IndexAny(line, "#;"); i >= 0 { + line = strings.TrimSpace(line[:i]) + } + rest, ok := strings.CutPrefix(line, "nameserver") + if !ok { + continue + } + address := strings.TrimSpace(rest) + if address == "" || net.ParseIP(address) == nil { + continue + } + out = append(out, net.JoinHostPort(address, "53")) + } + if err := scanner.Err(); err != nil { + return nil, fmt.Errorf("dns: reading resolv.conf: %w", err) + } + if len(out) == 0 { + return nil, fmt.Errorf("dns: %s names no usable nameserver", path) + } + return out, nil +} + // SandboxResolvConf replaces the pod's nameservers with nameserver, the // address the sandbox's DNS is served on, while preserving the pod's search // domains and options for Kubernetes DNS. diff --git a/internal/ateomnet/dns/resolvconf_test.go b/internal/ateomnet/dns/resolvconf_test.go index 68afa5660..edafe46c5 100644 --- a/internal/ateomnet/dns/resolvconf_test.go +++ b/internal/ateomnet/dns/resolvconf_test.go @@ -23,6 +23,53 @@ import ( "github.com/google/go-cmp/cmp" ) +func TestResolvConfNameservers(t *testing.T) { + for _, tc := range []struct { + name string + content string + want []string + wantErr bool + }{ + { + name: "cluster resolv.conf", + content: "search ate-system.svc.cluster.local svc.cluster.local\nnameserver 10.96.0.10\noptions ndots:5\n", + want: []string{"10.96.0.10:53"}, + }, + { + name: "several, in order", + content: "nameserver 10.96.0.10\nnameserver 8.8.8.8\n", + want: []string{"10.96.0.10:53", "8.8.8.8:53"}, + }, + { + name: "comments and blanks", + content: "# generated\n\n nameserver 10.96.0.10 # cluster\n;nameserver 1.1.1.1\n", + want: []string{"10.96.0.10:53"}, + }, + {name: "no nameserver", content: "search cluster.local\n", wantErr: true}, + {name: "unparsable address", content: "nameserver not-an-ip\n", wantErr: true}, + } { + t.Run(tc.name, func(t *testing.T) { + path := filepath.Join(t.TempDir(), "resolv.conf") + if err := os.WriteFile(path, []byte(tc.content), 0o644); err != nil { + t.Fatal(err) + } + got, err := resolvConfNameservers(path) + if tc.wantErr { + if err == nil { + t.Fatalf("resolvConfNameservers() = %v, want an error", got) + } + return + } + if err != nil { + t.Fatalf("resolvConfNameservers: %v", err) + } + if diff := cmp.Diff(tc.want, got); diff != "" { + t.Errorf("nameservers mismatch (-want +got):\n%s", diff) + } + }) + } +} + func TestSandboxResolvConf(t *testing.T) { // The shape kubelet writes into a pod. pod := "nameserver 10.96.0.10\n" + diff --git a/internal/ateomnet/dns/serve.go b/internal/ateomnet/dns/serve.go deleted file mode 100644 index 0b5c136d4..000000000 --- a/internal/ateomnet/dns/serve.go +++ /dev/null @@ -1,86 +0,0 @@ -//go:build linux - -// Copyright 2026 Google LLC -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package dns - -import ( - "context" - "fmt" - "io" - "log/slog" - "net" - "strconv" - - "github.com/agent-substrate/substrate/internal/ateomnet/netns" -) - -// Server answers an actor's DNS. Satisfied by atunnel.DNSRelay; an interface -// so this package does not depend on it. -type Server interface { - ServePacket(ctx context.Context, pc net.PacketConn) error - Serve(ctx context.Context, listener net.Listener) error -} - -// Serve serves UDP and TCP DNS in the sandbox's local gateway namespace. -func Serve(ctx context.Context, relay Server, ns netns.Handle, port uint16) ([]io.Closer, []func(), error) { - // Bind the wildcard because the microVM tap's gateway address is added later. - address := net.JoinHostPort("0.0.0.0", strconv.Itoa(int(port))) - - var packet net.PacketConn - var stream net.Listener - if err := netns.Do(ctx, ns, func(context.Context) error { - pc, err := net.ListenPacket("udp", address) - if err != nil { - return fmt.Errorf("while opening the actor DNS socket: %w", err) - } - packet = pc - l, err := net.Listen("tcp", address) - if err != nil { - _ = pc.Close() - return fmt.Errorf("while opening the actor DNS listener: %w", err) - } - stream = l - return nil - }); err != nil { - return nil, nil, err - } - - // Detached from the activation RPC's context but cancelable: the relay's - // capacity is the worker's, so teardown must drop queries still in flight. - serveCtx, stopServing := context.WithCancel(context.WithoutCancel(ctx)) - serve := []func(){ - func() { - if err := relay.ServePacket(serveCtx, packet); err != nil { - slog.WarnContext(ctx, "Actor DNS socket stopped", slog.Any("err", err)) - } - }, - func() { - if err := relay.Serve(serveCtx, stream); err != nil { - slog.WarnContext(ctx, "Actor DNS listener stopped", slog.Any("err", err)) - } - }, - } - // Cancel first: closing the sockets alone leaves the queries already being - // resolved holding the relay. - closers := []io.Closer{closerFunc(func() error { stopServing(); return nil }), packet, stream} - return closers, serve, nil -} - -// closerFunc adapts a cancel function to io.Closer, so a caller takes a -// sandbox's sockets and the work behind them down as one list. -type closerFunc func() error - -func (f closerFunc) Close() error { return f() } diff --git a/internal/ateomnet/dns/serve_linux_test.go b/internal/ateomnet/dns/serve_linux_test.go index 61c175d38..7573b0ed5 100644 --- a/internal/ateomnet/dns/serve_linux_test.go +++ b/internal/ateomnet/dns/serve_linux_test.go @@ -18,7 +18,6 @@ package dns import ( "context" - "net" "testing" "time" @@ -26,21 +25,6 @@ import ( "github.com/agent-substrate/substrate/internal/roottest" ) -// stoppableDNS records that its serving contexts were canceled. -type stoppableDNS struct{ packet, stream chan struct{} } - -func (d *stoppableDNS) ServePacket(ctx context.Context, pc net.PacketConn) error { - <-ctx.Done() - close(d.packet) - return pc.Close() -} - -func (d *stoppableDNS) Serve(ctx context.Context, l net.Listener) error { - <-ctx.Done() - close(d.stream) - return l.Close() -} - func TestClosingSandboxDNSStopsServing(t *testing.T) { roottest.Require(t, "creates network namespaces") const nsName = "dns-teardown-test" @@ -53,26 +37,30 @@ func TestClosingSandboxDNSStopsServing(t *testing.T) { _ = netns.RemoveNamed(nsName) }() - relay := &stoppableDNS{packet: make(chan struct{}), stream: make(chan struct{})} - closers, serve, err := Serve(context.Background(), relay, ns, 53) + relay, err := NewRelayForUpstreams([]string{"127.0.0.1:53"}) if err != nil { t.Fatal(err) } + closers, serve, err := relay.Serve(context.Background(), ns) + if err != nil { + t.Fatal(err) + } + stopped := make(chan struct{}, len(serve)) for _, fn := range serve { - go fn() + go func() { + fn() + stopped <- struct{}{} + }() } for _, c := range closers { _ = c.Close() } - for _, tc := range []struct { - name string - stopped chan struct{} - }{{"UDP", relay.packet}, {"TCP", relay.stream}} { + for range serve { select { - case <-tc.stopped: + case <-stopped: case <-time.After(5 * time.Second): - t.Errorf("%s serving outlived the sandbox's sockets", tc.name) + t.Fatal("DNS serving outlived the sandbox's sockets") } } } diff --git a/internal/ateomnet/netns/doc.go b/internal/ateomnet/netns/doc.go new file mode 100644 index 000000000..a4d7bf5b0 --- /dev/null +++ b/internal/ateomnet/netns/doc.go @@ -0,0 +1,16 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package netns contains Linux network namespace-related functionality. +package netns diff --git a/internal/ateomnet/sandbox.go b/internal/ateomnet/sandbox.go index c94422da3..83748d59a 100644 --- a/internal/ateomnet/sandbox.go +++ b/internal/ateomnet/sandbox.go @@ -68,9 +68,6 @@ type SandboxNetworkConfig struct { // EgressPort is where atunnel serves this actor. Every TCP connection the // actor makes is redirected to it, whatever port it was aimed at. EgressPort uint16 - - // DNSPort is where the DNS relay answers, on the sandbox's default gateway. - DNSPort uint16 } // SetupSandboxNetwork creates isolated networking with fixed sandbox addresses. @@ -353,7 +350,7 @@ type SandboxSession struct { // ServeSandbox builds a sandbox's network and serves egress and DNS from its // gateway namespace. A nil server leaves that unserved, which fails closed. -func ServeSandbox(ctx context.Context, cfg SandboxNetworkConfig, egress egressServer, resolver dns.Server) (_ *SandboxSession, retErr error) { +func ServeSandbox(ctx context.Context, cfg SandboxNetworkConfig, egress egressServer, resolver *dns.Relay) (_ *SandboxSession, retErr error) { network, err := SetupSandboxNetwork(ctx, cfg) if err != nil { return nil, err @@ -367,7 +364,7 @@ func ServeSandbox(ctx context.Context, cfg SandboxNetworkConfig, egress egressSe var serve []func() if resolver != nil { - closers, serveDNS, err := dns.Serve(ctx, resolver, network.GatewayNetNS, cfg.DNSPort) + closers, serveDNS, err := resolver.Serve(ctx, network.GatewayNetNS) if err != nil { return nil, err } diff --git a/internal/ateomnet/sandbox_linux_test.go b/internal/ateomnet/sandbox_linux_test.go index a07414a3d..4a39ac9d1 100644 --- a/internal/ateomnet/sandbox_linux_test.go +++ b/internal/ateomnet/sandbox_linux_test.go @@ -338,31 +338,26 @@ func TestCleanupClosesEachDescriptorOnce(t *testing.T) { } } -// slowDNS holds its serving goroutines open until released, so a test can tell -// whether Close waits for them or merely closes their sockets. -type slowDNS struct{ release chan struct{} } +// slowEgress holds its serving goroutine open until released, so a test can +// tell whether Close waits for serving to stop or merely closes the sockets. +type slowEgress struct{ release chan struct{} } -func (d *slowDNS) ServePacket(ctx context.Context, pc net.PacketConn) error { - <-ctx.Done() - <-d.release - return pc.Close() -} - -func (d *slowDNS) Serve(ctx context.Context, l net.Listener) error { - <-ctx.Done() - <-d.release - return l.Close() +func (e *slowEgress) Bind(string) (func(context.Context, net.Listener) error, error) { + return func(_ context.Context, l net.Listener) error { + <-e.release + return l.Close() + }, nil } // Close's contract is that serving has stopped when it returns, not just that -// the sockets are shut: a caller tearing an actor down needs the relay's +// the sockets are shut: a caller tearing an actor down needs the worker's // capacity back. func TestSessionCloseWaitsForServingToStop(t *testing.T) { roottest.Require(t, "creates network namespaces") - relay := &slowDNS{release: make(chan struct{})} + egress := &slowEgress{release: make(chan struct{})} session, err := ServeSandbox(context.Background(), SandboxNetworkConfig{ - ActorUID: "close-waits", EgressPort: testEgressPort, DNSPort: 53, - }, nil, relay) + ActorUID: "close-waits", EgressPort: testEgressPort, + }, egress, nil) if err != nil { t.Fatal(err) } @@ -371,11 +366,11 @@ func TestSessionCloseWaitsForServingToStop(t *testing.T) { go func() { returned <- session.Close(context.Background()) }() select { case <-returned: - t.Fatal("Close returned while the relay was still serving") + t.Fatal("Close returned while the session was still serving") case <-time.After(250 * time.Millisecond): } - close(relay.release) + close(egress.release) select { case err := <-returned: if err != nil { @@ -391,11 +386,11 @@ func TestSessionCloseWaitsForServingToStop(t *testing.T) { // activation of this actor. func TestSessionCloseReportsAWaitItCouldNotFinish(t *testing.T) { roottest.Require(t, "creates network namespaces") - relay := &slowDNS{release: make(chan struct{})} - defer close(relay.release) + egress := &slowEgress{release: make(chan struct{})} + defer close(egress.release) session, err := ServeSandbox(context.Background(), SandboxNetworkConfig{ - ActorUID: "close-deadline", EgressPort: testEgressPort, DNSPort: 53, - }, nil, relay) + ActorUID: "close-deadline", EgressPort: testEgressPort, + }, egress, nil) if err != nil { t.Fatal(err) } diff --git a/internal/ateomtunnel/ateomtunnel.go b/internal/ateomtunnel/ateomtunnel.go index cd83987a7..c434d65c3 100644 --- a/internal/ateomtunnel/ateomtunnel.go +++ b/internal/ateomtunnel/ateomtunnel.go @@ -28,6 +28,7 @@ import ( "net/url" "time" + "github.com/agent-substrate/substrate/internal/ateomnet/dns" "github.com/agent-substrate/substrate/internal/atunnel" "github.com/agent-substrate/substrate/internal/installdefaults" "github.com/agent-substrate/substrate/internal/nodepath" @@ -94,7 +95,7 @@ type Tunnel struct { // actor network's transparent TCP redirect. EgressPort uint16 // DNSRelay answers the sandbox's DNS from inside its own namespace. - DNSRelay *atunnel.DNSRelay + DNSRelay *dns.Relay cfg Config } @@ -107,18 +108,13 @@ func Start(ctx context.Context, cfg Config, upstream string) (*Tunnel, error) { if err != nil { return nil, fmt.Errorf("while parsing atunnel upstream: %w", err) } - // The pod's own resolvers, so an actor resolves exactly what the worker - // resolves, cluster DNS included. - nameservers, err := atunnel.ResolvConfNameservers(resolvConfPath) - if err != nil { - return nil, fmt.Errorf("while reading the worker pod resolvers: %w", err) - } - dnsRelay, err := atunnel.NewDNSRelay(nameservers) - if err != nil { - return nil, fmt.Errorf("while building the actor DNS relay: %w", err) - } - slog.InfoContext(ctx, "Actor DNS relay ready", slog.Any("upstreams", nameservers)) + dnsRelay, err := dns.NewRelay(resolvConfPath) + if err != nil { + return nil, fmt.Errorf("configuring DNS relay: %w", err) + } + + // Ingress ingress, err := atunnel.NewServer(atunnel.Config{ CredentialBundlePath: cfg.CredentialBundle, TrustBundlePath: cfg.TrustBundle, @@ -126,7 +122,7 @@ func Start(ctx context.Context, cfg Config, upstream string) (*Tunnel, error) { Upstream: upstreamURL, }) if err != nil { - return nil, fmt.Errorf("while configuring atunnel: %w", err) + return nil, fmt.Errorf("configuring ingress: %w", err) } if err := serve(ctx, "atunnel", cfg.ListenAddress, ingress.Serve); err != nil { return nil, err @@ -135,15 +131,23 @@ func Start(ctx context.Context, cfg Config, upstream string) (*Tunnel, error) { return nil, err } + // Egress egress, err := atunnel.NewEgress(atunnel.TCPOriginalDestination) if err != nil { - return nil, fmt.Errorf("while configuring atunnel egress: %w", err) + return nil, fmt.Errorf("configuring egress: %w", err) } egressPort, err := atunnel.EgressPort(cfg.EgressListenAddress) if err != nil { return nil, err } - return &Tunnel{Ingress: ingress, Egress: egress, EgressPort: egressPort, DNSRelay: dnsRelay, cfg: cfg}, nil + + return &Tunnel{ + Ingress: ingress, + Egress: egress, + EgressPort: egressPort, + DNSRelay: dnsRelay, + cfg: cfg}, + nil } // serve listens on address and runs serveFn on it, exiting the process if it diff --git a/internal/ateomtunnel/ateomtunnel_test.go b/internal/ateomtunnel/ateomtunnel_test.go index bf432e7c8..7f9954a31 100644 --- a/internal/ateomtunnel/ateomtunnel_test.go +++ b/internal/ateomtunnel/ateomtunnel_test.go @@ -198,7 +198,7 @@ func TestStartErrors(t *testing.T) { for name, tc := range map[string]struct { edit func(t *testing.T, cfg *Config) - wantErr string + wantErr bool }{ "no resolvers": { edit: func(t *testing.T, _ *Config) { @@ -208,23 +208,23 @@ func TestStartErrors(t *testing.T) { } resolvConfPath = empty }, - wantErr: "worker pod resolvers", + wantErr: true, }, "missing credentials": { edit: func(t *testing.T, cfg *Config) { cfg.CredentialBundle = filepath.Join(t.TempDir(), "absent.pem") }, - wantErr: "while configuring atunnel", + wantErr: true, }, "ingress address in use": { edit: func(_ *testing.T, cfg *Config) { cfg.ListenAddress = inUse.Addr().String() }, - wantErr: "while opening atunnel listener", + wantErr: true, }, "CONNECT address in use": { edit: func(_ *testing.T, cfg *Config) { cfg.ConnectListenAddress = inUse.Addr().String() }, - wantErr: "while opening atunnel CONNECT listener", + wantErr: true, }, "egress address without a port": { edit: func(_ *testing.T, cfg *Config) { cfg.EgressListenAddress = "0.0.0.0" }, - wantErr: "egress listen address", + wantErr: true, }, } { t.Run(name, func(t *testing.T) { @@ -233,16 +233,16 @@ func TestStartErrors(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() _, err := Start(ctx, cfg, "http://169.254.17.2:80") - if err == nil || !strings.Contains(err.Error(), tc.wantErr) { - t.Fatalf("Start error = %v, want one containing %q", err, tc.wantErr) + if (err != nil) != tc.wantErr { + t.Fatalf("Start error = %v, wantErr = %v", err, tc.wantErr) } }) } t.Run("bad upstream", func(t *testing.T) { _, err := Start(context.Background(), testConfig(t), "http://[::1") - if err == nil || !strings.Contains(err.Error(), "atunnel upstream") { - t.Fatalf("Start error = %v, want an upstream error", err) + if err == nil { + t.Fatalf("Start error = %v, want an error", err) } }) }