From a44de43e31e97251613e5f6633dad61906ec3026 Mon Sep 17 00:00:00 2001 From: Siavash Safi Date: Tue, 1 Sep 2026 13:31:47 +0200 Subject: [PATCH 1/2] api: harden Connect transports over TLS and h2c Exercise every supported RPC transport through exporter-toolkit TLS and plaintext h2c. Preserve HTTP/2 ALPN across reloads and isolate mutable server state per listener. Bind lifecycle timer callbacks to request teardown and report listener failures without waiting for other servers. Signed-off-by: Siavash Safi --- README.md | 6 +++ api/connect/connect.go | 11 ++++++ api/connect/status_test.go | 59 +++++++++++++++++++++++++++ app/app.go | 31 +++++++++------ app/lifecycle.go | 43 +++++++++++++++----- app/lifecycle_test.go | 21 ++++++++++ app/listen.go | 45 +++++++++++++++++++++ docs/https.md | 3 +- test/e2e/harness_test.go | 58 ++++++++++++++++++++++----- test/e2e/routing_test.go | 16 +++++--- test/e2e/status_test.go | 81 +++++++++++++++++++++++++++----------- 11 files changed, 313 insertions(+), 61 deletions(-) diff --git a/README.md b/README.md index 9c30662dde..acc4c9ca34 100644 --- a/README.md +++ b/README.md @@ -191,6 +191,12 @@ The v2 `/status` endpoint would be `/api/v2/status`. If `--web.route-prefix` is prefixed with that as well, so `--web.route-prefix=/alertmanager/` would relate to `/alertmanager/api/v2/status`. +The experimental ConnectRPC API serves Connect and gRPC-Web under the route-prefix-aware `/api/` +path and native gRPC, health, and reflection at the server root. The listener accepts native gRPC +over exporter-toolkit TLS with HTTP/2 ALPN and over plaintext h2c. Because h2c provides neither +encryption nor peer authentication, expose a plaintext listener only on a trusted network or behind +a trusted TLS-terminating proxy; use `--web.config.file` to configure TLS for direct exposure. + ## amtool `amtool` is a cli tool for interacting with the Alertmanager API. It is bundled with all releases of Alertmanager. diff --git a/api/connect/connect.go b/api/connect/connect.go index 06a3e413c2..0d23436e1f 100644 --- a/api/connect/connect.go +++ b/api/connect/connect.go @@ -304,12 +304,18 @@ type rpcLifecycle struct { controller *http.ResponseController mutex sync.Mutex idleTimer *time.Timer + finished bool decoded atomic.Bool observed atomic.Bool stream bool } func (l *rpcLifecycle) terminate(cause error) { + l.mutex.Lock() + defer l.mutex.Unlock() + if l.finished { + return + } l.cancel(cause) if l.controller == nil { return @@ -328,6 +334,9 @@ func (l *rpcLifecycle) terminate(cause error) { func (l *rpcLifecycle) touch() { l.mutex.Lock() defer l.mutex.Unlock() + if l.finished { + return + } if l.idleTimer != nil { l.idleTimer.Stop() l.idleTimer.Reset(l.idleTimeout) @@ -337,6 +346,7 @@ func (l *rpcLifecycle) touch() { func (l *rpcLifecycle) stop() { l.mutex.Lock() defer l.mutex.Unlock() + l.finished = true if l.idleTimer != nil { l.idleTimer.Stop() } @@ -519,6 +529,7 @@ func (i *admissionInterceptor) unaryContext(ctx context.Context, controller *htt if stopTimeout != nil { stopTimeout() } + lifecycle.stop() if timeoutCancel != nil { timeoutCancel() } diff --git a/api/connect/status_test.go b/api/connect/status_test.go index 23beec0c07..3297ea6aeb 100644 --- a/api/connect/status_test.go +++ b/api/connect/status_test.go @@ -103,6 +103,13 @@ type deadlineResponseWriter struct { writeDeadline time.Time } +type blockingDeadlineResponseWriter struct { + deadlineResponseWriter + entered chan struct{} + release chan struct{} + enteredOnce sync.Once +} + func (w *deadlineResponseWriter) Header() http.Header { return w.header } func (*deadlineResponseWriter) Write(p []byte) (int, error) { return len(p), nil } func (*deadlineResponseWriter) WriteHeader(int) {} @@ -116,6 +123,12 @@ func (w *deadlineResponseWriter) SetWriteDeadline(t time.Time) error { return nil } +func (w *blockingDeadlineResponseWriter) SetReadDeadline(t time.Time) error { + w.enteredOnce.Do(func() { close(w.entered) }) + <-w.release + return w.deadlineResponseWriter.SetReadDeadline(t) +} + var _ = Describe("StatusService", func() { It("returns status when clustering is disabled", func() { api := newTestAPI(Options{}) @@ -748,6 +761,52 @@ var _ = Describe("RPC admission", func() { Expect(context.Cause(ctx)).To(MatchError(context.DeadlineExceeded)) }) + It("waits for lifecycle termination before finishing cleanup", func() { + ctx, cancel := context.WithCancelCause(context.Background()) + writer := &blockingDeadlineResponseWriter{ + deadlineResponseWriter: deadlineResponseWriter{header: http.Header{}}, + entered: make(chan struct{}), + release: make(chan struct{}), + } + lifecycle := &rpcLifecycle{ + cancel: cancel, + controller: http.NewResponseController(writer), + stream: true, + } + terminated := make(chan struct{}) + go func() { + lifecycle.terminate(context.DeadlineExceeded) + close(terminated) + }() + Eventually(writer.entered).Should(BeClosed()) + stopped := make(chan struct{}) + go func() { + lifecycle.stop() + close(stopped) + }() + Consistently(stopped, 20*time.Millisecond).ShouldNot(BeClosed()) + close(writer.release) + Eventually(terminated).Should(BeClosed()) + Eventually(stopped).Should(BeClosed()) + Expect(writer.readDeadline).NotTo(BeZero()) + Expect(writer.writeDeadline).NotTo(BeZero()) + readDeadline := writer.readDeadline + writeDeadline := writer.writeDeadline + lifecycle.terminate(context.Canceled) + Expect(writer.readDeadline).To(Equal(readDeadline)) + Expect(writer.writeDeadline).To(Equal(writeDeadline)) + Expect(context.Cause(ctx)).To(MatchError(context.DeadlineExceeded)) + }) + + It("does not restart an idle timer after cleanup", func() { + var fired atomic.Bool + lifecycle := &rpcLifecycle{idleTimeout: time.Millisecond} + lifecycle.idleTimer = time.AfterFunc(time.Hour, func() { fired.Store(true) }) + lifecycle.stop() + lifecycle.touch() + Consistently(fired.Load, 20*time.Millisecond).Should(BeFalse()) + }) + It("releases stream capacity after lifetime expiration", func() { api := newTestAPI(Options{StreamConcurrency: 1, StreamLifetime: 10 * time.Millisecond}) wrapped := api.admission.WrapStreamingHandler(func(ctx context.Context, _ connect.StreamingHandlerConn) error { diff --git a/app/app.go b/app/app.go index 7e39dcd8c8..70494ba617 100644 --- a/app/app.go +++ b/app/app.go @@ -107,6 +107,7 @@ type App struct { coordinator *config.Coordinator tracingMgr *tracing.Manager server *http.Server + servers []*http.Server listeners []net.Listener // webReload is the channel exposed by httpserver.Register for the @@ -523,18 +524,26 @@ func (a *App) setup() error { mux := apih.Register(router, routePrefix) - protocols := new(http.Protocols) - protocols.SetHTTP1(true) - protocols.SetHTTP2(true) - protocols.SetUnencryptedHTTP2(true) - a.server = &http.Server{ - // Instrument all handlers with tracing. - Handler: tracing.Middleware(mux), - Protocols: protocols, - ReadHeaderTimeout: 10 * time.Second, - IdleTimeout: 90 * time.Second, + newServer := func() *http.Server { + protocols := new(http.Protocols) + protocols.SetHTTP1(true) + protocols.SetHTTP2(true) + protocols.SetUnencryptedHTTP2(true) + server := &http.Server{ + // Instrument all handlers with tracing. + Handler: tracing.Middleware(mux), + Protocols: protocols, + ReadHeaderTimeout: 10 * time.Second, + IdleTimeout: 90 * time.Second, + } + server.RegisterOnShutdown(apih.Shutdown) + return server + } + a.servers = make([]*http.Server, 0, len(a.listeners)) + for range a.listeners { + a.servers = append(a.servers, newServer()) } - a.server.RegisterOnShutdown(apih.Shutdown) + a.server = a.servers[0] return nil } diff --git a/app/lifecycle.go b/app/lifecycle.go index 6d3c74773d..f89ffa2126 100644 --- a/app/lifecycle.go +++ b/app/lifecycle.go @@ -17,6 +17,7 @@ import ( "context" "errors" "fmt" + "net" "net/http" "slices" "time" @@ -61,11 +62,22 @@ func (a *App) Start() error { // on an unbuffered channel has no receiver. go a.reloadRouter() + http2Enabled := configuredHTTP2(*a.opts.WebConfig.WebConfigFile) go func() { - err := web.ServeMultiple(a.listeners, a.server, a.opts.WebConfig, a.logger) - if err != nil && !errors.Is(err, http.ErrServerClosed) { - a.logger.Error("Listen error", "err", err) - a.serveErrc <- err + errCh := make(chan error, len(a.listeners)) + for idx, listener := range a.listeners { + server := a.servers[idx] + go func() { + errCh <- web.Serve(withReloadableTLSALPN([]net.Listener{listener}, server, http2Enabled)[0], server, a.opts.WebConfig, a.logger) + }() + } + reported := false + for range a.listeners { + if err := <-errCh; err != nil && !errors.Is(err, http.ErrServerClosed) && !reported { + a.logger.Error("Listen error", "err", err) + a.serveErrc <- err + reported = true + } } close(a.serveErrc) }() @@ -206,13 +218,24 @@ func (a *App) Stop(ctx context.Context) error { // reload router is still running at this point so any in-flight // /-/reload handler can complete its send/receive cycle and unblock // Shutdown. - if a.server != nil { - if err := a.server.Shutdown(shutdownCtx); err != nil { - a.logger.Warn("graceful HTTP shutdown failed", "err", err) - stopErr = err - if closeErr := a.server.Close(); closeErr != nil { - stopErr = errors.Join(stopErr, closeErr) + servers := a.servers + if len(servers) == 0 && a.server != nil { + servers = []*http.Server{a.server} + } + shutdownErrCh := make(chan error, len(servers)) + for _, server := range servers { + go func() { + err := server.Shutdown(shutdownCtx) + if err != nil { + err = errors.Join(err, server.Close()) } + shutdownErrCh <- err + }() + } + for range servers { + if err := <-shutdownErrCh; err != nil { + a.logger.Warn("graceful HTTP shutdown failed", "err", err) + stopErr = errors.Join(stopErr, err) } } // HTTP is fully drained; no new /-/reload requests can arrive. diff --git a/app/lifecycle_test.go b/app/lifecycle_test.go index f957aab8ad..ec729a3b6a 100644 --- a/app/lifecycle_test.go +++ b/app/lifecycle_test.go @@ -117,6 +117,27 @@ func TestApp_StartStop(t *testing.T) { require.NoError(t, a.Stop(t.Context())) } +func TestApp_StartReportsFirstListenerFailure(t *testing.T) { + opts := testOptions(t) + addrs := []string{"127.0.0.1:0", "127.0.0.1:0"} + opts.WebConfig.WebListenAddresses = &addrs + a, err := New(opts) + require.NoError(t, err) + require.Len(t, a.listeners, 2) + require.NoError(t, a.listeners[0].Close()) + require.NoError(t, a.Start()) + + done := make(chan error, 1) + go func() { done <- a.serveLoop(context.Background()) }() + select { + case err := <-done: + require.ErrorContains(t, err, "HTTP listener failed") + case <-time.After(time.Second): + t.Fatal("listener failure was not reported while another listener was active") + } + require.NoError(t, a.Stop(t.Context())) +} + func TestApp_ClusteredStartStop(t *testing.T) { // Bring up an instance with gossip clustering enabled so the // peer-dependent branches in setup (AddState/Join/Settle/ diff --git a/app/listen.go b/app/listen.go index c73b2a5362..aa8d64d3ba 100644 --- a/app/listen.go +++ b/app/listen.go @@ -17,13 +17,17 @@ import ( "errors" "fmt" "net" + "net/http" "net/url" + "os" "strconv" "strings" + "sync" "github.com/coreos/go-systemd/v22/activation" "github.com/mdlayher/vsock" "github.com/prometheus/exporter-toolkit/web" + "gopkg.in/yaml.v2" ) // listenAll eagerly binds every listener described by flags so that the @@ -94,3 +98,44 @@ func parseVsockPort(address string) (uint32, error) { } return uint32(port), nil } + +type reloadableTLSALPNListener struct { + net.Listener + server *http.Server + once *sync.Once + http2Enabled bool +} + +func (l reloadableTLSALPNListener) Accept() (net.Conn, error) { + l.once.Do(func() { + if !l.http2Enabled || l.server.TLSConfig == nil { + return + } + if _, ok := l.server.TLSNextProto["h2"]; ok { + l.server.TLSConfig.NextProtos = []string{"h2", "http/1.1"} + } + }) + return l.Listener.Accept() +} + +func configuredHTTP2(path string) bool { + config := struct { + HTTP struct { + HTTP2 *bool `yaml:"http2"` + } `yaml:"http_server_config"` + }{} + content, err := os.ReadFile(path) + if err != nil || yaml.Unmarshal(content, &config) != nil || config.HTTP.HTTP2 == nil { + return true + } + return *config.HTTP.HTTP2 +} + +func withReloadableTLSALPN(listeners []net.Listener, server *http.Server, http2Enabled bool) []net.Listener { + once := &sync.Once{} + wrapped := make([]net.Listener, 0, len(listeners)) + for _, listener := range listeners { + wrapped = append(wrapped, reloadableTLSALPNListener{Listener: listener, server: server, once: once, http2Enabled: http2Enabled}) + } + return wrapped +} diff --git a/docs/https.md b/docs/https.md index 6fa3ee8115..f78b1230ec 100644 --- a/docs/https.md +++ b/docs/https.md @@ -80,7 +80,8 @@ tls_server_config: [ - ] ] http_server_config: - # Enable HTTP/2 support. Note that HTTP/2 is only supported with TLS. + # Enable HTTP/2 support over TLS. Alertmanager also accepts plaintext h2c, + # independently of this setting; expose plaintext only on a trusted network. # This can not be changed on the fly. [ http2: | default = true ] # List of headers that can be added to HTTP responses. diff --git a/test/e2e/harness_test.go b/test/e2e/harness_test.go index a0fe5d0ed2..cd3de91246 100644 --- a/test/e2e/harness_test.go +++ b/test/e2e/harness_test.go @@ -15,6 +15,9 @@ package e2e import ( "context" + "crypto/tls" + "crypto/x509" + "fmt" "net/http" "os" "path/filepath" @@ -46,13 +49,18 @@ type instance struct { baseURL string routePrefix string httpClient *http.Client - h2cClient *http.Client + rpcClient *http.Client + tlsConfig *tls.Config } // startInstance boots an Alertmanager and registers its teardown (and // temp-dir removal) via Ginkgo's DeferCleanup. -func startInstance(routePrefix string) *instance { +func startInstance(routePrefix string, tlsEnabled bool, enableHTTP2 ...bool) *instance { GinkgoHelper() + http2Enabled := true + if len(enableHTTP2) > 0 { + http2Enabled = enableHTTP2[0] + } dir, err := os.MkdirTemp("", "am-e2e-") Expect(err).NotTo(HaveOccurred()) @@ -71,6 +79,22 @@ func startInstance(routePrefix string) *instance { addrs := []string{"127.0.0.1:0"} systemd := false webCfg := "" + var clientTLS *tls.Config + if tlsEnabled { + cert, err := os.ReadFile(filepath.Join("..", "..", "cluster", "testdata", "certs", "node1.pem")) + Expect(err).NotTo(HaveOccurred()) + key, err := os.ReadFile(filepath.Join("..", "..", "cluster", "testdata", "certs", "node1-key.pem")) + Expect(err).NotTo(HaveOccurred()) + ca, err := os.ReadFile(filepath.Join("..", "..", "cluster", "testdata", "certs", "ca.pem")) + Expect(err).NotTo(HaveOccurred()) + Expect(os.WriteFile(filepath.Join(dir, "server.pem"), cert, 0o600)).To(Succeed()) + Expect(os.WriteFile(filepath.Join(dir, "server-key.pem"), key, 0o600)).To(Succeed()) + webCfg = filepath.Join(dir, "web.yml") + Expect(os.WriteFile(webCfg, []byte(fmt.Sprintf("tls_server_config:\n cert_file: %s\n key_file: %s\nhttp_server_config:\n http2: %t\n", filepath.Join(dir, "server.pem"), filepath.Join(dir, "server-key.pem"), http2Enabled)), 0o600)).To(Succeed()) + roots := x509.NewCertPool() + Expect(roots.AppendCertsFromPEM(ca)).To(BeTrue()) + clientTLS = &tls.Config{RootCAs: roots, MinVersion: tls.VersionTLS12} + } opts := app.DefaultOptions() opts.ConfigFile = configPath @@ -95,19 +119,33 @@ func startInstance(routePrefix string) *instance { Expect(a.Start()).To(Succeed()) client := &http.Client{Timeout: 5 * time.Second} - DeferCleanup(client.CloseIdleConnections) - protocols := new(http.Protocols) - protocols.SetUnencryptedHTTP2(true) - h2cTransport := &http.Transport{Protocols: protocols} - h2cClient := &http.Client{Transport: h2cTransport, Timeout: 5 * time.Second} - DeferCleanup(h2cTransport.CloseIdleConnections) + var rpcClient *http.Client + scheme := "http://" + if tlsEnabled { + protocols := new(http.Protocols) + protocols.SetHTTP1(true) + protocols.SetHTTP2(true) + transport := &http.Transport{TLSClientConfig: clientTLS, Protocols: protocols} + client = &http.Client{Transport: transport, Timeout: 5 * time.Second} + rpcClient = client + scheme = "https://" + DeferCleanup(transport.CloseIdleConnections) + } else { + DeferCleanup(client.CloseIdleConnections) + protocols := new(http.Protocols) + protocols.SetUnencryptedHTTP2(true) + transport := &http.Transport{Protocols: protocols} + rpcClient = &http.Client{Transport: transport, Timeout: 5 * time.Second} + DeferCleanup(transport.CloseIdleConnections) + } inst := &instance{ app: a, - baseURL: "http://" + a.Addr(), + baseURL: scheme + a.Addr(), routePrefix: routePrefix, httpClient: client, - h2cClient: h2cClient, + rpcClient: rpcClient, + tlsConfig: clientTLS, } inst.waitHealthy() return inst diff --git a/test/e2e/routing_test.go b/test/e2e/routing_test.go index ffa476cf53..ab06716164 100644 --- a/test/e2e/routing_test.go +++ b/test/e2e/routing_test.go @@ -22,16 +22,20 @@ import ( var _ = Describe("API routing", func() { DescribeTable("serves v1 and v2 alongside the Connect API", - func(routePrefix, path string, expectedStatus int) { - inst := startInstance(routePrefix) + func(routePrefix, path string, tlsEnabled bool, expectedStatus int) { + inst := startInstance(routePrefix, tlsEnabled) resp, err := inst.httpClient.Get(inst.webURL(path)) Expect(err).NotTo(HaveOccurred()) DeferCleanup(resp.Body.Close) Expect(resp.StatusCode).To(Equal(expectedStatus)) }, - Entry("v2 at the root", "", "/api/v2/status", http.StatusOK), - Entry("v1 at the root", "", "/api/v1/status", http.StatusGone), - Entry("v2 under a route prefix", "/alertmanager", "/api/v2/status", http.StatusOK), - Entry("v1 under a route prefix", "/alertmanager", "/api/v1/status", http.StatusGone), + Entry("v2 over h2c at the root", "", "/api/v2/status", false, http.StatusOK), + Entry("v1 over h2c at the root", "", "/api/v1/status", false, http.StatusGone), + Entry("v2 over h2c under a route prefix", "/alertmanager", "/api/v2/status", false, http.StatusOK), + Entry("v1 over h2c under a route prefix", "/alertmanager", "/api/v1/status", false, http.StatusGone), + Entry("v2 over TLS at the root", "", "/api/v2/status", true, http.StatusOK), + Entry("v1 over TLS at the root", "", "/api/v1/status", true, http.StatusGone), + Entry("v2 over TLS under a route prefix", "/alertmanager", "/api/v2/status", true, http.StatusOK), + Entry("v1 over TLS under a route prefix", "/alertmanager", "/api/v1/status", true, http.StatusGone), ) }) diff --git a/test/e2e/status_test.go b/test/e2e/status_test.go index 695f3f60ab..9079090459 100644 --- a/test/e2e/status_test.go +++ b/test/e2e/status_test.go @@ -15,7 +15,7 @@ package e2e import ( "context" - "strings" + "crypto/tls" "time" "connectrpc.com/connect" @@ -23,8 +23,10 @@ import ( . "github.com/onsi/gomega" "github.com/prometheus/common/version" "google.golang.org/grpc" + "google.golang.org/grpc/credentials" "google.golang.org/grpc/credentials/insecure" healthv1 "google.golang.org/grpc/health/grpc_health_v1" + "google.golang.org/grpc/peer" reflectionv1 "google.golang.org/grpc/reflection/grpc_reflection_v1" statusv3alpha "github.com/prometheus/alertmanager/api/status/v3alpha" @@ -33,12 +35,12 @@ import ( var _ = Describe("StatusService", func() { DescribeTable("GetStatus succeeds over supported transports", - func(routePrefix string, nativeGRPC bool, opts []connect.ClientOption) { - inst := startInstance(routePrefix) + func(routePrefix string, tlsEnabled, nativeGRPC bool, opts []connect.ClientOption) { + inst := startInstance(routePrefix, tlsEnabled) httpClient := connect.HTTPClient(inst.httpClient) basePath := inst.apiPath() if nativeGRPC { - httpClient = inst.h2cClient + httpClient = inst.rpcClient basePath = "" } client := inst.statusClient(httpClient, basePath, opts...) @@ -54,25 +56,33 @@ var _ = Describe("StatusService", func() { Expect(status.GetStartTime().AsTime()).NotTo(BeZero()) Expect(status.GetCluster().GetState()).To(Equal(statusv3alpha.ClusterStatus_STATE_DISABLED)) }, - Entry("Connect POST at the root prefix", "", false, []connect.ClientOption{}), - Entry("Connect HTTP GET at the root prefix", "", false, []connect.ClientOption{connect.WithHTTPGet()}), - Entry("gRPC-Web at the root prefix", "", false, []connect.ClientOption{connect.WithGRPCWeb()}), - Entry("native gRPC at the server root", "", true, []connect.ClientOption{connect.WithGRPC()}), - Entry("Connect POST under a route prefix", "/alertmanager", false, []connect.ClientOption{}), - Entry("Connect HTTP GET under a route prefix", "/alertmanager", false, []connect.ClientOption{connect.WithHTTPGet()}), - Entry("gRPC-Web under a route prefix", "/alertmanager", false, []connect.ClientOption{connect.WithGRPCWeb()}), - Entry("native gRPC with a route prefix configured", "/alertmanager", true, []connect.ClientOption{connect.WithGRPC()}), + Entry("Connect POST over h2c at the root prefix", "", false, false, []connect.ClientOption{}), + Entry("Connect HTTP GET over h2c at the root prefix", "", false, false, []connect.ClientOption{connect.WithHTTPGet()}), + Entry("gRPC-Web over h2c at the root prefix", "", false, false, []connect.ClientOption{connect.WithGRPCWeb()}), + Entry("native gRPC over h2c at the server root", "", false, true, []connect.ClientOption{connect.WithGRPC()}), + Entry("Connect POST over h2c under a route prefix", "/alertmanager", false, false, []connect.ClientOption{}), + Entry("Connect HTTP GET over h2c under a route prefix", "/alertmanager", false, false, []connect.ClientOption{connect.WithHTTPGet()}), + Entry("gRPC-Web over h2c under a route prefix", "/alertmanager", false, false, []connect.ClientOption{connect.WithGRPCWeb()}), + Entry("native gRPC over h2c with a route prefix configured", "/alertmanager", false, true, []connect.ClientOption{connect.WithGRPC()}), + Entry("Connect POST over TLS at the root prefix", "", true, false, []connect.ClientOption{}), + Entry("Connect HTTP GET over TLS at the root prefix", "", true, false, []connect.ClientOption{connect.WithHTTPGet()}), + Entry("gRPC-Web over TLS at the root prefix", "", true, false, []connect.ClientOption{connect.WithGRPCWeb()}), + Entry("native gRPC over TLS at the server root", "", true, true, []connect.ClientOption{connect.WithGRPC()}), + Entry("Connect POST over TLS under a route prefix", "/alertmanager", true, false, []connect.ClientOption{}), + Entry("Connect HTTP GET over TLS under a route prefix", "/alertmanager", true, false, []connect.ClientOption{connect.WithHTTPGet()}), + Entry("gRPC-Web over TLS under a route prefix", "/alertmanager", true, false, []connect.ClientOption{connect.WithGRPCWeb()}), + Entry("native gRPC over TLS with a route prefix configured", "/alertmanager", true, true, []connect.ClientOption{connect.WithGRPC()}), ) DescribeTable("rejects transports outside their configured prefix", func(routePrefix, basePath string, nativeGRPC bool, opts []connect.ClientOption) { - inst := startInstance(routePrefix) + inst := startInstance(routePrefix, false) httpClient := connect.HTTPClient(inst.httpClient) if basePath == "api" { basePath = inst.apiPath() } if nativeGRPC { - httpClient = inst.h2cClient + httpClient = inst.rpcClient } client := inst.statusClient(httpClient, basePath, opts...) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) @@ -90,18 +100,31 @@ var _ = Describe("StatusService", func() { ) DescribeTable("exposes native health and reflection at the server root", - func(routePrefix string) { - inst := startInstance(routePrefix) + func(routePrefix string, tlsEnabled bool) { + inst := startInstance(routePrefix, tlsEnabled) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - conn, err := grpc.NewClient(strings.TrimPrefix(inst.baseURL, "http://"), grpc.WithTransportCredentials(insecure.NewCredentials())) + transportCredentials := credentials.TransportCredentials(insecure.NewCredentials()) + if tlsEnabled { + transportCredentials = credentials.NewTLS(inst.tlsConfig.Clone()) + } + conn, err := grpc.NewClient(inst.app.Addr(), grpc.WithTransportCredentials(transportCredentials)) Expect(err).NotTo(HaveOccurred()) DeferCleanup(conn.Close) - health, err := healthv1.NewHealthClient(conn).Check(ctx, &healthv1.HealthCheckRequest{}) - Expect(err).NotTo(HaveOccurred()) - Expect(health.GetStatus()).To(Equal(healthv1.HealthCheckResponse_SERVING)) + healthClient := healthv1.NewHealthClient(conn) + for _, service := range []string{"", statusv3alphaconnect.StatusServiceName} { + var remote peer.Peer + health, err := healthClient.Check(ctx, &healthv1.HealthCheckRequest{Service: service}, grpc.Peer(&remote)) + Expect(err).NotTo(HaveOccurred()) + Expect(health.GetStatus()).To(Equal(healthv1.HealthCheckResponse_SERVING)) + if tlsEnabled { + info, ok := remote.AuthInfo.(credentials.TLSInfo) + Expect(ok).To(BeTrue()) + Expect(info.State.NegotiatedProtocol).To(Equal("h2")) + } + } stream, err := reflectionv1.NewServerReflectionClient(conn).ServerReflectionInfo(ctx) Expect(err).NotTo(HaveOccurred()) @@ -118,12 +141,24 @@ var _ = Describe("StatusService", func() { } Expect(names).To(ContainElement(statusv3alphaconnect.StatusServiceName)) }, - Entry("without a route prefix", ""), - Entry("with a route prefix", "/alertmanager"), + Entry("over h2c without a route prefix", "", false), + Entry("over h2c with a route prefix", "/alertmanager", false), + Entry("over TLS without a route prefix", "", true), + Entry("over TLS with a route prefix", "/alertmanager", true), ) + It("honors exporter-toolkit HTTP/2 disablement", func() { + inst := startInstance("", true, false) + config := inst.tlsConfig.Clone() + config.NextProtos = []string{"h2", "http/1.1"} + conn, err := tls.Dial("tcp", inst.app.Addr(), config) + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(conn.Close) + Expect(conn.ConnectionState().NegotiatedProtocol).NotTo(Equal("h2")) + }) + It("cancels active streams during shutdown", func() { - inst := startInstance("") + inst := startInstance("", false) conn, err := grpc.NewClient(inst.app.Addr(), grpc.WithTransportCredentials(insecure.NewCredentials())) Expect(err).NotTo(HaveOccurred()) DeferCleanup(conn.Close) From 152f6ee3969d4f8309d62838e1e8057781faa16e Mon Sep 17 00:00:00 2001 From: Siavash Safi Date: Thu, 17 Sep 2026 15:09:41 +0200 Subject: [PATCH 2/2] api: serialize Connect read deadline updates Route read deadline changes through the RPC lifecycle mutex so request teardown cannot complete while a controller update is active. Signed-off-by: Siavash Safi --- api/connect/connect.go | 19 +++++++++++++------ api/connect/status_test.go | 28 ++++++++++++++++++++++++++++ 2 files changed, 41 insertions(+), 6 deletions(-) diff --git a/api/connect/connect.go b/api/connect/connect.go index 0d23436e1f..7d47c79ac8 100644 --- a/api/connect/connect.go +++ b/api/connect/connect.go @@ -331,6 +331,15 @@ func (l *rpcLifecycle) terminate(cause error) { } } +func (l *rpcLifecycle) setReadDeadline(deadline time.Time) { + l.mutex.Lock() + defer l.mutex.Unlock() + if l.finished || l.controller == nil { + return + } + _ = l.controller.SetReadDeadline(deadline) +} + func (l *rpcLifecycle) touch() { l.mutex.Lock() defer l.mutex.Unlock() @@ -466,9 +475,7 @@ func (i *admissionInterceptor) WrapUnary(next connect.UnaryFunc) connect.UnaryFu desc := i.descriptor(req.Spec().Procedure) state := unaryRequestStateFromContext(ctx) state.lifecycle.decoded.Store(true) - if state.lifecycle.controller != nil { - _ = state.lifecycle.controller.SetReadDeadline(time.Time{}) - } + state.lifecycle.setReadDeadline(time.Time{}) response, err := next(ctx, req) err = normalizeContextError(ctx, err) i.observe(desc, state.started, err) @@ -522,7 +529,7 @@ func (i *admissionInterceptor) unaryContext(ctx context.Context, controller *htt } }) if deadline, ok := unaryCtx.Deadline(); ok { - _ = controller.SetReadDeadline(deadline) + lifecycle.setReadDeadline(deadline) } } return context.WithValue(unaryCtx, rpcLifecycleContextKey{}, lifecycle), lifecycle, func() { @@ -718,8 +725,8 @@ func (api *API) controlHandler(next http.Handler, errorWriter *connect.ErrorWrit } }() defer func() { - if lifecycle.controller != nil && ctx.Err() == nil { - _ = lifecycle.controller.SetReadDeadline(time.Time{}) + if ctx.Err() == nil { + lifecycle.setReadDeadline(time.Time{}) } }() diff --git a/api/connect/status_test.go b/api/connect/status_test.go index 3297ea6aeb..327ab18558 100644 --- a/api/connect/status_test.go +++ b/api/connect/status_test.go @@ -798,6 +798,34 @@ var _ = Describe("RPC admission", func() { Expect(context.Cause(ctx)).To(MatchError(context.DeadlineExceeded)) }) + It("waits for read deadline updates before finishing cleanup", func() { + writer := &blockingDeadlineResponseWriter{ + deadlineResponseWriter: deadlineResponseWriter{header: http.Header{}}, + entered: make(chan struct{}), + release: make(chan struct{}), + } + lifecycle := &rpcLifecycle{controller: http.NewResponseController(writer)} + deadline := time.Now().Add(time.Minute) + updated := make(chan struct{}) + go func() { + lifecycle.setReadDeadline(deadline) + close(updated) + }() + Eventually(writer.entered).Should(BeClosed()) + stopped := make(chan struct{}) + go func() { + lifecycle.stop() + close(stopped) + }() + Consistently(stopped, 20*time.Millisecond).ShouldNot(BeClosed()) + close(writer.release) + Eventually(updated).Should(BeClosed()) + Eventually(stopped).Should(BeClosed()) + Expect(writer.readDeadline).To(Equal(deadline)) + lifecycle.setReadDeadline(time.Time{}) + Expect(writer.readDeadline).To(Equal(deadline)) + }) + It("does not restart an idle timer after cleanup", func() { var fired atomic.Bool lifecycle := &rpcLifecycle{idleTimeout: time.Millisecond}