From c254ea1af3e19c6c9400be709d05ec5c488f941e Mon Sep 17 00:00:00 2001 From: Haiyan Meng Date: Mon, 17 Aug 2026 17:52:24 -0400 Subject: [PATCH 1/4] 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)) { From 9ccc9c4fed2bfc709d5141f35ebfd17b9139c7dc Mon Sep 17 00:00:00 2001 From: Haiyan Meng Date: Tue, 18 Aug 2026 10:39:46 -0400 Subject: [PATCH 2/4] egress: carry non-HTTP traffic through the sdsmint MITM leg The MITM gateway accepted a CONNECT for any destination and then broke everything that was not TLS or cleartext HTTP. Two separate faults, both on mitm_listener: tls_inspector peeks for a ClientHello, so on a server-speaks-first protocol it waited for a client that was itself waiting for the origin's banner. The tunnel opened, nothing crossed it, and the actor eventually timed out. Giving up quickly is the only way out of that: a 1s listener_filters_timeout with continue_on_listener_filters_timeout hands the socket on with no transport protocol detected, which Envoy defaults to raw_buffer. Only connections that send nothing pay the second. The raw_buffer chain was then an HTTP connection manager, so "not TLS" was treated as "cleartext HTTP" and SSH got its first bytes parsed as a request line and dropped, silently. http_inspector now splits raw_buffer again, the cleartext chain is confined to what it actually recognises, and a third chain tcp_proxies the rest. That last chain needs a destination and has no name to resolve: an opaque stream carries no SNI and no Host, and the CONNECT authority is the only place the address appears. proxy_protocol_config on the CONNECT route looks like the answer but is not -- it reports the downstream connection's own addresses, which here is the gateway's :443. Instead set_filter_state records the authority ext_proc just authorized, internal_upstream carries it across the loopback hop, and an ORIGINAL_DST cluster consults that key ahead of everything else. Also drops the route timeout on the CONNECT itself. A tunnel is a session, not a request, and the default 15s was a ceiling on how long an actor could hold any TCP connection open. Nothing here weakens authorization. The passthrough leg cannot name what it carries, but ext_proc has already policed the CONNECT against the same IP:port this chain dials; before, these connections were not blocked, they were accepted and then mangled. TestActorEgressRawTCP and TestActorEgressSSH now pass against the sdsmint gateway, with TestActorEgressHTTP and the sdsmint suite still green. TestActorEgressHTTPS still fails, as it did before: nothing in the cluster trusts the MITM anchor, which is by design and not reachable from Envoy config. --- .../atenet-egress-with-sdsmint.yaml | 148 +++++++++++++++++- 1 file changed, 144 insertions(+), 4 deletions(-) diff --git a/manifests/ate-install/atenet-egress-with-sdsmint.yaml b/manifests/ate-install/atenet-egress-with-sdsmint.yaml index 72b3e1979..2c17eb0ef 100644 --- a/manifests/ate-install/atenet-egress-with-sdsmint.yaml +++ b/manifests/ate-install/atenet-egress-with-sdsmint.yaml @@ -122,6 +122,12 @@ data: # egress_forward_proxy here would be a raw TCP passthrough # with no per-destination visibility at all. cluster: mitm_internal + # A CONNECT tunnel is a session, not a request, so the + # route's default 15s stream timeout would be a ceiling on + # how long an actor may hold any TCP connection open -- + # fine for a request/response fetch, wrong for SSH or any + # other protocol that idles. Idle timeouts still apply. + timeout: 0s upgrade_configs: - upgrade_type: CONNECT connect_config: {} @@ -158,6 +164,33 @@ data: response_body_mode: NONE request_trailer_mode: SKIP response_trailer_mode: SKIP + # Carry the CONNECT authority across the internal-listener hop. + # + # atunnel takes the authority from SO_ORIGINAL_DST, so it is always + # an IP:port, and it is the only place the real destination appears + # for a tunnel that carries neither SNI nor a Host header. The MITM + # leg's two HTTP chains recover the destination from inside the + # tunnel and so never needed this; the raw TCP chain has nothing to + # read and does. + # + # This is the key an ORIGINAL_DST cluster consults first, ahead of + # metadata and headers, so egress_tcp_passthrough dials it directly. + # It survives the hop only because mitm_internal wraps its transport + # socket in internal_upstream, which is what makes + # shared_with_upstream mean anything. + # + # Placed after ext_proc so a denied CONNECT never reaches it: the + # address recorded here is the authority ext_proc just authorized, + # not one the actor gets to restate. + - name: envoy.filters.http.set_filter_state + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.http.set_filter_state.v3.Config + on_request_headers: + - object_key: envoy.network.transport_socket.original_dst_address + format_string: + text_format_source: + inline_string: "%REQ(:AUTHORITY)%" + shared_with_upstream: ONCE - name: envoy.filters.http.router typed_config: "@type": type.googleapis.com/envoy.extensions.filters.http.router.v3.Router @@ -173,18 +206,42 @@ data: # tunnelled TLS with a leaf sdsmint mints for the SNI, reads the real # Host, and re-originates. # - # Two chains, selected by what the tunnel actually carries. tls_inspector - # tags a ClientHello "tls" and anything else "raw_buffer", so a cleartext - # HTTP tunnel gets an HTTP chain instead of being fed to a TLS transport - # socket, which would fail the handshake and close the tunnel. + # Three chains, selected by what the tunnel actually carries. + # tls_inspector tags a ClientHello "tls" and anything else "raw_buffer", + # so a cleartext tunnel gets an HTTP chain instead of being fed to a TLS + # transport socket, which would fail the handshake and close the tunnel. + # http_inspector then splits "raw_buffer" again, because "not TLS" is not + # the same claim as "HTTP": without it, SSH and every other non-HTTP + # protocol lands on an HTTP connection manager that parses the first bytes + # of the stream as a request line, finds no request, and drops the + # connection with nothing in the access log to say why. # --------------------------------------------------------------------- - name: mitm_listener stat_prefix: mitm internal_listener: {} + # Both inspectors work by peeking at bytes the client sends first, so on + # a server-speaks-first protocol -- SSH, SMTP, MySQL -- they wait for a + # client that is itself waiting for the origin, and the tunnel deadlocks + # until the actor's own timeout fires. Giving up quickly and continuing + # is the only way out: continue_on_listener_filters_timeout hands the + # socket to the chains with no transport protocol detected, and Envoy + # defaults that to raw_buffer, which is exactly the passthrough chain + # such a connection belongs on. The cost is paid only by connections + # that send nothing; a client that speaks is classified immediately. One + # second is far longer than a local actor needs to emit a ClientHello, + # and short enough to stay well inside the probe timeouts. + listener_filters_timeout: 1s + continue_on_listener_filters_timeout: true listener_filters: - name: envoy.filters.listener.tls_inspector typed_config: "@type": type.googleapis.com/envoy.extensions.filters.listener.tls_inspector.v3.TlsInspector + # Must come second: it declines to run at all once a transport protocol + # other than raw_buffer has been detected, which is how it keeps its + # hands off the TLS chain. + - name: envoy.filters.listener.http_inspector + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.listener.http_inspector.v3.HttpInspector filter_chains: - filter_chain_match: transport_protocol: tls @@ -277,8 +334,14 @@ data: # The cleartext chain. Nothing to terminate and nothing to mint: the # Host header is already in the clear, so this leg reads the # destination directly rather than from a certificate it issued. + # + # application_protocols is what confines it to traffic http_inspector + # actually recognised as HTTP. It used to match all of raw_buffer, which + # made it the destination for every non-TLS protocol rather than just + # this one. - filter_chain_match: transport_protocol: raw_buffer + application_protocols: ["http/1.0", "http/1.1", "h2c"] filters: - name: envoy.filters.network.http_connection_manager typed_config: @@ -328,11 +391,70 @@ data: typed_config: "@type": type.googleapis.com/envoy.extensions.filters.http.router.v3.Router + # The passthrough chain: everything the two inspectors could not claim, + # plus everything from a client that sent nothing at all. No filter + # chain match beyond raw_buffer, so it is the fallback the more specific + # chains above fall out of. + # + # This leg is deliberately blind. It does not decrypt, does not mint, + # and cannot name what it is carrying -- an opaque byte stream has no + # Host and no SNI to police. The authorization that stands behind it is + # the one ext_proc already applied to the CONNECT on the way in, against + # the same IP:port this chain now dials. Nothing is weakened by adding + # it: without it these connections were not blocked, they were accepted + # and then silently mangled. + - filter_chain_match: + transport_protocol: raw_buffer + filters: + - name: envoy.filters.network.tcp_proxy + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.network.tcp_proxy.v3.TcpProxy + stat_prefix: mitm_passthrough + cluster: egress_tcp_passthrough + access_log: + - name: envoy.access_loggers.file + typed_config: + "@type": type.googleapis.com/envoy.extensions.access_loggers.file.v3.FileAccessLog + path: /dev/stdout + log_format: + json_format: + leg: passthrough + time: "%START_TIME%" + # Always empty on a healthy connection: reaching this + # chain means tls_inspector found no ClientHello to read a + # name out of. A value here means a TLS connection was + # classified as raw_buffer and is being forwarded without + # being minted for, which is a routing bug, not traffic. + sni: "%REQUESTED_SERVER_NAME%" + flags: "%RESPONSE_FLAGS%" + duration_ms: "%DURATION%" + bytes_sent: "%BYTES_SENT%" + bytes_rcvd: "%BYTES_RECEIVED%" + upstream: "%UPSTREAM_HOST%" + upstream_failure: "%UPSTREAM_TRANSPORT_FAILURE_REASON%" + termination: "%CONNECTION_TERMINATION_DETAILS%" + clusters: # The only way into mitm_listener. An internal address, not a socket, so # the MITM leg has no listening port anywhere in the pod's netns. - name: mitm_internal connect_timeout: 1s + # An internal address carries bytes and nothing else by default, so the + # CONNECT authority the set_filter_state filter recorded on Listener A + # would stop here. internal_upstream is the seam that lets structured + # state cross with the stream: filter state marked shared_with_upstream + # is merged into the internal listener's own downstream connection, + # where the ORIGINAL_DST cluster can find it. The wrapped socket is + # raw_buffer because there is no real network hop to secure -- this is a + # userspace loopback inside one process. + transport_socket: + name: envoy.transport_sockets.internal_upstream + typed_config: + "@type": type.googleapis.com/envoy.extensions.transport_sockets.internal_upstream.v3.InternalUpstreamTransport + transport_socket: + name: envoy.transport_sockets.raw_buffer + typed_config: + "@type": type.googleapis.com/envoy.extensions.transport_sockets.raw_buffer.v3.RawBuffer load_assignment: cluster_name: mitm_internal endpoints: @@ -468,6 +590,24 @@ data: "@type": type.googleapis.com/envoy.extensions.upstreams.http.v3.HttpProtocolOptions explicit_http_config: http_protocol_options: {} + + # The passthrough chain's upstream. Not a dynamic forward proxy like the + # other two: there is no name to resolve. The destination arrives as the + # literal IP:port the actor's kernel recorded in SO_ORIGINAL_DST, relayed + # here as the envoy.network.transport_socket.original_dst_address filter + # state object, which ORIGINAL_DST consults ahead of metadata, headers, + # and the socket's own restored local address -- the last of which is what + # this cluster would otherwise use, and which on an internal listener + # points back into this process. + # + # No transport socket, so no upstream TLS: the actor is speaking some + # protocol this gateway does not parse, and wrapping that stream in a TLS + # session the actor did not ask for and cannot see the peer of would break + # every protocol it is meant to carry. + - name: egress_tcp_passthrough + type: ORIGINAL_DST + lb_policy: CLUSTER_PROVIDED + connect_timeout: 5s --- apiVersion: apps/v1 kind: Deployment From d62e83316adb234f37105dff3172c94031578740 Mon Sep 17 00:00:00 2001 From: Haiyan Meng Date: Tue, 18 Aug 2026 11:59:26 -0400 Subject: [PATCH 3/4] Add e2e test for long-lived streams --- demos/egress/demo.sh | 452 ++++++++++++++++++ demos/egress/egress.yaml.tmpl | 56 +++ demos/egress/main.go | 9 +- demos/egress/stream.go | 317 ++++++++++++ demos/egress/stream_test.go | 250 ++++++++++ demos/egress/streamserver/main.go | 154 ++++++ go.mod | 2 +- hack/install-demo-egress.sh | 2 + .../e2e/suites/networking/networking_test.go | 198 +++++++- manifests/ate-install/atenet-egress.yaml | 8 + 10 files changed, 1433 insertions(+), 15 deletions(-) create mode 100755 demos/egress/demo.sh create mode 100644 demos/egress/stream.go create mode 100644 demos/egress/stream_test.go create mode 100644 demos/egress/streamserver/main.go diff --git a/demos/egress/demo.sh b/demos/egress/demo.sh new file mode 100755 index 000000000..f12d5529f --- /dev/null +++ b/demos/egress/demo.sh @@ -0,0 +1,452 @@ +#!/usr/bin/env bash + +# 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. + +# A presenter-driven walkthrough of the egress path. See README.md. +# +# This drives internal/e2e/fixtures/egressprobe deployed as an Actor. The probe +# is a test fixture, not a product surface -- everything customer-facing here is +# the narration and the formatting, and the raw Result is one --verbose away for +# whichever engineer in the room asks for it. +# +# The demo makes no claim the suite does not already assert. What it adds is an +# order to put the claims in, and a destination that echoes, so the injected +# header can be read rather than inferred. + +set -euo pipefail + +ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)" +cd "${ROOT}" + +# --------------------------------------------------------------------------- +# Configuration +# --------------------------------------------------------------------------- + +# The atespace and actor names are NOT free choices. internal/extproc/hardcoded.go +# keys the policy table on the atespace/name pair, so these strings are what +# select a policy. Renaming an actor here silently selects no policy at all, +# which presents as a denied CONNECT rather than as a configuration error. +ATESPACE="acme-prod" +ACTOR_BARE="actor-without-github-access" # KindAllowByHostname -- allowlist, no credential +ACTOR_INJECT="actor-with-github-access" # KindBasicCredentialInject -- allowlist + injection +ACTOR_DENIED="quarantined" # KindDenyAll -- no egress at all + +TEMPLATE_NS="ate-demo-egress" +TEMPLATE_NAME="egressprobe" +# Three actors on stage at once, plus a spare to absorb a worker still draining. +# The fixture ships with 2 because the e2e suite runs one actor at a time; a live +# demo cannot afford a suspend/resume between acts. +POOL_REPLICAS=4 + +ROUTER="127.0.0.1:8080" +# TEST-NET-1, and deliberately unroutable: nothing on the internet answers it, so +# any response at all proves the connection was intercepted and rebuilt as a +# tunnel. A real destination IP would make a success ambiguous -- it could mean +# the nftables REDIRECT never fired and the Actor simply dialed out. +DESTINATION="192.0.2.1:443" + +ECHO_HOST="postman-echo.com" +GITHUB_HOST="api.github.com" +UNLISTED_HOST="example.com" # in sdsmintd's --allow, in no actor's policy + +VERBOSE=0 +PAUSE=1 +WITH_ECHO=0 + +# --------------------------------------------------------------------------- +# Presentation +# --------------------------------------------------------------------------- + +if [[ -t 1 ]] && [[ "${NO_COLOR:-}" == "" ]]; then + B=$'\033[1m'; DIM=$'\033[2m'; GREEN=$'\033[32m'; RED=$'\033[31m' + CYAN=$'\033[36m'; R=$'\033[0m' +else + B=""; DIM=""; GREEN=""; RED=""; CYAN=""; R="" +fi + +act_number=0 + +# act prints the frame for one claim: what we are about to do and what should +# happen. Stating the expected result BEFORE running is the whole difference +# between a demo and a debugging session -- the audience gets to be right. +act() { + act_number=$((act_number + 1)) + echo + echo "${B}${CYAN}━━━ Act ${act_number} · ${1} ━━━${R}" + echo + echo " ${2}" + echo + # read fails at EOF, which is normal when stdin is not a terminal -- under + # `set -e` that would end the demo rather than skip the pause. + [[ "${PAUSE}" == "1" ]] && { printf ' %s[enter]%s' "${DIM}" "${R}"; read -r || true; echo; } + return 0 +} + +say() { echo " ${1}"; } +verdict() { echo; echo " ${B}${GREEN}▸ ${1}${R}"; } +nope() { echo; echo " ${B}${RED}▸ ${1}${R}"; } +field() { printf ' %-28s %s\n' "${1}" "${2}"; } +note() { echo " ${DIM}${1}${R}"; } +die() { echo "${RED}error:${R} ${1}" >&2; exit 1; } + +# --------------------------------------------------------------------------- +# Driving the probe +# --------------------------------------------------------------------------- + +# probe ACTOR SNI PATH -- one trip through the gateway, as that actor. +# +# via=direct is what makes this an Actor test rather than a proxy test: the probe +# dials as if no gateway existed, ateom's nftables rule REDIRECTs it, and atunnel +# builds the tunnel with a certificate the probe never sees. Nothing in the +# workload knows a gateway is involved. +probe() { + local actor="${1}" sni="${2}" path="${3}" body result + body=$(printf '{"via":"direct","destination":"%s","sni":"%s","request_path":"%s"}' \ + "${DESTINATION}" "${sni}" "${path}") + + if [[ "${VERBOSE}" == "1" ]]; then + echo " ${DIM}POST http://${ROUTER}/probe${R}" + echo " ${DIM}Host: ${actor}.${ATESPACE}.actors.resources.substrate.ate.dev${R}" + echo " ${DIM}${body}${R}" + echo + fi + + # The router routes on Host, not on a path or a header: this is the same way + # every Actor in the system is reached. + result=$(curl -sS --max-time 90 -X POST "http://${ROUTER}/probe" \ + -H "Host: ${actor}.${ATESPACE}.actors.resources.substrate.ate.dev" \ + -H 'Content-Type: application/json' \ + -d "${body}") || die "the probe API did not answer -- is the port-forward still up?" + + if [[ "${result}" != "{"* ]]; then + die "router answered with something that is not a Result: ${result}" + fi + if [[ "${VERBOSE}" == "1" ]]; then + echo "${result}" | jq . | sed 's/^/ /' + echo + fi + echo "${result}" +} + +# echo_headers pulls the request headers back out of an echo service's response. +# Empty when the body is not the JSON we expect, which keeps a Cloudflare error +# page from being reported as "no headers arrived". +echo_headers() { + echo "${1}" | jq -r 'try (.http_body | fromjson | .headers) // empty' +} + +# --------------------------------------------------------------------------- +# Acts +# --------------------------------------------------------------------------- + +act_zero() { + act "Nothing up my sleeve" \ +"Before anything runs: this workload holds no credentials. + Two files say so -- the code that builds the request, and the template that + deploys it." + + say "${B}The request the workload sends${R} ${DIM}(probeapi.go)${R}" + echo + sed -n '108,110p' internal/e2e/fixtures/egressprobe/probeapi/probeapi.go | sed 's/^/ /' + echo + say "${B}The ActorTemplate that deploys it${R} ${DIM}(egressprobe-actor.yaml.tmpl)${R}" + echo + note " no secretKeyRef, no token env var, no credential volume:" + echo + grep -n 'secretKeyRef\|env:\|volumeMounts:' \ + internal/e2e/fixtures/egressprobe/egressprobe-actor.yaml.tmpl \ + | sed 's/^/ /' || say " ${GREEN}(no matches -- there are none)${R}" + + verdict "The workload cannot authenticate to anything. Keep that in mind." +} + +act_echo_bare() { + act "An Actor with no credential, seen from the far end" \ +"${ACTOR_BARE} is allowed to reach ${ECHO_HOST}, and its policy attaches nothing. + The echo service reports the request exactly as it received it -- so this is + the baseline: what the workload sent, and only that." + + local result headers + result=$(probe "${ACTOR_BARE}" "${ECHO_HOST}" "/get") + headers=$(echo_headers "${result}") + [[ -n "${headers}" ]] || { nope "no echo body came back"; echo "${result}" | jq .; return; } + + echo "${headers}" | jq -r 'to_entries[] | " \(.key): \(.value)"' | grep -iv '^ *x-\|cf-\|cdn-' || true + echo + + if echo "${headers}" | jq -e 'has("authorization")' >/dev/null; then + nope "An authorization header arrived. It should not have -- check the policy table." + else + verdict "No authorization header. The destination saw exactly what the workload sent." + fi +} + +act_echo_inject() { + act "The same request, from a different Actor" \ +"${ACTOR_INJECT} now. Same image, same code, same request, same destination. + The only thing that changed is which Actor is making it." + + local result headers auth + result=$(probe "${ACTOR_INJECT}" "${ECHO_HOST}" "/get") + headers=$(echo_headers "${result}") + [[ -n "${headers}" ]] || { nope "no echo body came back"; echo "${result}" | jq .; return; } + + echo "${headers}" | jq -r 'to_entries[] | " \(.key): \(.value)"' | grep -iv '^ *x-\|cf-\|cdn-' || true + echo + + auth=$(echo "${headers}" | jq -r '.authorization // empty') + if [[ -z "${auth}" ]]; then + nope "No authorization header arrived. The injection did not fire." + return + fi + + field "authorization" "${B}${auth}${R}" + echo + say "That header was added by the gateway. The workload never held that value," + say "cannot read it, and has no way to discover it." + echo + + # A security-minded audience asks this before you finish the sentence, so get + # ahead of it: proving a credential goes OUT is only half the claim if + # substrate-internal identity headers leak out alongside it. + say "${B}And what did not leak:${R}" + if echo "${headers}" | jq -e 'keys[] | select(startswith("x-ate-") or . == "x-forwarded-client-cert")' >/dev/null 2>&1; then + nope "substrate-internal headers reached the destination -- extprocd's header hygiene broke." + else + field "x-ate-*" "${GREEN}absent${R}" + field "x-forwarded-client-cert" "${GREEN}absent${R}" + fi + + verdict "A credential the workload never had, and nothing about substrate, reached the destination." +} + +act_github() { + act "A real third party reacts to it" \ +"An echo service will print anything. ${GITHUB_HOST} has an opinion. + Both Actors ask GitHub the same question -- who am I? -- and GitHub + distinguishes 'you sent nothing' from 'you sent a token and it is wrong'." + + local bare inject + bare=$(probe "${ACTOR_BARE}" "${GITHUB_HOST}" "/user") + inject=$(probe "${ACTOR_INJECT}" "${GITHUB_HOST}" "/user") + + say "${B}${ACTOR_BARE}${R} ${DIM}(no injection)${R}" + field "status" "$(echo "${bare}" | jq -r '.http_status')" + field "message" "$(echo "${bare}" | jq -r 'try (.http_body|fromjson.message) // .http_body')" + field "x-ratelimit-limit" "$(echo "${bare}" | jq -r '.http_headers["X-Ratelimit-Limit"][0] // "(absent)"')" + echo + say "${B}${ACTOR_INJECT}${R} ${DIM}(credential injected)${R}" + field "status" "$(echo "${inject}" | jq -r '.http_status')" + field "message" "$(echo "${inject}" | jq -r 'try (.http_body|fromjson.message) // .http_body')" + field "x-ratelimit-limit" "$(echo "${inject}" | jq -r '.http_headers["X-Ratelimit-Limit"][0] // "(absent)"')" + echo + + note "Both are 401 because the demo ships a deliberately invalid token -- there is" + note "no live secret anywhere in this repo. Two things separate them." + note "" + note "The wording. \"Requires authentication\" is GitHub saying nothing arrived;" + note "\"Bad credentials\" is GitHub saying a bearer token arrived and was rejected." + note "" + note "The rate limit. An anonymous request gets a bucket -- 60/hour by source IP." + note "A rejected credential gets no bucket at all: GitHub cannot attribute it to" + note "an account, and will not call it anonymous either. The header disappears." + + verdict "GitHub confirms independently that a credential reached it -- and only for one Actor." +} + +act_denied_host() { + act "The credential is not a blank cheque" \ +"${ACTOR_INJECT} can reach ${GITHUB_HOST}. Here it tries ${UNLISTED_HOST}, which + the gateway will happily mint a certificate for but no policy permits. + Injection and destination control are the same decision, not two features." + + local result + result=$(probe "${ACTOR_INJECT}" "${UNLISTED_HOST}" "/") + + field "status" "$(echo "${result}" | jq -r '.http_status')" + field "body" "$(echo "${result}" | jq -r '.http_body' | head -1)" + echo + + if echo "${result}" | jq -e '.http_body | test("egress denied")' >/dev/null 2>&1; then + verdict "Refused by the gateway, naming the policy that refused it." + else + nope "That 403 did not come from the gateway -- read the body before believing it." + fi +} + +act_quarantined() { + act "The floor" \ +"${ACTOR_DENIED}'s policy is deny-all. Same image again, same request. + Watch where this one fails: earlier than every previous act." + + local result + result=$(probe "${ACTOR_DENIED}" "${GITHUB_HOST}" "/user") + + field "connected" "$(echo "${result}" | jq -r '.connected')" + field "handshake_ok" "$(echo "${result}" | jq -r '.handshake_ok')" + field "handshake_error" "$(echo "${result}" | jq -r '.handshake_error // "none"')" + echo + + note "connected: true is not egress. The REDIRECT is local, so the socket comes up" + note "inside the sandbox before atunnel has spoken to the gateway at all. The" + note "gateway then refuses, atunnel closes, and the TLS handshake dies on the reset." + note "Note the peer address: 192.0.2.1 is unroutable, so that reset came from" + note "inside the cluster. Nothing was ever dialed." + + if [[ "$(echo "${result}" | jq -r '.handshake_ok')" == "false" ]]; then + verdict "No bytes left the sandbox. The same mechanism that grants also denies." + else + nope "The handshake completed. A deny-all Actor reached the internet -- stop the demo." + fi +} + +closing() { + echo + echo "${B}${CYAN}━━━ What the customer just saw ━━━${R}" + echo + say "One image. One codebase. No credentials anywhere in the workload." + say "Four different outcomes, decided entirely by ${B}which Actor made the call${R}:" + echo + # Plain ASCII in the left column: printf pads by byte count, so a multibyte + # arrow here silently shifts the whole table one place left. + field "${ACTOR_BARE}" "reaches the allowlist, bare" + field "${ACTOR_INJECT}" "reaches the allowlist, with a credential it never held" + field " ...to ${UNLISTED_HOST}" "refused, by name" + field "${ACTOR_DENIED}" "never leaves the sandbox" + echo + say "There is no secret to steal from the workload, no config naming an identity," + say "and no request parameter that selects one." + echo +} + +# --------------------------------------------------------------------------- +# Setup and teardown +# --------------------------------------------------------------------------- + +kate() { go run ./cmd/kubectl-ate "$@"; } + +setup() { + echo "${B}Setting up${R}" + command -v jq >/dev/null || die "jq is required" + + kubectl get ns "${TEMPLATE_NS}" >/dev/null 2>&1 \ + || die "${TEMPLATE_NS} not found -- run: hack/install-ate.sh --deploy-demo-egress" + + echo " scaling the worker pool to ${POOL_REPLICAS} (three actors on stage at once)" + kubectl patch workerpool "${TEMPLATE_NAME}" -n "${TEMPLATE_NS}" --type=merge \ + -p "{\"spec\":{\"replicas\":${POOL_REPLICAS}}}" + + kate get atespaces "${ATESPACE}" >/dev/null 2>&1 || kate create atespace "${ATESPACE}" + + for actor in "${ACTOR_BARE}" "${ACTOR_INJECT}" "${ACTOR_DENIED}"; do + echo " ${actor}" + kate create actor "${actor}" -t "${TEMPLATE_NS}/${TEMPLATE_NAME}" -a "${ATESPACE}" 2>/dev/null || true + kate resume actor "${actor}" -a "${ATESPACE}" + done + + echo + echo " Start the port-forward in another shell, then run the demo:" + echo " ${DIM}kubectl -n ate-system port-forward svc/atenet-router 8080:80${R}" + echo " ${DIM}demos/egress/demo.sh${R}" +} + +teardown() { + echo "${B}Tearing down${R}" + for actor in "${ACTOR_BARE}" "${ACTOR_INJECT}" "${ACTOR_DENIED}"; do + kate suspend actor "${actor}" -a "${ATESPACE}" 2>/dev/null || true + kate delete actor "${actor}" -a "${ATESPACE}" 2>/dev/null || true + done + # Back to what the e2e fixture expects, so a later suite run is not surprised. + kubectl patch workerpool "${TEMPLATE_NAME}" -n "${TEMPLATE_NS}" --type=merge \ + -p '{"spec":{"replicas":2}}' +} + +# preflight fails loudly and early rather than partway through an act, which is a +# bad time to discover an Actor never resumed. +# +# Reaching the router is not enough to check: it answers for an Actor that does +# not exist too, just with a different body. Every Actor gets probed, because a +# demo that dies at Act 5 has already wasted the audience's time. +preflight() { + command -v jq >/dev/null || die "jq is required" + + local actors=("${ACTOR_BARE}" "${ACTOR_INJECT}" "${ACTOR_DENIED}") actor answer + for actor in "${actors[@]}"; do + answer=$(curl -sS --max-time 10 "http://${ROUTER}/healthz" \ + -H "Host: ${actor}.${ATESPACE}.actors.resources.substrate.ate.dev" 2>&1) \ + || die "no answer from ${ROUTER} + Start it with: kubectl -n ate-system port-forward svc/atenet-router 8080:80" + + case "${answer}" in + *"not found"*) + die "Actor ${ATESPACE}/${actor} does not exist. + Create all three with: demos/egress/demo.sh --setup" ;; + *"no healthy upstream"*|*"upstream connect error"*) + die "Actor ${ATESPACE}/${actor} exists but is not running. + Resume it with: go run ./cmd/kubectl-ate resume actor ${actor} -a ${ATESPACE}" ;; + esac + done +} + +usage() { + cat <= testHold { + t.Errorf("stream stayed open %v, want it to have ended before the %v hold", elapsed, testHold) + } + if status.Events != eventsBeforeClose { + t.Errorf("events = %d, want %d", status.Events, eventsBeforeClose) + } +} + +func TestStreamProbeHoldsWebSocketOpen(t *testing.T) { + origin := httptest.NewServer(webSocketTickHandler(t, 0)) + defer origin.Close() + + status := runStreamProbe(t, streamProtocolWebSocket, webSocketURL(origin.URL), testHold) + if status.Error != "" { + t.Fatalf("probe reported an error after %dms: %s", status.ElapsedMs, status.Error) + } + if elapsed := time.Duration(status.ElapsedMs) * time.Millisecond; elapsed < testHold { + t.Errorf("stream stayed open %v, want at least %v", elapsed, testHold) + } + if status.Events == 0 { + t.Error("probe recorded no events, so it never read the stream") + } +} + +func TestStreamProbeReportsWebSocketCutShort(t *testing.T) { + const eventsBeforeClose = 3 + origin := httptest.NewServer(webSocketTickHandler(t, eventsBeforeClose)) + defer origin.Close() + + status := runStreamProbe(t, streamProtocolWebSocket, webSocketURL(origin.URL), testHold) + if status.Error == "" { + t.Fatalf("probe reported success after %dms and %d events, want an error", status.ElapsedMs, status.Events) + } + if status.Events != eventsBeforeClose { + t.Errorf("events = %d, want %d", status.Events, eventsBeforeClose) + } +} + +// TestStreamProbeRefusesUpgradeFailure covers what a proxy that will not carry a +// WebSocket upgrade looks like from here: the handshake is answered with a plain +// HTTP status, and that status is what the probe must report. +func TestStreamProbeRefusesUpgradeFailure(t *testing.T) { + origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + http.Error(w, "no upgrades here", http.StatusBadGateway) + })) + defer origin.Close() + + status := runStreamProbe(t, streamProtocolWebSocket, webSocketURL(origin.URL), testHold) + if status.Error == "" { + t.Fatal("probe reported success against an origin that refused the upgrade") + } + if !strings.Contains(status.Error, "502") { + t.Errorf("error = %q, want it to name the 502 the origin answered with", status.Error) + } +} + +func TestStreamProbeInvalidRequests(t *testing.T) { + tests := []struct { + name string + method string + path string + body string + status int + }{ + {name: "method", method: http.MethodDelete, path: "/stream", body: `{}`, status: http.StatusMethodNotAllowed}, + {name: "malformed JSON", method: http.MethodPost, path: "/stream", body: `{`, status: http.StatusBadRequest}, + {name: "unknown protocol", method: http.MethodPost, path: "/stream", body: `{"url":"http://o/","protocol":"grpc"}`, status: http.StatusBadRequest}, + // The scheme has to agree with the protocol: a ws:// URL handed to the + // HTTP client, or an http:// URL to the dialer, fails deep in a library + // rather than as a bad request. + {name: "sse with a ws URL", method: http.MethodPost, path: "/stream", body: `{"url":"ws://o/","protocol":"sse"}`, status: http.StatusBadRequest}, + {name: "websocket with an http URL", method: http.MethodPost, path: "/stream", body: `{"url":"http://o/","protocol":"websocket"}`, status: http.StatusBadRequest}, + {name: "missing hostname", method: http.MethodPost, path: "/stream", body: `{"url":"http:///sse","protocol":"sse"}`, status: http.StatusBadRequest}, + {name: "invalid hold", method: http.MethodPost, path: "/stream", body: `{"url":"http://o/","protocol":"sse","hold":"soon"}`, status: http.StatusBadRequest}, + {name: "unknown probe id", method: http.MethodGet, path: "/stream?id=stream-404", body: "", status: http.StatusNotFound}, + } + + handler := newHandler(http.DefaultClient) + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + recorder := httptest.NewRecorder() + handler.ServeHTTP(recorder, httptest.NewRequest(test.method, test.path, strings.NewReader(test.body))) + if recorder.Code != test.status { + t.Errorf("status = %d, want %d; body = %s", recorder.Code, test.status, recorder.Body.String()) + } + }) + } +} + +// runStreamProbe drives one probe through the handler the way the e2e test +// drives it through the ingress: start it, then poll until it is done. +func runStreamProbe(t *testing.T, protocol, url string, hold time.Duration) streamProbeStatus { + t.Helper() + handler := newHandler(http.DefaultClient) + + payload, err := json.Marshal(streamProbeRequest{URL: url, Protocol: protocol, Hold: hold.String()}) + if err != nil { + t.Fatalf("marshaling the stream probe request: %v", err) + } + recorder := httptest.NewRecorder() + handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodPost, "/stream", strings.NewReader(string(payload)))) + if recorder.Code != http.StatusAccepted { + t.Fatalf("starting the probe returned %d, want 202; body = %s", recorder.Code, recorder.Body.String()) + } + var started streamProbeStatus + if err := json.NewDecoder(recorder.Body).Decode(&started); err != nil { + t.Fatalf("decoding the start response: %v", err) + } + if started.ID == "" { + t.Fatal("start response carried no probe id") + } + + deadline := time.Now().Add(hold + 10*time.Second) + for { + recorder := httptest.NewRecorder() + handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/stream?id="+started.ID, nil)) + if recorder.Code != http.StatusOK { + t.Fatalf("polling probe %s returned %d; body = %s", started.ID, recorder.Code, recorder.Body.String()) + } + var status streamProbeStatus + if err := json.NewDecoder(recorder.Body).Decode(&status); err != nil { + t.Fatalf("decoding probe %s status: %v", started.ID, err) + } + if status.Done { + return status + } + if time.Now().After(deadline) { + t.Fatalf("probe %s never finished; last seen with %d events over %dms", started.ID, status.Events, status.ElapsedMs) + } + time.Sleep(testTickInterval) + } +} + +// sseTickHandler serves the same event stream as demos/egress/streamserver. A +// nonzero stopAfter makes it hang up mid-stream, standing in for a proxy that +// cuts the connection. +func sseTickHandler(t *testing.T, stopAfter int) http.HandlerFunc { + t.Helper() + return func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + w.(http.Flusher).Flush() + for n := 0; stopAfter == 0 || n < stopAfter; n++ { + select { + case <-r.Context().Done(): + return + case <-time.After(testTickInterval): + } + if _, err := fmt.Fprintf(w, "data: tick-%d\n\n", n); err != nil { + return + } + w.(http.Flusher).Flush() + } + } +} + +func webSocketTickHandler(t *testing.T, stopAfter int) http.HandlerFunc { + t.Helper() + upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} + return func(w http.ResponseWriter, r *http.Request) { + connection, err := upgrader.Upgrade(w, r, nil) + if err != nil { + t.Errorf("upgrading the test origin's connection: %v", err) + return + } + defer connection.Close() + for n := 0; stopAfter == 0 || n < stopAfter; n++ { + select { + case <-r.Context().Done(): + return + case <-time.After(testTickInterval): + } + if err := connection.WriteMessage(websocket.TextMessage, fmt.Appendf(nil, "tick-%d", n)); err != nil { + return + } + } + } +} + +func webSocketURL(httpURL string) string { + return "ws" + strings.TrimPrefix(httpURL, "http") +} diff --git a/demos/egress/streamserver/main.go b/demos/egress/streamserver/main.go new file mode 100644 index 000000000..c41867ac8 --- /dev/null +++ b/demos/egress/streamserver/main.go @@ -0,0 +1,154 @@ +// 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 streamserver is a long-lived-stream origin for the egress tests. It +// emits the same tick sequence two ways -- Server-Sent Events on /sse and +// WebSocket text frames on /ws -- so a test can hold a stream open across a +// proxy's timeout boundary and find out whether the proxy cut it. +// +// It ticks until the peer goes away rather than sending a fixed number of +// events. What these tests ask is how long a stream survives, and an origin +// that ended the stream itself would be answering that question for them. +package main + +import ( + "fmt" + "log/slog" + "net/http" + "os" + "time" + + "github.com/gorilla/websocket" +) + +func main() { + slog.SetDefault(slog.New(slog.NewJSONHandler(os.Stdout, nil))) + + const defaultListenAddress = ":8080" + // One tick a second is frequent enough that a test can tell a live stream + // from a stalled one within a couple of ticks, and sparse enough that a + // half-minute hold is tens of events rather than thousands. + const defaultTickInterval = time.Second + + address := defaultListenAddress + if value := os.Getenv("LISTEN_ADDRESS"); value != "" { + address = value + } + interval := defaultTickInterval + if value := os.Getenv("TICK_INTERVAL"); value != "" { + parsed, err := time.ParseDuration(value) + if err != nil { + slog.Error("invalid TICK_INTERVAL", "value", value, "error", err) + os.Exit(1) + } + interval = parsed + } + + slog.Info("starting stream server", "address", address, "tickInterval", interval) + if err := http.ListenAndServe(address, newHandler(interval)); err != nil { + slog.Error("stream server stopped", "error", err) + os.Exit(1) + } +} + +func newHandler(interval time.Duration) http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("/readyz", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + _, _ = fmt.Fprintln(w, "ok") + }) + mux.HandleFunc("/sse", handleSSE(interval)) + mux.HandleFunc("/ws", handleWebSocket(interval)) + return mux +} + +// tick is the payload of the nth event. Numbering them lets a reader tell a +// stream that was cut and resumed from one that ran unbroken. +func tick(n int) string { + return fmt.Sprintf("tick-%d", n) +} + +func handleSSE(interval time.Duration) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + flusher, ok := w.(http.Flusher) + if !ok { + http.Error(w, "streaming unsupported", http.StatusInternalServerError) + return + } + + header := w.Header() + header.Set("Content-Type", "text/event-stream") + // Without this an intermediary is free to buffer the whole stream, + // which would make a cut one indistinguishable from a slow one. + header.Set("Cache-Control", "no-cache") + w.WriteHeader(http.StatusOK) + flusher.Flush() + + ticker := time.NewTicker(interval) + defer ticker.Stop() + for n := 0; ; n++ { + select { + case <-r.Context().Done(): + return + case <-ticker.C: + } + if _, err := fmt.Fprintf(w, "data: %s\n\n", tick(n)); err != nil { + return + } + flusher.Flush() + } + } +} + +func handleWebSocket(interval time.Duration) http.HandlerFunc { + // Same-origin checking is the default and would reject the actor, which + // sends no Origin header of its own. This fixture is reachable only from + // inside the test namespace and serves one hard-coded tick sequence. + upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} + + return func(w http.ResponseWriter, r *http.Request) { + connection, err := upgrader.Upgrade(w, r, nil) + if err != nil { + // Upgrade has already written a response by this point. + slog.Error("websocket upgrade failed", "error", err) + return + } + defer connection.Close() + + // gorilla only processes close and ping frames from inside a read call, + // so without draining, a peer that has left is noticed only when a write + // eventually fails. + go func() { + for { + if _, _, err := connection.ReadMessage(); err != nil { + connection.Close() + return + } + } + }() + + ticker := time.NewTicker(interval) + defer ticker.Stop() + for n := 0; ; n++ { + select { + case <-r.Context().Done(): + return + case <-ticker.C: + } + if err := connection.WriteMessage(websocket.TextMessage, []byte(tick(n))); err != nil { + return + } + } + } +} diff --git a/go.mod b/go.mod index cc6d8d97d..903562961 100644 --- a/go.mod +++ b/go.mod @@ -23,6 +23,7 @@ require ( github.com/google/go-containerregistry v0.21.7 github.com/google/nftables v0.3.0 github.com/google/uuid v1.6.0 + github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 github.com/hashicorp/go-reap v0.0.0-20260220095743-4e27870b4f51 github.com/jackc/pgx/v5 v5.10.0 github.com/klauspost/compress v1.18.6 @@ -144,7 +145,6 @@ require ( github.com/google/s2a-go v0.1.9 // indirect github.com/googleapis/enterprise-certificate-proxy v0.3.14 // indirect github.com/googleapis/gax-go/v2 v2.21.0 // indirect - github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect diff --git a/hack/install-demo-egress.sh b/hack/install-demo-egress.sh index ec74950c4..12df26f6b 100644 --- a/hack/install-demo-egress.sh +++ b/hack/install-demo-egress.sh @@ -42,6 +42,8 @@ demo-egress_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 + # The SSE and WebSocket origin for the streaming-duration egress tests. + run_kubectl rollout status deployment/streamserver -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 473530565..c4e6c6c0d 100644 --- a/internal/e2e/suites/networking/networking_test.go +++ b/internal/e2e/suites/networking/networking_test.go @@ -232,24 +232,188 @@ func TestActorEgressSSH(t *testing.T) { 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. +// TestActorEgressStreamingDuration covers egress for connections that stay open +// long after the request that opened them. Every other egress test finishes in +// a second or two, so all of them pass against a gateway that silently cuts a +// connection once some timer expires -- and several timers on this path do +// exactly that unless they are explicitly disabled: Envoy applies a route +// timeout to a CONNECT tunnel's whole lifetime rather than just its headers, +// and the MITM gateway's HTTP chains apply one to a response body that, for a +// stream, never ends. +// +// The two subtests are the two ways a long-lived stream is normally built. SSE +// is one HTTP response that never finishes; WebSocket is an upgrade that leaves +// HTTP behind entirely, which a proxy carries only if it has been configured to +// proxy that upgrade. +// +// The Actor reads the stream in the background and is polled for progress, +// because the ingress route in front of it has a timeout of its own: a request +// that waited out the whole hold would be cut by the ingress before it could +// report on the egress. +func TestActorEgressStreamingDuration(t *testing.T) { + // The hold has to clear every timeout on the egress path that a stream can + // trip -- Envoy's 15s default route timeout on the CONNECT tunnel and the + // 30s route timeout on the MITM leg's HTTP chains -- or the test passes + // against a gateway that would still cut a real stream. + const hold = 35 * time.Second + // The origin ticks once a second, so a stream that survives the hold + // delivers roughly one event per second of it. The margin absorbs connection + // setup and the first tick's interval; the assertion that matters is that + // events kept arriving for the whole hold, not their exact count. + const minEvents = 25 + + tests := []struct { + name string + protocol string + // url builds the origin address from the fixture's cluster IP. The + // sandbox does not resolve cluster Service names, so it must be dialed + // by address. + url func(clusterIP string) string + }{ + { + name: "server-sent events", + protocol: "sse", + url: func(ip string) string { return fmt.Sprintf("http://%s:%d/sse", ip, streamServerPort) }, + }, + { + name: "websocket", + protocol: "websocket", + url: func(ip string) string { return fmt.Sprintf("ws://%s:%d/ws", ip, streamServerPort) }, + }, + } + + ctx := context.Background() + clusterIP := egressFixtureClusterIP(t, ctx, "streamserver") + actorName, _ := createAndResumeActor(t, ctx, "egress-stream", egressTemplate) + router := mustRouterClient(t, ctx) + defer router.Close() + + actorRef := resources.ActorRef{Atespace: networkingAtespace, Name: actorName} + 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)) + url := test.url(clusterIP) + + payload, err := json.Marshal(map[string]any{"url": url, "protocol": test.protocol, "hold": hold.String()}) + if err != nil { + t.Fatalf("marshaling the stream probe request for %s: %v", url, err) + } + // The Actor answers 202: it has started reading, not finished. A + // retry on anything else must not treat 202 as the failure, or every + // pass would start another stream and leave it running. + status, body := postThroughEgressActorExpecting(t, ctx, router, actorRef, "/stream", payload, http.StatusAccepted) + if status != http.StatusAccepted { + t.Fatalf("starting the stream probe of %s returned HTTP %d, want 202; body: %s", url, status, body) + } + var started struct { + ID string `json:"id"` + } + if err := json.Unmarshal(body, &started); err != nil { + t.Fatalf("decoding the stream probe start response %s: %v", body, err) + } + + probe := awaitStreamProbe(t, ctx, router, actorRef, started.ID, hold) + if probe.Error != "" { + t.Fatalf("stream of %s ended after %v with %d events: %s (want it held for %v)", + url, time.Duration(probe.ElapsedMs)*time.Millisecond, probe.Events, probe.Error, hold) + } + // A stream can be cut without anyone reporting an error -- a proxy + // that closes a CONNECT tunnel on its route timeout looks to the + // reader like an origin that hung up -- so check the clock too. + if elapsed := time.Duration(probe.ElapsedMs) * time.Millisecond; elapsed < hold { + t.Fatalf("stream of %s stayed open %v, want at least %v (%d events, last %q)", + url, elapsed, hold, probe.Events, probe.Last) + } + if probe.Events < minEvents { + t.Fatalf("stream of %s delivered %d events over %v, want at least %d; it stalled rather than closed (first %q, last %q)", + url, probe.Events, hold, minEvents, probe.First, probe.Last) + } + t.Logf("Actor held the %s stream of %s open for %dms across %d events (first %q, last %q)", + test.protocol, url, probe.ElapsedMs, probe.Events, probe.First, probe.Last) + + port := strconv.Itoa(streamServerPort) + assertEgressGatewayConnect(t, ctx, since, actorName, port) + }) + } +} + +// streamProbeStatus mirrors the response of the egress demo Actor's /stream +// endpoint, from demos/egress/stream.go. +type streamProbeStatus struct { + Events int `json:"events"` + First string `json:"first"` + Last string `json:"last"` + ElapsedMs int64 `json:"elapsedMs"` + Done bool `json:"done"` + Error string `json:"error"` +} + +// awaitStreamProbe polls the Actor until the probe it started reports itself +// done, and returns its final state. A probe that never finishes is a failure +// in its own right, so the wait is bounded well above the hold it was given. +func awaitStreamProbe(t *testing.T, ctx context.Context, router *e2e.RouterClient, actorRef resources.ActorRef, id string, hold time.Duration) streamProbeStatus { + t.Helper() + + // Slack over the hold for the reader's own setup and for the poll interval. + deadline := time.Now().Add(hold + 30*time.Second) + const pollInterval = 2 * time.Second + for { + response, err := router.Get(ctx, actorRef, "/stream?id="+id) + if err != nil { + t.Fatalf("polling stream probe %s: %v", id, err) + } + body, err := io.ReadAll(response.Body) + response.Body.Close() + if err != nil { + t.Fatalf("reading stream probe %s status (HTTP %d): %v", id, response.StatusCode, err) + } + if response.StatusCode != http.StatusOK { + t.Fatalf("polling stream probe %s returned HTTP %d; body: %s", id, response.StatusCode, body) + } + + var status streamProbeStatus + if err := json.Unmarshal(body, &status); err != nil { + t.Fatalf("decoding stream probe %s status %s: %v", id, body, err) + } + if status.Done { + return status + } + if time.Now().After(deadline) { + t.Fatalf("stream probe %s never finished; last seen with %d events over %dms", id, status.Events, status.ElapsedMs) + } + time.Sleep(pollInterval) + } +} + +// The ports the egress fixture Services publish, from +// demos/egress/egress.yaml.tmpl. The banner origin greets on the first and +// stays silent until spoken to on the second. const ( bannerServerPort = 2222 bannerServerQuietPort = 2223 + streamServerPort = 8080 ) // 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{}) + return egressFixtureClusterIP(t, ctx, "bannerserver") +} + +// egressFixtureClusterIP returns the address of one of the egress demo's origin +// Services, which the Actor must dial by address because the sandbox does not +// resolve cluster Service names. +func egressFixtureClusterIP(t *testing.T, ctx context.Context, name string) string { + t.Helper() + service, err := e2e.GetClients().K8s.CoreV1().Services(egressTemplate.namespace).Get(ctx, name, metav1.GetOptions{}) if err != nil { - t.Fatalf("getting Service %s/bannerserver: %v (deploy the fixture with %s)", egressTemplate.namespace, err, egressTemplate.deployFlag) + t.Fatalf("getting Service %s/%s: %v (deploy the fixture with %s)", egressTemplate.namespace, name, 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) + t.Fatalf("Service %s/%s has no cluster IP to dial: %q", egressTemplate.namespace, name, service.Spec.ClusterIP) } return service.Spec.ClusterIP } @@ -266,11 +430,23 @@ func fetchThroughEgressActor(t *testing.T, ctx context.Context, router *e2e.Rout } // 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. +// returns the status and body it answers with, retrying anything other than a +// 200 as postThroughEgressActorExpecting describes. func postThroughEgressActor(t *testing.T, ctx context.Context, router *e2e.RouterClient, actorRef resources.ActorRef, path string, payload []byte) (int, []byte) { t.Helper() + return postThroughEgressActorExpecting(t, ctx, router, actorRef, path, payload, http.StatusOK) +} + +// postThroughEgressActorExpecting POSTs payload to path on the egress demo +// Actor and returns the status and body it answers with. Retries any status +// other than want 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. +// +// want is a parameter rather than a fixed 200 because a POST that only starts +// work answers 202, and retrying that would start the work again on every pass. +func postThroughEgressActorExpecting(t *testing.T, ctx context.Context, router *e2e.RouterClient, actorRef resources.ActorRef, path string, payload []byte, want int) (int, []byte) { + t.Helper() const timeout = 30 * time.Second deadline := time.Now().Add(timeout) @@ -284,10 +460,10 @@ func postThroughEgressActor(t *testing.T, ctx context.Context, router *e2e.Route if err != nil { t.Fatalf("reading egress response body (HTTP %d): %v", response.StatusCode, err) } - if response.StatusCode == http.StatusOK || time.Now().After(deadline) { + if response.StatusCode == want || time.Now().After(deadline) { return response.StatusCode, body } - t.Logf("POST %s to egress Actor returned HTTP %d; retrying...", path, response.StatusCode) + t.Logf("POST %s to egress Actor returned HTTP %d, want %d; retrying...", path, response.StatusCode, want) time.Sleep(1 * time.Second) } } diff --git a/manifests/ate-install/atenet-egress.yaml b/manifests/ate-install/atenet-egress.yaml index 591ef31fc..2f803ed8d 100644 --- a/manifests/ate-install/atenet-egress.yaml +++ b/manifests/ate-install/atenet-egress.yaml @@ -104,6 +104,14 @@ data: upgrade_configs: - upgrade_type: CONNECT connect_config: {} + # A CONNECT tunnel is a session, not a request. Unlike a + # WebSocket upgrade, Envoy never disables the route timeout + # once the tunnel is established, so the 15s default would + # be a ceiling on how long an actor may hold any outbound + # connection open -- fine for a fetch, wrong for SSH, a + # stream, or anything else that idles. 0 disables it; the + # tunnel is still bounded by idle timeouts and by the peers. + timeout: 0s http_filters: # Actor-identity authorization: on every egress CONNECT, ext_proc # reads the actor certificate out of x-forwarded-client-cert, From dfd3e4bfb7de176088b945d40b079f8c8f7edd22 Mon Sep 17 00:00:00 2001 From: Haiyan Meng Date: Tue, 18 Aug 2026 12:01:43 -0400 Subject: [PATCH 4/4] Update atenet-egress-with-sdsmint.yaml to support long-lived streams --- demos/egress/egress.yaml.tmpl | 6 +-- internal/e2e/suites/sdsmint/sdsmint_test.go | 6 +-- .../atenet-egress-with-sdsmint.yaml | 38 +++++++++++++++++-- manifests/ate-install/atenet-egress.yaml | 10 ++++- 4 files changed, 48 insertions(+), 12 deletions(-) diff --git a/demos/egress/egress.yaml.tmpl b/demos/egress/egress.yaml.tmpl index 42f425e7e..0c02a93ce 100644 --- a/demos/egress/egress.yaml.tmpl +++ b/demos/egress/egress.yaml.tmpl @@ -112,10 +112,8 @@ spec: --- -# A long-lived-stream origin for the streaming-duration egress tests. It serves -# cleartext HTTP on purpose: that is the path the MITM egress gateway parses as -# HTTP, and so the path where a route timeout or a missing upgrade config can -# cut a stream that the raw CONNECT path would have carried. +# A long-lived-stream origin for the streaming-duration egress tests. +# It serves cleartext HTTP. apiVersion: apps/v1 kind: Deployment metadata: diff --git a/internal/e2e/suites/sdsmint/sdsmint_test.go b/internal/e2e/suites/sdsmint/sdsmint_test.go index 92adc69a6..f5308328a 100644 --- a/internal/e2e/suites/sdsmint/sdsmint_test.go +++ b/internal/e2e/suites/sdsmint/sdsmint_test.go @@ -123,7 +123,7 @@ func skipUntilPresubmit(t *testing.T) { // from Envoy's live secret set and the test would pass without sdsmint having // minted anything. func TestSdsmintMintsALeafPerSNI(t *testing.T) { - skipUntilPresubmit(t) + // skipUntilPresubmit(t) ctx := context.Background() @@ -228,7 +228,7 @@ func TestSdsmintMintsALeafPerSNI(t *testing.T) { // failed to dial, or that got as far as the CONNECT, would be reporting // something other than the front door turning it away. func TestGatewayRefusesANonActorWorkload(t *testing.T) { - skipUntilPresubmit(t) + // skipUntilPresubmit(t) ctx := context.Background() @@ -254,7 +254,7 @@ func TestGatewayRefusesANonActorWorkload(t *testing.T) { // on the certificate alone from one that authorizes on control-plane state, and // the difference is whether a deleted actor's credential still works. func TestGatewayRefusesAnUnknownActor(t *testing.T) { - skipUntilPresubmit(t) + // skipUntilPresubmit(t) ctx := context.Background() diff --git a/manifests/ate-install/atenet-egress-with-sdsmint.yaml b/manifests/ate-install/atenet-egress-with-sdsmint.yaml index 2c17eb0ef..00eaa8d9c 100644 --- a/manifests/ate-install/atenet-egress-with-sdsmint.yaml +++ b/manifests/ate-install/atenet-egress-with-sdsmint.yaml @@ -126,7 +126,14 @@ data: # route's default 15s stream timeout would be a ceiling on # how long an actor may hold any TCP connection open -- # fine for a request/response fetch, wrong for SSH or any - # other protocol that idles. Idle timeouts still apply. + # other protocol that idles. + # + # 0 removes the ceiling on the tunnel's lifetime. What + # still bounds it is inactivity: this HCM leaves + # stream_idle_timeout at its 5 minute default, and every + # tunnelled byte in either direction resets that timer. So + # a busy tunnel lives as long as its peers do, and one with + # nothing on it for 5 minutes is reset. timeout: 0s upgrade_configs: - upgrade_type: CONNECT @@ -250,6 +257,13 @@ data: typed_config: "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager stat_prefix: mitm_http + # Absent this, Envoy answers an Upgrade request itself instead of + # proxying it, so every WebSocket through this leg fails at the + # handshake. The raw CONNECT gateway never sees the upgrade at all + # and so never needed it; parsing the traffic is what creates the + # obligation to understand it. + upgrade_configs: + - upgrade_type: websocket # This is the leg that knows where the traffic actually went. # REQUESTED_SERVER_NAME is the SNI the leaf was minted for and it # should always equal the authority -- dynamic_forward_proxy @@ -286,7 +300,17 @@ data: - match: { prefix: "/" } route: cluster: egress_forward_proxy - timeout: 30s + # The route timeout runs until the response is completely + # processed, which for a stream is never: at 30s it cut + # every Server-Sent Events response at half a minute. + # + # 0 removes the ceiling on the response's duration. What + # still bounds it is inactivity: this HCM leaves + # stream_idle_timeout at its 5 minute default, and every + # chunk of the response body resets that timer. So a + # ticking stream runs indefinitely and a stalled one is + # reset after 5 minutes. + timeout: 0s http_filters: - name: envoy.filters.http.dynamic_forward_proxy typed_config: @@ -347,6 +371,10 @@ data: typed_config: "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager stat_prefix: mitm_cleartext + # Same reasoning as the TLS chain: this leg parses HTTP, so it has + # to be told which upgrades to carry rather than answer. + upgrade_configs: + - upgrade_type: websocket access_log: - name: envoy.access_loggers.file typed_config: @@ -377,7 +405,11 @@ data: - match: { prefix: "/" } route: cluster: egress_forward_proxy_cleartext - timeout: 30s + # Same reasoning as the TLS chain: a streaming response + # never completes, so any route timeout is a cap on how + # long a stream may run, and stream_idle_timeout's 5 + # minute default is what bounds a stalled one instead. + timeout: 0s http_filters: # Same reasoning as the TLS chain: resolve from the request's own # Host, so the name that was policed is the name that is dialled. diff --git a/manifests/ate-install/atenet-egress.yaml b/manifests/ate-install/atenet-egress.yaml index 2f803ed8d..113994153 100644 --- a/manifests/ate-install/atenet-egress.yaml +++ b/manifests/ate-install/atenet-egress.yaml @@ -109,8 +109,14 @@ data: # once the tunnel is established, so the 15s default would # be a ceiling on how long an actor may hold any outbound # connection open -- fine for a fetch, wrong for SSH, a - # stream, or anything else that idles. 0 disables it; the - # tunnel is still bounded by idle timeouts and by the peers. + # stream, or anything else that idles. + # + # 0 removes the ceiling on the tunnel's lifetime. What + # still bounds it is inactivity: this HCM leaves + # stream_idle_timeout at its 5 minute default, and every + # tunnelled byte in either direction resets that timer. So + # a busy tunnel lives as long as its peers do, and one with + # nothing on it for 5 minutes is reset. timeout: 0s http_filters: # Actor-identity authorization: on every egress CONNECT, ext_proc