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)) {