e2e: add a egress test on http request to non 80 port

internal/e2e/fixtures/egressprobe/httptarget.yaml.tmpl

@@ -0,0 +1,86 @@
+# 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 plain HTTP origin on a non-80 port, used by internal/e2e/suites/networking
+# as the destination for TestActorEgressNonStandardPort. ${NAMESPACE} is
+# substituted by the suite with the randomized namespace it created, so the
+# target is torn down with that namespace and leaves nothing behind.
+#
+# It runs the egressprobe binary rather than a target of its own: egressprobe's
+# main only parses flags and serves, and its --listen default is already a
+# non-privileged :8080, so serving /healthz here needs no new fixture. The one
+# difference from egressprobe.yaml.tmpl is that nothing is mounted -- the
+# credential bundles that manifest projects are read lazily inside /handshake,
+# which this target never calls, so the pod starts without them.
+#
+# The port is the whole point of the fixture: the actor dials this Service's
+# ClusterIP:8080, and that address is what SO_ORIGINAL_DST hands atunnel for the
+# CONNECT authority. Keep containerPort, the Service port and --listen in step.
+
+apiVersion: v1
+kind: Pod
+metadata:
+  name: httptarget
+  namespace: ${NAMESPACE}
+  labels:
+    app: httptarget
+spec:
+  restartPolicy: Never
+  containers:
+  - name: httptarget
+    image: ko://github.com/agent-substrate/substrate/internal/e2e/fixtures/egressprobe
+    args:
+    - "--listen=:8080"
+    ports:
+    - name: http
+      containerPort: 8080
+    # The suite waits on this before pointing an actor at the Service. Without a
+    # readiness gate the actor's fetch can arrive before the listener exists and
+    # come back as a 502 that looks like an egress failure.
+    readinessProbe:
+      httpGet:
+        path: /healthz
+        port: 8080
+      periodSeconds: 2
+    resources:
+      requests:
+        cpu: 10m
+        memory: 32Mi
+    # runAsUser must be spelled out: ko's distroless static base declares no
+    # USER, so runAsNonRoot on its own makes kubelet refuse to start the
+    # container ("image will run as root") rather than pick a uid. 65532 is
+    # distroless's nonroot uid, and the same one the egress gateway's pod uses.
+    securityContext:
+      allowPrivilegeEscalation: false
+      readOnlyRootFilesystem: true
+      runAsNonRoot: true
+      runAsUser: 65532
+      runAsGroup: 65532
+      capabilities:
+        drop: ["ALL"]
+
+---
+
+apiVersion: v1
+kind: Service
+metadata:
+  name: httptarget
+  namespace: ${NAMESPACE}
+spec:
+  selector:
+    app: httptarget
+  ports:
+  - name: http
+    port: 8080
+    targetPort: 8080

internal/e2e/suites/networking/httptarget_test.go

@@ -0,0 +1,140 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package networking
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/agent-substrate/substrate/internal/e2e"
+	"github.com/agent-substrate/substrate/internal/portforward"
+)
+
+const (
+	// httpTargetName names both the Pod and the Service in
+	// internal/e2e/fixtures/egressprobe/httptarget.yaml.tmpl.
+	httpTargetName = "httptarget"
+	// httpTargetPort is the target's Service port, and therefore the port the
+	// Actor dials and the one that has to show up in the egress gateway's
+	// CONNECT authority. Keep it in step with the manifest.
+	httpTargetPort = 8080
+)
+
+// startHTTPTarget deploys a plain HTTP origin listening on httpTargetPort into
+// a namespace of its own and returns the ClusterIP to dial it at.
+//
+// Callers must dial that IP rather than the Service's DNS name. atunnel takes
+// the CONNECT authority from SO_ORIGINAL_DST, which is always an address, so
+// the name would buy nothing; it would only add the actor sandbox's DNS path
+// (UDP over the compatibility masquerade, see InstallActorNftablesRules in
+// internal/ateomnet/net.go) to what a failure could mean. demos/egress's
+// test-egress.sh targets a ClusterIP for the same reason.
+func startHTTPTarget(t *testing.T, ctx context.Context) string {
+	t.Helper()
+	if _, err := e2e.CheckEnv("KO_DOCKER_REPO"); err != nil {
+		t.Fatalf("CheckEnv failed: %v", err)
+	}
+	root, err := e2e.FindRepoRoot()
+	if err != nil {
+		t.Fatalf("FindRepoRoot: %v", err)
+	}
+	namespace := e2e.CreateNamespace(t).Name
+
+	// Render the manifest to a file so ko consumes it with no shell involved.
+	tmpl, err := os.ReadFile(filepath.Join(root, "internal/e2e/fixtures/egressprobe/httptarget.yaml.tmpl"))
+	if err != nil {
+		t.Fatalf("reading httptarget manifest template: %v", err)
+	}
+	manifest := filepath.Join(t.TempDir(), "httptarget.yaml")
+	rendered := strings.ReplaceAll(string(tmpl), "${NAMESPACE}", namespace)
+	if err := os.WriteFile(manifest, []byte(rendered), 0o644); err != nil {
+		t.Fatalf("writing rendered httptarget manifest: %v", err)
+	}
+
+	// Build/push the image and apply through the repo's pinned ko
+	// (hack/run-tool.sh ko); CI does not install ko on PATH, and every other
+	// deploy in this repo goes through this wrapper. The trailing
+	// `-- --context=...` mirrors run_ko in hack/install-ate.sh: ko's apply
+	// subcommand forwards args after `--` to kubectl. KO_CONFIG_PATH is
+	// required because ko resolves .ko.yaml from its working directory, which
+	// is the test's package dir, not the repo root; without it the build
+	// silently loses defaultPlatforms (and produces amd64-only images that
+	// cannot run on arm64 nodes).
+	applyArgs := []string{"ko", "apply", "-f", manifest}
+	if e2e.KubeContext != "" {
+		applyArgs = append(applyArgs, "--", "--context="+e2e.KubeContext)
+	}
+	e2e.RunCmdWithEnv(t, []string{"KO_CONFIG_PATH=" + root}, filepath.Join(root, "hack/run-tool.sh"), applyArgs...)
+
+	// No cleanup registered: everything the manifest creates is namespaced, so
+	// it goes with the namespace CreateNamespace already tears down (and keeps
+	// on failure, which is when its logs are worth reading).
+	waitForHTTPTargetReady(t, ctx, namespace)
+
+	service, err := e2e.GetClients().K8s.CoreV1().Services(namespace).Get(ctx, httpTargetName, metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("getting Service %s/%s: %v", namespace, httpTargetName, err)
+	}
+	if service.Spec.ClusterIP == "" || service.Spec.ClusterIP == corev1.ClusterIPNone {
+		t.Fatalf("Service %s/%s has no ClusterIP to dial: %q", namespace, httpTargetName, service.Spec.ClusterIP)
+	}
+	t.Logf("HTTP target ready at %s:%d (namespace %s)", service.Spec.ClusterIP, httpTargetPort, namespace)
+	return service.Spec.ClusterIP
+}
+
+// waitForHTTPTargetReady blocks until the target pod passes its readiness
+// probe, so a fetch through it cannot race the listener coming up.
+func waitForHTTPTargetReady(t *testing.T, ctx context.Context, namespace string) {
+	t.Helper()
+	const timeout = 3 * time.Minute
+	deadline := time.Now().Add(timeout)
+	var lastState string
+	for time.Now().Before(deadline) {
+		pod, err := e2e.GetClients().K8s.CoreV1().Pods(namespace).Get(ctx, httpTargetName, metav1.GetOptions{})
+		switch {
+		case err != nil:
+			lastState = err.Error()
+		case portforward.IsPodReady(pod):
+			return
+		default:
+			lastState = describeHTTPTargetState(pod)
+		}
+		time.Sleep(2 * time.Second)
+	}
+	t.Fatalf("timed out after %v waiting for pod %s/%s to become ready: %s", timeout, namespace, httpTargetName, lastState)
+}
+
+func describeHTTPTargetState(pod *corev1.Pod) string {
+	parts := []string{"phase=" + string(pod.Status.Phase)}
+	for _, cs := range pod.Status.ContainerStatuses {
+		switch {
+		case cs.State.Waiting != nil:
+			parts = append(parts, fmt.Sprintf("%s waiting: %s: %s", cs.Name, cs.State.Waiting.Reason, cs.State.Waiting.Message))
+		case cs.State.Terminated != nil:
+			parts = append(parts, fmt.Sprintf("%s terminated: %s: %s", cs.Name, cs.State.Terminated.Reason, cs.State.Terminated.Message))
+		default:
+			parts = append(parts, fmt.Sprintf("%s running, ready=%t", cs.Name, cs.Ready))
+		}
+	}
+	return strings.Join(parts, "; ")
+}

internal/e2e/suites/networking/networking_test.go

@@ -19,7 +19,9 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"net"
 	"net/http"
+	"strconv"
 	"strings"
 	"testing"
 	"time"
@@ -111,6 +113,47 @@ func TestActorEgressHTTPS(t *testing.T) {
 	assertEgressGatewayConnect(t, ctx, since, actorName, "443")
 }
 
+// TestActorEgressNonStandardPort covers plaintext HTTP/1.1 egress to a port
+// that is neither 80 nor 443, the shape most in-cluster services actually take.
+//
+// The port is worth its own test because nothing in the egress path holds it as
+// a constant or derives it from the scheme: it is the Actor's own TCP
+// destination port, recovered from SO_ORIGINAL_DST by TCPOriginalDestination
+// after the prerouting REDIRECT that InstallActorNftablesRules adds inside the
+// worker pod's netns, and then written verbatim into the CONNECT authority by
+// atunnel's Client.DialContext. The other two tests would still pass if that
+// port were defaulted from the scheme, because 80 and 443 are exactly what such
+// a default would produce.
+func TestActorEgressNonStandardPort(t *testing.T) {
+	ctx := context.Background()
+
+	// Stand the target up first: a fixture failure here should not leave a
+	// resumed Actor idling in the cluster waiting for a destination.
+	targetIP := startHTTPTarget(t, ctx)
+
+	actorName, _ := createAndResumeActor(t, ctx, "egress-port", egressTemplate)
+	router := mustRouterClient(t, ctx)
+	defer router.Close()
+
+	since := metav1.NewTime(time.Now().Add(-1 * time.Minute))
+
+	// The ClusterIP literal, not the Service's DNS name: the authority atunnel
+	// sends is always an address, so the name would add nothing but a
+	// dependency on the sandbox's DNS-over-UDP masquerade path -- turning a DNS
+	// failure into something that reads as an egress-port failure. kube-proxy's
+	// service DNAT happens later, in the host netns, so <ClusterIP>:8080 is
+	// what SO_ORIGINAL_DST returns and what has to reach the gateway.
+	url := fmt.Sprintf("http://%s/healthz", net.JoinHostPort(targetIP, strconv.Itoa(httpTargetPort)))
+	actorRef := resources.ActorRef{Atespace: networkingAtespace, Name: actorName}
+	status, body := fetchThroughEgressActor(t, ctx, router, actorRef, url)
+	if status != http.StatusOK {
+		t.Fatalf("Actor egress fetch of %s returned HTTP %d, want 200; body: %s", url, status, body)
+	}
+	t.Logf("Actor egress fetch of %s succeeded", url)
+
+	assertEgressGatewayConnect(t, ctx, since, actorName, strconv.Itoa(httpTargetPort))
+}
+
 // 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