From f859011595b88286fad2b70b6a9120f8d4e7c234 Mon Sep 17 00:00:00 2001 From: Haiyan Meng Date: Mon, 17 Aug 2026 17:52:24 -0400 Subject: [PATCH] e2e: cover egress for non-HTTP traffic, in both speaking orders TestActorEgress and TestActorEgressHTTPS both have the client send the first bytes, so neither notices an egress path that waits for downstream data before dialing upstream, or that inspects those first bytes to route. SSH does not work that way: the server announces itself on accept. Add that shape to the egress demo and the networking suite: - /tcp on the egress demo opens a raw TCP connection and reads before it writes, so an empty banner really does mean the peer stayed silent rather than that the probe got the ordering wrong. - bannerserver is an in-cluster TCP origin that echoes on two ports: on one it greets on accept, on the other it stays silent until spoken to. Two ports, because the server has to decide whether to greet before it has read anything, so nothing in the request could select the behavior -- only the address dialed can. - TestActorEgressRawTCP dials both ports through one Actor and requires, per port, both the CONNECT record and the byte counters on the gateway's close-time access log, which is what shows the gateway relayed the payload rather than the Actor having reached the origin some other way. - TestActorEgressSSH is the same probe against github.com:22. --- demos/egress/bannerserver/main.go | 118 ++++++++++ demos/egress/egress.yaml.tmpl | 56 +++++ demos/egress/main.go | 121 +++++++++- demos/egress/main_test.go | 140 ++++++++++++ hack/install-demo-egress.sh | 2 + .../e2e/suites/networking/networking_test.go | 215 +++++++++++++++++- 6 files changed, 643 insertions(+), 9 deletions(-) create mode 100644 demos/egress/bannerserver/main.go diff --git a/demos/egress/bannerserver/main.go b/demos/egress/bannerserver/main.go new file mode 100644 index 000000000..467e8ccbe --- /dev/null +++ b/demos/egress/bannerserver/main.go @@ -0,0 +1,118 @@ +// 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. + +// Command bannerserver is a TCP origin for egress tests. It echoes what it is +// sent on two ports: on one it greets the peer first, on the other it stays +// silent until spoken to. +// +// Which port a test dials is how it selects the behavior. Nothing in the +// request can select it, because the server has to decide whether to greet +// before it has read anything -- that is what speaking first means. +package main + +import ( + "errors" + "io" + "log/slog" + "net" + "os" + "sync" + "time" +) + +// Banner is what the server writes on accept. Tests match on it, so it is a +// fixed string rather than anything derived from the connection. +const Banner = "TESTBANNER/1.0\r\n" + +func main() { + slog.SetDefault(slog.New(slog.NewJSONHandler(os.Stdout, nil))) + + greeting := listenerAddress("LISTEN_ADDRESS", ":2222") + quiet := listenerAddress("QUIET_LISTEN_ADDRESS", ":2223") + + var group sync.WaitGroup + group.Add(2) + go func() { defer group.Done(); accept(greeting, true) }() + go func() { defer group.Done(); accept(quiet, false) }() + group.Wait() +} + +func listenerAddress(variable, fallback string) string { + if address := os.Getenv(variable); address != "" { + return address + } + return fallback +} + +// accept serves address until it stops accepting, greeting each peer first when +// greet is set. A dead listener takes the process down rather than leaving it +// half-serving: a test that reached the surviving port would pass while the +// other silently answered nothing. +func accept(address string, greet bool) { + listener, err := net.Listen("tcp", address) + if err != nil { + slog.Error("banner server failed to listen", "address", address, "error", err) + os.Exit(1) + } + slog.Info("starting banner server", "address", address, "greets", greet) + + for { + connection, err := listener.Accept() + if err != nil { + slog.Error("banner server stopped accepting", "address", address, "error", err) + os.Exit(1) + } + go serve(connection, greet) + } +} + +// serve echoes until the peer goes away, announcing itself first when greet is +// set. +func serve(connection net.Conn, greet bool) { + // 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 connection.Close() + + if greet { + if err := connection.SetWriteDeadline(time.Now().Add(idleTimeout)); err != nil { + slog.Error("setting write deadline", "error", err) + return + } + if _, err := io.WriteString(connection, Banner); err != nil { + slog.Error("writing banner", "error", err) + return + } + } + + buffer := make([]byte, 4<<10) + for { + if err := connection.SetDeadline(time.Now().Add(idleTimeout)); err != nil { + return + } + n, err := connection.Read(buffer) + if n > 0 { + if _, writeErr := connection.Write(buffer[:n]); writeErr != nil { + return + } + } + if err != nil { + if !errors.Is(err, io.EOF) && !errors.Is(err, os.ErrDeadlineExceeded) { + slog.Error("reading from peer", "error", err) + } + return + } + } +} diff --git a/demos/egress/egress.yaml.tmpl b/demos/egress/egress.yaml.tmpl index b37a1f5d6..9ca4c0cb1 100644 --- a/demos/egress/egress.yaml.tmpl +++ b/demos/egress/egress.yaml.tmpl @@ -53,3 +53,59 @@ spec: onPause: Full onCommit: Full location: gs://${BUCKET_NAME}/ate-demo-egress/ + +--- + +# A TCP origin for the non-HTTP egress tests. +apiVersion: apps/v1 +kind: Deployment +metadata: + name: bannerserver + namespace: ate-demo-egress + labels: + app: bannerserver +spec: + replicas: 1 + selector: + matchLabels: + app: bannerserver + template: + metadata: + labels: + app: bannerserver + spec: + containers: + - name: bannerserver + image: ko://github.com/agent-substrate/substrate/demos/egress/bannerserver + ports: + # Two ports, because whether the origin speaks first has to be selected + # by the address the actor dials: the server decides before it reads. + - containerPort: 2222 + name: banner + - containerPort: 2223 + name: quiet + resources: + limits: + cpu: 100m + memory: 64Mi + requests: + cpu: 10m + memory: 64Mi + +--- + +apiVersion: v1 +kind: Service +metadata: + name: bannerserver + namespace: ate-demo-egress +spec: + selector: + app: bannerserver + ports: + - name: banner + port: 2222 + targetPort: banner + - name: quiet + port: 2223 + targetPort: quiet diff --git a/demos/egress/main.go b/demos/egress/main.go index 4c0288e12..ff8ca7947 100644 --- a/demos/egress/main.go +++ b/demos/egress/main.go @@ -13,14 +13,18 @@ // 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. +// policy. It accepts a URL, fetches it, and returns the upstream response, and +// on /tcp it opens a raw TCP connection so that egress can be exercised with +// something other than HTTP. package main import ( "encoding/json" + "errors" "fmt" "io" "log/slog" + "net" "net/http" "net/url" "os" @@ -101,9 +105,124 @@ func newHandler(client *http.Client) http.Handler { } writeJSON(w, response.StatusCode, fetchResponse{StatusCode: response.StatusCode, Body: string(body)}) }) + mux.HandleFunc("/tcp", handleTCPProbe) return mux } +// 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) + writeTCPProbeJSON(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 { + writeTCPProbeJSON(w, http.StatusBadRequest, tcpProbeResponse{Error: fmt.Sprintf("invalid JSON payload: %v", err)}) + return + } + if _, _, err := net.SplitHostPort(input.Address); err != nil { + writeTCPProbeJSON(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 { + writeTCPProbeJSON(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 { + writeTCPProbeJSON(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 { + writeTCPProbeJSON(w, http.StatusBadGateway, tcpProbeResponse{Error: fmt.Sprintf("reading banner from %s: %v", input.Address, err)}) + return + } + + response := tcpProbeResponse{Banner: string(banner)} + if input.Send == "" { + writeTCPProbeJSON(w, http.StatusOK, response) + return + } + + if err := connection.SetWriteDeadline(time.Now().Add(timeout)); err != nil { + writeTCPProbeJSON(w, http.StatusBadGateway, tcpProbeResponse{Error: fmt.Sprintf("setting write deadline: %v", err)}) + return + } + if _, err := io.WriteString(connection, input.Send); err != nil { + writeTCPProbeJSON(w, http.StatusBadGateway, tcpProbeResponse{Error: fmt.Sprintf("writing to %s: %v", input.Address, err)}) + return + } + received, err := readWithTimeout(connection, readBytes, timeout) + if err != nil { + writeTCPProbeJSON(w, http.StatusBadGateway, tcpProbeResponse{Error: fmt.Sprintf("reading reply from %s: %v", input.Address, err)}) + return + } + response.Received = string(received) + writeTCPProbeJSON(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 writeTCPProbeJSON(w http.ResponseWriter, status int, response tcpProbeResponse) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(response) +} + 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 6cd203732..f2de0c6a2 100644 --- a/demos/egress/main_test.go +++ b/demos/egress/main_test.go @@ -18,6 +18,7 @@ import ( "encoding/json" "errors" "io" + "net" "net/http" "net/http/httptest" "strings" @@ -102,6 +103,145 @@ 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 records that silence is a result, not an error: a +// client-speaks-first peer yields an empty banner and a 200, so a test can tell +// the two shapes apart. +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/hack/install-demo-egress.sh b/hack/install-demo-egress.sh index e4e0074a9..ec74950c4 100644 --- a/hack/install-demo-egress.sh +++ b/hack/install-demo-egress.sh @@ -40,6 +40,8 @@ demo-egress_deploy() { # ("egress"), the same way demo-counter gets "deployment/counter". The old # "egress-deployment" name was NotFound on every successful deploy. run_kubectl rollout status deployment/egress -n ate-demo-egress --timeout=300s + # The TCP origin for the non-HTTP egress tests. + run_kubectl rollout status deployment/bannerserver -n ate-demo-egress --timeout=300s run_kubectl wait --for=condition=Ready actortemplate/egress -n ate-demo-egress --timeout=300s } diff --git a/internal/e2e/suites/networking/networking_test.go b/internal/e2e/suites/networking/networking_test.go index 59871dc76..473530565 100644 --- a/internal/e2e/suites/networking/networking_test.go +++ b/internal/e2e/suites/networking/networking_test.go @@ -20,6 +20,7 @@ import ( "fmt" "io" "net/http" + "strconv" "strings" "testing" "time" @@ -66,13 +67,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", egressTemplate) router := mustRouterClient(t, ctx) @@ -111,23 +112,172 @@ func TestActorEgressHTTPS(t *testing.T) { assertEgressGatewayConnect(t, ctx, since, actorName, "443") } +// 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. +// +// The two subtests dial different ports of the same origin, because that is the +// only way to select the behavior: the server decides whether to greet before +// it has read anything, so no field in the request could choose for it. Distinct +// ports also keep the access-log assertions unambiguous while both subtests +// share one Actor. +func TestActorEgressRawTCP(t *testing.T) { + // The greeting from demos/egress/bannerserver, which must stay in step with + // the Banner constant there. + const banner = "TESTBANNER/1.0\r\n" + + tests := []struct { + name string + port int + // 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", port: bannerServerPort, wantBanner: banner, timeout: "10s"}, + {name: "client speaks first", port: bannerServerQuietPort, wantBanner: "", timeout: "2s"}, + } + + ctx := context.Background() + clusterIP := bannerServerClusterIP(t, ctx) + actorName, _ := createAndResumeActor(t, ctx, "egress-tcp", egressTemplate) + 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. The slack absorbs clock skew with the gateway's node. + since := metav1.NewTime(time.Now().Add(-1 * time.Minute)) + // Dial by address: the sandbox does not resolve cluster Service names. + address := fmt.Sprintf("%s:%d", clusterIP, test.port) + + // 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) + } + // The probe reads before it writes, so this distinguishes an origin + // that spoke unprompted through the tunnel from one that did not. + 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.port) + assertEgressGatewayConnect(t, ctx, since, actorName, port) + assertEgressGatewayTunneledBytes(t, ctx, since, actorName, port) + }) + } +} + +// TestActorEgressSSH is TestActorEgressRawTCP against 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", egressTemplate) + 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") +} + +// The ports the banner server Service publishes, from +// demos/egress/egress.yaml.tmpl. The origin greets on the first and stays +// silent until spoken to on the second. +const ( + bannerServerPort = 2222 + bannerServerQuietPort = 2223 +) + +// bannerServerClusterIP returns the address of the in-cluster TCP origin the +// raw-TCP test dials. +func bannerServerClusterIP(t *testing.T, ctx context.Context) string { + t.Helper() + service, err := e2e.GetClients().K8s.CoreV1().Services(egressTemplate.namespace).Get(ctx, "bannerserver", metav1.GetOptions{}) + if err != nil { + t.Fatalf("getting Service %s/bannerserver: %v (deploy the fixture with %s)", egressTemplate.namespace, err, egressTemplate.deployFlag) + } + if service.Spec.ClusterIP == "" || service.Spec.ClusterIP == corev1.ClusterIPNone { + t.Fatalf("Service %s/bannerserver has no cluster IP to dial: %q", egressTemplate.namespace, service.Spec.ClusterIP) + } + return service.Spec.ClusterIP +} + // fetchThroughEgressActor asks the egress demo Actor to fetch url and returns -// the status and body it echoes back. Retries a non-200 response for up to -// 30s: ResumeActor can return before its route reaches atenet-router's xDS -// snapshot, and a request sent in that window sees a transient 503. +// 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) { t.Helper() payload, err := json.Marshal(map[string]string{"url": url}) if err != nil { t.Fatalf("marshaling the fetch request for %s: %v", url, err) } + return postThroughEgressActor(t, ctx, router, actorRef, "/", payload) +} + +// postThroughEgressActor POSTs payload to path on the egress demo Actor and +// returns the status and body it answers with. Retries a non-200 response for +// up to 30s: ResumeActor can return before its route reaches atenet-router's +// xDS snapshot, and a request sent in that window sees a transient 503. +func postThroughEgressActor(t *testing.T, ctx context.Context, router *e2e.RouterClient, actorRef resources.ActorRef, path string, payload []byte) (int, []byte) { + t.Helper() const timeout = 30 * time.Second deadline := time.Now().Add(timeout) for { - response, err := router.PostJSON(ctx, actorRef, "/", payload) + response, err := router.PostJSON(ctx, actorRef, path, payload) if err != nil { - t.Fatalf("POST %s to egress Actor through ingress: %v", url, err) + t.Fatalf("POST %s to egress Actor through ingress: %v", path, err) } body, err := io.ReadAll(response.Body) response.Body.Close() @@ -137,7 +287,7 @@ func fetchThroughEgressActor(t *testing.T, ctx context.Context, router *e2e.Rout if response.StatusCode == http.StatusOK || time.Now().After(deadline) { return response.StatusCode, body } - t.Logf("fetch through egress Actor returned HTTP %d; retrying...", response.StatusCode) + t.Logf("POST %s to egress Actor returned HTTP %d; retrying...", path, response.StatusCode) time.Sleep(1 * time.Second) } } @@ -163,6 +313,55 @@ func assertEgressGatewayConnect(t *testing.T, ctx context.Context, since metav1. }) } +// 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 +} + // waitForAccessLog polls the atenet-egress access log, across every gateway // replica, until predicate accepts the lines written since. func waitForAccessLog(t *testing.T, ctx context.Context, since metav1.Time, want string, predicate func(lines []string) (bool, error)) {