From 05a137a28fb69f9f0d5b255f9751c18a0296f090 Mon Sep 17 00:00:00 2001 From: Bo-Cheng Chu Date: Mon, 12 Oct 2026 21:17:25 -0700 Subject: [PATCH] e2e: cover actor UDP and raw TCP egress Combines two egress e2e changes, rebased onto main: - UDP egress: a udpecho origin publishing 53, 443 and 9999 onto one listener, and TestActorUDPEgress. - Raw TCP egress: a tcpecho origin that greets or stays silent, a /tcp endpoint in the egress demo, TCP readiness probes and batched ServerPod deploys, and TestActorEgressRawTCPBlocked. Each sandbox's only route out is its gateway address, where a relay serves DNS, so UDP to any other address, port 53 included, goes nowhere. TestActorUDPEgress first runs the probe binary as a plain pod outside any sandbox, which must get an echo on every port, so a broken origin or Service cannot make the drops pass. It retries while kube-proxy has not yet programmed the new Service's UDP ports. The probe actor must then send all of its attempts on every port and see each end in a read timeout, the signature of a silent drop rather than a failed dial or send or an ICMP error, and must still resolve the origin's Service name. The probe grows --listen so it can serve on an unprivileged port as that pod. Actor TCP that is neither HTTP nor TLS is refused by the gateway, and what the actor sees depends on who speaks first. If the actor waits for the server, the gateway has nothing to classify, gives up on its listener timeout and closes the connection with no bytes returned. If the actor sends first, the gateway treats bytes that are not TLS as cleartext HTTP and its HTTP handler answers with its own 400 Bad Request. The test covers both: against an in-cluster origin that greets and one that echoes, the actor must never get the greeting or its payload back, and the gateway must have logged the CONNECT, so the refusal is the gateway's. When the actor waits, the connection must also be closed with no bytes, and on Envoy both the establish and close records of the tunnel must show down_bytes=0. When it sends a text line or binary bytes, it must get HTTP/1.1 400 Bad Request. The demo's /tcp endpoint grows a skipBanner option so the client-speaks- first cases write immediately rather than waiting out the gateway's listener timeout, and reports closed when the peer closes or resets the connection, which an empty read alone cannot tell from a timeout. Other adjustments made while combining: - The tests create their actors with an allow-all policy and use the atespace-aware assertEgressGatewayConnect. - renderServerPod returns manifest text, so renderServerPodDocs and CreateServerService use it directly. - serveLocal in the tcpecho tests waits for its serve goroutine in cleanup, so it can no longer log after the test returns (a race under -race). --- demos/egress/main.go | 148 +++++++++- demos/egress/main_test.go | 200 ++++++++++++++ internal/e2e/fixtures/probe/main.go | 142 +++++++++- internal/e2e/fixtures/probe/main_test.go | 26 ++ internal/e2e/fixtures/probe/resolve_test.go | 71 +++++ internal/e2e/fixtures/probe/udpecho_test.go | 125 +++++++++ internal/e2e/fixtures/serverpod.yaml.tmpl | 2 + internal/e2e/fixtures/testserver/main.go | 4 +- internal/e2e/fixtures/testserver/tcpecho.go | 119 ++++++++ .../e2e/fixtures/testserver/tcpecho_test.go | 158 +++++++++++ internal/e2e/fixtures/testserver/udpecho.go | 91 ++++++ .../e2e/fixtures/testserver/udpecho_test.go | 101 +++++++ internal/e2e/manifest.go | 46 +++- internal/e2e/serverpod.go | 177 ++++++++++-- internal/e2e/serverpod_test.go | 256 ++++++++++++++--- .../e2e/suites/networking/networking_test.go | 205 ++++++++++++++ .../e2e/suites/networking/udpegress_test.go | 259 ++++++++++++++++++ 17 files changed, 2053 insertions(+), 77 deletions(-) create mode 100644 internal/e2e/fixtures/probe/resolve_test.go create mode 100644 internal/e2e/fixtures/probe/udpecho_test.go create mode 100644 internal/e2e/fixtures/testserver/tcpecho.go create mode 100644 internal/e2e/fixtures/testserver/tcpecho_test.go create mode 100644 internal/e2e/fixtures/testserver/udpecho.go create mode 100644 internal/e2e/fixtures/testserver/udpecho_test.go create mode 100644 internal/e2e/suites/networking/udpegress_test.go diff --git a/demos/egress/main.go b/demos/egress/main.go index da40483663..c8b486f58d 100644 --- a/demos/egress/main.go +++ b/demos/egress/main.go @@ -13,8 +13,10 @@ // limitations under the License. // Command egress is a small HTTP service for demonstrating per-Actor egress -// policy. It accepts a URL, fetches it, and returns the upstream response, and -// on a second endpoint it makes gRPC calls and returns what came back. +// policy. It accepts a URL, fetches it, and returns the upstream response. On +// /grpc it makes gRPC calls and returns what came back, and on /tcp it opens a +// raw TCP connection, so that egress can be exercised with something other than +// HTTP. package main import ( @@ -34,6 +36,7 @@ import ( "os" "strconv" "sync" + "syscall" "time" "github.com/gorilla/websocket" @@ -210,6 +213,7 @@ func newHandler(client *http.Client) http.Handler { mux.HandleFunc("/mux", handleMux) mux.HandleFunc("/websocket", handleWebSocket) mux.HandleFunc("/grpc", handleGRPC) + mux.HandleFunc("/tcp", handleTCPProbe) return mux } @@ -862,6 +866,146 @@ func writeGRPCFailure(w http.ResponseWriter, rpc string, err error) { }) } +// tcpProbeRequest asks for one raw TCP exchange. +type tcpProbeRequest struct { + Address string `json:"address"` + Send string `json:"send,omitempty"` + ReadBytes int `json:"readBytes,omitempty"` + Timeout string `json:"timeout,omitempty"` + // SkipBanner writes Send as soon as the connection is up instead of first + // waiting for the peer to speak, so a path that classifies traffic by the + // client's first bytes sees them right away. + SkipBanner bool `json:"skipBanner,omitempty"` +} + +type tcpProbeResponse struct { + // Banner is whatever the peer sent before being spoken to. + Banner string `json:"banner,omitempty"` + Received string `json:"received,omitempty"` + // Closed reports that the peer closed or reset the connection, as opposed + // to a read that only ran out its timeout. + Closed bool `json:"closed,omitempty"` + Error string `json:"error,omitempty"` +} + +// handleTCPProbe opens a TCP connection and, unless SkipBanner is set, reads +// before it writes. +func handleTCPProbe(w http.ResponseWriter, r *http.Request) { + const defaultProbeTimeout = 5 * time.Second + // Enough for an SSH identification string or a test banner. + const defaultProbeReadBytes = 512 + const maxProbeReadBytes = 8 << 10 // 8 KiB + + if r.Method != http.MethodPost { + w.Header().Set("Allow", http.MethodPost) + writeJSON(w, http.StatusMethodNotAllowed, tcpProbeResponse{Error: "method must be POST"}) + return + } + + var input tcpProbeRequest + decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, maxRequestBody)) + if err := decoder.Decode(&input); err != nil { + writeJSON(w, http.StatusBadRequest, tcpProbeResponse{Error: fmt.Sprintf("invalid JSON payload: %v", err)}) + return + } + if _, _, err := net.SplitHostPort(input.Address); err != nil { + writeJSON(w, http.StatusBadRequest, tcpProbeResponse{Error: fmt.Sprintf("address must be host:port: %v", err)}) + return + } + + timeout := defaultProbeTimeout + if input.Timeout != "" { + parsed, err := time.ParseDuration(input.Timeout) + if err != nil { + writeJSON(w, http.StatusBadRequest, tcpProbeResponse{Error: fmt.Sprintf("invalid timeout: %v", err)}) + return + } + timeout = parsed + } + readBytes := input.ReadBytes + if readBytes <= 0 { + readBytes = defaultProbeReadBytes + } + readBytes = min(readBytes, maxProbeReadBytes) + + dialer := net.Dialer{Timeout: timeout} + connection, err := dialer.DialContext(r.Context(), "tcp", input.Address) + if err != nil { + writeJSON(w, http.StatusBadGateway, tcpProbeResponse{Error: fmt.Sprintf("dialing %s: %v", input.Address, err)}) + return + } + defer connection.Close() + + // Read before writing anything at all, so an empty banner really does mean + // the peer stayed silent. + var response tcpProbeResponse + if !input.SkipBanner { + banner, closed, err := readWithTimeout(connection, readBytes, timeout) + if err != nil { + writeJSON(w, http.StatusBadGateway, tcpProbeResponse{Error: fmt.Sprintf("reading banner from %s: %v", input.Address, err)}) + return + } + response.Banner = string(banner) + if closed { + response.Closed = true + writeJSON(w, http.StatusOK, response) + return + } + } + if input.Send == "" { + writeJSON(w, http.StatusOK, response) + return + } + + if err := connection.SetWriteDeadline(time.Now().Add(timeout)); err != nil { + writeJSON(w, http.StatusBadGateway, tcpProbeResponse{Error: fmt.Sprintf("setting write deadline: %v", err)}) + return + } + if _, err := io.WriteString(connection, input.Send); err != nil { + if peerClosed(err) { + response.Closed = true + writeJSON(w, http.StatusOK, response) + return + } + writeJSON(w, http.StatusBadGateway, tcpProbeResponse{Error: fmt.Sprintf("writing to %s: %v", input.Address, err)}) + return + } + received, closed, err := readWithTimeout(connection, readBytes, timeout) + if err != nil { + writeJSON(w, http.StatusBadGateway, tcpProbeResponse{Error: fmt.Sprintf("reading reply from %s: %v", input.Address, err)}) + return + } + response.Received = string(received) + response.Closed = closed + writeJSON(w, http.StatusOK, response) +} + +// readWithTimeout returns the bytes of a single read, capped at limit, and +// whether the peer closed or reset the connection. A peer that says nothing +// within timeout yields no bytes, closed false and no error, since silence is a +// legitimate answer to "does this peer speak first?". +func readWithTimeout(connection net.Conn, limit int, timeout time.Duration) ([]byte, bool, error) { + if err := connection.SetReadDeadline(time.Now().Add(timeout)); err != nil { + return nil, false, fmt.Errorf("setting read deadline: %w", err) + } + buffer := make([]byte, limit) + n, err := connection.Read(buffer) + switch { + case err == nil, errors.Is(err, os.ErrDeadlineExceeded): + return buffer[:n], false, nil + case peerClosed(err): + return buffer[:n], true, nil + default: + return nil, false, err + } +} + +// peerClosed reports whether err means the peer closed or reset the +// connection. +func peerClosed(err error) bool { + return errors.Is(err, io.EOF) || errors.Is(err, syscall.ECONNRESET) || errors.Is(err, syscall.EPIPE) +} + func validateURL(raw string) error { parsed, err := url.Parse(raw) if err != nil { diff --git a/demos/egress/main_test.go b/demos/egress/main_test.go index 82c1dc72d8..40ba30d196 100644 --- a/demos/egress/main_test.go +++ b/demos/egress/main_test.go @@ -850,3 +850,203 @@ func TestSlowHeadersInvalidRequests(t *testing.T) { } }) } + +// TestTCPProbeServerSpeaksFirst covers the ordering the probe exists for: the +// peer's greeting is reported without the probe having written anything, and +// the reply to Send comes back separately. +func TestTCPProbeServerSpeaksFirst(t *testing.T) { + const banner = "TESTBANNER/1.0\r\n" + address := startTestPeer(t, func(connection net.Conn) { + if _, err := io.WriteString(connection, banner); err != nil { + return + } + buffer := make([]byte, 64) + n, err := connection.Read(buffer) + if err != nil { + return + } + _, _ = connection.Write(buffer[:n]) + }) + + got := probe(t, tcpProbeRequest{Address: address, Send: "ping", Timeout: "5s"}, http.StatusOK) + if got.Banner != banner { + t.Errorf("banner = %q, want %q", got.Banner, banner) + } + if got.Received != "ping" { + t.Errorf("received = %q, want %q", got.Received, "ping") + } +} + +// TestTCPProbeSilentPeer verifies that a client-speaks-first peer yields an empty banner and a 200. +func TestTCPProbeSilentPeer(t *testing.T) { + address := startTestPeer(t, func(connection net.Conn) { + buffer := make([]byte, 64) + n, err := connection.Read(buffer) + if err != nil { + return + } + _, _ = connection.Write(buffer[:n]) + }) + + got := probe(t, tcpProbeRequest{Address: address, Send: "ping", Timeout: "250ms"}, http.StatusOK) + if got.Banner != "" { + t.Errorf("banner = %q, want empty for a peer that does not speak first", got.Banner) + } + if got.Received != "ping" { + t.Errorf("received = %q, want %q", got.Received, "ping") + } +} + +// TestTCPProbeSkipBannerWritesFirst verifies that SkipBanner sends before +// reading: the peer only answers once it has read, and the probe's timeout is +// far shorter than any banner wait would need, so the bytes must already be +// on the wire when the peer reads. +func TestTCPProbeSkipBannerWritesFirst(t *testing.T) { + firstRead := make(chan time.Duration, 1) + address := startTestPeer(t, func(connection net.Conn) { + accepted := time.Now() + buffer := make([]byte, 64) + n, err := connection.Read(buffer) + if err != nil { + return + } + firstRead <- time.Since(accepted) + _, _ = connection.Write(buffer[:n]) + }) + + got := probe(t, tcpProbeRequest{Address: address, Send: "ping", SkipBanner: true, Timeout: "2s"}, http.StatusOK) + if got.Banner != "" { + t.Errorf("banner = %q, want empty when the banner read is skipped", got.Banner) + } + if got.Received != "ping" { + t.Errorf("received = %q, want %q", got.Received, "ping") + } + // Reading first would have held the write back for the whole 2s timeout. + if elapsed := <-firstRead; elapsed >= time.Second { + t.Errorf("peer waited %v for the first bytes, want them sent without a banner read", elapsed) + } +} + +func TestTCPProbeReadBytesCapsTheBanner(t *testing.T) { + address := startTestPeer(t, func(connection net.Conn) { + _, _ = io.WriteString(connection, "0123456789") + }) + + got := probe(t, tcpProbeRequest{Address: address, ReadBytes: 4, Timeout: "5s"}, http.StatusOK) + if got.Banner != "0123" { + t.Errorf("banner = %q, want %q", got.Banner, "0123") + } +} + +func TestTCPProbeDialFailure(t *testing.T) { + // Port 0 is not connectable, so this fails without depending on which + // ports happen to be free. + got := probe(t, tcpProbeRequest{Address: "127.0.0.1:0", Timeout: "2s"}, http.StatusBadGateway) + if got.Error == "" { + t.Error("error = empty, want a dial failure") + } +} + +func TestTCPProbeInvalidRequests(t *testing.T) { + tests := []struct { + name string + method string + body string + status int + }{ + {name: "method", method: http.MethodGet, body: `{}`, status: http.StatusMethodNotAllowed}, + {name: "malformed JSON", method: http.MethodPost, body: `{`, status: http.StatusBadRequest}, + {name: "address without port", method: http.MethodPost, body: `{"address":"example.com"}`, status: http.StatusBadRequest}, + {name: "invalid timeout", method: http.MethodPost, body: `{"address":"127.0.0.1:9","timeout":"soon"}`, status: http.StatusBadRequest}, + } + + handler := newHandler(http.DefaultClient) + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + recorder := httptest.NewRecorder() + request := httptest.NewRequest(test.method, "/tcp", strings.NewReader(test.body)) + handler.ServeHTTP(recorder, request) + if recorder.Code != test.status { + t.Errorf("status = %d, want %d; body = %s", recorder.Code, test.status, recorder.Body.String()) + } + }) + } +} + +// TestTCPProbeReportsClosed separates a peer that closes the connection from +// one that only stays quiet, which an empty read alone cannot. +func TestTCPProbeReportsClosed(t *testing.T) { + tests := []struct { + name string + serve func(net.Conn) + skipBanner bool + wantClosed bool + }{ + {name: "closes before speaking", serve: func(c net.Conn) { _ = c.Close() }, wantClosed: true}, + {name: "closes after reading", skipBanner: true, wantClosed: true, serve: func(c net.Conn) { + _, _ = c.Read(make([]byte, 64)) + _ = c.Close() + }}, + {name: "stays silent", skipBanner: true, wantClosed: false, serve: func(c net.Conn) { + _, _ = c.Read(make([]byte, 64)) + time.Sleep(time.Second) + }}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + address := startTestPeer(t, tt.serve) + got := probe(t, tcpProbeRequest{Address: address, Send: "ping\r\n", SkipBanner: tt.skipBanner, Timeout: "300ms"}, http.StatusOK) + if got.Closed != tt.wantClosed { + t.Errorf("closed = %v, want %v (response %+v)", got.Closed, tt.wantClosed, got) + } + if got.Banner != "" || got.Received != "" { + t.Errorf("response = %+v, want no bytes", got) + } + }) + } +} + +func probe(t *testing.T, input tcpProbeRequest, wantStatus int) tcpProbeResponse { + t.Helper() + payload, err := json.Marshal(input) + if err != nil { + t.Fatal(err) + } + recorder := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, "/tcp", strings.NewReader(string(payload))) + newHandler(http.DefaultClient).ServeHTTP(recorder, request) + + if recorder.Code != wantStatus { + t.Fatalf("status = %d, want %d; body = %s", recorder.Code, wantStatus, recorder.Body.String()) + } + var got tcpProbeResponse + if err := json.NewDecoder(recorder.Body).Decode(&got); err != nil { + t.Fatalf("decoding response: %v", err) + } + return got +} + +// startTestPeer listens on loopback and hands each connection to serve. It +// returns the address to probe. +func startTestPeer(t *testing.T, serve func(net.Conn)) string { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listening: %v", err) + } + t.Cleanup(func() { listener.Close() }) + + go func() { + for { + connection, err := listener.Accept() + if err != nil { + return + } + go func() { + defer connection.Close() + serve(connection) + }() + } + }() + return listener.Addr().String() +} diff --git a/internal/e2e/fixtures/probe/main.go b/internal/e2e/fixtures/probe/main.go index e1c3a113b0..9e25c6ebd8 100644 --- a/internal/e2e/fixtures/probe/main.go +++ b/internal/e2e/fixtures/probe/main.go @@ -26,6 +26,7 @@ import ( "crypto/tls" "crypto/x509" "encoding/json" + "flag" "fmt" "io" "log" @@ -452,6 +453,128 @@ func fetch(w http.ResponseWriter, r *http.Request) { writeJSON(w, resp) } +// resolveTimeout bounds a lookup. The resolver retries on its own schedule +// when nothing answers, so without a bound a request made while DNS is broken +// hangs for whatever /etc/resolv.conf adds up to instead of reporting it. +const resolveTimeout = 10 * time.Second + +// resolveResult is the /resolve response body. +type resolveResult struct { + Host string `json:"host"` + // Addresses is what the name resolved to. Returned rather than a bare + // "it worked" so a caller can check the answer is the one it deployed, + // which a stale or hijacked resolver would fail. + Addresses []string `json:"addresses,omitempty"` + // Error is the lookup failure. A failure is a result -- it is what an + // actor that cannot reach DNS looks like -- so it is not an HTTP error. + Error string `json:"error,omitempty"` +} + +// resolve looks ?host= up through the sandbox's own resolver, which reaches the +// cluster's DNS over UDP. It is the narrowest check that actor DNS still +// leaves the worker pod: unlike /fetch it opens no connection afterwards, so +// nothing about the answer depends on the egress path. +func resolve(w http.ResponseWriter, r *http.Request) { + host := r.URL.Query().Get("host") + result := resolveResult{Host: host} + if host == "" { + result.Error = "missing host parameter" + writeJSON(w, result) + return + } + + ctx, cancel := context.WithTimeout(r.Context(), resolveTimeout) + defer cancel() + addresses, err := net.DefaultResolver.LookupHost(ctx, host) + if err != nil { + result.Error = err.Error() + } else { + result.Addresses = addresses + } + writeJSON(w, result) +} + +// UDP probe budget. A datagram that is dropped on its way out of the worker +// pod is silent -- nothing answers, and nothing reports the drop -- so the wait +// IS the measurement, and a delivered-but-slow echo read as a drop would be a +// false accusation. Retrying inside the budget covers the other direction: UDP +// may lose a single datagram for reasons that have nothing to do with policy, +// and one lost datagram must not read as "blocked". +const ( + udpEchoAttempts = 5 + udpEchoWait = time.Second +) + +// udpEchoResult is the /udpecho response body. +type udpEchoResult struct { + Addr string `json:"addr"` + // Echoed reports whether the datagram came back. False is a normal, + // assertable outcome -- it is what a dropped datagram looks like from in + // here -- so it is never an HTTP error. + Echoed bool `json:"echoed"` + // Attempts is how many datagrams were sent before giving up or getting one + // back, so a flaky pass (echoed on the last try) is visible in the log + // rather than indistinguishable from a clean one. + Attempts int `json:"attempts"` + // Error is the last send or receive failure. A read timeout is the + // expected one when Echoed is false; an ICMP-driven "connection refused" + // means the datagram reached a host that had nothing listening, which is a + // different finding and worth not confusing with a drop. + Error string `json:"error,omitempty"` +} + +// udpecho sends datagrams to ?addr= and reports whether one was echoed back, +// so a test can assert which of an actor's UDP destinations actually leave the +// worker pod. It needs an echo server on the far end: reachable-but-silent and +// dropped are the same observation from inside the sandbox. +func udpecho(w http.ResponseWriter, r *http.Request) { + addr := r.URL.Query().Get("addr") + result := udpEchoResult{Addr: addr} + if addr == "" { + result.Error = "missing addr parameter" + writeJSON(w, result) + return + } + + conn, err := net.Dial("udp", addr) + if err != nil { + result.Error = "dialing " + addr + ": " + err.Error() + writeJSON(w, result) + return + } + defer conn.Close() + + // A token unique to this request, so a straggler from an earlier one + // cannot be counted as this one's echo. + payload := fmt.Sprintf("probe-%d", time.Now().UnixNano()) + for result.Attempts < udpEchoAttempts { + result.Attempts++ + if _, err := conn.Write([]byte(payload)); err != nil { + result.Error = "sending to " + addr + ": " + err.Error() + continue + } + if err := conn.SetReadDeadline(time.Now().Add(udpEchoWait)); err != nil { + result.Error = "setting the read deadline: " + err.Error() + break + } + // Oversized on purpose: a wrong-length reply should be reported as the + // mismatch it is, not silently truncated into a match. + buf := make([]byte, 512) + n, err := conn.Read(buf) + if err != nil { + result.Error = "waiting for the echo from " + addr + ": " + err.Error() + continue + } + if string(buf[:n]) == payload { + result.Echoed = true + result.Error = "" + break + } + result.Error = fmt.Sprintf("echo from %s was %q, want %q", addr, buf[:n], payload) + } + writeJSON(w, result) +} + func writeJSON(w http.ResponseWriter, v any) { w.Header().Set("Content-Type", "application/json") if err := json.NewEncoder(w).Encode(v); err != nil { @@ -472,16 +595,33 @@ func main() { mux := http.NewServeMux() mux.HandleFunc("/whoami", whoami) mux.HandleFunc("/fetch", fetch) + mux.HandleFunc("/udpecho", udpecho) + mux.HandleFunc("/resolve", resolve) mux.HandleFunc("/readfile", readfile) mux.HandleFunc("/writefile", writefile) mux.HandleFunc("/resources", resources) mux.HandleFunc("/capabilities", capabilities) mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) }) - const addr = ":80" + addr, err := listenAddress(os.Args[1:]) + if err != nil { + log.Fatalf("probe: %v", err) + } log.Printf("probe listening on %s", addr) server := &http.Server{Addr: addr, Handler: mux} if err := server.ListenAndServe(); err != nil { log.Fatalf("probe server: %v", err) } } + +// listenAddress parses --listen. An actor runs the probe with no arguments and +// gets :80; a suite that runs it as a plain pod, outside any sandbox, passes an +// unprivileged port the way e2e.ServerPod does for every server. +func listenAddress(args []string) (string, error) { + flags := flag.NewFlagSet("probe", flag.ContinueOnError) + listen := flags.String("listen", ":80", "address to serve on") + if err := flags.Parse(args); err != nil { + return "", err + } + return *listen, nil +} diff --git a/internal/e2e/fixtures/probe/main_test.go b/internal/e2e/fixtures/probe/main_test.go index cf10505e29..765b3b2e73 100644 --- a/internal/e2e/fixtures/probe/main_test.go +++ b/internal/e2e/fixtures/probe/main_test.go @@ -363,3 +363,29 @@ func TestCapabilityNamesTable(t *testing.T) { } } } + +// TestListenAddress covers the two ways the probe is started: bare, inside an +// actor, and with the --listen e2e.ServerPod appends when it runs as a pod. +func TestListenAddress(t *testing.T) { + tests := []struct { + name string + args []string + want string + wantErr bool + }{ + {name: "actor default", args: nil, want: ":80"}, + {name: "server pod", args: []string{"--listen=:8080"}, want: ":8080"}, + {name: "unknown flag", args: []string{"--port=8080"}, wantErr: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := listenAddress(tt.args) + if (err != nil) != tt.wantErr { + t.Fatalf("listenAddress(%q) error = %v, wantErr %v", tt.args, err, tt.wantErr) + } + if got != tt.want { + t.Errorf("listenAddress(%q) = %q, want %q", tt.args, got, tt.want) + } + }) + } +} diff --git a/internal/e2e/fixtures/probe/resolve_test.go b/internal/e2e/fixtures/probe/resolve_test.go new file mode 100644 index 0000000000..e8b830dcc5 --- /dev/null +++ b/internal/e2e/fixtures/probe/resolve_test.go @@ -0,0 +1,71 @@ +// 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 main + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "slices" + "testing" +) + +// callResolve invokes the handler for host and decodes its response. +func callResolve(t *testing.T, host string) resolveResult { + t.Helper() + recorder := httptest.NewRecorder() + resolve(recorder, httptest.NewRequest(http.MethodGet, "/resolve?host="+host, nil)) + + if recorder.Code != http.StatusOK { + t.Fatalf("/resolve for %q answered HTTP %d, want 200: a failed lookup is a result, not an error", host, recorder.Code) + } + var out resolveResult + if err := json.Unmarshal(recorder.Body.Bytes(), &out); err != nil { + t.Fatalf("decoding the /resolve response: %v (body %q)", err, recorder.Body) + } + return out +} + +// TestResolve covers the answer shape the UDP-egress e2e suite asserts on: the +// addresses come back, so a test can check the name resolved to the Service it +// deployed rather than settling for "the lookup returned". +// +// localhost, because it resolves from /etc/hosts with no nameserver involved -- +// this pins the handler, not the machine's DNS. +func TestResolve(t *testing.T) { + got := callResolve(t, "localhost") + if got.Error != "" { + t.Fatalf("resolving localhost failed: %s", got.Error) + } + if !slices.Contains(got.Addresses, "127.0.0.1") && !slices.Contains(got.Addresses, "::1") { + t.Errorf("localhost resolved to %v, want a loopback address", got.Addresses) + } + if got.Host != "localhost" { + t.Errorf("Host = %q, want the name that was asked for", got.Host) + } +} + +// TestResolveMissingHost covers the caller error. It must not look like a +// resolver failure to a test asserting DNS is broken -- both carry Error, but +// only this one carries it without a lookup having happened. +func TestResolveMissingHost(t *testing.T) { + got := callResolve(t, "") + if got.Error == "" { + t.Error("Error is empty with no host to look up") + } + if len(got.Addresses) != 0 { + t.Errorf("Addresses = %v with no host to look up, want none", got.Addresses) + } +} diff --git a/internal/e2e/fixtures/probe/udpecho_test.go b/internal/e2e/fixtures/probe/udpecho_test.go new file mode 100644 index 0000000000..8e14516f03 --- /dev/null +++ b/internal/e2e/fixtures/probe/udpecho_test.go @@ -0,0 +1,125 @@ +// 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 main + +import ( + "encoding/json" + "net" + "net/http" + "net/http/httptest" + "testing" + "time" +) + +// TestActorUDPEgress in the networking e2e suite reads Echoed=false as "the +// worker pod dropped the datagram", so a probe that reported it for its own +// reasons would manufacture a passing security assertion. These tests pin both +// answers against loopback, where nothing is dropping anything. + +// callUDPEcho invokes the handler for addr and decodes its response. +func callUDPEcho(t *testing.T, addr string) udpEchoResult { + t.Helper() + recorder := httptest.NewRecorder() + udpecho(recorder, httptest.NewRequest(http.MethodGet, "/udpecho?addr="+addr, nil)) + + if recorder.Code != http.StatusOK { + t.Fatalf("/udpecho for %q answered HTTP %d, want 200: a drop is a result, not an error", addr, recorder.Code) + } + var out udpEchoResult + if err := json.Unmarshal(recorder.Body.Bytes(), &out); err != nil { + t.Fatalf("decoding the /udpecho response: %v (body %q)", err, recorder.Body) + } + return out +} + +// TestUDPEchoAnswered covers the positive case, which is the e2e test's gate: +// an origin that answers must be reported as echoed. +func TestUDPEchoAnswered(t *testing.T) { + conn, err := net.ListenPacket("udp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listening on loopback: %v", err) + } + stopped := make(chan struct{}) + go func() { + defer close(stopped) + buf := make([]byte, 512) + for { + n, from, err := conn.ReadFrom(buf) + if err != nil { + return + } + if _, err := conn.WriteTo(buf[:n], from); err != nil { + return + } + } + }() + t.Cleanup(func() { + conn.Close() + <-stopped + }) + + got := callUDPEcho(t, conn.LocalAddr().String()) + if !got.Echoed { + t.Errorf("Echoed = false against a live echo server after %d attempts: %s", got.Attempts, got.Error) + } + if got.Attempts != 1 { + t.Errorf("Attempts = %d against a live echo server, want the first datagram to be answered", got.Attempts) + } + if got.Error != "" { + t.Errorf("Error = %q on a successful echo, want it cleared", got.Error) + } +} + +// TestUDPEchoUnanswered covers what the e2e test's drop assertions actually +// observe: every attempt spent, nothing back, and an error that says why. +func TestUDPEchoUnanswered(t *testing.T) { + // A discard address: nothing routes there, so no reply and no ICMP either. + // Blackholed rather than refused is exactly the shape a dropped datagram + // has, and it is what the probe must report as not echoed. + addr := "192.0.2.1:9" + + start := time.Now() + got := callUDPEcho(t, addr) + if got.Echoed { + t.Fatalf("Echoed = true for %s, which nothing answers", addr) + } + if got.Attempts != udpEchoAttempts { + t.Errorf("Attempts = %d, want every one of the %d spent before giving up", got.Attempts, udpEchoAttempts) + } + if got.Error == "" { + t.Error("Error is empty on a datagram that was never answered, leaving a failing e2e assertion nothing to explain itself with") + } + // The retries have to actually wait: a budget spent instantly would read a + // slow-but-delivered echo as a drop. + if elapsed := time.Since(start); elapsed < udpEchoWait { + t.Errorf("gave up after %v, want at least one full %v wait", elapsed, udpEchoWait) + } +} + +// TestUDPEchoMissingAddr covers the caller error, which must not look like a +// drop to a test asserting one -- hence the error rather than a bare +// Echoed=false. +func TestUDPEchoMissingAddr(t *testing.T) { + got := callUDPEcho(t, "") + if got.Echoed { + t.Error("Echoed = true with no addr to send to") + } + if got.Error == "" { + t.Error("Error is empty with no addr to send to") + } + if got.Attempts != 0 { + t.Errorf("Attempts = %d with no addr to send to, want 0", got.Attempts) + } +} diff --git a/internal/e2e/fixtures/serverpod.yaml.tmpl b/internal/e2e/fixtures/serverpod.yaml.tmpl index 0159e22154..96aef10cff 100644 --- a/internal/e2e/fixtures/serverpod.yaml.tmpl +++ b/internal/e2e/fixtures/serverpod.yaml.tmpl @@ -45,6 +45,7 @@ ${ARGS} containerPort: ${TARGET_PORT} # The HTTPS listener, present when the ServerPod sets TLSPort. ${TLS_CONTAINER_PORT} +${UDP_CONTAINER_PORTS} # The suite starts dialing as soon as it has an address, so without a # readiness gate the first connection can land before the listener exists # and come back refused -- which reads as a failure of whatever sits between @@ -86,3 +87,4 @@ spec: port: ${PORT} targetPort: ${TARGET_PORT} ${TLS_SERVICE_PORT} +${UDP_SERVICE_PORTS} diff --git a/internal/e2e/fixtures/testserver/main.go b/internal/e2e/fixtures/testserver/main.go index 177c89e0c1..947570c1ec 100644 --- a/internal/e2e/fixtures/testserver/main.go +++ b/internal/e2e/fixtures/testserver/main.go @@ -19,8 +19,10 @@ // // testserver grpc --listen=:50051 a cleartext HTTP/2 gRPC echo origin // testserver http --listen=:8080 an HTTP(S) origin serving /healthz and /echoheaders +// testserver tcpecho --listen=:2222 a raw TCP echo origin, greeting optional // testserver egressprobe --listen=:8080 a client that drives the egress gateway // testserver websocket --listen=:8080 a websocket server that responds to PINGs +// testserver udpecho --listen=:8053 a UDP echo origin, with TCP /healthz beside it // // Each pod runs exactly one subcommand, so the wire behavior of any given pod is // unchanged from when these were separate binaries -- the grpc pod still speaks @@ -45,7 +47,7 @@ func main() { Use: "testserver", Short: "Multi-mode helper server for the egress e2e suites.", } - root.AddCommand(newGRPCCmd(), newHTTPCmd(), newEgressProbeCmd(), newWebsocketCmd()) + root.AddCommand(newGRPCCmd(), newHTTPCmd(), newTCPEchoCmd(), newEgressProbeCmd(), newWebsocketCmd(), newUDPEchoCmd()) if err := root.Execute(); err != nil { fmt.Fprintln(os.Stderr, err) os.Exit(1) diff --git a/internal/e2e/fixtures/testserver/tcpecho.go b/internal/e2e/fixtures/testserver/tcpecho.go new file mode 100644 index 0000000000..7ed9e9ccb2 --- /dev/null +++ b/internal/e2e/fixtures/testserver/tcpecho.go @@ -0,0 +1,119 @@ +// 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 main + +import ( + "errors" + "fmt" + "io" + "log" + "net" + "os" + "time" + + "github.com/spf13/cobra" +) + +// newTCPEchoCmd is a raw TCP origin for the egress suites: it echoes whatever +// it is sent, and with --banner it writes that greeting before reading a byte. +// +// The greeting is what this mode exists for. HTTP and gRPC both have the client +// send the first bytes, so neither notices an egress path that waits for +// downstream data before dialing upstream -- the shape a server-speaks-first +// protocol like SSH breaks on. Whether to greet is fixed at startup rather than +// chosen per connection because the server has to decide before it has read +// anything, which is what speaking first means: a suite covering both orders +// deploys two of these rather than sending two kinds of request. +// +// The banner is a flag rather than a constant here so the suite that asserts on +// it is the same place that supplies it, leaving no second copy to drift. +func newTCPEchoCmd() *cobra.Command { + var ( + listenAddress string + banner string + ) + cmd := &cobra.Command{ + Use: "tcpecho", + Short: "Serve a raw TCP echo origin, optionally greeting each peer first.", + Args: cobra.NoArgs, + RunE: func(_ *cobra.Command, _ []string) error { + listener, err := net.Listen("tcp", listenAddress) + if err != nil { + return fmt.Errorf("listening on %s: %w", listenAddress, err) + } + log.Printf("testserver tcpecho: serving on %s, banner %q", listener.Addr(), banner) + return serveTCPEcho(listener, banner) + }, + } + cmd.Flags().StringVar(&listenAddress, "listen", ":2222", "Address the echo origin listens on.") + cmd.Flags().StringVar(&banner, "banner", "", "Greeting written on accept, before reading anything. Empty stays silent until spoken to.") + return cmd +} + +// serveTCPEcho accepts until the listener fails, which it treats as terminal: a +// pod that stopped accepting but stayed up would fail a test as though the +// egress path had dropped the connection. +func serveTCPEcho(listener net.Listener, banner string) error { + defer listener.Close() + for { + conn, err := listener.Accept() + if err != nil { + return fmt.Errorf("accepting on %s: %w", listener.Addr(), err) + } + go echoTCP(conn, banner) + } +} + +// echoTCP greets the peer when banner is non-empty, then echoes until it goes +// away. +func echoTCP(conn net.Conn, banner string) { + // A stuck peer must not hold a goroutine and a socket forever. Long enough + // that a tunneled round trip is never the thing that trips it. + const idleTimeout = 60 * time.Second + + defer conn.Close() + + if banner != "" { + if err := conn.SetWriteDeadline(time.Now().Add(idleTimeout)); err != nil { + log.Printf("testserver tcpecho: setting write deadline: %v", err) + return + } + if _, err := io.WriteString(conn, banner); err != nil { + log.Printf("testserver tcpecho: writing banner: %v", err) + return + } + } + + buffer := make([]byte, 4<<10) + for { + if err := conn.SetDeadline(time.Now().Add(idleTimeout)); err != nil { + return + } + n, err := conn.Read(buffer) + if n > 0 { + if _, writeErr := conn.Write(buffer[:n]); writeErr != nil { + return + } + } + if err != nil { + // A peer that hangs up or falls silent is the normal end of a + // probe, not something worth a line in the pod's log. + if !errors.Is(err, io.EOF) && !errors.Is(err, os.ErrDeadlineExceeded) { + log.Printf("testserver tcpecho: reading from peer: %v", err) + } + return + } + } +} diff --git a/internal/e2e/fixtures/testserver/tcpecho_test.go b/internal/e2e/fixtures/testserver/tcpecho_test.go new file mode 100644 index 0000000000..7db47c98dc --- /dev/null +++ b/internal/e2e/fixtures/testserver/tcpecho_test.go @@ -0,0 +1,158 @@ +// 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 main + +import ( + "errors" + "io" + "net" + "os" + "testing" + "time" +) + +// The egress suite reads whether this origin spoke first as evidence about the +// tunnel, so getting it wrong here would be read as a broken gateway. These +// tests pin the two orders against a loopback listener, with no gateway +// involved. + +// serveLocal starts the echo origin on loopback and returns its address. +func serveLocal(t *testing.T, banner string) string { + t.Helper() + + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listening on loopback: %v", err) + } + // The serve goroutine may only log while the test is running, so cleanup + // closes the listener and then waits for it to return. + done := make(chan struct{}) + t.Cleanup(func() { + _ = listener.Close() + <-done + }) + go func() { + defer close(done) + if err := serveTCPEcho(listener, banner); err != nil && !errors.Is(err, net.ErrClosed) { + t.Logf("serving: %v", err) + } + }() + return listener.Addr().String() +} + +// dialLocalEcho connects to address and closes the connection with the test. +func dialLocalEcho(t *testing.T, address string) net.Conn { + t.Helper() + conn, err := net.DialTimeout("tcp", address, 5*time.Second) + if err != nil { + t.Fatalf("dialing %s: %v", address, err) + } + t.Cleanup(func() { _ = conn.Close() }) + if err := conn.SetDeadline(time.Now().Add(10 * time.Second)); err != nil { + t.Fatalf("setting deadline: %v", err) + } + return conn +} + +// readAvailable returns the bytes of a single read, treating a timeout as an +// empty answer rather than a failure: silence is what the quiet origin is +// supposed to produce. +func readAvailable(t *testing.T, conn net.Conn, within time.Duration) string { + t.Helper() + if err := conn.SetReadDeadline(time.Now().Add(within)); err != nil { + t.Fatalf("setting read deadline: %v", err) + } + buffer := make([]byte, 512) + n, err := conn.Read(buffer) + if err != nil && !errors.Is(err, os.ErrDeadlineExceeded) && !errors.Is(err, io.EOF) { + t.Fatalf("reading: %v", err) + } + return string(buffer[:n]) +} + +// TestTCPEchoGreetsBeforeReading is the server-speaks-first order: the banner +// has to arrive without the peer having written anything. +func TestTCPEchoGreetsBeforeReading(t *testing.T) { + const banner = "TESTBANNER/1.0\r\n" + conn := dialLocalEcho(t, serveLocal(t, banner)) + + if got := readAvailable(t, conn, 5*time.Second); got != banner { + t.Fatalf("banner = %q, want %q", got, banner) + } + + if _, err := io.WriteString(conn, "ping"); err != nil { + t.Fatalf("writing: %v", err) + } + if got := readAvailable(t, conn, 5*time.Second); got != "ping" { + t.Fatalf("echo = %q, want %q", got, "ping") + } +} + +// TestTCPEchoStaysSilentWithoutBanner is the client-speaks-first order. An +// origin that greeted anyway would make the other subtest pass for the wrong +// reason, so the silence is asserted rather than assumed. +func TestTCPEchoStaysSilentWithoutBanner(t *testing.T) { + conn := dialLocalEcho(t, serveLocal(t, "")) + + if got := readAvailable(t, conn, 250*time.Millisecond); got != "" { + t.Fatalf("origin volunteered %q before being spoken to, want silence", got) + } + + if _, err := io.WriteString(conn, "ping"); err != nil { + t.Fatalf("writing: %v", err) + } + if got := readAvailable(t, conn, 5*time.Second); got != "ping" { + t.Fatalf("echo = %q, want %q", got, "ping") + } +} + +// TestTCPEchoServesConcurrentPeers covers the accept loop handing each +// connection to its own goroutine: a serial one would leave the second peer +// waiting on the first, which reads as a stalled tunnel. +func TestTCPEchoServesConcurrentPeers(t *testing.T) { + const banner = "HELLO\r\n" + address := serveLocal(t, banner) + + first := dialLocalEcho(t, address) + second := dialLocalEcho(t, address) + + for i, conn := range []net.Conn{first, second} { + if got := readAvailable(t, conn, 5*time.Second); got != banner { + t.Fatalf("banner on connection %d = %q, want %q", i, got, banner) + } + } + + if _, err := io.WriteString(first, "ping"); err != nil { + t.Fatalf("writing: %v", err) + } + if got := readAvailable(t, first, 5*time.Second); got != "ping" { + t.Fatalf("echo = %q, want %q", got, "ping") + } +} + +// TestTCPEchoListenFailure covers the subcommand reporting a bad --listen +// rather than serving nothing: a pod that came up and never listened would fail +// its readiness probe with nothing in the log to say why. +func TestTCPEchoListenFailure(t *testing.T) { + cmd := newTCPEchoCmd() + cmd.SetArgs([]string{"--listen=not-an-address"}) + cmd.SetOut(io.Discard) + cmd.SetErr(io.Discard) + cmd.SilenceUsage = true + + if err := cmd.Execute(); err == nil { + t.Fatal("tcpecho --listen=not-an-address returned no error, want a listen failure") + } +} diff --git a/internal/e2e/fixtures/testserver/udpecho.go b/internal/e2e/fixtures/testserver/udpecho.go new file mode 100644 index 0000000000..08ea5a25d3 --- /dev/null +++ b/internal/e2e/fixtures/testserver/udpecho.go @@ -0,0 +1,91 @@ +// 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 main + +import ( + "fmt" + "log" + "net" + "net/http" + "time" + + "github.com/spf13/cobra" +) + +// udpEchoMaxDatagram bounds one read. Callers send a short token, so a larger +// datagram is not one of theirs and truncating it costs nothing. +const udpEchoMaxDatagram = 1500 + +// serveUDPEcho echoes every datagram back to its sender until conn is closed, +// returning the read error that stopped it. +func serveUDPEcho(conn net.PacketConn) error { + buf := make([]byte, udpEchoMaxDatagram) + for { + n, from, err := conn.ReadFrom(buf) + if err != nil { + return err + } + if _, err := conn.WriteTo(buf[:n], from); err != nil { + // One unanswered sender is not a reason to stop answering the + // rest, and the caller reads a missing echo as a drop anyway. + log.Printf("testserver udpecho: replying to %s: %v", from, err) + } + } +} + +// newUDPEchoCmd is the UDP origin an Actor's datagrams land on, for a test that +// asserts which of them are forwarded out of the worker pod at all. An echo +// rather than a sink: a datagram that is dropped on the way out and one that is +// delivered to a server that never answers look identical from inside the +// sandbox, so the origin has to answer for a delivery to be observable. +// +// It also serves /healthz over TCP on the same port number, because the shared +// ServerPod manifest gates readiness on an HTTP GET and kubelet has no UDP +// probe. The two listeners do not collide: they are different protocols. +func newUDPEchoCmd() *cobra.Command { + var listenAddress string + cmd := &cobra.Command{ + Use: "udpecho", + Short: "Echo UDP datagrams, and answer /healthz on the same port over TCP.", + Args: cobra.NoArgs, + RunE: func(_ *cobra.Command, _ []string) error { + conn, err := net.ListenPacket("udp", listenAddress) + if err != nil { + return fmt.Errorf("listening for UDP on %s: %w", listenAddress, err) + } + defer conn.Close() + go func() { + log.Printf("testserver udpecho: echoing UDP on %s", listenAddress) + log.Printf("testserver udpecho: UDP echo stopped: %v", serveUDPEcho(conn)) + }() + + mux := http.NewServeMux() + mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + }) + + server := &http.Server{ + Addr: listenAddress, + Handler: mux, + ReadHeaderTimeout: 10 * time.Second, + WriteTimeout: 2 * time.Minute, + } + log.Printf("testserver udpecho: serving readiness on %s", listenAddress) + return server.ListenAndServe() + }, + } + cmd.Flags().StringVar(&listenAddress, "listen", ":8053", "Address the UDP echo and its HTTP readiness endpoint listen on.") + return cmd +} diff --git a/internal/e2e/fixtures/testserver/udpecho_test.go b/internal/e2e/fixtures/testserver/udpecho_test.go new file mode 100644 index 0000000000..b9a2775336 --- /dev/null +++ b/internal/e2e/fixtures/testserver/udpecho_test.go @@ -0,0 +1,101 @@ +// 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 main + +import ( + "bytes" + "net" + "testing" + "time" +) + +// The UDP-egress e2e assertion reads a missing echo as a dropped datagram, so +// an origin that quietly fails to answer would read as a policy that is working. +// These tests pin the answering against a loopback listener, where no sandbox +// and no nftables rule are involved. + +// serveLocalEcho starts the echo on loopback and returns its address. +func serveLocalEcho(t *testing.T) string { + t.Helper() + conn, err := net.ListenPacket("udp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listening on loopback: %v", err) + } + stopped := make(chan struct{}) + go func() { + defer close(stopped) + // Closing conn is how the cleanup below ends the loop, so the error + // this returns is the expected one and says nothing. + _ = serveUDPEcho(conn) + }() + // Joining the goroutine, not just closing under it, so no echo server + // outlives the test that started it. + t.Cleanup(func() { + conn.Close() + <-stopped + }) + return conn.LocalAddr().String() +} + +// exchange sends payload to addr and returns the reply, or "" if none arrived +// before the deadline. +func exchange(t *testing.T, addr string, payload []byte) []byte { + t.Helper() + conn, err := net.Dial("udp", addr) + if err != nil { + t.Fatalf("dialing %s: %v", addr, err) + } + defer conn.Close() + if _, err := conn.Write(payload); err != nil { + t.Fatalf("sending to %s: %v", addr, err) + } + if err := conn.SetReadDeadline(time.Now().Add(10 * time.Second)); err != nil { + t.Fatalf("setting the read deadline: %v", err) + } + buf := make([]byte, udpEchoMaxDatagram) + n, err := conn.Read(buf) + if err != nil { + t.Fatalf("waiting for the echo from %s: %v", addr, err) + } + return buf[:n] +} + +func TestServeUDPEcho(t *testing.T) { + addr := serveLocalEcho(t) + + // Byte-for-byte, because the probe matches the echo against the exact + // token it sent: anything else is reported as no echo at all. + payload := []byte("probe-1234567890") + if got := exchange(t, addr, payload); !bytes.Equal(got, payload) { + t.Errorf("echo = %q, want %q", got, payload) + } + + // The e2e probe retries within its budget, so the server has to answer + // more than the first datagram it ever sees. + second := []byte("probe-2345678901") + if got := exchange(t, addr, second); !bytes.Equal(got, second) { + t.Errorf("echo of a second datagram = %q, want %q", got, second) + } +} + +// TestServeUDPEchoEmptyDatagram covers the degenerate payload: a zero-length +// datagram is a datagram, and answering it keeps "nothing came back" meaning +// only that nothing came back. +func TestServeUDPEchoEmptyDatagram(t *testing.T) { + addr := serveLocalEcho(t) + if got := exchange(t, addr, nil); len(got) != 0 { + t.Errorf("echo of an empty datagram = %q, want empty", got) + } +} diff --git a/internal/e2e/manifest.go b/internal/e2e/manifest.go index 5714973ba6..11d315f5d7 100644 --- a/internal/e2e/manifest.go +++ b/internal/e2e/manifest.go @@ -23,10 +23,16 @@ import ( "sigs.k8s.io/yaml" ) -// renderManifest substitutes placeholders into the manifest template at relPath -// (repo-relative), writes the result into the test's temp dir and returns that -// path. Both an apply and a later delete can then consume the same file, with -// no shell involved. +// renderManifest substitutes placeholders into the manifest template at relPath, +// writes the result into the test's temp dir and returns that path. +func renderManifest(t *testing.T, relPath string, inline, blocks map[string]string) string { + t.Helper() + name := strings.TrimSuffix(filepath.Base(relPath), ".tmpl") + return writeManifest(t, name, renderManifestText(t, relPath, inline, blocks)) +} + +// renderManifestText is renderManifest without the file, for a caller that +// concatenates several rendered templates into one manifest before applying it. // // Templates carry two kinds of ${...} placeholder: // @@ -36,7 +42,7 @@ import ( // the whole line with it — the same trick the installer's renderer plays // with its drop list. Requiring the placeholder to be the whole line // is what lets a comment mention one without being deleted. -func renderManifest(t *testing.T, relPath string, inline, blocks map[string]string) string { +func renderManifestText(t *testing.T, relPath string, inline, blocks map[string]string) string { t.Helper() root, err := FindRepoRoot() if err != nil { @@ -60,12 +66,18 @@ func renderManifest(t *testing.T, relPath string, inline, blocks map[string]stri } out = append(out, line) } + return strings.Join(out, "\n") +} - rendered := strings.TrimSuffix(filepath.Join(t.TempDir(), filepath.Base(relPath)), ".tmpl") - if err := os.WriteFile(rendered, []byte(strings.Join(out, "\n")), 0o644); err != nil { - t.Fatalf("writing rendered manifest %s: %v", rendered, err) +// writeManifest writes content into a fresh temp dir under name and returns the +// path. +func writeManifest(t *testing.T, name, content string) string { + t.Helper() + path := filepath.Join(t.TempDir(), name) + if err := os.WriteFile(path, []byte(content), 0o644); err != nil { + t.Fatalf("writing rendered manifest %s: %v", path, err) } - return rendered + return path } // yamlListBlock renders items as `key:` followed by their YAML, every line @@ -78,16 +90,28 @@ func renderManifest(t *testing.T, relPath string, inline, blocks map[string]stri // what keeps a fragment honest: a misspelled field is a compile error here, // where in a template it would apply cleanly and do nothing. func yamlListBlock[T any](t *testing.T, key string, items []T, indent int) string { + t.Helper() + if len(items) == 0 { + return "" + } + pad := strings.Repeat(" ", indent) + return pad + key + ":\n" + yamlItemsBlock(t, items, indent) +} + +// yamlItemsBlock is yamlListBlock without the `key:` line, for a block +// placeholder that extends a list the template has already opened — where the +// key is spelled in the template beside the entries that are always there. +func yamlItemsBlock[T any](t *testing.T, items []T, indent int) string { t.Helper() if len(items) == 0 { return "" } raw, err := yaml.Marshal(items) if err != nil { - t.Fatalf("marshaling %s for the manifest: %v", key, err) + t.Fatalf("marshaling manifest list entries: %v", err) } pad := strings.Repeat(" ", indent) - out := []string{pad + key + ":"} + var out []string for line := range strings.SplitSeq(strings.TrimRight(string(raw), "\n"), "\n") { out = append(out, pad+line) } diff --git a/internal/e2e/serverpod.go b/internal/e2e/serverpod.go index 33c062ef06..706eb0b134 100644 --- a/internal/e2e/serverpod.go +++ b/internal/e2e/serverpod.go @@ -18,7 +18,6 @@ import ( "context" "fmt" "net" - "os" "path" "slices" "strconv" @@ -29,6 +28,7 @@ import ( corev1 "k8s.io/api/core/v1" discoveryv1 "k8s.io/api/discovery/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/intstr" "sigs.k8s.io/yaml" ) @@ -64,6 +64,20 @@ type ServerPod struct { // publish a port the container -- uid 65532, every capability dropped -- // cannot bind, such as 80; the Service maps Port down to it. TargetPort int + // UDPPorts are Service ports published over UDP alongside the TCP port + // above, every one of them mapped to the same UDPTargetPort. Several + // published ports onto one listener is the point: a caller that expects a + // datagram to arrive on one port and not another can then attribute the + // difference to the port, not to two servers behaving differently. + // + // A port here can be one the container could never bind, 53 included: the + // Service maps it down to UDPTargetPort, and kube-proxy does that in the + // host netns, after whatever the sender's own netns did with the datagram. + UDPPorts []int + // UDPTargetPort is the UDP port the binary listens on, defaulting to the + // TCP target port -- which is where a server that takes one --listen flag + // for both protocols ends up. + UDPTargetPort int // TLSPort, when set, publishes the HTTPS listener testserver's http and // websocket subcommands open on container port 8443 when given a // certificate, beside the cleartext one on Port. The readiness probe then @@ -78,6 +92,10 @@ type ServerPod struct { // an HTTP GET. A gRPC server answers an HTTP request with a protocol error, // so a server speaking grpc must set this and register the health service. GRPCProbe bool + // TCPProbe asks kubelet to probe by opening a TCP connection and closing + // it, which is all readiness can mean for a server that speaks neither HTTP + // nor gRPC. Ignored when GRPCProbe is set. + TCPProbe bool // HealthPath is the HTTP readiness path, defaulting to /healthz. Ignored // when GRPCProbe is set. HealthPath string @@ -146,7 +164,15 @@ type Server struct { // Address is the host:port to dial the server at. func (s Server) Address() string { - return net.JoinHostPort(s.ClusterIP, strconv.Itoa(s.Port)) + return s.AddressOnPort(s.Port) +} + +// AddressOnPort is the host:port to dial one of the server's other published +// ports at -- a ServerPod.UDPPorts entry, say. The port is not checked against +// what was published: an address for a port the Service does not carry is +// exactly what a test asserting unreachability wants. +func (s Server) AddressOnPort(port int) string { + return net.JoinHostPort(s.ClusterIP, strconv.Itoa(port)) } // TLSAddress is the host:port to dial the server's HTTPS listener at. @@ -163,38 +189,100 @@ func (s Server) SvcName() string { // DeployServerPod builds spec's image, applies the shared server manifest, waits // for readiness and returns the address to dial. +func DeployServerPod(t *testing.T, ctx context.Context, spec ServerPod) Server { + t.Helper() + return DeployServerPods(t, ctx, spec)[0] +} + +// DeployServerPods deploys several servers at once, into one namespace and +// through one ko invocation, and returns their addresses in the order given. +// +// Worth having rather than a loop over DeployServerPod, because a serial loop +// pays for each server end to end: ko starts, builds and pushes, then the pod +// schedules, pulls and goes ready, and only then does the next one begin. One +// apply overlaps all of that. ko caches builds and pushes by image reference +// within an invocation, so servers sharing an ImportPath -- which is the common +// case, since one testserver binary backs them all -- build once; and every pod +// is admitted at the same instant, so the second readiness wait has usually +// already been satisfied by the time the first returns. // // It registers no cleanup: everything the manifest creates is namespaced, so it // goes with the namespace CreateNamespace made — and, on failure, is retained // with it for `kubectl logs`. -func DeployServerPod(t *testing.T, ctx context.Context, spec ServerPod) Server { +func DeployServerPods(t *testing.T, ctx context.Context, specs ...ServerPod) []Server { t.Helper() + if len(specs) == 0 { + t.Fatalf("DeployServerPods was given no servers to deploy") + } if _, err := CheckEnv("KO_DOCKER_REPO"); err != nil { t.Fatalf("CheckEnv failed: %v", err) } - namespace := spec.Namespace - if namespace == "" { - namespace = CreateNamespace(t).Name + namespace := serverPodNamespace(t, specs) + + koApply(t, writeManifest(t, "serverpods.yaml", renderServerPods(t, specs, namespace))) + + servers := make([]Server, 0, len(specs)) + for _, spec := range specs { + WaitForPodReady(t, ctx, namespace, spec.Name, serverPodReadyTimeout) + + service, err := GetClients().K8s.CoreV1().Services(namespace).Get(ctx, spec.Name, metav1.GetOptions{}) + if err != nil { + t.Fatalf("getting service %s/%s: %v", namespace, spec.Name, err) + } + if service.Spec.ClusterIP == "" || service.Spec.ClusterIP == corev1.ClusterIPNone { + t.Fatalf("service %s/%s has no ClusterIP to dial: %q", namespace, spec.Name, service.Spec.ClusterIP) + } + + server := Server{Name: spec.Name, Namespace: namespace, ClusterIP: service.Spec.ClusterIP, Port: spec.Port, TLSPort: spec.TLSPort} + if server.TLSPort != 0 { + t.Logf("server %s is serving at %s and HTTPS at %s (namespace %s)", spec.Name, server.Address(), server.TLSAddress(), namespace) + } else { + t.Logf("server %s is serving at %s (namespace %s)", spec.Name, server.Address(), namespace) + } + servers = append(servers, server) } + return servers +} - koApply(t, renderServerPod(t, spec, namespace)) - WaitForPodReady(t, ctx, namespace, spec.Name, serverPodReadyTimeout) +// serverPodNamespace picks the one namespace specs are deployed into: the one +// they name, or a fresh one when none does. Sharing it is what lets a single +// apply cover them all, so specs that disagree are a mistake in the caller +// rather than something to paper over by applying twice. +func serverPodNamespace(t *testing.T, specs []ServerPod) string { + t.Helper() + names := map[string]bool{} + namespace := "" + for _, spec := range specs { + // Distinct names, because the Pod, the Service and the app label are all + // spec.Name: two servers sharing one would apply over each other and + // leave the suite dialing a Service that selects both. + if names[spec.Name] { + t.Fatalf("two servers are both named %q", spec.Name) + } + names[spec.Name] = true - service, err := GetClients().K8s.CoreV1().Services(namespace).Get(ctx, spec.Name, metav1.GetOptions{}) - if err != nil { - t.Fatalf("getting service %s/%s: %v", namespace, spec.Name, err) + if spec.Namespace == "" { + continue + } + if namespace != "" && namespace != spec.Namespace { + t.Fatalf("servers deployed together must share a namespace, got %q and %q", namespace, spec.Namespace) + } + namespace = spec.Namespace } - if service.Spec.ClusterIP == "" || service.Spec.ClusterIP == corev1.ClusterIPNone { - t.Fatalf("service %s/%s has no ClusterIP to dial: %q", namespace, spec.Name, service.Spec.ClusterIP) + if namespace == "" { + namespace = CreateNamespace(t).Name } + return namespace +} - server := Server{Name: spec.Name, Namespace: namespace, ClusterIP: service.Spec.ClusterIP, Port: spec.Port, TLSPort: spec.TLSPort} - if server.TLSPort != 0 { - t.Logf("server %s is serving at %s and HTTPS at %s (namespace %s)", spec.Name, server.Address(), server.TLSAddress(), namespace) - } else { - t.Logf("server %s is serving at %s (namespace %s)", spec.Name, server.Address(), namespace) +// renderServerPods renders specs into one multi-document manifest. +func renderServerPods(t *testing.T, specs []ServerPod, namespace string) string { + t.Helper() + docs := make([]string, 0, len(specs)) + for _, spec := range specs { + docs = append(docs, renderServerPod(t, spec, namespace)) } - return server + return strings.Join(docs, "\n---\n") } // renderServerPod writes spec's manifest into the test's temp dir and returns @@ -217,6 +305,7 @@ func renderServerPod(t *testing.T, spec ServerPod, namespace string) string { "${PORT}": strconv.Itoa(spec.Port), "${TARGET_PORT}": targetPortStr, } + servicePorts, containerPorts := serverUDPPorts(t, spec, targetPort) blocks := map[string]string{ "${ARGS}": serverArgs(spec, targetPortStr), "${READINESS_PROBE}": serverReadinessProbe(spec, targetPortStr), @@ -227,8 +316,12 @@ func renderServerPod(t *testing.T, spec ServerPod, namespace string) string { // a pod one. An empty list takes its whole line, key included. "${VOLUME_MOUNTS}": yamlListBlock(t, "volumeMounts", spec.VolumeMounts, 4), "${VOLUMES}": yamlListBlock(t, "volumes", spec.Volumes, 2), + // Extensions of the two `ports:` lists the template already opens, so + // these carry no key of their own. + "${UDP_SERVICE_PORTS}": servicePorts, + "${UDP_CONTAINER_PORTS}": containerPorts, } - return renderManifest(t, serverPodTemplate, inline, blocks) + return renderManifestText(t, serverPodTemplate, inline, blocks) } // serverArgs renders the container's `args:` list -- spec.Args followed by the @@ -244,14 +337,49 @@ func serverArgs(spec ServerPod, targetPort string) string { return strings.Join(out, "\n") } +// serverUDPPorts renders spec's UDP ports as fragments extending the Service's +// and the container's `ports:` lists. Every published port maps to the one +// listener, so the container side is a single entry however many the Service +// publishes. +func serverUDPPorts(t *testing.T, spec ServerPod, tcpTargetPort int) (service, container string) { + t.Helper() + if len(spec.UDPPorts) == 0 { + return "", "" + } + target := spec.UDPTargetPort + if target == 0 { + target = tcpTargetPort + } + + servicePorts := make([]corev1.ServicePort, 0, len(spec.UDPPorts)) + for _, port := range spec.UDPPorts { + servicePorts = append(servicePorts, corev1.ServicePort{ + // Named because a Service with more than one port requires it. + Name: fmt.Sprintf("udp-%d", port), + Protocol: corev1.ProtocolUDP, + Port: int32(port), + TargetPort: intstr.FromInt32(int32(target)), + }) + } + containerPorts := []corev1.ContainerPort{{ + Name: "serve-udp", + ContainerPort: int32(target), + Protocol: corev1.ProtocolUDP, + }} + return yamlItemsBlock(t, servicePorts, 2), yamlItemsBlock(t, containerPorts, 4) +} + // serverReadinessProbe renders the probe fragment for spec, indented to sit // under the template's `readinessProbe:` key. kubelet dials the container // directly, so the probe names the listen port rather than the published one. // A server with TLSPort is probed over HTTPS, on the listener that depends on // its certificate. func serverReadinessProbe(spec ServerPod, targetPort string) string { - if spec.GRPCProbe { + switch { + case spec.GRPCProbe: return " grpc:\n port: " + targetPort + case spec.TCPProbe: + return " tcpSocket:\n port: " + targetPort } healthPath := spec.HealthPath if healthPath == "" { @@ -311,11 +439,8 @@ func CreateServerService(t *testing.T, ctx context.Context, spec ServerPod) Serv if spec.Namespace == "" { t.Fatal("CreateServerService requires a test namespace") } - raw, err := os.ReadFile(renderServerPod(t, spec, spec.Namespace)) - if err != nil { - t.Fatalf("reading origin manifest: %v", err) - } - for doc := range strings.SplitSeq(string(raw), "\n---\n") { + raw := renderServerPod(t, spec, spec.Namespace) + for doc := range strings.SplitSeq(raw, "\n---\n") { var kind metav1.TypeMeta if err := yaml.Unmarshal([]byte(doc), &kind); err != nil { t.Fatalf("decoding origin kind: %v", err) diff --git a/internal/e2e/serverpod_test.go b/internal/e2e/serverpod_test.go index df6ad7db9b..6c9d28abba 100644 --- a/internal/e2e/serverpod_test.go +++ b/internal/e2e/serverpod_test.go @@ -15,7 +15,6 @@ package e2e import ( - "os" "slices" "strconv" "strings" @@ -37,42 +36,11 @@ import ( // silently has no readiness gate or no credentials. func renderServerPodDocs(t *testing.T, spec ServerPod) (*corev1.Pod, *corev1.Service) { t.Helper() - raw, err := os.ReadFile(renderServerPod(t, spec, "test-namespace")) - if err != nil { - t.Fatalf("reading the rendered server manifest: %v", err) + pods, services := decodeServerPodManifest(t, renderServerPod(t, spec, "test-namespace")) + if len(pods) != 1 || len(services) != 1 { + t.Fatalf("rendered %d Pods and %d Services, want one of each", len(pods), len(services)) } - if strings.Contains(string(raw), "${") { - t.Errorf("rendered server manifest still carries a placeholder:\n%s", raw) - } - - pod, service := &corev1.Pod{}, &corev1.Service{} - for doc := range strings.SplitSeq(string(raw), "\n---\n") { - if strings.TrimSpace(doc) == "" { - continue - } - var meta struct { - Kind string `json:"kind"` - } - if err := yaml.Unmarshal([]byte(doc), &meta); err != nil { - t.Fatalf("rendered server manifest is not valid YAML: %v\n%s", err, doc) - } - var into any - switch meta.Kind { - case "Pod": - into = pod - case "Service": - into = service - default: - continue - } - if err := yaml.UnmarshalStrict([]byte(doc), into); err != nil { - t.Fatalf("rendered server %s does not match the API type: %v\n%s", meta.Kind, err, doc) - } - } - if pod.Name == "" || service.Name == "" { - t.Fatalf("rendered server manifest is missing a Pod or a Service:\n%s", raw) - } - return pod, service + return pods[0], services[0] } // TestRenderServerPod_GRPCProbe covers the shape the networking suite deploys: @@ -220,6 +188,98 @@ func TestRenderServerPod(t *testing.T) { } } +// TestRenderServerPod_UDPPorts covers the shape the UDP-egress test deploys: +// several published UDP ports onto one listener, beside the TCP port that +// carries readiness. The assertion that matters is that the published ports +// stay distinct while every target port is the same, since a test that reads a +// missing datagram as policy depends on all of them reaching one server. +func TestRenderServerPod_UDPPorts(t *testing.T) { + pod, service := renderServerPodDocs(t, ServerPod{ + Name: "udpecho", + ImportPath: "github.com/agent-substrate/substrate/internal/e2e/fixtures/testserver", + Args: []string{"udpecho"}, + Port: 8053, + UDPPorts: []int{53, 443, 9999}, + }) + + // The TCP port keeps its place at the head of both lists: the readiness + // probe and anything dialing Address() still address it. + ports := service.Spec.Ports + if len(ports) != 4 { + t.Fatalf("rendered %d service ports, want the TCP one plus three UDP: %+v", len(ports), ports) + } + if got := ports[0].Port; got != 8053 || ports[0].Protocol == corev1.ProtocolUDP { + t.Errorf("first service port = %+v, want TCP 8053", ports[0]) + } + names := map[string]bool{} + for _, port := range ports[1:] { + if port.Protocol != corev1.ProtocolUDP { + t.Errorf("service port %+v is not UDP", port) + } + if got := port.TargetPort.IntValue(); got != 8053 { + t.Errorf("service port %d targets %d, want the one listener on 8053", port.Port, got) + } + // Kubernetes requires a name once a Service carries more than one + // port, and rejects a duplicate. + if port.Name == "" || names[port.Name] { + t.Errorf("service port %d has an empty or duplicate name %q", port.Port, port.Name) + } + names[port.Name] = true + } + if got := []int32{ports[1].Port, ports[2].Port, ports[3].Port}; !slices.Equal(got, []int32{53, 443, 9999}) { + t.Errorf("published UDP ports = %v, want [53 443 9999]", got) + } + + // One container port for the one listener, however many the Service + // publishes onto it. + containerPorts := pod.Spec.Containers[0].Ports + if len(containerPorts) != 2 { + t.Fatalf("rendered %d container ports, want the TCP one plus one UDP: %+v", len(containerPorts), containerPorts) + } + udp := containerPorts[1] + if udp.Protocol != corev1.ProtocolUDP || udp.ContainerPort != 8053 { + t.Errorf("UDP container port = %+v, want UDP 8053", udp) + } +} + +// TestRenderServerPod_NoUDPPorts pins the default: a server that asked for no +// UDP gets neither list extended, and in particular no bare entry left behind +// by the placeholder's own line. +func TestRenderServerPod_NoUDPPorts(t *testing.T) { + pod, service := renderServerPodDocs(t, ServerPod{ + Name: "httporigin", + ImportPath: "github.com/agent-substrate/substrate/internal/e2e/fixtures/testserver", + Args: []string{"http"}, + Port: 8080, + }) + + if got := len(service.Spec.Ports); got != 1 { + t.Errorf("rendered %d service ports, want only the TCP one: %+v", got, service.Spec.Ports) + } + if got := len(pod.Spec.Containers[0].Ports); got != 1 { + t.Errorf("rendered %d container ports, want only the TCP one: %+v", got, pod.Spec.Containers[0].Ports) + } +} + +// TestRenderServerPod_UDPTargetPort covers a listener that is not the TCP one. +func TestRenderServerPod_UDPTargetPort(t *testing.T) { + pod, service := renderServerPodDocs(t, ServerPod{ + Name: "udpecho", + ImportPath: "github.com/agent-substrate/substrate/internal/e2e/fixtures/testserver", + Args: []string{"udpecho"}, + Port: 8080, + UDPPorts: []int{53}, + UDPTargetPort: 9053, + }) + + if got := service.Spec.Ports[1].TargetPort.IntValue(); got != 9053 { + t.Errorf("service targetPort for UDP 53 = %d, want the UDP listener on 9053", got) + } + if got := pod.Spec.Containers[0].Ports[1].ContainerPort; got != 9053 { + t.Errorf("UDP containerPort = %d, want 9053", got) + } +} + // TestRenderServerPod_Volumes covers the credential-carrying shape the egressmitm // suite deploys, with both volume kinds it needs: a plain Secret and a // projection. A projection is the interesting one — it nests three levels, so @@ -413,3 +473,127 @@ func TestServerEndpointReady(t *testing.T) { }) } } + +// TestRenderServerPod_TCPProbe covers the third probe kind, for an origin that +// speaks neither HTTP nor gRPC, and the arg quoting a raw-TCP server needs: its +// greeting is a command-line flag, and the CRLF in it has to survive the +// round trip through YAML or the origin greets with something the suite is not +// expecting. +func TestRenderServerPod_TCPProbe(t *testing.T) { + const banner = "TESTBANNER/1.0\r\n" + pod, _ := renderServerPodDocs(t, ServerPod{ + Name: "tcpbanner", + ImportPath: "github.com/agent-substrate/substrate/internal/e2e/fixtures/testserver", + Args: []string{"tcpecho", "--banner=" + banner}, + Port: 2222, + TCPProbe: true, + }) + + container := pod.Spec.Containers[0] + if got, want := container.Args, []string{"tcpecho", "--banner=" + banner, "--listen=:2222"}; !slices.Equal(got, want) { + t.Errorf("container args = %q, want %q", got, want) + } + + probe := container.ReadinessProbe + if probe == nil || probe.TCPSocket == nil { + t.Fatalf("readinessProbe = %+v, want a tcpSocket probe", probe) + } + if got := probe.TCPSocket.Port.IntValue(); got != 2222 { + t.Errorf("tcpSocket probe port = %d, want 2222", got) + } + // Either of the others would probe an origin that speaks no HTTP and no + // gRPC, so the pod would never go ready. + if probe.HTTPGet != nil || probe.GRPC != nil { + t.Errorf("readinessProbe also carries httpGet %+v / gRPC %+v", probe.HTTPGet, probe.GRPC) + } +} + +// TestRenderServerPods covers the manifest a batched deploy applies: the whole +// point of it is that one ko invocation and one apply cover every server, so +// what matters is that the documents concatenate into something that still +// decodes as separate objects, each keeping the name and port its caller asked +// for and all of them landing in the shared namespace. +func TestRenderServerPods(t *testing.T) { + specs := []ServerPod{{ + Name: "tcpbanner", + ImportPath: "github.com/agent-substrate/substrate/internal/e2e/fixtures/testserver", + Args: []string{"tcpecho", "--banner=TESTBANNER/1.0\r\n"}, + Port: 2222, + TCPProbe: true, + }, { + Name: "tcpquiet", + ImportPath: "github.com/agent-substrate/substrate/internal/e2e/fixtures/testserver", + Args: []string{"tcpecho"}, + Port: 2223, + TCPProbe: true, + }} + pods, services := decodeServerPodManifest(t, renderServerPods(t, specs, "test-namespace")) + + if len(pods) != len(specs) || len(services) != len(specs) { + t.Fatalf("rendered %d Pods and %d Services, want %d of each", len(pods), len(services), len(specs)) + } + for i, spec := range specs { + pod, service := pods[i], services[i] + if pod.Name != spec.Name || service.Name != spec.Name { + t.Errorf("document %d is Pod %q / Service %q, want both named %q", i, pod.Name, service.Name, spec.Name) + } + // A shared namespace is what makes one apply legal in the first place. + if pod.Namespace != "test-namespace" || service.Namespace != "test-namespace" { + t.Errorf("%s landed in namespace %q / %q, want test-namespace", spec.Name, pod.Namespace, service.Namespace) + } + // Distinct ports per server, since the suite tells them apart by the + // port in the egress gateway's access log. + if got := int(pod.Spec.Containers[0].Ports[0].ContainerPort); got != spec.Port { + t.Errorf("%s containerPort = %d, want %d", spec.Name, got, spec.Port) + } + if got := int(service.Spec.Ports[0].Port); got != spec.Port { + t.Errorf("%s service port = %d, want %d", spec.Name, got, spec.Port) + } + if got := service.Spec.Selector["app"]; got != pod.Labels["app"] { + t.Errorf("%s Service selects app=%q but its Pod is labelled app=%q", spec.Name, got, pod.Labels["app"]) + } + } + // The Services select on a label that is the server's name, so two servers + // sharing a namespace must not share it. + if services[0].Spec.Selector["app"] == services[1].Spec.Selector["app"] { + t.Errorf("both Services select app=%q, so each would reach either pod", services[0].Spec.Selector["app"]) + } +} + +// decodeServerPodManifest decodes every Pod and Service out of a rendered +// server manifest, in the order they appear. +func decodeServerPodManifest(t *testing.T, raw string) ([]*corev1.Pod, []*corev1.Service) { + t.Helper() + if strings.Contains(raw, "${") { + t.Errorf("rendered server manifest still carries a placeholder:\n%s", raw) + } + + var pods []*corev1.Pod + var services []*corev1.Service + for doc := range strings.SplitSeq(raw, "\n---\n") { + if strings.TrimSpace(doc) == "" { + continue + } + var meta struct { + Kind string `json:"kind"` + } + if err := yaml.Unmarshal([]byte(doc), &meta); err != nil { + t.Fatalf("rendered server manifest is not valid YAML: %v\n%s", err, doc) + } + var into any + switch meta.Kind { + case "Pod": + pod := &corev1.Pod{} + pods, into = append(pods, pod), pod + case "Service": + service := &corev1.Service{} + services, into = append(services, service), service + default: + t.Fatalf("rendered server manifest carries an unexpected %q document:\n%s", meta.Kind, doc) + } + if err := yaml.UnmarshalStrict([]byte(doc), into); err != nil { + t.Fatalf("rendered server %s does not match the API type: %v\n%s", meta.Kind, err, doc) + } + } + return pods, services +} diff --git a/internal/e2e/suites/networking/networking_test.go b/internal/e2e/suites/networking/networking_test.go index 3655e053b2..521bbb25ff 100644 --- a/internal/e2e/suites/networking/networking_test.go +++ b/internal/e2e/suites/networking/networking_test.go @@ -1008,3 +1008,208 @@ func protocolGatewayAccessLogMatches(entry gatewayAccessLogLine, actorRef resour } return false } + +// TestActorEgressRawTCPBlocked covers TCP that is neither HTTP nor TLS, from +// both sides of the who-speaks-first divide. The gateway refuses it either way; +// what the actor sees depends on whether it sends first: +// +// - If the actor waits for the server, the gateway has nothing to classify, +// gives up on its listener timeout and closes the connection with no bytes +// returned. +// - If the actor sends first, the gateway treats bytes that are not TLS as +// cleartext HTTP, its HTTP handler cannot parse them, and it answers with +// its own "400 Bad Request" and closes. +// +// The origins answer whenever anything reaches them -- one greets on connect, +// the other echoes what it reads -- so a greeting or an echo back means the +// gateway relayed the connection. The CONNECT record shows the attempt did +// reach the gateway, so the refusal is the gateway's and not a connection +// that never left the sandbox. +func TestActorEgressRawTCPBlocked(t *testing.T) { + ctx := context.Background() + + // Stand the targets up first in one call so they share a ko build and start together, + // rather than the second waiting out the first's build, push and readiness. + targets := e2e.DeployServerPods(t, ctx, tcpBannerTarget, tcpQuietTarget) + bannerTarget, quietTarget := targets[0], targets[1] + + tests := []struct { + name string + target e2e.Server + // skipBanner makes the actor write before reading, so the gateway sees + // the client's bytes at once instead of waiting for them. + skipBanner bool + // payload is what the actor sends, given a token unique to the run. + payload func(token string) string + // wantBadRequest is whether the gateway answers with its own 400 + // rather than closing with no bytes. + wantBadRequest bool + }{ + {name: "server speaks first", target: bannerTarget, payload: func(token string) string { return token + "\r\n" }}, + // A line-based protocol: complete lines that are not HTTP. + {name: "client speaks first, text line", target: quietTarget, skipBanner: true, wantBadRequest: true, payload: func(token string) string { return token + "\r\n" }}, + // A binary protocol: bytes no HTTP parser accepts as a request line. + {name: "client speaks first, binary", target: quietTarget, skipBanner: true, wantBadRequest: true, payload: func(token string) string { return "\x00\x01\x02\x03" + token }}, + } + + actorAtespace, actorName, _ := createAndResumeActorWithEgress(t, ctx, "egress-tcp", egressFixture(), e2e.EgressAllowAll()...) + router := mustRouterClient(t, ctx) + defer router.Close() + actorRef := resources.ActorRef{Atespace: networkingAtespace, Name: actorName} + waitForActorRoute(t, ctx, router, actorRef) + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + // Bound the access-log scan to lines this subtest could have produced. + since := metav1.NewTime(time.Now().Add(-1 * time.Minute)) + address := test.target.Address() + + // Distinct per run, so an echo cannot be satisfied by anything stale. + token := fmt.Sprintf("ping-%d", time.Now().UnixNano()) + probe := probeRawTCP(t, ctx, router, actorRef, map[string]any{ + "address": address, "send": test.payload(token), "skipBanner": test.skipBanner, "timeout": "5s", + }) + if probe.Banner != "" { + t.Fatalf("raw TCP to %s got the origin's greeting %q: the gateway relayed the connection", address, probe.Banner) + } + if strings.Contains(probe.Received, token) { + t.Fatalf("raw TCP to %s got its payload echoed back: the gateway relayed the connection (received %q)", address, probe.Received) + } + port := strconv.Itoa(test.target.Port) + if test.wantBadRequest { + if !strings.HasPrefix(probe.Received, "HTTP/1.1 400 Bad Request\r\n") { + t.Fatalf("raw TCP to %s received %q, want the gateway's HTTP/1.1 400 Bad Request", address, probe.Received) + } + t.Logf("raw TCP to %s was answered by the gateway with 400 Bad Request", address) + assertEgressGatewayConnect(t, ctx, since, actorAtespace, actorName, port) + return + } + if probe.Received != "" { + t.Fatalf("raw TCP to %s received %q, want the connection closed with no bytes", address, probe.Received) + } + // No bytes alone is also what a connection the gateway holds open + // until the actor gives up looks like. + if !probe.Closed { + t.Fatalf("raw TCP to %s got no bytes but was never closed: the gateway held it open until the actor's 5s read timeout", address) + } + t.Logf("raw TCP to %s was closed with no bytes returned", address) + assertEgressGatewayConnect(t, ctx, since, actorAtespace, actorName, port) + assertEgressGatewayReturnedNoBytes(t, ctx, since, actorAtespace, actorName, port) + }) + } +} + +// rawTCPProbe is the egress demo's /tcp answer. +type rawTCPProbe struct { + Banner string `json:"banner"` + Received string `json:"received"` + Closed bool `json:"closed"` + Error string `json:"error"` +} + +// probeRawTCP asks the egress demo actor for one raw TCP exchange. The demo +// reports a closed connection as a 200 with Closed set, so anything but a 200 +// means the probe itself failed. Only a 503 from the router, while the actor's +// route propagates, is retried. +func probeRawTCP(t *testing.T, ctx context.Context, router *e2e.RouterClient, actorRef resources.ActorRef, request map[string]any) rawTCPProbe { + t.Helper() + payload, err := json.Marshal(request) + if err != nil { + t.Fatalf("marshaling the TCP probe request: %v", err) + } + status, body := postThroughEgressActorUntil(t, ctx, router, actorRef, "/tcp", payload, notTransient) + if status != http.StatusOK { + t.Fatalf("TCP probe returned HTTP %d, want 200; body: %s", status, body) + } + var probe rawTCPProbe + if err := json.Unmarshal(body, &probe); err != nil { + t.Fatalf("decoding the TCP probe response %s: %v", body, err) + } + return probe +} + +// accessLogCount parses the field named key as a count. A missing field and an +// unparseable one are both reported as absent, since either means the caller's +// expectation of the log format no longer holds. +func accessLogCount(line, key string) (int, bool) { + raw, ok := accessLogField(line, key) + if !ok { + return 0, false + } + value, err := strconv.Atoi(raw) + if err != nil { + return 0, false + } + return value, true +} + +// assertEgressGatewayReturnedNoBytes checks, on Envoy, that the actor's tunnels +// to port carried nothing back. Envoy logs each CONNECT twice: once when the +// tunnel is established, with zero counters, and once when it closes, with the +// final ones. Waiting for both records is what makes a zero down_bytes mean the +// gateway sent nothing, rather than that the close had not been logged yet. +// AgentGateway's tunnel log carries no byte counters, so there the probe's own +// result is the only evidence. +func assertEgressGatewayReturnedNoBytes(t *testing.T, ctx context.Context, since metav1.Time, atespace, actorName, port string) { + t.Helper() + spiffeSlug := "/ateom-for-actor/" + atespace + "/" + actorName + want := fmt.Sprintf("the establish and close records of a tunnel to port %s by actor %s/%s", port, atespace, actorName) + var records []string + agentGateway := false + waitForAccessLog(t, ctx, since, want, func(lines []gatewayAccessLogLine) bool { + records = records[:0] + for _, line := range lines { + switch line.container { + case "agentgateway": + agentGateway = true + case "envoy": + if !strings.Contains(line.text, spiffeSlug) { + continue + } + if authority, ok := accessLogField(line.text, "authority"); ok && strings.HasSuffix(authority, ":"+port) { + records = append(records, line.text) + } + } + } + return agentGateway || len(records) >= 2 + }) + if agentGateway { + return + } + for _, record := range records { + down, ok := accessLogCount(record, "down_bytes") + if !ok { + t.Fatalf("access-log line has no down_bytes counter, so the log format changed: %s", record) + } + if down != 0 { + t.Fatalf("egress gateway sent %d bytes back through a tunnel that should have been closed with none: %s", down, record) + } + } + t.Logf("egress gateway returned no bytes on %d tunnel records to port %s", len(records), port) +} + +// tcpEchoBanner is the greeting tcpBannerTarget is started with; the actor must +// never receive it. +const tcpEchoBanner = "TESTBANNER/1.0\r\n" + +// tcpBannerTarget and tcpQuietTarget are the origins TestActorEgressRawTCPBlocked +// dials. They differ only in whether they greet. +// +// Distinct ports as well as distinct pods, so that the access-log assertions +// stay unambiguous while both subtests share one Actor. +var ( + tcpBannerTarget = e2e.ServerPod{ + Name: "tcpbanner", + ImportPath: "github.com/agent-substrate/substrate/internal/e2e/fixtures/testserver", + Args: []string{"tcpecho", "--banner=" + tcpEchoBanner}, + Port: 2222, + TCPProbe: true, + } + tcpQuietTarget = e2e.ServerPod{ + Name: "tcpquiet", + ImportPath: "github.com/agent-substrate/substrate/internal/e2e/fixtures/testserver", + Args: []string{"tcpecho"}, + Port: 2223, + TCPProbe: true, + } +) diff --git a/internal/e2e/suites/networking/udpegress_test.go b/internal/e2e/suites/networking/udpegress_test.go new file mode 100644 index 0000000000..20c5db24bf --- /dev/null +++ b/internal/e2e/suites/networking/udpegress_test.go @@ -0,0 +1,259 @@ +// 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 networking + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/url" + "slices" + "strconv" + "strings" + "testing" + "time" + + "github.com/agent-substrate/substrate/internal/e2e" + "github.com/agent-substrate/substrate/internal/resources" +) + +// The destination ports TestActorUDPEgress sends to. They are Service ports, +// all published by the one echo origin and all mapped to udpEchoListenPort, so +// the only thing that differs between them is the number a datagram is +// addressed to. +const ( + // dnsPort is the port a rule that spared DNS by number would let through. + // The sandbox serves DNS from a relay at its own gateway address instead, + // so UDP to port 53 anywhere else must be blocked like any other. + dnsPort = 53 + // quicPort is HTTP/3: an actor that could send it would have HTTPS egress + // with no CONNECT authority, no access log and no policy hook. + quicPort = 443 + // arbitraryUDPPort stands for every other port. + arbitraryUDPPort = 9999 + // udpEchoListenPort is the single listener behind all three. It is also the + // origin's TCP port, where it answers the readiness probe. + udpEchoListenPort = 8053 + // udpClientPort is where the control probe pod serves its HTTP API. + udpClientPort = 8080 +) + +// udpEchoPorts are the UDP ports udpEchoTarget publishes. +var udpEchoPorts = []int{dnsPort, quicPort, arbitraryUDPPort} + +// udpEchoTarget is the origin TestActorUDPEgress addresses: testserver's +// udpecho subcommand, published on udpEchoPorts plus its own TCP port. +func udpEchoTarget() e2e.ServerPod { + return e2e.ServerPod{ + Name: "udpecho", + ImportPath: "github.com/agent-substrate/substrate/internal/e2e/fixtures/testserver", + Args: []string{"udpecho"}, + Port: udpEchoListenPort, + UDPPorts: udpEchoPorts, + } +} + +// udpClientControl runs the probe binary as a plain pod, outside any sandbox, +// so the same /udpecho code the actor runs can show the origin answers on +// every port. +func udpClientControl() e2e.ServerPod { + return e2e.ServerPod{ + Name: "udpclient", + ImportPath: "github.com/agent-substrate/substrate/internal/e2e/fixtures/probe", + Port: udpClientPort, + } +} + +// TestActorUDPEgress covers docs/egress-traffic.md's UDP rows: DNS is answered, +// and every other datagram is dropped. +// +// Each actor runs in its own network namespace whose only way out is its +// gateway address (internal/ateomnet.SetupSandboxNetwork): TCP is redirected +// to atunnel, and DNS over UDP and TCP is answered by the relay listening on +// the gateway address. A datagram for any other address, port 53 included, +// has nowhere to go. So the claim is that none of an actor's UDP reaches an +// origin that answers everything, while name resolution still works. +// +// A datagram dropped on the way out and one delivered to a server that stays +// quiet look the same from inside the sandbox, so a control pod outside any +// sandbox, running the same probe code, first has to get an echo on every +// port. Without it, a broken origin or Service would let the drop assertions +// pass having proven nothing. +func TestActorUDPEgress(t *testing.T) { + env, err := e2e.CheckEnv("BUCKET_NAME", "KO_DOCKER_REPO") + if err != nil { + t.Fatalf("CheckEnv failed: %v", err) + } + ctx := context.Background() + + // The origins first, so a fixture failure costs no Actor resume. + servers := e2e.DeployServerPods(t, ctx, udpEchoTarget(), udpClientControl()) + echo, control := servers[0], servers[1] + + // kube-proxy programs a new Service's UDP ports on its own schedule, and + // until it does a datagram to the ClusterIP draws an ICMP port-unreachable + // ("connection refused"). Wait the control out rather than fail on it. + for _, port := range udpEchoPorts { + got := waitForUDPEchoFromPod(t, ctx, control.Namespace, udpClientControl().Name, echo.AddressOnPort(port)) + t.Logf("UDP from the control pod to port %d was echoed after %d attempt(s)", port, got.Attempts) + } + + atespace, _ := e2e.DeployProbe(t, env["BUCKET_NAME"], "networking") + _, actorName, _ := createAndResumeSubstrateActor(t, ctx, "udpegress", e2e.SubstrateFixture{ + Atespace: atespace, + Name: e2e.ProbeName, + // Unlike the demo fixtures this suite's other tests build actors from, + // the probe is deployed by the line above rather than by an install + // script, so a failure here is a bug and not a missing prerequisite. + DeployWith: "e2e.DeployProbe, which this test already ran", + }) + router := mustRouterClient(t, ctx) + defer router.Close() + actorRef := resources.ActorRef{Atespace: networkingAtespace, Name: actorName} + + for _, port := range udpEchoPorts { + t.Run(fmt.Sprintf("port %d dropped", port), func(t *testing.T) { + got := probeUDPEcho(t, ctx, router, actorRef, echo.AddressOnPort(port)) + assertUDPDropped(t, port, got) + t.Logf("UDP from the actor to port %d went unanswered after %d attempts, as expected: %s", port, got.Attempts, got.Error) + }) + } + + // DNS is the one UDP the sandbox answers, through its own relay. Resolving + // a name is what checks it; a lookup rather than a fetch, so the answer + // depends on nothing but DNS. + t.Run("DNS still resolves", func(t *testing.T) { + host := fmt.Sprintf("%s.%s.svc.cluster.local", udpEchoTarget().Name, echo.Namespace) + got := probeResolve(t, ctx, router, actorRef, host) + if got.Error != "" { + t.Fatalf("resolving %s from the actor failed: %s", host, got.Error) + } + if !slices.Contains(got.Addresses, echo.ClusterIP) { + t.Fatalf("%s resolved to %v, want it to include the Service's ClusterIP %s", host, got.Addresses, echo.ClusterIP) + } + t.Logf("%s resolved to %v from inside the actor", host, got.Addresses) + }) +} + +// udpEchoResult mirrors the probe's /udpecho response body. +type udpEchoResult struct { + Addr string `json:"addr"` + Echoed bool `json:"echoed"` + Attempts int `json:"attempts"` + Error string `json:"error"` +} + +// probeUDPEcho asks the probe Actor to send UDP to addr and reports whether it +// was echoed back. A datagram that never comes back is a result, not a probe +// failure: the probe answers 200 either way, so only a router-level failure +// retries here. +func probeUDPEcho(t *testing.T, ctx context.Context, router *e2e.RouterClient, actorRef resources.ActorRef, addr string) udpEchoResult { + t.Helper() + var out udpEchoResult + probeGet(t, ctx, router, actorRef, "/udpecho?addr="+url.QueryEscape(addr), "UDP echo probe of "+addr, &out) + return out +} + +// probeUDPEchoAttempts is the probe's udpEchoAttempts: how many datagrams +// /udpecho sends before it reports no echo. +const probeUDPEchoAttempts = 5 + +// assertUDPDropped requires got to be what a silently dropped datagram looks +// like: every attempt sent, and the last one ending in a read timeout. No echo +// alone would also be the result of a dial or send that failed, or of an ICMP +// error ("connection refused"), none of which shows a drop. +func assertUDPDropped(t *testing.T, port int, got udpEchoResult) { + t.Helper() + if got.Echoed { + t.Fatalf("UDP from the actor to port %d was echoed: actor UDP left the sandbox", port) + } + if got.Attempts != probeUDPEchoAttempts { + t.Fatalf("UDP from the actor to port %d stopped after %d of %d attempts (%s): the probe did not get to send and wait every time", port, got.Attempts, probeUDPEchoAttempts, got.Error) + } + if !strings.HasPrefix(got.Error, "waiting for the echo from ") || !strings.HasSuffix(got.Error, "i/o timeout") { + t.Fatalf("UDP from the actor to port %d failed with %q, want the read timeout of a datagram that was silently dropped", port, got.Error) + } +} + +// waitForUDPEchoFromPod repeats probeUDPEchoFromPod until the datagram is +// echoed, failing after udpControlTimeout. +func waitForUDPEchoFromPod(t *testing.T, ctx context.Context, namespace, podName, addr string) udpEchoResult { + t.Helper() + const udpControlTimeout = 60 * time.Second + deadline := time.Now().Add(udpControlTimeout) + for { + got := probeUDPEchoFromPod(t, ctx, namespace, podName, addr) + if got.Echoed { + return got + } + if time.Now().After(deadline) { + t.Fatalf("UDP from the control pod to %s was not echoed within %v (last: %d attempts, %s): the echo origin or its Service is broken, so the actor's drops would prove nothing", + addr, udpControlTimeout, got.Attempts, got.Error) + } + t.Logf("UDP from the control pod to %s not echoed yet (%s); retrying", addr, got.Error) + time.Sleep(2 * time.Second) + } +} + +// probeUDPEchoFromPod is probeUDPEcho against the probe running as the plain +// pod podName, reached through the API server's pod proxy. +func probeUDPEchoFromPod(t *testing.T, ctx context.Context, namespace, podName, addr string) udpEchoResult { + t.Helper() + body, err := e2e.GetClients().K8s.CoreV1().RESTClient().Get(). + Namespace(namespace). + Resource("pods"). + Name(podName+":"+strconv.Itoa(udpClientPort)). + SubResource("proxy"). + Suffix("udpecho"). + Param("addr", addr). + DoRaw(ctx) + if err != nil { + t.Fatalf("UDP echo probe of %s from pod %s/%s: %v (body %q)", addr, namespace, podName, err, body) + } + var out udpEchoResult + if err := json.Unmarshal(body, &out); err != nil { + t.Fatalf("decoding pod %s/%s's response to the UDP echo probe of %s: %v (body %q)", namespace, podName, addr, err, body) + } + return out +} + +// probeResolveResult mirrors the probe's /resolve response body. +type probeResolveResult struct { + Host string `json:"host"` + Addresses []string `json:"addresses"` + Error string `json:"error"` +} + +// probeResolve asks the probe Actor to look host up through the sandbox's own +// resolver. +func probeResolve(t *testing.T, ctx context.Context, router *e2e.RouterClient, actorRef resources.ActorRef, host string) probeResolveResult { + t.Helper() + var out probeResolveResult + probeGet(t, ctx, router, actorRef, "/resolve?host="+url.QueryEscape(host), "lookup of "+host, &out) + return out +} + +// probeGet GETs path on the probe Actor through the router and decodes the JSON +// body into out. +func probeGet(t *testing.T, ctx context.Context, router *e2e.RouterClient, actorRef resources.ActorRef, path, what string, out any) { + t.Helper() + body := waitForRouteReady(t, what, func() (*http.Response, error) { + return router.Get(ctx, actorRef, path) + }) + if err := json.Unmarshal([]byte(body), out); err != nil { + t.Fatalf("decoding the probe's response to the %s: %v (body %q)", what, err, body) + } +}