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) + } +}