diff --git a/CHANGELOG.md b/CHANGELOG.md index 3dd2b73..0af5c73 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,12 @@ Gatekeeper is a standalone credential-injecting TLS-intercepting proxy. It trans Gatekeeper is pre-1.0. The configuration schema and credential source interface may change between minor versions. +## v0.25.1 — 2026-10-05 + +### Fixed + +- **HTTP/2 clients can now reach HTTP/1.1-only TLS upstreams** (`proxy/proxy.go`, `proxy/upstream.go`) — the proxy previously selected the upstream protocol from the client's ALPN result. This returned `502` for HTTP/1.1-only services such as Inngest ingest (`inn.gs`, `POST /e/`). The upstream transport now offers `h2` and `http/1.1`, uses HTTP/2 only when the upstream selects it, and falls back to HTTP/1.1 when the upstream selects `http/1.1` or no protocol. Shared host:port connection pools reuse the negotiated protocol across client tunnels without a separate probe handshake. Updating upstream CAs invalidates the shared pool for new tunnels, and server shutdown closes its idle connections. HTTP/2 upstreams still support gRPC, streaming, and trailers. The h2-capable transport retains the previous absence of a response-header deadline, including on fallback; HTTP/1.1 clients retain their five-minute response-header timeout and otherwise unchanged behavior. + ## v0.25.0 — 2026-09-27 ### Changed diff --git a/gatekeeper.go b/gatekeeper.go index f60ca7a..3818d6b 100644 --- a/gatekeeper.go +++ b/gatekeeper.go @@ -965,6 +965,9 @@ func (s *Server) Start(ctx context.Context) error { // Stop gracefully shuts down the proxy server and all background refresh goroutines. func (s *Server) Stop(ctx context.Context) error { + if s.proxy != nil { + defer s.proxy.CloseIdleConnections() + } if s.refreshCancel != nil { s.refreshCancel() } diff --git a/gatekeeper_test.go b/gatekeeper_test.go index 0add2eb..f979349 100644 --- a/gatekeeper_test.go +++ b/gatekeeper_test.go @@ -3500,3 +3500,63 @@ func TestHTTPSTokenExchangeOutrankedByStatic(t *testing.T) { t.Errorf("backend received X-Gatekeeper-Subject = %q, want stripped (declared strip headers must apply when the resolver is skipped)", backendSubjectHeader) } } + +func TestServerStopClosesIdleHTTP2Upstream(t *testing.T) { + closed := make(chan struct{}, 1) + backend := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.ProtoMajor != 2 { + t.Errorf("upstream protocol = %s, want HTTP/2", r.Proto) + } + io.WriteString(w, "ok") + })) + backend.EnableHTTP2 = true + backend.Config.ConnState = func(_ net.Conn, state http.ConnState) { + if state == http.StateClosed { + closed <- struct{}{} + } + } + backend.StartTLS() + defer backend.Close() + ca, err := proxy.NewCA(t.TempDir()) + if err != nil { + t.Fatal(err) + } + p := proxy.NewProxy() + p.SetCA(ca) + upstreamCAs := x509.NewCertPool() + upstreamCAs.AddCert(backend.Certificate()) + p.SetUpstreamCAs(upstreamCAs) + backendURL, _ := url.Parse(backend.URL) + port, _ := net.LookupPort("tcp", backendURL.Port()) + p.SetContextResolver(func(string) (*proxy.RunContextData, bool) { + return &proxy.RunContextData{Policy: "permissive", HostGateway: "127.0.0.1", HostGatewayIP: "127.0.0.1", AllowedHostPorts: []int{port}}, true + }) + front := httptest.NewServer(p) + defer front.Close() + proxyURL, _ := url.Parse(front.URL) + proxyURL.User = url.UserPassword("test", "test-token") + clientCAs := x509.NewCertPool() + clientCAs.AppendCertsFromPEM(ca.CertPEM()) + transport := &http.Transport{Proxy: http.ProxyURL(proxyURL), ForceAttemptHTTP2: true, TLSClientConfig: &tls.Config{RootCAs: clientCAs}} + defer transport.CloseIdleConnections() + client := &http.Client{Transport: transport, Timeout: 5 * time.Second} + resp, err := client.Get(backend.URL) + if err != nil { + t.Fatal(err) + } + io.Copy(io.Discard, resp.Body) + resp.Body.Close() + if resp.StatusCode != 200 || resp.ProtoMajor != 2 { + t.Fatalf("response = %s %s, want HTTP/2 200", resp.Proto, resp.Status) + } + transport.CloseIdleConnections() + s := &Server{proxy: p, proxyServer: front.Config} + if err := s.Stop(context.Background()); err != nil { + t.Fatal(err) + } + select { + case <-closed: + case <-time.After(2 * time.Second): + t.Fatal("standalone stop left an idle upstream HTTP/2 connection open") + } +} diff --git a/proxy/http2_test.go b/proxy/http2_test.go index bf64eb3..6e70748 100644 --- a/proxy/http2_test.go +++ b/proxy/http2_test.go @@ -20,7 +20,9 @@ import ( "net/http/httptest" "net/url" "strings" + "sync/atomic" "testing" + "time" "golang.org/x/net/http2" ) @@ -81,24 +83,33 @@ func newGRPCServer(t *testing.T, receivedHeaders *http.Header) *httptest.Server func newGRPCProxySetup(t *testing.T, receivedHeaders *http.Header) (transport *http2.Transport, backendURL string) { t.Helper() + backend := newGRPCServer(t, receivedHeaders) + t.Cleanup(backend.Close) + transport, backendURL, _ = newHTTP2ProxySetup(t, backend) + return transport, backendURL +} + +func newHTTP2ProxySetup(t *testing.T, backends ...*httptest.Server) (transport *http2.Transport, backendURL string, p *Proxy) { + t.Helper() ca, err := generateCA() if err != nil { t.Fatal(err) } - backend := newGRPCServer(t, receivedHeaders) - t.Cleanup(backend.Close) - upstreamCAs := x509.NewCertPool() - upstreamCAs.AddCert(backend.Certificate()) - - backendAddr, _ := url.Parse(backend.URL) - backendPort := 0 - fmt.Sscanf(backendAddr.Port(), "%d", &backendPort) + var backendPorts []int + for _, backend := range backends { + upstreamCAs.AddCert(backend.Certificate()) + backendAddr, _ := url.Parse(backend.URL) + backendPort := 0 + fmt.Sscanf(backendAddr.Port(), "%d", &backendPort) + backendPorts = append(backendPorts, backendPort) + } - p := NewProxy() + p = NewProxy() p.SetCA(ca) p.SetUpstreamCAs(upstreamCAs) + t.Cleanup(p.CloseIdleConnections) p.SetContextResolver(func(token string) (*RunContextData, bool) { if token != "grpctest" { return nil, false @@ -107,7 +118,7 @@ func newGRPCProxySetup(t *testing.T, receivedHeaders *http.Header) (transport *h Policy: "permissive", HostGateway: "api.modal.com", HostGatewayIP: "127.0.0.1", - AllowedHostPorts: []int{backendPort}, + AllowedHostPorts: backendPorts, Credentials: map[string][]credentialHeader{ "api.modal.com": { {Name: "x-modal-token-id", Value: "token-id-test", Grant: "modal"}, @@ -171,8 +182,9 @@ func newGRPCProxySetup(t *testing.T, receivedHeaders *http.Header) (transport *h }, } - backendURL = fmt.Sprintf("https://api.modal.com:%d", backendPort) - return transport, backendURL + t.Cleanup(transport.CloseIdleConnections) + backendURL = fmt.Sprintf("https://api.modal.com:%d", backendPorts[0]) + return transport, backendURL, p } // TestHTTP2_GRPCCredentialInjection verifies that HTTP/2 (gRPC) requests @@ -223,3 +235,167 @@ func TestHTTP2_GRPCCredentialInjection(t *testing.T) { t.Errorf("x-modal-token-secret = %q, want token-secret-test", got) } } + +func TestHTTP2_UpstreamALPN(t *testing.T) { + for _, tc := range []struct { + name string + protos []string + wantProto int + }{ + {"h1-only", []string{"http/1.1"}, 1}, + {"no-alpn", []string{}, 1}, + {"h2", []string{"h2", "http/1.1"}, 2}, + } { + t.Run(tc.name, func(t *testing.T) { + release := make(chan struct{}) + defer close(release) + backend := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.ProtoMajor != tc.wantProto { + t.Errorf("upstream protocol = %s, want HTTP/%d", r.Proto, tc.wantProto) + } + body, err := io.ReadAll(r.Body) + if err != nil || string(body) != `{"name":"box.event"}` { + t.Errorf("event body = %q, error = %v", body, err) + } + if r.Header.Get("x-modal-token-id") != "token-id-test" { + t.Error("missing injected credential") + } + if r.Header.Get("Proxy-Authorization") != "" || r.Header.Get("Proxy-Connection") != "" { + t.Error("proxy headers leaked upstream") + } + w.Header().Set("Trailer", "X-Event-Status") + if tc.wantProto == 1 { + w.Header().Set("Connection", "X-Hop") + w.Header().Set("X-Hop", "must-not-forward") + } + w.Write([]byte("accepted\n")) + w.(http.Flusher).Flush() + <-release + w.Header().Set("X-Event-Status", "ok") + })) + backend.EnableHTTP2 = tc.wantProto == 2 + backend.TLS = &tls.Config{NextProtos: tc.protos} + backend.StartTLS() + t.Cleanup(backend.Close) + transport, backendURL, p := newHTTP2ProxySetup(t, backend) + logs := make(chan RequestLogData, 1) + p.SetLogger(func(data RequestLogData) { logs <- data }) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + req, _ := http.NewRequestWithContext(ctx, "POST", backendURL+"/e/test-key", strings.NewReader(`{"name":"box.event"}`)) + req.Header.Set("Proxy-Connection", "keep-alive") + resp, err := transport.RoundTrip(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + b, _ := io.ReadAll(resp.Body) + t.Fatalf("POST /e/test-key status = %d, want 200: %s", resp.StatusCode, b) + } + if resp.ProtoMajor != 2 { + t.Errorf("client protocol = %s, want HTTP/2", resp.Proto) + } + if resp.Header.Get("X-Hop") != "" || resp.Header.Get("Connection") != "" { + t.Error("hop-by-hop response headers leaked") + } + first := make([]byte, len("accepted\n")) + if _, err := io.ReadFull(resp.Body, first); err != nil || string(first) != "accepted\n" { + t.Fatalf("streamed body = %q, error = %v", first, err) + } + release <- struct{}{} + if _, err := io.Copy(io.Discard, resp.Body); err != nil { + t.Fatal(err) + } + if got := resp.Trailer.Get("X-Event-Status"); got != "ok" { + t.Errorf("trailer = %q, want ok", got) + } + select { + case data := <-logs: + if data.StatusCode != 200 || !data.AuthInjected || !data.InjectedHeaders["x-modal-token-id"] { + t.Errorf("request log status = %d, injected = %v, headers = %v", data.StatusCode, data.AuthInjected, data.InjectedHeaders) + } + if data.RequestHeaders.Get("x-modal-token-id") != "" || data.RequestHeaders.Get("x-modal-token-secret") != "" { + t.Error("injected credential values leaked into request log") + } + case <-ctx.Done(): + t.Fatal("missing request log") + } + }) + } +} + +func TestHTTP2_UpstreamProtocolCache(t *testing.T) { + var connections [2]atomic.Int32 + var backends []*httptest.Server + for i, wantProto := range []int{1, 2} { + backend := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.ProtoMajor != wantProto { + t.Errorf("upstream protocol = %s, want HTTP/%d", r.Proto, wantProto) + } + io.WriteString(w, "ok") + })) + backend.EnableHTTP2 = wantProto == 2 + backend.Config.ConnState = func(_ net.Conn, state http.ConnState) { + if state == http.StateNew { + connections[i].Add(1) + } + } + backend.StartTLS() + t.Cleanup(backend.Close) + backends = append(backends, backend) + } + transport, _, _ := newHTTP2ProxySetup(t, backends...) + client := &http.Client{Transport: transport, Timeout: 10 * time.Second} + for range 3 { + for _, backend := range backends { + u, _ := url.Parse(backend.URL) + resp, err := client.Get("https://api.modal.com:" + u.Port() + "/cached") + if err != nil { + t.Fatal(err) + } + body, err := io.ReadAll(resp.Body) + resp.Body.Close() + if err != nil || resp.StatusCode != 200 || string(body) != "ok" { + t.Fatalf("cached request status = %d, body = %q, error = %v", resp.StatusCode, body, err) + } + transport.CloseIdleConnections() + } + } + for i := range connections { + if got := connections[i].Load(); got != 1 { + t.Errorf("upstream %d TLS connections = %d, want 1 across client tunnels", i, got) + } + } +} + +func TestHTTP2_UpstreamCAChange(t *testing.T) { + backend := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + io.WriteString(w, "ok") + })) + backend.EnableHTTP2 = true + backend.StartTLS() + t.Cleanup(backend.Close) + transport, backendURL, p := newHTTP2ProxySetup(t, backend) + client := &http.Client{Transport: transport, Timeout: 10 * time.Second} + request := func(wantStatus int) { + t.Helper() + resp, err := client.Get(backendURL + "/trust") + if err != nil { + t.Fatal(err) + } + io.Copy(io.Discard, resp.Body) + resp.Body.Close() + if resp.StatusCode != wantStatus { + t.Fatalf("status after upstream CA change = %d, want %d", resp.StatusCode, wantStatus) + } + transport.CloseIdleConnections() + } + request(http.StatusOK) + p.SetUpstreamCAs(x509.NewCertPool()) + request(http.StatusBadGateway) + trusted := x509.NewCertPool() + trusted.AddCert(backend.Certificate()) + p.SetUpstreamCAs(trusted) + request(http.StatusOK) +} diff --git a/proxy/postgres.go b/proxy/postgres.go index ccc63fe..af091f7 100644 --- a/proxy/postgres.go +++ b/proxy/postgres.go @@ -952,7 +952,7 @@ func (s *PostgresServer) connectWithRetry(ctx context.Context, resolver Postgres return connectPostgresUpstream(ctx, upstreamParams{ dialAddr: dialAddr, serverName: host, - rootCAs: s.proxy.upstreamCAs, + rootCAs: s.proxy.getUpstreamCAs(), user: user, password: password, startupParameters: startupParams, diff --git a/proxy/proxy.go b/proxy/proxy.go index 8b4eff1..00f0352 100644 --- a/proxy/proxy.go +++ b/proxy/proxy.go @@ -698,6 +698,8 @@ type Proxy struct { policyLogger PolicyLogger // optional policy decision logger upstreamCAs *x509.CertPool // optional CA pool for upstream TLS verification captureHeaders []string // headers to capture in logs and strip before forwarding + + h2UpstreamTransport *http.Transport } // NewProxy creates a new auth proxy. @@ -738,7 +740,16 @@ func (p *Proxy) SetCA(ca *CA) { // system root certificates are used. This is useful for environments with // private PKI or for testing. func (p *Proxy) SetUpstreamCAs(pool *x509.CertPool) { + p.mu.Lock() p.upstreamCAs = pool + transport := p.h2UpstreamTransport + p.h2UpstreamTransport = nil + p.mu.Unlock() + if transport != nil { + // Existing tunnels retain their trust settings. Retired connections + // still in use are reclaimed by the transport's timeout once idle. + transport.CloseIdleConnections() + } } // SetLogger sets the request logger. @@ -2837,52 +2848,16 @@ func (p *Proxy) handleConnectWithInterception(w http.ResponseWriter, r *http.Req } }() - // Shared TLS config for upstream connections (both h2 and h1 paths). - upstreamTLS := &tls.Config{ - MinVersion: tls.VersionTLS12, - RootCAs: p.upstreamCAs, - } - - // Build an upstream transport matching the negotiated protocol. - // When the client negotiated h2 (e.g., gRPC), the request object is an - // h2 request and cannot be round-tripped via an HTTP/1.1 transport - // without framing errors, so we must forward upstream over h2 as well. - // - // Limitation: http2.Transport never falls back to HTTP/1.1, so if the - // upstream only speaks HTTP/1.1 the connection will fail when the client - // has negotiated h2. For gRPC this is always correct (gRPC requires h2); - // for general h2 clients hitting h1-only upstreams it is a known - // limitation of the current implementation. var transport http.RoundTripper - dialer := &net.Dialer{ - Timeout: 30 * time.Second, - KeepAlive: 30 * time.Second, - } if tlsClientConn.ConnectionState().NegotiatedProtocol == http2.NextProtoTLS { - transport = &http2.Transport{ - TLSClientConfig: upstreamTLS, - DialTLSContext: func(ctx context.Context, network, addr string, cfg *tls.Config) (net.Conn, error) { - if cfg == nil { - cfg = upstreamTLS - } - return (&tls.Dialer{NetDialer: dialer, Config: cfg}).DialContext(ctx, network, addr) - }, - ReadIdleTimeout: 30 * time.Second, - PingTimeout: 15 * time.Second, + transport, err = p.getH2UpstreamTransport() + if err != nil { + slog.Debug("failed to configure upstream transport", + "subsystem", "proxy", "host", host, "error", err) + return } } else { - transport = &http.Transport{ - Proxy: nil, - DialContext: dialer.DialContext, - TLSClientConfig: upstreamTLS, - // Do NOT set ForceAttemptHTTP2: this path handles HTTP/1.1 - // requests. Enabling h2 upstream for h1 clients causes - // framing mismatches. - TLSHandshakeTimeout: 10 * time.Second, - ResponseHeaderTimeout: 5 * time.Minute, - MaxIdleConns: 100, - IdleConnTimeout: 90 * time.Second, - } + transport = newUpstreamTransport(p.getUpstreamCAs()) } // Extract port from the CONNECT request for rule checking. diff --git a/proxy/server.go b/proxy/server.go index 38168f8..91b5baf 100644 --- a/proxy/server.go +++ b/proxy/server.go @@ -77,6 +77,9 @@ func (s *Server) Port() string { // Stop stops the proxy server. func (s *Server) Stop(ctx context.Context) error { + if s.proxy != nil { + defer s.proxy.CloseIdleConnections() + } if s.server != nil { return s.server.Shutdown(ctx) } diff --git a/proxy/server_test.go b/proxy/server_test.go index 0e9f246..866ae88 100644 --- a/proxy/server_test.go +++ b/proxy/server_test.go @@ -2,7 +2,11 @@ package proxy import ( "context" + "crypto/x509" + "io" "net" + "net/http" + "net/http/httptest" "strings" "testing" "time" @@ -72,3 +76,53 @@ func TestServer_Port(t *testing.T) { t.Errorf("Addr() = %q doesn't end with Port() = %q", addr, port) } } + +func TestServer_StopClosesIdleUpstreamConnections(t *testing.T) { + closed := make(chan struct{}, 1) + backend := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + io.WriteString(w, "ok") + })) + backend.EnableHTTP2 = true + backend.Config.ConnState = func(_ net.Conn, state http.ConnState) { + if state == http.StateClosed { + closed <- struct{}{} + } + } + backend.StartTLS() + defer backend.Close() + p := NewProxy() + pool := x509.NewCertPool() + pool.AddCert(backend.Certificate()) + p.SetUpstreamCAs(pool) + transport, err := p.getH2UpstreamTransport() + if err != nil { + t.Fatal(err) + } + defer transport.CloseIdleConnections() + client := &http.Client{Transport: transport, Timeout: 5 * time.Second} + resp, err := client.Get(backend.URL) + if err != nil { + t.Fatal(err) + } + io.Copy(io.Discard, resp.Body) + resp.Body.Close() + s := NewServer(p) + if err := s.Start(); err != nil { + t.Fatal(err) + } + if err := s.Stop(context.Background()); err != nil { + t.Fatal(err) + } + select { + case <-closed: + case <-time.After(2 * time.Second): + t.Fatal("proxy stop left an idle upstream HTTP/2 connection open") + } +} + +func TestServer_StopBeforeStart(t *testing.T) { + s := &Server{} + if err := s.Stop(context.Background()); err != nil { + t.Fatal(err) + } +} diff --git a/proxy/upstream.go b/proxy/upstream.go new file mode 100644 index 0000000..2a119bd --- /dev/null +++ b/proxy/upstream.go @@ -0,0 +1,71 @@ +package proxy + +import ( + "crypto/tls" + "crypto/x509" + "net" + "net/http" + "time" + + "golang.org/x/net/http2" +) + +func newUpstreamTransport(rootCAs *x509.CertPool) *http.Transport { + return &http.Transport{ + Proxy: nil, + DialContext: (&net.Dialer{ + Timeout: 30 * time.Second, + KeepAlive: 30 * time.Second, + }).DialContext, + TLSClientConfig: &tls.Config{ + MinVersion: tls.VersionTLS12, + RootCAs: rootCAs, + }, + TLSHandshakeTimeout: 10 * time.Second, + ResponseHeaderTimeout: 5 * time.Minute, + MaxIdleConns: 100, + IdleConnTimeout: 90 * time.Second, + } +} + +func (p *Proxy) getUpstreamCAs() *x509.CertPool { + p.mu.RLock() + defer p.mu.RUnlock() + return p.upstreamCAs +} + +func (p *Proxy) getH2UpstreamTransport() (*http.Transport, error) { + p.mu.RLock() + transport := p.h2UpstreamTransport + p.mu.RUnlock() + if transport != nil { + return transport, nil + } + p.mu.Lock() + defer p.mu.Unlock() + if p.h2UpstreamTransport != nil { + return p.h2UpstreamTransport, nil + } + transport = newUpstreamTransport(p.upstreamCAs) + transport.ForceAttemptHTTP2 = true + transport.ResponseHeaderTimeout = 0 + h2Transport, err := http2.ConfigureTransports(transport) + if err != nil { + return nil, err + } + h2Transport.ReadIdleTimeout = 30 * time.Second + h2Transport.PingTimeout = 15 * time.Second + p.h2UpstreamTransport = transport + return transport, nil +} + +// CloseIdleConnections closes idle connections in the shared upstream pool. +// Active requests and per-tunnel HTTP/1.1 transports are unaffected. +func (p *Proxy) CloseIdleConnections() { + p.mu.RLock() + transport := p.h2UpstreamTransport + p.mu.RUnlock() + if transport != nil { + transport.CloseIdleConnections() + } +} diff --git a/proxy/upstream_test.go b/proxy/upstream_test.go new file mode 100644 index 0000000..a04e52a --- /dev/null +++ b/proxy/upstream_test.go @@ -0,0 +1,23 @@ +package proxy + +import ( + "testing" + "time" +) + +func TestUpstreamResponseHeaderTimeouts(t *testing.T) { + p := NewProxy() + transport, err := p.getH2UpstreamTransport() + if err != nil { + t.Fatal(err) + } + defer p.CloseIdleConnections() + if got := transport.ResponseHeaderTimeout; got != 0 { + t.Errorf("h2 response header timeout = %s, want no deadline for long-lived gRPC calls", got) + } + h1Transport := newUpstreamTransport(nil) + defer h1Transport.CloseIdleConnections() + if got := h1Transport.ResponseHeaderTimeout; got != 5*time.Minute { + t.Errorf("h1 response header timeout = %s, want existing 5m0s", got) + } +}