mirror of
https://github.com/k3s-io/kubernetes.git
synced 2025-08-05 02:09:56 +00:00
Merge pull request #127485 from seans3/websocket-upgrade-error-info
Upgrade websocket failure add extra error info
This commit is contained in:
commit
5c96d8d34d
@ -20,11 +20,17 @@ import (
|
|||||||
"crypto/tls"
|
"crypto/tls"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"io"
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/url"
|
"net/url"
|
||||||
|
"strings"
|
||||||
|
|
||||||
gwebsocket "github.com/gorilla/websocket"
|
gwebsocket "github.com/gorilla/websocket"
|
||||||
|
|
||||||
|
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
||||||
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||||
|
"k8s.io/apimachinery/pkg/runtime"
|
||||||
|
"k8s.io/apimachinery/pkg/runtime/serializer"
|
||||||
"k8s.io/apimachinery/pkg/util/httpstream"
|
"k8s.io/apimachinery/pkg/util/httpstream"
|
||||||
"k8s.io/apimachinery/pkg/util/httpstream/wsstream"
|
"k8s.io/apimachinery/pkg/util/httpstream/wsstream"
|
||||||
utilnet "k8s.io/apimachinery/pkg/util/net"
|
utilnet "k8s.io/apimachinery/pkg/util/net"
|
||||||
@ -37,6 +43,17 @@ var (
|
|||||||
_ http.RoundTripper = &RoundTripper{}
|
_ http.RoundTripper = &RoundTripper{}
|
||||||
)
|
)
|
||||||
|
|
||||||
|
var (
|
||||||
|
statusScheme = runtime.NewScheme()
|
||||||
|
statusCodecs = serializer.NewCodecFactory(statusScheme)
|
||||||
|
)
|
||||||
|
|
||||||
|
func init() {
|
||||||
|
statusScheme.AddUnversionedTypes(metav1.SchemeGroupVersion,
|
||||||
|
&metav1.Status{},
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
// ConnectionHolder defines functions for structure providing
|
// ConnectionHolder defines functions for structure providing
|
||||||
// access to the websocket connection.
|
// access to the websocket connection.
|
||||||
type ConnectionHolder interface {
|
type ConnectionHolder interface {
|
||||||
@ -110,12 +127,33 @@ func (rt *RoundTripper) RoundTrip(request *http.Request) (retResp *http.Response
|
|||||||
}
|
}
|
||||||
wsConn, resp, err := dialer.DialContext(request.Context(), request.URL.String(), request.Header)
|
wsConn, resp, err := dialer.DialContext(request.Context(), request.URL.String(), request.Header)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
// BadHandshake error becomes an "UpgradeFailureError" (used for streaming fallback).
|
||||||
if errors.Is(err, gwebsocket.ErrBadHandshake) {
|
if errors.Is(err, gwebsocket.ErrBadHandshake) {
|
||||||
// Enhance the error message with the response status if possible.
|
cause := err
|
||||||
|
// Enhance the error message with the error response if possible.
|
||||||
if resp != nil && len(resp.Status) > 0 {
|
if resp != nil && len(resp.Status) > 0 {
|
||||||
err = fmt.Errorf("%w (%s)", err, resp.Status)
|
defer resp.Body.Close() //nolint:errcheck
|
||||||
|
cause = fmt.Errorf("%w (%s)", err, resp.Status) // Always add the response status
|
||||||
|
responseError := ""
|
||||||
|
responseErrorBytes, readErr := io.ReadAll(io.LimitReader(resp.Body, 64*1024))
|
||||||
|
if readErr != nil {
|
||||||
|
cause = fmt.Errorf("%w: unable to read error from server response", cause)
|
||||||
|
} else {
|
||||||
|
// If returned error can be decoded as "metav1.Status", return a "StatusError".
|
||||||
|
responseError = strings.TrimSpace(string(responseErrorBytes))
|
||||||
|
if len(responseError) > 0 {
|
||||||
|
if obj, _, decodeErr := statusCodecs.UniversalDecoder().Decode(responseErrorBytes, nil, &metav1.Status{}); decodeErr == nil {
|
||||||
|
if status, ok := obj.(*metav1.Status); ok {
|
||||||
|
cause = &apierrors.StatusError{ErrStatus: *status}
|
||||||
}
|
}
|
||||||
return nil, &httpstream.UpgradeFailureError{Cause: err}
|
} else {
|
||||||
|
// Otherwise, append the responseError string.
|
||||||
|
cause = fmt.Errorf("%w: %s", cause, responseError)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nil, &httpstream.UpgradeFailureError{Cause: cause}
|
||||||
}
|
}
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
@ -18,6 +18,7 @@ package websocket
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"errors"
|
||||||
"io"
|
"io"
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/http/httptest"
|
"net/http/httptest"
|
||||||
@ -28,6 +29,9 @@ import (
|
|||||||
"github.com/stretchr/testify/assert"
|
"github.com/stretchr/testify/assert"
|
||||||
"github.com/stretchr/testify/require"
|
"github.com/stretchr/testify/require"
|
||||||
|
|
||||||
|
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
||||||
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||||
|
"k8s.io/apimachinery/pkg/runtime"
|
||||||
"k8s.io/apimachinery/pkg/util/httpstream"
|
"k8s.io/apimachinery/pkg/util/httpstream"
|
||||||
"k8s.io/apimachinery/pkg/util/httpstream/wsstream"
|
"k8s.io/apimachinery/pkg/util/httpstream/wsstream"
|
||||||
"k8s.io/apimachinery/pkg/util/remotecommand"
|
"k8s.io/apimachinery/pkg/util/remotecommand"
|
||||||
@ -64,12 +68,58 @@ func TestWebSocketRoundTripper_RoundTripperSucceeds(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func TestWebSocketRoundTripper_RoundTripperFails(t *testing.T) {
|
func TestWebSocketRoundTripper_RoundTripperFails(t *testing.T) {
|
||||||
|
testCases := map[string]struct {
|
||||||
|
statusCode int
|
||||||
|
body string
|
||||||
|
status *metav1.Status
|
||||||
|
expectedError string
|
||||||
|
}{
|
||||||
|
"Empty response status still returns basic websocket error": {
|
||||||
|
statusCode: -1,
|
||||||
|
body: "",
|
||||||
|
expectedError: "websocket: bad handshake",
|
||||||
|
},
|
||||||
|
"Empty response body still returns status": {
|
||||||
|
statusCode: http.StatusForbidden,
|
||||||
|
body: "",
|
||||||
|
expectedError: "(403 Forbidden)",
|
||||||
|
},
|
||||||
|
"Error response body returned as string when can not be cast as metav1.Status": {
|
||||||
|
statusCode: http.StatusForbidden,
|
||||||
|
body: "RBAC violated",
|
||||||
|
expectedError: "(403 Forbidden): RBAC violated",
|
||||||
|
},
|
||||||
|
"Error returned as metav1.Status within response body": {
|
||||||
|
statusCode: http.StatusBadRequest,
|
||||||
|
body: "",
|
||||||
|
status: &metav1.Status{
|
||||||
|
TypeMeta: metav1.TypeMeta{
|
||||||
|
APIVersion: "meta.k8s.io/v1",
|
||||||
|
Kind: "Status",
|
||||||
|
},
|
||||||
|
Status: "Failure",
|
||||||
|
Reason: "Unable to negotiate sub-protocol",
|
||||||
|
Code: http.StatusBadRequest,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
encoder := statusCodecs.LegacyCodec(metav1.SchemeGroupVersion)
|
||||||
|
for testName, testCase := range testCases {
|
||||||
|
t.Run(testName, func(t *testing.T) {
|
||||||
// Create fake WebSocket server.
|
// Create fake WebSocket server.
|
||||||
websocketServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
|
websocketServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
|
||||||
// Bad handshake means websocket server will not completely initialize.
|
if testCase.statusCode > 0 {
|
||||||
_, err := webSocketServerStreams(req, w)
|
w.WriteHeader(testCase.statusCode)
|
||||||
require.Error(t, err)
|
}
|
||||||
assert.ErrorContains(t, err, "websocket server finished before becoming ready")
|
if testCase.status != nil {
|
||||||
|
statusBytes, err := runtime.Encode(encoder, testCase.status)
|
||||||
|
require.NoError(t, err)
|
||||||
|
_, err = w.Write(statusBytes)
|
||||||
|
require.NoError(t, err)
|
||||||
|
} else if len(testCase.body) > 0 {
|
||||||
|
_, err := w.Write([]byte(testCase.body))
|
||||||
|
require.NoError(t, err)
|
||||||
|
}
|
||||||
}))
|
}))
|
||||||
defer websocketServer.Close()
|
defer websocketServer.Close()
|
||||||
|
|
||||||
@ -80,15 +130,24 @@ func TestWebSocketRoundTripper_RoundTripperFails(t *testing.T) {
|
|||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
rt, _, err := RoundTripperFor(&restclient.Config{Host: websocketLocation.Host})
|
rt, _, err := RoundTripperFor(&restclient.Config{Host: websocketLocation.Host})
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
// Requested subprotocol version 1 is not supported by test websocket server.
|
|
||||||
requestedProtocol := remotecommand.StreamProtocolV1Name
|
|
||||||
req.Header[wsstream.WebSocketProtocolHeader] = []string{requestedProtocol}
|
|
||||||
_, err = rt.RoundTrip(req)
|
_, err = rt.RoundTrip(req)
|
||||||
// Ensure a "bad handshake" error is returned, since requested protocol is not supported.
|
|
||||||
require.Error(t, err)
|
require.Error(t, err)
|
||||||
assert.ErrorContains(t, err, "websocket: bad handshake")
|
|
||||||
assert.ErrorContains(t, err, "403 Forbidden")
|
|
||||||
assert.True(t, httpstream.IsUpgradeFailure(err))
|
assert.True(t, httpstream.IsUpgradeFailure(err))
|
||||||
|
if testCase.status != nil {
|
||||||
|
upgradeErr := &httpstream.UpgradeFailureError{}
|
||||||
|
validErr := errors.As(err, &upgradeErr)
|
||||||
|
assert.True(t, validErr, "could not cast error as httpstream.UpgradeFailureError")
|
||||||
|
statusErr := upgradeErr.Cause
|
||||||
|
apiErr := &apierrors.StatusError{}
|
||||||
|
validErr = errors.As(statusErr, &apiErr)
|
||||||
|
assert.True(t, validErr, "could not cast error as apierrors.StatusError")
|
||||||
|
assert.Equal(t, *testCase.status, apiErr.ErrStatus)
|
||||||
|
} else {
|
||||||
|
assert.Contains(t, err.Error(), testCase.expectedError,
|
||||||
|
"expected (%s), got (%s)", testCase.expectedError, err.Error())
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestWebSocketRoundTripper_NegotiateCreatesConnection(t *testing.T) {
|
func TestWebSocketRoundTripper_NegotiateCreatesConnection(t *testing.T) {
|
||||||
|
Loading…
Reference in New Issue
Block a user