From 381c845319de386c7e21e6ebcc2b8fa78092dabf Mon Sep 17 00:00:00 2001 From: Haiyan Meng Date: Mon, 31 Aug 2026 09:30:12 -0400 Subject: [PATCH] e2e: cover egress for non-HTTP traffic, in both speaking orders HTTP and HTTPS egress both have the client speak first, so neither notices a path that waits for downstream bytes before dialing upstream. Add raw-TCP and SSH tests covering both sides of that divide, asserting from the gateway's own byte counters that it relayed the payload. The origin is a new `testserver tcpecho` subcommand whose greeting is fixed at startup by --banner, because a server decides whether to greet before it has read a byte: the two orders are two ServerPods, not two kinds of request. ServerPod grows a TCPProbe readiness option, since an origin that speaks neither HTTP nor gRPC can only be probed by opening a connection. The demo egress app grows a /tcp endpoint that reads before it writes, so an empty banner really does mean the origin stayed silent. Deploying those two origins one after the other cost 31.5s before the first subtest ran, because a serial DeployServerPod 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 begin. DeployServerPods renders every spec into one manifest in one namespace and applies it once. ko caches builds and pushes by image reference within an invocation, so servers sharing an ImportPath -- the common case, since one testserver binary backs them all -- build once, and the pods are admitted together, so later readiness waits are usually already satisfied when they are reached. Setup drops to 13.4s, with the second server going from 11.0s to 2.3s. DeployServerPod becomes a call into it, so there is one code path rather than two. Sharing a namespace is what makes a single apply legal, so specs that name different ones are rejected rather than papered over by applying twice; names have to be distinct as well, since spec.Name is the Pod, the Service and the app selector alike. --- demos/egress/main.go | 115 ++++++++++- demos/egress/main_test.go | 137 +++++++++++++ internal/e2e/fixtures/testserver/main.go | 3 +- internal/e2e/fixtures/testserver/tcpecho.go | 119 +++++++++++ .../e2e/fixtures/testserver/tcpecho_test.go | 151 ++++++++++++++ internal/e2e/manifest.go | 30 ++- internal/e2e/serverpod.go | 106 ++++++++-- internal/e2e/serverpod_test.go | 120 +++++++++-- .../e2e/suites/networking/networking_test.go | 189 +++++++++++++++++- 9 files changed, 923 insertions(+), 47 deletions(-) create mode 100644 internal/e2e/fixtures/testserver/tcpecho.go create mode 100644 internal/e2e/fixtures/testserver/tcpecho_test.go 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) {