api: harden Connect transports over TLS and h2c

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.

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
@@ -325,9 +331,21 @@ 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()
+	if l.finished {
+		return
+	}
 	if l.idleTimer != nil {
 		l.idleTimer.Stop()
 		l.idleTimer.Reset(l.idleTimeout)
@@ -337,6 +355,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()
 	}
@@ -456,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)
@@ -512,13 +529,14 @@ 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() {
 		if stopTimeout != nil {
 			stopTimeout()
 		}
+		lifecycle.stop()
 		if timeoutCancel != nil {
 			timeoutCancel()
 		}
@@ -707,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{})
 			}
 		}()
 

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,80 @@ 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("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}
+		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 {

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
 }

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.

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/