e2e: cover egress for raw TCP and SSH traffic, in both speaking orders

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
+		}
+	}
+}

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

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 {