diff --git a/proxy/internal/roundtrip/upstream.go b/proxy/internal/roundtrip/upstream.go index d6d11a26a..902cc1bef 100644 --- a/proxy/internal/roundtrip/upstream.go +++ b/proxy/internal/roundtrip/upstream.go @@ -1,7 +1,10 @@ package roundtrip import ( + "errors" + "net" "net/http" + "net/url" "strings" "sync" "time" @@ -93,7 +96,7 @@ func (t *upstreamTransport) RoundTrip(req *http.Request) (*http.Response, error) return t.primary.RoundTrip(req) } - host := req.URL.Host + host := upstreamKey(req.URL) if t.isDowngraded(host) { return t.http1().RoundTrip(req) } @@ -109,11 +112,22 @@ func (t *upstreamTransport) RoundTrip(req *http.Request) (*http.Response, error) // the cause is site configuration rather than a passing fault. t.markDowngraded(host, isHTTP11Required(err)) + if !safeToRetry(req, err) { + // The upstream may have carried out the request before failing + // to answer over h2, and repeating it could duplicate whatever + // it did. The host is pinned either way, so the next request + // goes out over HTTP/1.1. + t.logger.WithFields(log.Fields{ + "upstream": host, + "method": req.Method, + }).Debug("not retrying over HTTP/1.1: the upstream may already have applied this request") + return nil, err + } + retry, ok := replayable(req) if !ok { // The body is already consumed and cannot be regenerated, so - // this request fails. The host is pinned either way, so the - // next one goes out over HTTP/1.1. + // this request fails, and the pin carries the next one. return nil, err } return t.http1().RoundTrip(retry) @@ -136,8 +150,6 @@ func (t *upstreamTransport) mayDowngrade(req *http.Request) bool { } func (t *upstreamTransport) isDowngraded(host string) bool { - now := time.Now() - t.mu.RLock() pin, ok := t.downgraded[host] t.mu.RUnlock() @@ -145,17 +157,25 @@ func (t *upstreamTransport) isDowngraded(host string) bool { if !ok { return false } - if pin.active(now) { + if pin.active(time.Now()) { return true } t.mu.Lock() - // Re-check under the write lock: a concurrent request may have - // re-pinned the host after the read above. - if pin, ok := t.downgraded[host]; ok && !pin.active(time.Now()) { - delete(t.downgraded, host) + defer t.mu.Unlock() + + // Re-read under the write lock rather than trusting the expired pin + // from above: a concurrent request may have re-pinned the host since, + // and that pin decides this request too. Reporting the stale read + // would send one request back to h2 against a live pin. + pin, ok = t.downgraded[host] + if !ok { + return false } - t.mu.Unlock() + if pin.active(time.Now()) { + return true + } + delete(t.downgraded, host) return false } @@ -174,6 +194,7 @@ func (t *upstreamTransport) markDowngraded(host string, permanent bool) { previous, pinned := t.downgraded[host] // A permanent pin is never weakened back into an expiring one: the // upstream has already said h2 is not on offer. + promoted := pinned && !previous.permanent() && permanent if !pinned || !previous.permanent() { t.downgraded[host] = pin } @@ -184,7 +205,10 @@ func (t *upstreamTransport) markDowngraded(host string, permanent bool) { } t.mu.Unlock() - if pinned { + // Log a new pin, and a pin the upstream has since asked to make + // permanent — otherwise an operator would only ever see the "for the + // next 10m" line and never learn the upstream settled the question. + if pinned && !promoted { return } @@ -221,6 +245,71 @@ func (t *upstreamTransport) existingHTTP1() *http.Transport { return t.fallback } +// upstreamKey normalizes an authority for use as a pin key, so one +// upstream cannot end up with two independent pins. DNS labels compare +// case-insensitively, and the default HTTPS port is implied — every +// downgrade path is TLS-only, so a bare host and the same host on :443 +// are the same upstream. +func upstreamKey(u *url.URL) string { + host := strings.ToLower(u.Hostname()) + + port := u.Port() + if port == "" || port == "443" { + return host + } + + // JoinHostPort rather than concatenation: an IPv6 literal needs its + // brackets back after Hostname stripped them. + return net.JoinHostPort(host, port) +} + +// safeToRetry reports whether req may be sent a second time over +// HTTP/1.1 after err ended its h2 attempt. +// +// A failure at the h2 layer does not say whether the upstream already +// carried out the request, so replaying one that changes state could +// duplicate it. Two cases are safe: a request whose repetition is +// harmless by definition, and an upstream that told us it processed +// nothing on the connection. The second is what makes the IIS case work +// for every method — a site requiring HTTP/1.1 refuses at stream 0, +// before the request is looked at. +func safeToRetry(req *http.Request, err error) bool { + return idempotent(req) || upstreamProcessedNothing(err) +} + +// idempotent reports whether repeating req is defined to be harmless. +// It mirrors net/http's own retry rule (Request.isReplayable): a method +// with no side effects, or a caller that promised the upstream +// deduplicates by key. +func idempotent(req *http.Request) bool { + if req.Header.Get("Idempotency-Key") != "" || req.Header.Get("X-Idempotency-Key") != "" { + return true + } + + switch req.Method { + // An empty method means GET, as in net/http. + case "", http.MethodGet, http.MethodHead, http.MethodOptions, http.MethodTrace: + return true + } + + return false +} + +// upstreamProcessedNothing reports whether err describes a GOAWAY that +// named this request's stream as one the upstream had not received, so +// it cannot have acted on it. A stream error says the opposite: the +// stream was open, so the request had been delivered. +func upstreamProcessedNothing(err error) bool { + if err == nil { + return false + } + + msg := transportError(err).Error() + + return strings.Contains(msg, goAwayStreamNotReceivedMarker) || + strings.Contains(msg, goAwayNothingProcessedMarker) +} + // replayable returns a request that can be sent a second time, or // ok=false when the body is gone. A RoundTripper consumes and closes // the body it was given, so a retry needs either no body at all or @@ -265,15 +354,41 @@ var http2ErrorMarkers = []string{ http11RequiredMarker, } -// http11RequiredMarker is the error code an upstream sends to say the -// request belongs on HTTP/1.1. Unlike the other markers it is not a -// fault: the upstream is describing its own configuration. -const http11RequiredMarker = "HTTP_1_1_REQUIRED" +const ( + // http11RequiredMarker is the error code an upstream sends to say the + // request belongs on HTTP/1.1. Unlike the other markers it is not a + // fault: the upstream is describing its own configuration. + http11RequiredMarker = "HTTP_1_1_REQUIRED" + + // A GOAWAY carrying NO_ERROR closes a connection without complaint: + // a server draining before shutdown, recycling an application pool, + // capping requests per connection. The upstream speaks h2 perfectly + // well, so this must never pin it. The two spellings are the two + // formats the bundled transport prints the code in. + goAwayNoErrorEqualsMarker = "ErrCode=NO_ERROR" + goAwayNoErrorColonMarker = "ErrCode:NO_ERROR" + // gracefulGoAwayMarker is errClientConnGotGoAway, which the bundled + // transport raises for a stream the server never received on a + // connection it is shutting down gracefully. It normally retries + // those itself on a new connection and this never surfaces. + gracefulGoAwayMarker = "Transport received Server's graceful shutdown GOAWAY" + + // goAwayStreamNotReceivedMarker is the bundled transport's abort for + // the first stream on a connection whose GOAWAY carried a real error + // code — the IIS case. It sits in the same "streamID > LastStreamID" + // branch as the graceful abort, so the server had not received the + // stream (see net/http's h2_bundle.go). + goAwayStreamNotReceivedMarker = "Transport received GOAWAY from server ErrCode:" + // goAwayNothingProcessedMarker is a GoAwayError naming stream 0 as + // the last one received, which says the same thing. The trailing + // comma keeps it from matching LastStreamID=10 and the rest. + goAwayNothingProcessedMarker = "LastStreamID=0," +) // isHTTP11Required reports whether the upstream itself asked for // HTTP/1.1, rather than merely failing at h2. func isHTTP11Required(err error) bool { - return err != nil && strings.Contains(err.Error(), http11RequiredMarker) + return err != nil && strings.Contains(transportError(err).Error(), http11RequiredMarker) } // isHTTP2ProtocolError reports whether err says the upstream cannot @@ -283,7 +398,14 @@ func isHTTP2ProtocolError(err error) bool { return false } - msg := err.Error() + msg := transportError(err).Error() + + // A graceful GOAWAY is routine connection management, not an + // upstream that cannot serve h2. + if isGracefulGoAway(msg) { + return false + } + for _, marker := range http2ErrorMarkers { if strings.Contains(msg, marker) { return true @@ -292,3 +414,24 @@ func isHTTP2ProtocolError(err error) bool { return false } + +// isGracefulGoAway reports whether msg describes a GOAWAY sent to close +// a healthy connection rather than to report an inability to serve h2. +func isGracefulGoAway(msg string) bool { + return strings.Contains(msg, goAwayNoErrorEqualsMarker) || + strings.Contains(msg, goAwayNoErrorColonMarker) || + strings.Contains(msg, gracefulGoAwayMarker) +} + +// transportError strips a *url.Error wrapper, which prefixes the request +// URL to the message. Markers are matched as substrings, so a URL left +// in place could classify an ordinary dial or TLS failure as an h2 one +// on the strength of the path alone. +func transportError(err error) error { + var urlErr *url.Error + if errors.As(err, &urlErr) && urlErr.Err != nil { + return urlErr.Err + } + + return err +} diff --git a/proxy/internal/roundtrip/upstream_test.go b/proxy/internal/roundtrip/upstream_test.go index 74acfa94b..f41c69998 100644 --- a/proxy/internal/roundtrip/upstream_test.go +++ b/proxy/internal/roundtrip/upstream_test.go @@ -10,10 +10,12 @@ import ( "crypto/x509" "crypto/x509/pkix" "errors" + "fmt" "io" "math/big" "net" "net/http" + "net/url" "strings" "testing" "time" @@ -87,6 +89,146 @@ func TestUpstreamTransport_ExplicitHTTP2NeverDowngrades(t *testing.T) { assert.False(t, mt.insecure.isDowngraded(srv.addr), "an explicit version must never pin an upstream") } +// TestUpstreamTransport_AutoDoesNotReplayUnsafeRequests covers the +// other half of the fallback: an h2 failure says nothing about whether +// the upstream already applied the request, so a state-changing one is +// not replayed. The host is still pinned, so the next request rides +// HTTP/1.1 without a second h2 attempt. +func TestUpstreamTransport_AutoDoesNotReplayUnsafeRequests(t *testing.T) { + t.Setenv(EnvUpstreamHTTPVersion, string(upstreamHTTPAuto)) + srv := startBrokenHTTP2Server(t) + // A stream error means the stream was open, so the upstream had the + // request in hand — unlike the GOAWAY at stream 0 the fake server + // sends, which states it processed nothing. + streamErr := http2.StreamError{StreamID: 1, Code: http2.ErrCodeProtocol} + + mt := NewDirectOnly(nil) + transport := mt.insecure + ctx := WithSkipTLSVerify(WithDirectUpstream(context.Background())) + + assert.False(t, safeToRetry(newTestRequest(t, ctx, http.MethodPost, srv.addr), streamErr), + "a POST must not be replayed after a failure that may have been applied") + assert.True(t, safeToRetry(newTestRequest(t, ctx, http.MethodGet, srv.addr), streamErr), + "a GET is safe to replay whatever the failure was") + + // The fake server's GOAWAY names stream 0, so even a POST is safe + // there and the request must succeed over HTTP/1.1. + resp, err := transport.RoundTrip(newTestRequest(t, ctx, http.MethodPost, srv.addr)) + require.NoError(t, err, "a GOAWAY at stream 0 means the upstream applied nothing, so the POST may be replayed") + body, err := io.ReadAll(resp.Body) + _ = resp.Body.Close() + require.NoError(t, err) + assert.Equal(t, "http/1.1", string(body), "the retry must reach the upstream over http/1.1") + + h2Attempts := srv.http2Handshakes() + resp, err = transport.RoundTrip(newTestRequest(t, ctx, http.MethodPost, srv.addr)) + require.NoError(t, err) + _ = resp.Body.Close() + assert.Equal(t, h2Attempts, srv.http2Handshakes(), + "the pin must carry later requests without another h2 attempt") +} + +func newTestRequest(t *testing.T, ctx context.Context, method, addr string) *http.Request { + t.Helper() + + req, err := http.NewRequestWithContext(ctx, method, "https://"+addr, strings.NewReader("payload")) + require.NoError(t, err) + + return req +} + +func TestUpstreamKey(t *testing.T) { + tests := []struct { + name string + url string + want string + }{ + {name: "host", url: "https://backend.invalid/path", want: "backend.invalid"}, + // DNS is case-insensitive, so these are one upstream and must + // share one pin. + {name: "mixed case", url: "https://Backend.INVALID/path", want: "backend.invalid"}, + // The default port is implied on every path that can downgrade. + {name: "explicit default port", url: "https://backend.invalid:443/", want: "backend.invalid"}, + {name: "non-default port", url: "https://backend.invalid:8443/", want: "backend.invalid:8443"}, + // An IPv6 literal needs its brackets back after Hostname strips + // them, or the key is not a dialable authority. + {name: "ipv6 default port", url: "https://[2001:db8::1]/", want: "2001:db8::1"}, + {name: "ipv6 with port", url: "https://[2001:db8::1]:8443/", want: "[2001:db8::1]:8443"}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + parsed, err := url.Parse(tc.url) + require.NoError(t, err) + + assert.Equal(t, tc.want, upstreamKey(parsed)) + }) + } +} + +func TestSafeToRetry(t *testing.T) { + streamErr := http2.StreamError{StreamID: 1, Code: http2.ErrCodeProtocol} + goAwayAtZero := errors.New(`http2: server sent GOAWAY and closed the connection; LastStreamID=0, ErrCode=HTTP_1_1_REQUIRED, debug=""`) + goAwayLater := errors.New(`http2: server sent GOAWAY and closed the connection; LastStreamID=11, ErrCode=PROTOCOL_ERROR, debug=""`) + + tests := []struct { + name string + method string + headers map[string]string + err error + want bool + }{ + {name: "get", method: http.MethodGet, err: streamErr, want: true}, + {name: "head", method: http.MethodHead, err: streamErr, want: true}, + {name: "options", method: http.MethodOptions, err: streamErr, want: true}, + {name: "trace", method: http.MethodTrace, err: streamErr, want: true}, + {name: "post", method: http.MethodPost, err: streamErr, want: false}, + {name: "put", method: http.MethodPut, err: streamErr, want: false}, + {name: "patch", method: http.MethodPatch, err: streamErr, want: false}, + {name: "delete", method: http.MethodDelete, err: streamErr, want: false}, + // The upstream reported it handled nothing, so repeating the + // request cannot duplicate anything. + {name: "post with goaway at stream 0", method: http.MethodPost, err: goAwayAtZero, want: true}, + // What the bundled transport actually raises for the first + // stream on a connection the upstream GOAWAYs with a real error + // code, which is the shape a real IIS site produces. + { + name: "post with first-stream goaway abort", + method: http.MethodPost, + err: errors.New("http2: Transport received GOAWAY from server ErrCode:HTTP_1_1_REQUIRED"), + want: true, + }, + // It handled earlier streams, so this one may have been applied. + {name: "post with goaway after other streams", method: http.MethodPost, err: goAwayLater, want: false}, + { + name: "post with idempotency key", + method: http.MethodPost, + headers: map[string]string{"Idempotency-Key": "abc"}, + err: streamErr, + want: true, + }, + { + name: "post with prefixed idempotency key", + method: http.MethodPost, + headers: map[string]string{"X-Idempotency-Key": "abc"}, + err: streamErr, + want: true, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + req, err := http.NewRequest(tc.method, "https://backend.invalid", nil) + require.NoError(t, err) + for k, v := range tc.headers { + req.Header.Set(k, v) + } + + assert.Equal(t, tc.want, safeToRetry(req, tc.err)) + }) + } +} + func TestUpstreamTransport_MayDowngrade(t *testing.T) { tests := []struct { name string @@ -227,6 +369,30 @@ func TestIsHTTP2ProtocolError(t *testing.T) { err: errors.New("Get \"https://backend.invalid\": http2: client connection lost"), want: true, }, + // A GOAWAY with NO_ERROR is a server draining a connection — + // recycling an application pool, capping requests per + // connection, shutting down gracefully. It speaks h2 fine. + { + name: "graceful goaway", + err: errors.New(`http2: server sent GOAWAY and closed the connection; LastStreamID=9, ErrCode=NO_ERROR, debug=""`), + want: false, + }, + { + name: "graceful shutdown abort", + err: errors.New("http2: Transport received Server's graceful shutdown GOAWAY"), + want: false, + }, + // Markers are substrings, so a URL carried by a *url.Error must + // not be able to classify a plain failure as an h2 one. + { + name: "url error whose path looks like a marker", + err: &url.Error{ + Op: "Get", + URL: "https://backend.invalid/http2:/connection error: x", + Err: errors.New("dial tcp 10.0.0.1:443: connect: connection refused"), + }, + want: false, + }, // Retrying these on HTTP/1.1 fixes nothing, so they must never // pin an upstream. {name: "dial failure", err: errors.New("dial tcp 10.0.0.1:443: connect: connection refused"), want: false}, @@ -332,7 +498,8 @@ func (s *brokenHTTP2Server) handle(conn net.Conn) { return } - if tlsConn.ConnectionState().NegotiatedProtocol == "h2" { + proto := tlsConn.ConnectionState().NegotiatedProtocol + if proto == "h2" { select { case s.handshakes <- struct{}{}: default: @@ -341,7 +508,7 @@ func (s *brokenHTTP2Server) handle(conn net.Conn) { return } - s.serveHTTP1(tlsConn) + s.serveHTTP1(tlsConn, proto) } // refuseHTTP2 completes just enough of the h2 handshake for the client @@ -370,16 +537,18 @@ func (s *brokenHTTP2Server) refuseHTTP2(conn net.Conn) { _, _ = io.Copy(io.Discard, conn) } -// serveHTTP1 answers a single request with the protocol the upstream -// saw, so the test can tell which transport carried it. -func (s *brokenHTTP2Server) serveHTTP1(conn net.Conn) { +// serveHTTP1 answers a single request with the ALPN protocol the +// upstream actually settled on, so a test asserting on the body is +// checking what the upstream saw rather than a constant. +func (s *brokenHTTP2Server) serveHTTP1(conn net.Conn, alpn string) { reader := bufio.NewReader(conn) if _, err := http.ReadRequest(reader); err != nil { return } - const body = "http/1.1" - _, _ = io.WriteString(conn, "HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: 8\r\nConnection: close\r\n\r\n"+body) + _, _ = fmt.Fprintf(conn, + "HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: %d\r\nConnection: close\r\n\r\n%s", + len(alpn), alpn) } func selfSignedCert(t *testing.T) tls.Certificate {