diff --git a/demos/egress/main.go b/demos/egress/main.go index 09c7b26772..609682b2e6 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 ( @@ -150,6 +152,7 @@ func newHandler(client *http.Client) http.Handler { writeJSON(w, response.StatusCode, fetchResponse{StatusCode: response.StatusCode, Body: string(body)}) }) mux.HandleFunc("/grpc", handleGRPC) + mux.HandleFunc("/tcp", handleTCPProbe) return mux } @@ -290,6 +293,114 @@ 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"` +} + +type tcpProbeResponse struct { + // Banner is whatever the peer sent before being spoken to. + Banner string `json:"banner,omitempty"` + Received string `json:"received,omitempty"` + Error string `json:"error,omitempty"` +} + +// handleTCPProbe opens a TCP connection and 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. + banner, 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 := tcpProbeResponse{Banner: string(banner)} + 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 { + writeJSON(w, http.StatusBadGateway, tcpProbeResponse{Error: fmt.Sprintf("writing to %s: %v", input.Address, err)}) + return + } + received, 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) + writeJSON(w, http.StatusOK, response) +} + +// readWithTimeout returns the bytes of a single read, capped at limit. A peer +// that says nothing within timeout yields no bytes 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, error) { + if err := connection.SetReadDeadline(time.Now().Add(timeout)); err != nil { + return nil, fmt.Errorf("setting read deadline: %w", err) + } + buffer := make([]byte, limit) + n, err := connection.Read(buffer) + if err != nil && !errors.Is(err, os.ErrDeadlineExceeded) && !errors.Is(err, io.EOF) { + return nil, err + } + return buffer[:n], nil +} + 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 28f50371a5..445091a14c 100644 --- a/demos/egress/main_test.go +++ b/demos/egress/main_test.go @@ -110,6 +110,143 @@ func TestOutboundFailure(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") + } +} + +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 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()) + } + }) + } +} + +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") + } +} + +// 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() +} + +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 +} + type roundTripFunc func(*http.Request) (*http.Response, error) func (f roundTripFunc) RoundTrip(request *http.Request) (*http.Response, error) { diff --git a/internal/e2e/fixtures/testserver/main.go b/internal/e2e/fixtures/testserver/main.go index 6393712f61..3d0b3e5048 100644 --- a/internal/e2e/fixtures/testserver/main.go +++ b/internal/e2e/fixtures/testserver/main.go @@ -19,6 +19,7 @@ // // testserver grpc --listen=:50051 a cleartext HTTP/2 gRPC echo origin // testserver http --listen=:8080 a plain HTTP origin serving /healthz +// 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 // @@ -43,7 +44,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()) 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..7d32627f5a --- /dev/null +++ b/internal/e2e/fixtures/testserver/tcpecho_test.go @@ -0,0 +1,151 @@ +// 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) + } + t.Cleanup(func() { _ = listener.Close() }) + go func() { + if err := serveTCPEcho(listener, banner); err != nil { + 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/manifest.go b/internal/e2e/manifest.go index 85d306eb26..d5f544e33c 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 hack/install-demo-counter.sh // plays with `sed /.../d`. 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 diff --git a/internal/e2e/serverpod.go b/internal/e2e/serverpod.go index 09a95f7f8e..b09639c2a2 100644 --- a/internal/e2e/serverpod.go +++ b/internal/e2e/serverpod.go @@ -64,6 +64,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 @@ -93,39 +97,100 @@ func (s Server) Address() 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{Namespace: namespace, ClusterIP: service.Spec.ClusterIP, Port: spec.Port} + 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{Namespace: namespace, ClusterIP: service.Spec.ClusterIP, Port: spec.Port} - t.Logf("server %s is serving at %s (namespace %s)", spec.Name, server.Address(), namespace) - return server +// 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 strings.Join(docs, "\n---\n") } -// renderServerPod writes spec's manifest into the test's temp dir and returns -// the path. Split out of DeployServerPod so the rendering has a unit test that -// does not need a cluster. +// renderServerPod renders spec's Pod and Service. Split out of DeployServerPods +// so the rendering has a unit test that does not need a cluster. func renderServerPod(t *testing.T, spec ServerPod, namespace string) string { t.Helper() port := strconv.Itoa(spec.Port) @@ -143,7 +208,7 @@ func renderServerPod(t *testing.T, spec ServerPod, namespace string) string { "${VOLUME_MOUNTS}": yamlListBlock(t, "volumeMounts", spec.VolumeMounts, 4), "${VOLUMES}": yamlListBlock(t, "volumes", spec.Volumes, 2), } - return renderManifest(t, serverPodTemplate, inline, blocks) + return renderManifestText(t, serverPodTemplate, inline, blocks) } // serverArgs renders the container's `args:` list -- spec.Args followed by the @@ -162,8 +227,11 @@ func serverArgs(spec ServerPod, port string) string { // serverReadinessProbe renders the probe fragment for spec, indented to sit // under the template's `readinessProbe:` key. func serverReadinessProbe(spec ServerPod, port string) string { - if spec.GRPCProbe { + switch { + case spec.GRPCProbe: return " grpc:\n port: " + port + case spec.TCPProbe: + return " tcpSocket:\n port: " + port } path := spec.HealthPath if path == "" { diff --git a/internal/e2e/serverpod_test.go b/internal/e2e/serverpod_test.go index 4186ac3e04..7f191997b0 100644 --- a/internal/e2e/serverpod_test.go +++ b/internal/e2e/serverpod_test.go @@ -15,7 +15,6 @@ package e2e import ( - "os" "slices" "strings" "testing" @@ -35,16 +34,24 @@ 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), "${") { + return pods[0], services[0] +} + +// 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) } - pod, service := &corev1.Pod{}, &corev1.Service{} - for doc := range strings.SplitSeq(string(raw), "\n---\n") { + var pods []*corev1.Pod + var services []*corev1.Service + for doc := range strings.SplitSeq(raw, "\n---\n") { if strings.TrimSpace(doc) == "" { continue } @@ -57,20 +64,19 @@ func renderServerPodDocs(t *testing.T, spec ServerPod) (*corev1.Pod, *corev1.Ser var into any switch meta.Kind { case "Pod": - into = pod + pod := &corev1.Pod{} + pods, into = append(pods, pod), pod case "Service": - into = service + service := &corev1.Service{} + services, into = append(services, service), service default: - continue + 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) } } - 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, services } // TestRenderServerPod_GRPCProbe covers the shape the networking suite deploys: @@ -155,6 +161,92 @@ func TestRenderServerPod_HTTPProbe(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"]) + } +} + // TestRenderServerPod_Volumes covers the credential-carrying shape the sdsmint // 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 diff --git a/internal/e2e/suites/networking/networking_test.go b/internal/e2e/suites/networking/networking_test.go index 35e4912836..f7ca6fa84e 100644 --- a/internal/e2e/suites/networking/networking_test.go +++ b/internal/e2e/suites/networking/networking_test.go @@ -82,13 +82,13 @@ func TestActorDirectAccess(t *testing.T) { }) } -// TestActorEgress exercises the full egress path. The Actor's outbound TCP +// TestActorEgressHTTP exercises the full egress path. The Actor's outbound TCP // connection is transparently redirected by nftables into atunnel, wrapped in // mTLS with the Actor's own actor-identity certificate plus an HTTP CONNECT to // atenet-egress, authorized there against that certificate, and only then // dialed out. A masqueraded (pre-gateway) egress would also return 200, so this // asserts the gateway is deployed and that it did not reject the Actor. -func TestActorEgress(t *testing.T) { +func TestActorEgressHTTP(t *testing.T) { ctx := context.Background() actorName, _ := createAndResumeActor(t, ctx, "egress", egressFixture()) router := mustRouterClient(t, ctx) @@ -179,6 +179,142 @@ func TestActorEgressNonStandardPort(t *testing.T) { assertEgressGatewayConnect(t, ctx, since, actorName, strconv.Itoa(httpTarget.Port)) } +// tcpEchoBanner is the greeting tcpBannerTarget is started with and the exact +// bytes TestActorEgressRawTCP expects to come back through the tunnel. +const tcpEchoBanner = "TESTBANNER/1.0\r\n" + +// tcpBannerTarget and tcpQuietTarget are the origins TestActorEgressRawTCP +// 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, + } +) + +// TestActorEgressRawTCP covers egress for a payload that is neither HTTP nor +// TLS, from both sides of the who-speaks-first divide. HTTP and HTTPS both have +// the client send the first bytes, so neither notices a path that waits for +// downstream data before dialing upstream, or that inspects those first bytes +// to route. +func TestActorEgressRawTCP(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 + // wantBanner is what the origin volunteers before being spoken to, so + // empty means it stayed silent. + wantBanner string + // timeout bounds each read the probe does. + timeout string + }{ + {name: "server speaks first", target: bannerTarget, wantBanner: tcpEchoBanner, timeout: "10s"}, + {name: "client speaks first", target: quietTarget, wantBanner: "", timeout: "2s"}, + } + + actorName, _ := createAndResumeActor(t, ctx, "egress-tcp", egressFixture()) + router := mustRouterClient(t, ctx) + defer router.Close() + + 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 the echo cannot be satisfied by anything stale. + sent := fmt.Sprintf("ping-%d", time.Now().UnixNano()) + payload, err := json.Marshal(map[string]any{"address": address, "send": sent, "timeout": test.timeout}) + if err != nil { + t.Fatalf("marshaling the TCP probe request for %s: %v", address, err) + } + status, body := postThroughEgressActor(t, ctx, router, resources.ActorRef{Atespace: networkingAtespace, Name: actorName}, "/tcp", payload) + if status != http.StatusOK { + t.Fatalf("Actor raw TCP probe of %s returned HTTP %d, want 200; body: %s", address, status, body) + } + + var probe struct { + Banner string `json:"banner"` + Received string `json:"received"` + Error string `json:"error"` + } + if err := json.Unmarshal(body, &probe); err != nil { + t.Fatalf("decoding the TCP probe response %s: %v", body, err) + } + if probe.Banner != test.wantBanner { + t.Fatalf("banner from %s = %q, want %q (error: %q)", address, probe.Banner, test.wantBanner, probe.Error) + } + if probe.Received != sent { + t.Fatalf("echo from %s = %q, want %q (error: %q)", address, probe.Received, sent, probe.Error) + } + t.Logf("Actor raw TCP probe of %s succeeded; banner: %q", address, probe.Banner) + + port := strconv.Itoa(test.target.Port) + assertEgressGatewayConnect(t, ctx, since, actorName, port) + assertEgressGatewayTunneledBytes(t, ctx, since, actorName, port) + }) + } +} + +// TestActorEgressSSH tests against SSH, a real server-speaks-first protocol. +func TestActorEgressSSH(t *testing.T) { + // RFC 4253 ยง4.2: the SSH server sends its identification string first. + const identificationPrefix = "SSH-2.0-" + const address = "github.com:22" + + ctx := context.Background() + actorName, _ := createAndResumeActor(t, ctx, "egress-ssh", egressFixture()) + router := mustRouterClient(t, ctx) + defer router.Close() + + since := metav1.NewTime(time.Now().Add(-1 * time.Minute)) + + // No Send: the identification string alone shows the transport carried the + // server's first bytes, and anything written after it would start a key + // exchange this test has no reason to hold up its end of. + payload, err := json.Marshal(map[string]any{"address": address, "timeout": "10s"}) + if err != nil { + t.Fatalf("marshaling the SSH probe request: %v", err) + } + status, body := postThroughEgressActor(t, ctx, router, resources.ActorRef{Atespace: networkingAtespace, Name: actorName}, "/tcp", payload) + if status != http.StatusOK { + t.Fatalf("Actor SSH probe of %s returned HTTP %d, want 200; body: %s", address, status, body) + } + + var probe struct { + Banner string `json:"banner"` + Error string `json:"error"` + } + if err := json.Unmarshal(body, &probe); err != nil { + t.Fatalf("decoding the SSH probe response %s: %v", body, err) + } + if !strings.HasPrefix(probe.Banner, identificationPrefix) { + t.Fatalf("banner from %s = %q, want a %q prefix (error: %q)", address, probe.Banner, identificationPrefix, probe.Error) + } + t.Logf("Actor SSH probe of %s succeeded; identification string: %q", address, strings.TrimSpace(probe.Banner)) + + assertEgressGatewayConnect(t, ctx, since, actorName, "22") +} + // fetchThroughEgressActor asks the egress demo Actor to fetch url and returns // the status and body it echoes back. func fetchThroughEgressActor(t *testing.T, ctx context.Context, router *e2e.RouterClient, actorRef resources.ActorRef, url string) (int, []byte) { @@ -297,6 +433,55 @@ func waitForAccessLog(t *testing.T, ctx context.Context, since metav1.Time, want } } +// assertEgressGatewayTunneledBytes waits for the access-log record the gateway +// writes when the tunnel closes, and requires bytes to have crossed it in both +// directions. The CONNECT record alone only says the tunnel was authorized and +// opened; these counters are the gateway's own evidence that it relayed the +// payload, rather than the Actor having reached the origin some other way. +func assertEgressGatewayTunneledBytes(t *testing.T, ctx context.Context, since metav1.Time, actorName, port string) { + t.Helper() + want := fmt.Sprintf("a closed tunnel to port %s by actor %s carrying bytes both ways", port, actorName) + waitForAccessLog(t, ctx, since, want, func(lines []string) (bool, error) { + for _, line := range lines { + authority, ok := accessLogField(line, "authority") + if !ok || !strings.HasSuffix(authority, ":"+port) { + continue + } + if !strings.Contains(line, "/actor/"+actorName) { + continue + } + // The counters only carry their final values on the record flushed + // at close; the one flushed on establishment reports zeroes. + up, upOK := accessLogCount(line, "up_bytes") + down, downOK := accessLogCount(line, "down_bytes") + if !upOK || !downOK { + return false, fmt.Errorf("access-log line has no byte counters, so the log format changed: %s", line) + } + if up == 0 || down == 0 { + continue + } + t.Logf("egress gateway relayed %d bytes up and %d down: %s", up, down, line) + return true, nil + } + return false, nil + }) +} + +// 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 +} + // accessLogField returns the value of the key=value field named key in an Envoy // access log line whose fields are separated by spaces. func accessLogField(line, key string) (string, bool) {