diff --git a/cmd/ateapi/internal/controlapi/syncer_test.go b/cmd/ateapi/internal/controlapi/syncer_test.go index 1db18baa4..6adf22389 100644 --- a/cmd/ateapi/internal/controlapi/syncer_test.go +++ b/cmd/ateapi/internal/controlapi/syncer_test.go @@ -968,11 +968,12 @@ func TestReleaseActorOnDeadWorker_StatusTransitions(t *testing.T) { type conflictStore struct { store.Interface conflictTriggered atomic.Bool + shouldInject func(worker *ateapipb.Worker) bool onUpdate func(ctx context.Context, worker *ateapipb.Worker) } func (c *conflictStore) UpdateWorker(ctx context.Context, worker *ateapipb.Worker, expectedVersion int64) error { - if c.onUpdate != nil && c.conflictTriggered.CompareAndSwap(false, true) { + if c.shouldInject != nil && c.shouldInject(worker) && c.conflictTriggered.CompareAndSwap(false, true) { c.onUpdate(ctx, worker) } return c.Interface.UpdateWorker(ctx, worker, expectedVersion) @@ -998,7 +999,20 @@ func TestSyncer_UpdateWorker_RetryOnVersionConflict(t *testing.T) { var cs *conflictStore persistence, fakeK8s, fakeAte, syncer, cleanup := setupSyncerTestWithStore(t, ctx, func(s store.Interface) store.Interface { - cs = &conflictStore{Interface: s} + // Configure the injector before the syncer starts. It only fires for the + // update under test, so the initial worker creation cannot consume it. + cs = &conflictStore{ + Interface: s, + shouldInject: func(w *ateapipb.Worker) bool { + return w.GetSandboxClass() == "microvm" + }, + onUpdate: func(c context.Context, w *ateapipb.Worker) { + if cw, err := s.GetWorker(c, ns, poolName, podName); err == nil { + cw.NodeName = "node2" + _ = s.UpdateWorker(c, cw, cw.Version) + } + }, + } return cs }, pool) defer func() { @@ -1068,14 +1082,6 @@ func TestSyncer_UpdateWorker_RetryOnVersionConflict(t *testing.T) { t.Fatalf("pool informer cache failed to update: %v", err) } - // Configure conflictStore to inject a concurrent version bump in Redis when the syncer calls UpdateWorker. - cs.onUpdate = func(c context.Context, w *ateapipb.Worker) { - if cw, err := cs.Interface.GetWorker(c, ns, poolName, podName); err == nil { - cw.NodeName = "node2" - _ = cs.Interface.UpdateWorker(c, cw, cw.Version) - } - } - // Touch the pod ONCE in K8s so the syncer reconciles it. The first reconcile's // UpdateWorker hits ErrVersionConflict (injected by conflictStore), which requeues // the key with backoff; the retry re-fetches the latest version from Redis. diff --git a/cmd/atecontroller/internal/controllers/workerpool_apply.go b/cmd/atecontroller/internal/controllers/workerpool_apply.go index 0128ab3d4..1153605e3 100644 --- a/cmd/atecontroller/internal/controllers/workerpool_apply.go +++ b/cmd/atecontroller/internal/controllers/workerpool_apply.go @@ -78,6 +78,7 @@ func buildDeploymentApplyConfig(wp *atev1alpha1.WorkerPool, otel ateomOTelSettin WithArgs( "--pod-uid=$(POD_UID)", "--atunnel-listen-address=0.0.0.0:443", + "--atunnel-connect-listen-address=0.0.0.0:444", "--atunnel-credential-bundle="+atunnelIdentityMountPath+"/credential-bundle.pem", "--atunnel-trust-bundle="+atunnelIdentityMountPath+"/trust-bundle.pem", "--atunnel-egress-listen-address=0.0.0.0:15001", @@ -86,7 +87,11 @@ func buildDeploymentApplyConfig(wp *atev1alpha1.WorkerPool, otel ateomOTelSettin WithPorts(corev1ac.ContainerPort(). WithName("https"). WithContainerPort(443). - WithProtocol(corev1.ProtocolTCP)). + WithProtocol(corev1.ProtocolTCP), + corev1ac.ContainerPort(). + WithName("connect"). + WithContainerPort(444). + WithProtocol(corev1.ProtocolTCP)). WithSecurityContext(ateomSecurityContext(wp.Spec.SandboxClass)). WithEnv(ateomContainerEnv(otel)...). WithVolumeMounts( diff --git a/cmd/atecontroller/internal/controllers/workerpool_apply_test.go b/cmd/atecontroller/internal/controllers/workerpool_apply_test.go index 9e73c6d73..02310337e 100644 --- a/cmd/atecontroller/internal/controllers/workerpool_apply_test.go +++ b/cmd/atecontroller/internal/controllers/workerpool_apply_test.go @@ -735,6 +735,7 @@ func expectedDeploymentApplyConfig(mutatePodSpec func(*corev1ac.PodSpecApplyConf WithArgs( "--pod-uid=$(POD_UID)", "--atunnel-listen-address=0.0.0.0:443", + "--atunnel-connect-listen-address=0.0.0.0:444", "--atunnel-credential-bundle="+atunnelIdentityMountPath+"/credential-bundle.pem", "--atunnel-trust-bundle="+atunnelIdentityMountPath+"/trust-bundle.pem", "--atunnel-egress-listen-address=0.0.0.0:15001", @@ -743,7 +744,11 @@ func expectedDeploymentApplyConfig(mutatePodSpec func(*corev1ac.PodSpecApplyConf WithPorts(corev1ac.ContainerPort(). WithName("https"). WithContainerPort(443). - WithProtocol(corev1.ProtocolTCP)). + WithProtocol(corev1.ProtocolTCP), + corev1ac.ContainerPort(). + WithName("connect"). + WithContainerPort(444). + WithProtocol(corev1.ProtocolTCP)). WithSecurityContext(corev1ac.SecurityContext(). WithRunAsUser(0). WithRunAsGroup(0). diff --git a/cmd/ateom-gvisor/main.go b/cmd/ateom-gvisor/main.go index dc59c6fd8..ce165ca18 100644 --- a/cmd/ateom-gvisor/main.go +++ b/cmd/ateom-gvisor/main.go @@ -65,12 +65,13 @@ var ( podUID = pflag.String("pod-uid", "", "The UID of the current pod") // TODO(liorlieberman) have a sub package for all atunnel releated things like that - atunnelListenAddress = pflag.String("atunnel-listen-address", "0.0.0.0:443", "Address for actor ingress HTTPS") - workerCredentialBundle = pflag.String("atunnel-credential-bundle", "/run/podidentity.podcert.ate.dev/credential-bundle.pem", "Worker Pod credential bundle used by atunnel for inbound serving and outbound mTLS") - podIdentityTrustBundle = pflag.String("atunnel-trust-bundle", "/run/podidentity.podcert.ate.dev/trust-bundle.pem", "Pod identity trust bundle used for router clients and the node-local atelet") - atunnelClientIdentity = pflag.String("atunnel-client-identity", "spiffe://cluster.local/ns/ate-system/sa/atenet-router", "SPIFFE identity allowed to call actor ingress HTTPS") - atunnelEgressListenAddress = pflag.String("atunnel-egress-listen-address", "0.0.0.0:15001", "Address for transparently intercepted actor egress TCP") - egressGatewayTrustBundle = pflag.String("atunnel-egress-trust-bundle", "/run/servicedns.podcert.ate.dev/trust-bundle.pem", "Service DNS trust bundle for the remote egress gateway") + atunnelListenAddress = pflag.String("atunnel-listen-address", "0.0.0.0:443", "Address for actor ingress HTTPS") + atunnelConnectListenAddress = pflag.String("atunnel-connect-listen-address", "0.0.0.0:444", "Address for actor ingress mTLS CONNECT") + workerCredentialBundle = pflag.String("atunnel-credential-bundle", "/run/podidentity.podcert.ate.dev/credential-bundle.pem", "Worker Pod credential bundle used by atunnel for inbound serving and outbound mTLS") + podIdentityTrustBundle = pflag.String("atunnel-trust-bundle", "/run/podidentity.podcert.ate.dev/trust-bundle.pem", "Pod identity trust bundle used for router clients and the node-local atelet") + atunnelClientIdentity = pflag.String("atunnel-client-identity", "spiffe://cluster.local/ns/ate-system/sa/atenet-router", "SPIFFE identity allowed to call actor ingress HTTPS") + atunnelEgressListenAddress = pflag.String("atunnel-egress-listen-address", "0.0.0.0:15001", "Address for transparently intercepted actor egress TCP") + egressGatewayTrustBundle = pflag.String("atunnel-egress-trust-bundle", "/run/servicedns.podcert.ate.dev/trust-bundle.pem", "Service DNS trust bundle for the remote egress gateway") showVersion = pflag.Bool("version", false, "Print version and exit.") logLevelFlag = pflag.String("log-level", "info", "Minimum log level: debug, info, warn, or error.") @@ -261,6 +262,16 @@ func runAtunnel(ctx context.Context, upstream *url.URL) (*atunnel.Server, *atunn } }() slog.InfoContext(ctx, "atunnel serving", slog.String("address", *atunnelListenAddress)) + atunnelConnectListener, err := net.Listen("tcp", *atunnelConnectListenAddress) + if err != nil { + return nil, nil, 0, fmt.Errorf("while opening atunnel CONNECT listener: %w", err) + } + go func() { + if err := atunnelIngress.ServeConnect(ctx, atunnelConnectListener); err != nil { + serverboot.Fatal(ctx, "Failed to serve actor CONNECT ingress", err) + } + }() + slog.InfoContext(ctx, "atunnel CONNECT serving", slog.String("address", *atunnelConnectListenAddress)) atunnelEgress, err := atunnel.NewEgress(atunnel.TCPOriginalDestination) if err != nil { diff --git a/cmd/ateom-microvm/main.go b/cmd/ateom-microvm/main.go index f94b2eb1a..af9c4c443 100644 --- a/cmd/ateom-microvm/main.go +++ b/cmd/ateom-microvm/main.go @@ -71,12 +71,13 @@ var ( otlpRelaySocket = flag.String("otlp-relay-socket", ateompath.AteletOTLPSocketPath(), "Unix socket of atelet's OTLP relay to export telemetry through, keeping it off the pod network. Empty, or absent at startup, exports directly to OTEL_EXPORTER_OTLP_ENDPOINT instead.") - atunnelListenAddress = flag.String("atunnel-listen-address", "0.0.0.0:443", "Address for actor ingress HTTPS") - workerCredentialBundle = flag.String("atunnel-credential-bundle", "/run/podidentity.podcert.ate.dev/credential-bundle.pem", "Worker Pod credential bundle used by atunnel for inbound serving and outbound mTLS") - podIdentityTrustBundle = flag.String("atunnel-trust-bundle", "/run/podidentity.podcert.ate.dev/trust-bundle.pem", "Pod identity trust bundle used for router clients and the node-local atelet") - atunnelClientIdentity = flag.String("atunnel-client-identity", "spiffe://cluster.local/ns/ate-system/sa/atenet-router", "SPIFFE identity allowed to call actor ingress HTTPS") - atunnelEgressListenAddress = flag.String("atunnel-egress-listen-address", "0.0.0.0:15001", "Address for transparently intercepted actor egress TCP") - egressGatewayTrustBundle = flag.String("atunnel-egress-trust-bundle", "/run/servicedns.podcert.ate.dev/trust-bundle.pem", "Service DNS trust bundle for the remote egress gateway") + atunnelListenAddress = flag.String("atunnel-listen-address", "0.0.0.0:443", "Address for actor ingress HTTPS") + atunnelConnectListenAddress = flag.String("atunnel-connect-listen-address", "0.0.0.0:444", "Address for actor ingress mTLS CONNECT") + workerCredentialBundle = flag.String("atunnel-credential-bundle", "/run/podidentity.podcert.ate.dev/credential-bundle.pem", "Worker Pod credential bundle used by atunnel for inbound serving and outbound mTLS") + podIdentityTrustBundle = flag.String("atunnel-trust-bundle", "/run/podidentity.podcert.ate.dev/trust-bundle.pem", "Pod identity trust bundle used for router clients and the node-local atelet") + atunnelClientIdentity = flag.String("atunnel-client-identity", "spiffe://cluster.local/ns/ate-system/sa/atenet-router", "SPIFFE identity allowed to call actor ingress HTTPS") + atunnelEgressListenAddress = flag.String("atunnel-egress-listen-address", "0.0.0.0:15001", "Address for transparently intercepted actor egress TCP") + egressGatewayTrustBundle = flag.String("atunnel-egress-trust-bundle", "/run/servicedns.podcert.ate.dev/trust-bundle.pem", "Service DNS trust bundle for the remote egress gateway") ) const ( @@ -221,6 +222,16 @@ func do(ctx context.Context) error { } }() slog.InfoContext(ctx, "atunnel serving", slog.String("address", *atunnelListenAddress)) + atunnelConnectListener, err := net.Listen("tcp", *atunnelConnectListenAddress) + if err != nil { + return fmt.Errorf("while opening atunnel CONNECT listener: %w", err) + } + go func() { + if err := atunnelIngress.ServeConnect(ctx, atunnelConnectListener); err != nil { + serverboot.Fatal(ctx, "Failed to serve actor CONNECT ingress", err) + } + }() + slog.InfoContext(ctx, "atunnel CONNECT serving", slog.String("address", *atunnelConnectListenAddress)) atunnelEgress, err := atunnel.NewEgress(atunnel.TCPOriginalDestination) if err != nil { return fmt.Errorf("while configuring atunnel egress: %w", err) diff --git a/demos/counter/counter.go b/demos/counter/counter.go index d5e9ad637..a3927a6dc 100644 --- a/demos/counter/counter.go +++ b/demos/counter/counter.go @@ -69,6 +69,7 @@ func main() { secondFileCounterDirectory := pflag.String("second-file-counter-directory", "", "Directory for a second file counter; empty disables it. Used to exercise an Actor with more than one durable volume") validateExistingFilePath := pflag.String("validate-existing-file-path", "", "Path to existing file to validate reading") extraPort := pflag.Int("extra-port", 0, "Additional port to listen on, for exercising atenet-router's arbitrary-port ingress support; 0 disables it") + tcpPort := pflag.Int("tcp-port", 0, "Plain TCP echo port for exercising atunnel CONNECT ingress; 0 disables it") pflag.Parse() ctx := context.Background() @@ -177,6 +178,28 @@ func main() { }() } + if *tcpPort > 0 { + go func() { + listener, err := net.Listen("tcp", fmt.Sprintf(":%d", *tcpPort)) + if err != nil { + slog.ErrorContext(ctx, "Error starting counter TCP echo server", slog.Any("err", err)) + os.Exit(1) + } + slog.InfoContext(ctx, "Starting counter TCP echo server", slog.Int("port", *tcpPort)) + for { + conn, err := listener.Accept() + if err != nil { + slog.ErrorContext(ctx, "Counter TCP echo accept failed", slog.Any("err", err)) + return + } + go func() { + defer conn.Close() + _, _ = io.Copy(conn, conn) + }() + } + }() + } + // Write some random data to a file in the root filesystem, to test // filesystem checkpoint/restore. if err := writeRandomFile(); err != nil { diff --git a/demos/counter/counter.yaml.tmpl b/demos/counter/counter.yaml.tmpl index 10b60ff6a..db92de4fe 100644 --- a/demos/counter/counter.yaml.tmpl +++ b/demos/counter/counter.yaml.tmpl @@ -69,6 +69,7 @@ spec: # listener a test can address by CONNECTing to :9090, distinct # from the primary port 80 every other assertion in this demo uses. - --extra-port=9090 + - --tcp-port=9091 ${VALIDATE_EXISTING_FILE_PATH_ARG} readyz: httpGet: diff --git a/internal/atunnel/ingress.go b/internal/atunnel/ingress.go index f00d2bb7b..9cf824eb1 100644 --- a/internal/atunnel/ingress.go +++ b/internal/atunnel/ingress.go @@ -21,6 +21,7 @@ import ( "crypto/x509" "errors" "fmt" + "io" "log/slog" "net" "net/http" @@ -36,6 +37,10 @@ import ( ) const ( + // DefaultConnectPort is the worker port on which atunnel accepts inbound + // mTLS CONNECT tunnels from the ingress router. + DefaultConnectPort = 444 + // StaleAssignmentHeader distinguishes an atunnel routing rejection from a // 421 returned by the actor application itself. StaleAssignmentHeader = "X-Ate-Assignment-Stale" @@ -78,6 +83,7 @@ type Server struct { credentialBundlePath string tlsConfig *tls.Config proxy *httputil.ReverseProxy + upstream *url.URL mu sync.Mutex active *activation @@ -146,6 +152,7 @@ func NewServer(cfg Config) (*Server, error) { s := &Server{ credentialBundlePath: cfg.CredentialBundlePath, proxy: proxy, + upstream: cfg.Upstream, } s.tlsConfig = &tls.Config{ MinVersion: tls.VersionTLS12, @@ -188,9 +195,25 @@ func loadCredentialBundle(path string) (*tls.Certificate, error) { // Serve serves HTTPS on lis until ctx is canceled or the server fails. func (s *Server) Serve(ctx context.Context, lis net.Listener) error { + return s.serve(ctx, lis, s, s.tlsConfig) +} + +// ServeConnect serves the mTLS CONNECT endpoint. CONNECT is deliberately on a +// separate listener so ordinary actor ingress remains a request proxy, while +// the router can use this listener for a bidirectional tunnel. +func (s *Server) ServeConnect(ctx context.Context, lis net.Listener) error { + // Keep ALPN confined to the CONNECT listener. The established ingress + // listener is explicitly HTTP/1.1 in the router's upstream cluster, while + // the CONNECT listener supports either HTTP/1.1 or HTTP/2. + tlsConfig := s.tlsConfig.Clone() + tlsConfig.NextProtos = []string{"h2", "http/1.1"} + return s.serve(ctx, lis, http.HandlerFunc(s.ServeConnectHTTP), tlsConfig) +} + +func (s *Server) serve(ctx context.Context, lis net.Listener, handler http.Handler, tlsConfig *tls.Config) error { httpServer := &http.Server{ - Handler: s, - TLSConfig: s.tlsConfig, + Handler: handler, + TLSConfig: tlsConfig, ReadHeaderTimeout: 10 * time.Second, } done := make(chan struct{}) @@ -209,6 +232,123 @@ func (s *Server) Serve(ctx context.Context, lis net.Listener) error { return err } +// ServeConnectHTTP accepts a router-authenticated CONNECT request and relays +// its tunnel to the named port on the currently active actor. +func (s *Server) ServeConnectHTTP(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodConnect { + http.Error(w, "CONNECT required", http.StatusMethodNotAllowed) + return + } + ref, ctx, release, ok := s.authorize(r) + if !ok { + s.reject(w) + return + } + defer release() + + _, port, err := net.SplitHostPort(r.Host) + if err != nil { + http.Error(w, "CONNECT authority must include a port", http.StatusBadRequest) + return + } + if _, ok := ParsePort(port); !ok { + http.Error(w, "invalid CONNECT port", http.StatusBadRequest) + return + } + + dialer := &net.Dialer{Timeout: 5 * time.Second} + upstream, err := dialer.DialContext(ctx, "tcp", net.JoinHostPort(s.upstream.Hostname(), port)) + if err != nil { + slog.WarnContext(r.Context(), "atunnel CONNECT upstream failed", slog.Any("actor", ref), slog.Any("err", err)) + http.Error(w, "bad gateway", http.StatusBadGateway) + return + } + defer upstream.Close() + + if r.ProtoMajor == 2 { + s.serveH2Connect(w, r, upstream, ctx) + return + } + s.serveH1Connect(w, upstream, ctx) +} + +func (s *Server) serveH1Connect(w http.ResponseWriter, upstream net.Conn, ctx context.Context) { + hj, ok := w.(http.Hijacker) + if !ok { + http.Error(w, "CONNECT hijacking unsupported", http.StatusInternalServerError) + return + } + client, rw, err := hj.Hijack() + if err != nil { + return + } + defer client.Close() + if _, err := rw.WriteString("HTTP/1.1 200 Connection Established\r\n\r\n"); err != nil { + return + } + if err := rw.Flush(); err != nil { + return + } + + relayIngressWithHalfClose(ctx, upstream, rw, client, client) +} + +func (s *Server) serveH2Connect(w http.ResponseWriter, r *http.Request, upstream net.Conn, ctx context.Context) { + // A HTTP/2 CONNECT tunnel is a pair of streams, not a hijackable TCP + // socket. Send the response headers before copying so the peer can start + // sending DATA frames, then flush each upstream write promptly. + w.WriteHeader(http.StatusOK) + if err := http.NewResponseController(w).Flush(); err != nil { + return + } + + relayIngressWithHalfClose(ctx, upstream, r.Body, flushingWriter{ResponseWriter: w}, r.Body) +} + +// relayIngressWithHalfClose copies a CONNECT stream. When the client request +// stream ends, it half-closes the actor connection and continues forwarding +// actor output until that stream ends too. +func relayIngressWithHalfClose(ctx context.Context, upstream net.Conn, clientReader io.Reader, clientWriter io.Writer, clientCloser io.Closer) { + stop := context.AfterFunc(ctx, func() { + _ = upstream.Close() + _ = clientCloser.Close() + }) + defer stop() + + done := make(chan struct{}, 2) + go func() { + _, _ = io.Copy(upstream, clientReader) + closeWrite(upstream) + done <- struct{}{} + }() + go func() { + _, _ = io.Copy(clientWriter, upstream) + done <- struct{}{} + }() + for range 2 { + select { + case <-ctx.Done(): + return + case <-done: + } + } +} + +type flushingWriter struct { + http.ResponseWriter +} + +func (w flushingWriter) Write(p []byte) (int, error) { + n, err := w.ResponseWriter.Write(p) + if err != nil { + return n, err + } + if err := http.NewResponseController(w.ResponseWriter).Flush(); err != nil { + return n, err + } + return n, nil +} + // Activate allows requests for actorName in atespace. There can be only one // active actor per worker. func (s *Server) Activate(atespace, actorName string) error { @@ -266,48 +406,58 @@ func (s *Server) closeIdleUpstreamConnections() { // ServeHTTP validates the actor hostname on every request before proxying it. func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { + _, requestCtx, release, ok := s.authorize(r) + if !ok { + s.reject(w) + return + } + defer release() + + // Do not expose the router-only routing header to actor code. Restore Host + // so dataplanes that route dynamically on worker IP still give the actor its + // stable actor DNS name. + actorHost := r.Header.Get(OriginalHostHeader) + if actorHost == "" { + actorHost = r.Host + } + r.Header.Del(OriginalHostHeader) + r.Host = actorHost + + // ReverseProxy changes the URL destination but intentionally retains Host, + // allowing the actor application to observe its stable actor DNS name. + s.proxy.ServeHTTP(w, r.WithContext(requestCtx)) +} + +func (s *Server) authorize(r *http.Request) (resources.ActorRef, context.Context, func(), bool) { actorHost := r.Header.Get(OriginalHostHeader) if actorHost == "" { actorHost = r.Host } host, err := requestHostname(actorHost) if err != nil { - s.reject(w) - return + return resources.ActorRef{}, nil, nil, false } ref, err := resources.ParseActorDNSName(host) if err != nil { - s.reject(w) - return + return resources.ActorRef{}, nil, nil, false } s.mu.Lock() active := s.active if active == nil || active.ref != ref { s.mu.Unlock() - s.reject(w) - return + return resources.ActorRef{}, nil, nil, false } active.wg.Add(1) s.mu.Unlock() - defer active.wg.Done() - requestCtx, cancel := context.WithCancel(r.Context()) stop := context.AfterFunc(active.ctx, cancel) - defer func() { + release := func() { + active.wg.Done() stop() cancel() - }() - - // Do not expose the router-only routing header to actor code. Restore Host - // so dataplanes that route dynamically on worker IP still give the actor its - // stable actor DNS name. - r.Header.Del(OriginalHostHeader) - r.Host = actorHost - - // ReverseProxy changes the URL destination but intentionally retains Host, - // allowing the actor application to observe its stable actor DNS name. - s.proxy.ServeHTTP(w, r.WithContext(requestCtx)) + } + return ref, requestCtx, release, true } func (s *Server) reject(w http.ResponseWriter) { diff --git a/internal/atunnel/ingress_test.go b/internal/atunnel/ingress_test.go index 56deac913..a121a39fc 100644 --- a/internal/atunnel/ingress_test.go +++ b/internal/atunnel/ingress_test.go @@ -15,6 +15,7 @@ package atunnel import ( + "bytes" "context" "crypto/ecdsa" "crypto/elliptic" @@ -23,6 +24,7 @@ import ( "crypto/x509" "crypto/x509/pkix" "encoding/pem" + "io" "math/big" "net" "net/http" @@ -34,6 +36,99 @@ import ( "time" ) +func TestRelayIngressWithHalfClose(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer listener.Close() + + accepted := make(chan net.Conn, 1) + go func() { + conn, err := listener.Accept() + if err == nil { + accepted <- conn + } + }() + actor, err := net.Dial("tcp", listener.Addr().String()) + if err != nil { + t.Fatal(err) + } + defer actor.Close() + upstream := <-accepted + defer upstream.Close() + + clientReader, clientInput := io.Pipe() + defer clientReader.Close() + var clientOutput bytes.Buffer + done := make(chan struct{}) + go func() { + relayIngressWithHalfClose(context.Background(), upstream, clientReader, &clientOutput, clientReader) + close(done) + }() + + if _, err := io.WriteString(clientInput, "request"); err != nil { + t.Fatal(err) + } + if err := clientInput.Close(); err != nil { + t.Fatal(err) + } + if err := actor.SetReadDeadline(time.Now().Add(time.Second)); err != nil { + t.Fatal(err) + } + request, err := io.ReadAll(actor) + if err != nil { + t.Fatal(err) + } + if got := string(request); got != "request" { + t.Fatalf("actor received %q, want request", got) + } + + if _, err := io.WriteString(actor, "response"); err != nil { + t.Fatal(err) + } + if err := actor.Close(); err != nil { + t.Fatal(err) + } + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("relay did not finish after the actor response ended") + } + if got := clientOutput.String(); got != "response" { + t.Errorf("client received %q, want response", got) + } +} + +func TestRelayIngressCancellationClosesBothSides(t *testing.T) { + upstream, actor := net.Pipe() + defer actor.Close() + clientReader, clientInput := io.Pipe() + defer clientInput.Close() + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + relayIngressWithHalfClose(ctx, upstream, clientReader, io.Discard, clientReader) + close(done) + }() + + cancel() + if err := actor.SetReadDeadline(time.Now().Add(time.Second)); err != nil { + t.Fatal(err) + } + if _, err := actor.Read(make([]byte, 1)); err == nil { + t.Fatal("actor connection remained open after relay cancellation") + } + if _, err := io.WriteString(clientInput, "request"); err == nil { + t.Fatal("client stream remained open after relay cancellation") + } + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("relay did not return after cancellation") + } +} + func TestServeHTTP(t *testing.T) { upstreamHost := make(chan string, 4) upstreamURL, err := url.Parse("http://actor.internal:80") @@ -152,6 +247,40 @@ func TestServeHTTPHonorsTargetPortHeader(t *testing.T) { } } +func TestServeConnectHTTPValidatesMethodAndAuthority(t *testing.T) { + upstreamURL, err := url.Parse("http://actor.internal:80") + if err != nil { + t.Fatal(err) + } + s := newTestServer(t, upstreamURL) + if err := s.Activate("team-a", "actor-1"); err != nil { + t.Fatal(err) + } + + tests := []struct { + name string + method string + host string + want int + }{ + {name: "rejects non CONNECT", method: http.MethodGet, host: "actor-1.team-a.actors.resources.substrate.ate.dev:9090", want: http.StatusMethodNotAllowed}, + {name: "requires authority port", method: http.MethodConnect, host: "actor-1.team-a.actors.resources.substrate.ate.dev", want: http.StatusBadRequest}, + {name: "rejects invalid authority port", method: http.MethodConnect, host: "actor-1.team-a.actors.resources.substrate.ate.dev:70000", want: http.StatusMisdirectedRequest}, + {name: "rejects inactive actor", method: http.MethodConnect, host: "actor-2.team-a.actors.resources.substrate.ate.dev:9090", want: http.StatusMisdirectedRequest}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + req := httptest.NewRequest(tt.method, "https://worker/", nil) + req.Host = tt.host + rec := httptest.NewRecorder() + s.ServeConnectHTTP(rec, req) + if rec.Code != tt.want { + t.Fatalf("status = %d, want %d", rec.Code, tt.want) + } + }) + } +} + type roundTripFunc func(*http.Request) (*http.Response, error) func (f roundTripFunc) RoundTrip(r *http.Request) (*http.Response, error) { @@ -252,6 +381,9 @@ func TestMutualTLSClientIdentity(t *testing.T) { if err != nil { t.Fatal(err) } + if got := s.tlsConfig.NextProtos; len(got) != 0 { + t.Fatalf("ordinary ingress ALPN protocols = %v, want none", got) + } untrustedCA := newTestCA(t) tests := []struct { diff --git a/internal/e2e/suites/networking/arbitraryport_test.go b/internal/e2e/suites/networking/arbitraryport_test.go index 59a28750d..7121637ce 100644 --- a/internal/e2e/suites/networking/arbitraryport_test.go +++ b/internal/e2e/suites/networking/arbitraryport_test.go @@ -17,6 +17,7 @@ package networking import ( "bufio" "context" + "fmt" "io" "net/http" "regexp" @@ -24,6 +25,7 @@ import ( "testing" "time" + "github.com/agent-substrate/substrate/internal/e2e" "github.com/agent-substrate/substrate/internal/resources" ) @@ -62,16 +64,7 @@ func TestActorArbitraryPortAccess(t *testing.T) { }) t.Run("arbitrary port reachable", func(t *testing.T) { - conn, err := router.Connect(ctx, actorRef, counterExtraPort) - if err != nil { - t.Fatalf("CONNECT to the actor's extra port: %v", err) - } - defer conn.Close() - - resp, body := sendTunneledRequest(t, conn, resources.ActorDNSName(actorRef)) - if resp.StatusCode != http.StatusOK { - t.Fatalf("tunneled request returned HTTP %d, want 200; body: %s", resp.StatusCode, body) - } + body := waitForTunneledRouteReady(t, ctx, router, actorRef, counterExtraPort) if !strings.Contains(body, "extra port 9090") { t.Fatalf("tunneled response body = %q, want it to mention extra port 9090", body) } @@ -131,28 +124,70 @@ func TestActorArbitraryPortAccess(t *testing.T) { }) } -// sendTunneledRequest issues a plain GET / over conn (an established CONNECT -// tunnel) and returns the parsed response with its body already read and the -// connection's read side left consumed accordingly. -func sendTunneledRequest(t *testing.T, conn interface { +// waitForTunneledRouteReady retries an entire CONNECT exchange while the +// resumed actor's route propagates through atenet-router. A CONNECT tunnel is +// tied to one upstream connection, so each retry must establish a fresh tunnel +// rather than reusing a response that Envoy has already closed. +func waitForTunneledRouteReady(t *testing.T, ctx context.Context, router *e2e.RouterClient, actorRef resources.ActorRef, port int) string { + t.Helper() + const timeout = 30 * time.Second + deadline := time.Now().Add(timeout) + var lastErr error + var lastStatus int + var lastBody string + + for { + conn, err := router.Connect(ctx, actorRef, port) + if err == nil { + resp, body, requestErr := requestTunneled(conn, resources.ActorDNSName(actorRef)) + _ = conn.Close() + if requestErr == nil && resp.StatusCode == http.StatusOK { + return body + } + lastErr = requestErr + if resp != nil { + lastStatus = resp.StatusCode + lastBody = body + } + } else { + lastErr = err + } + + if time.Now().After(deadline) { + if lastErr != nil { + t.Fatalf("tunneled request to port %d did not become ready within %s: %v", port, timeout, lastErr) + } + t.Fatalf("tunneled request to port %d returned HTTP %d after %s, want 200; body: %s", port, lastStatus, timeout, lastBody) + } + if lastErr != nil { + t.Logf("tunneled request to port %d failed: %v; retrying...", port, lastErr) + } else { + t.Logf("tunneled request to port %d returned HTTP %d; retrying...", port, lastStatus) + } + time.Sleep(time.Second) + } +} + +func requestTunneled(conn interface { io.ReadWriter SetDeadline(time.Time) error -}, host string) (*http.Response, string) { - t.Helper() - conn.SetDeadline(time.Now().Add(10 * time.Second)) +}, host string) (*http.Response, string, error) { + if err := conn.SetDeadline(time.Now().Add(10 * time.Second)); err != nil { + return nil, "", fmt.Errorf("setting tunneled request deadline: %w", err) + } if _, err := conn.Write([]byte("GET / HTTP/1.1\r\nHost: " + host + "\r\nConnection: close\r\n\r\n")); err != nil { - t.Fatalf("writing tunneled request: %v", err) + return nil, "", fmt.Errorf("writing tunneled request: %w", err) } resp, err := http.ReadResponse(bufio.NewReader(conn), nil) if err != nil { - t.Fatalf("reading tunneled response: %v", err) + return nil, "", fmt.Errorf("reading tunneled response: %w", err) } body, err := io.ReadAll(resp.Body) resp.Body.Close() if err != nil { - t.Fatalf("reading tunneled response body (HTTP %d): %v", resp.StatusCode, err) + return resp, "", fmt.Errorf("reading tunneled response body (HTTP %d): %w", resp.StatusCode, err) } - return resp, string(body) + return resp, string(body), nil } // counterDefaultPortIPPattern matches the "hello from: | ..." response diff --git a/manifests/ate-install/components/agentgateway/kustomization.yaml b/manifests/ate-install/components/agentgateway/kustomization.yaml index d1cfddc96..9c9150653 100644 --- a/manifests/ate-install/components/agentgateway/kustomization.yaml +++ b/manifests/ate-install/components/agentgateway/kustomization.yaml @@ -55,7 +55,7 @@ patches: path: /spec/template/spec/containers/1 value: name: agentgateway - image: cr.agentgateway.dev/agentgateway:v1.4.1 + image: cr.agentgateway.dev/agentgateway:v0.0.0-alpha.a6c0e366 args: - -f - /etc/agentgateway/config.yaml