c6c2c525a6
Also fix some tests that were broken by bumping golang and dependencies in the previous commits. Note that in addition to changes made to satisfy the linter which do not impact the behavior of the code, this commit also adds ReadHeaderTimeout to all usages of http.Server to satisfy the linter (and because it seemed like a good suggestion).
389 lines
12 KiB
Go
389 lines
12 KiB
Go
// Copyright 2021-2022 the Pinniped contributors. All Rights Reserved.
|
|
// SPDX-License-Identifier: Apache-2.0
|
|
|
|
package kubeclient
|
|
|
|
import (
|
|
"bytes"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
|
|
apiequality "k8s.io/apimachinery/pkg/api/equality"
|
|
"k8s.io/apimachinery/pkg/api/meta"
|
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
|
"k8s.io/apimachinery/pkg/runtime"
|
|
"k8s.io/apimachinery/pkg/runtime/schema"
|
|
"k8s.io/apimachinery/pkg/runtime/serializer"
|
|
genericapirequest "k8s.io/apiserver/pkg/endpoints/request"
|
|
"k8s.io/apiserver/pkg/server"
|
|
restclient "k8s.io/client-go/rest"
|
|
"k8s.io/client-go/transport"
|
|
|
|
"go.pinniped.dev/internal/httputil/roundtripper"
|
|
"go.pinniped.dev/internal/plog"
|
|
)
|
|
|
|
func configWithWrapper(config *restclient.Config, scheme *runtime.Scheme, negotiatedSerializer runtime.NegotiatedSerializer, middlewares []Middleware, wrapper transport.WrapperFunc) *restclient.Config {
|
|
hostURL, apiPathPrefix, err := getHostAndAPIPathPrefix(config)
|
|
if err != nil {
|
|
plog.DebugErr("invalid rest config", err)
|
|
return config // invalid input config, will fail existing client-go validation
|
|
}
|
|
|
|
// no need for any wrapping when we have no middleware to inject
|
|
if len(middlewares) == 0 {
|
|
return config
|
|
}
|
|
|
|
info, ok := runtime.SerializerInfoForMediaType(negotiatedSerializer.SupportedMediaTypes(), config.ContentType)
|
|
if !ok {
|
|
panic(fmt.Errorf("unknown content type: %s ", config.ContentType)) // static input, programmer error
|
|
}
|
|
regSerializer := info.Serializer // should perform no conversion
|
|
|
|
resolver := server.NewRequestInfoResolver(server.NewConfig(serializer.CodecFactory{}))
|
|
|
|
schemeRestMapperFunc := schemeRestMapper(scheme)
|
|
|
|
f := newWrapper(hostURL, apiPathPrefix, config, resolver, regSerializer, negotiatedSerializer, schemeRestMapperFunc, middlewares)
|
|
|
|
cc := restclient.CopyConfig(config)
|
|
cc.Wrap(f)
|
|
if wrapper != nil {
|
|
cc.Wrap(wrapper)
|
|
}
|
|
return cc
|
|
}
|
|
|
|
type roundTripperFunc func(req *http.Request) (bool, *http.Response, error)
|
|
|
|
func (f roundTripperFunc) RoundTrip(req *http.Request) (*http.Response, error) {
|
|
// always attempt to close the body, as long as we are the ones that handled the request
|
|
// see http.RoundTripper doc:
|
|
// "RoundTrip must always close the body, including on errors, ..."
|
|
handled, resp, err := f(req)
|
|
if handled && req.Body != nil {
|
|
_ = req.Body.Close()
|
|
}
|
|
return resp, err
|
|
}
|
|
|
|
func newWrapper(
|
|
hostURL, apiPathPrefix string,
|
|
config *restclient.Config,
|
|
resolver genericapirequest.RequestInfoResolver,
|
|
regSerializer runtime.Serializer,
|
|
negotiatedSerializer runtime.NegotiatedSerializer,
|
|
schemeRestMapperFunc func(schema.GroupVersionResource, Verb) (schema.GroupVersionKind, bool),
|
|
middlewares []Middleware,
|
|
) transport.WrapperFunc {
|
|
return func(rt http.RoundTripper) http.RoundTripper {
|
|
return roundtripper.WrapFunc(rt, roundTripperFunc(func(req *http.Request) (bool, *http.Response, error) {
|
|
reqInfo, err := resolver.NewRequestInfo(reqWithoutPrefix(req, hostURL, apiPathPrefix))
|
|
if err != nil || !reqInfo.IsResourceRequest {
|
|
resp, err := rt.RoundTrip(req) // we only handle kube resource requests
|
|
return false, resp, err
|
|
}
|
|
|
|
middlewareReq := &request{
|
|
verb: verb(reqInfo.Verb),
|
|
namespace: reqInfo.Namespace,
|
|
resource: schema.GroupVersionResource{
|
|
Group: reqInfo.APIGroup,
|
|
Version: reqInfo.APIVersion,
|
|
Resource: reqInfo.Resource,
|
|
},
|
|
subresource: reqInfo.Subresource,
|
|
}
|
|
|
|
for _, middleware := range middlewares {
|
|
middleware := middleware
|
|
middleware.Handle(req.Context(), middlewareReq)
|
|
}
|
|
|
|
if len(middlewareReq.reqFuncs) == 0 && len(middlewareReq.respFuncs) == 0 {
|
|
resp, err := rt.RoundTrip(req) // no middleware wanted to mutate this request
|
|
return false, resp, err
|
|
}
|
|
|
|
switch v := middlewareReq.Verb(); v {
|
|
case VerbCreate, VerbUpdate:
|
|
return handleCreateOrUpdate(req, middlewareReq, regSerializer, rt, apiPathPrefix, reqInfo, config, negotiatedSerializer)
|
|
|
|
case VerbGet, VerbList, VerbDelete, VerbDeleteCollection, VerbPatch, VerbWatch:
|
|
return handleOtherVerbs(v, req, middlewareReq, schemeRestMapperFunc, rt, apiPathPrefix, reqInfo, config, negotiatedSerializer)
|
|
|
|
case VerbProxy: // for now we do not support proxy interception
|
|
fallthrough
|
|
|
|
default:
|
|
resp, err := rt.RoundTrip(req) // we only handle certain verbs
|
|
return false, resp, err
|
|
}
|
|
}).RoundTrip)
|
|
}
|
|
}
|
|
|
|
func handleOtherVerbs(
|
|
v Verb,
|
|
req *http.Request,
|
|
middlewareReq *request,
|
|
schemeRestMapperFunc func(schema.GroupVersionResource, Verb) (schema.GroupVersionKind, bool),
|
|
rt http.RoundTripper,
|
|
apiPathPrefix string,
|
|
reqInfo *genericapirequest.RequestInfo,
|
|
config *restclient.Config,
|
|
negotiatedSerializer runtime.NegotiatedSerializer,
|
|
) (bool, *http.Response, error) {
|
|
mapperGVK, ok := schemeRestMapperFunc(middlewareReq.Resource(), v)
|
|
if !ok {
|
|
return true, nil, fmt.Errorf("unable to determine GVK for middleware request %#v", middlewareReq)
|
|
}
|
|
|
|
// no need to do anything with object meta since we only support GVK changes
|
|
obj := &metav1.PartialObjectMetadata{}
|
|
obj.APIVersion, obj.Kind = mapperGVK.ToAPIVersionAndKind()
|
|
|
|
result, err := middlewareReq.mutateRequest(obj)
|
|
if err != nil {
|
|
return true, nil, fmt.Errorf("middleware request for %#v failed to mutate: %w", middlewareReq, err)
|
|
}
|
|
|
|
if !result.mutated {
|
|
resp, err := rt.RoundTrip(req) // no middleware mutated the request
|
|
return false, resp, err
|
|
}
|
|
|
|
// sanity check to make sure mutation is to type meta and/or the response
|
|
unexpectedMutation := len(middlewareReq.respFuncs) == 0 && !result.gvkChanged
|
|
metaIsZero := apiequality.Semantic.DeepEqual(obj.ObjectMeta, metav1.ObjectMeta{})
|
|
if unexpectedMutation || !metaIsZero {
|
|
return true, nil, fmt.Errorf("invalid object meta mutation: %#v", middlewareReq)
|
|
}
|
|
|
|
reqURL, err := updatePathNewGVK(req.URL, result, apiPathPrefix, reqInfo)
|
|
if err != nil {
|
|
return true, nil, err
|
|
}
|
|
|
|
// shallow copy because we want to preserve all the headers and such but not mutate the original request
|
|
newReq := req.WithContext(req.Context())
|
|
|
|
// replace the body and path with the new data
|
|
newReq.URL = reqURL
|
|
|
|
glogBody("mutated request url", []byte(reqURL.String()))
|
|
|
|
resp, err := rt.RoundTrip(newReq)
|
|
if err != nil {
|
|
return false, nil, fmt.Errorf("middleware request for %#v failed: %w", middlewareReq, err)
|
|
}
|
|
|
|
switch v {
|
|
case VerbDelete, VerbDeleteCollection:
|
|
return false, resp, nil // we do not need to fix the response on delete
|
|
|
|
case VerbWatch:
|
|
resp, err := handleWatchResponseNewGVK(config, negotiatedSerializer, resp, middlewareReq, result)
|
|
return false, resp, err
|
|
|
|
default: // VerbGet, VerbList, VerbPatch
|
|
resp, err := handleResponseNewGVK(config, negotiatedSerializer, resp, middlewareReq, result)
|
|
return false, resp, err
|
|
}
|
|
}
|
|
|
|
func handleCreateOrUpdate(
|
|
req *http.Request,
|
|
middlewareReq *request,
|
|
regSerializer runtime.Serializer,
|
|
rt http.RoundTripper,
|
|
apiPathPrefix string,
|
|
reqInfo *genericapirequest.RequestInfo,
|
|
config *restclient.Config,
|
|
negotiatedSerializer runtime.NegotiatedSerializer,
|
|
) (bool, *http.Response, error) {
|
|
if req.GetBody == nil {
|
|
return true, nil, fmt.Errorf("unreadible body for request: %#v", middlewareReq) // this should never happen
|
|
}
|
|
|
|
body, err := req.GetBody()
|
|
if err != nil {
|
|
return true, nil, fmt.Errorf("get body failed: %w", err)
|
|
}
|
|
defer body.Close()
|
|
data, err := io.ReadAll(body)
|
|
if err != nil {
|
|
return true, nil, fmt.Errorf("read body failed: %w", err)
|
|
}
|
|
|
|
// attempt to decode with no defaults or into specified, i.e. defer to the decoder
|
|
// this should result in the a straight decode with no conversion
|
|
decodedObj, err := runtime.Decode(regSerializer, data)
|
|
if err != nil {
|
|
return true, nil, fmt.Errorf("body decode failed: %w", err)
|
|
}
|
|
|
|
obj, ok := decodedObj.(Object)
|
|
if !ok {
|
|
return true, nil, fmt.Errorf("middleware request for %#v has invalid object semantics: %T", middlewareReq, decodedObj)
|
|
}
|
|
|
|
result, err := middlewareReq.mutateRequest(obj)
|
|
if err != nil {
|
|
return true, nil, fmt.Errorf("middleware request for %#v failed to mutate: %w", middlewareReq, err)
|
|
}
|
|
|
|
if !result.mutated {
|
|
resp, err := rt.RoundTrip(req) // no middleware mutated the request
|
|
return false, resp, err
|
|
}
|
|
|
|
reqURL, err := updatePathNewGVK(req.URL, result, apiPathPrefix, reqInfo)
|
|
if err != nil {
|
|
return true, nil, err
|
|
}
|
|
|
|
newData, err := runtime.Encode(regSerializer, obj)
|
|
if err != nil {
|
|
return true, nil, fmt.Errorf("new body encode failed: %w", err)
|
|
}
|
|
|
|
// simplest way to reuse the body creation logic
|
|
newReqForBody, err := http.NewRequest(req.Method, reqURL.String(), bytes.NewReader(newData))
|
|
if err != nil {
|
|
return true, nil, fmt.Errorf("failed to create new req for body: %w", err) // this should never happen
|
|
}
|
|
|
|
// shallow copy because we want to preserve all the headers and such but not mutate the original request
|
|
newReq := req.WithContext(req.Context())
|
|
|
|
// replace the body and path with the new data
|
|
newReq.URL = reqURL
|
|
newReq.ContentLength = newReqForBody.ContentLength
|
|
newReq.Body = newReqForBody.Body
|
|
newReq.GetBody = newReqForBody.GetBody
|
|
|
|
glogBody("mutated request", newData)
|
|
|
|
resp, err := rt.RoundTrip(newReq)
|
|
if err != nil {
|
|
return true, nil, fmt.Errorf("middleware request for %#v failed: %w", middlewareReq, err)
|
|
}
|
|
|
|
if !result.gvkChanged && len(middlewareReq.respFuncs) == 0 {
|
|
return true, resp, nil // we did not change the GVK, so we do not need to mess with the incoming data
|
|
}
|
|
|
|
resp, err = handleResponseNewGVK(config, negotiatedSerializer, resp, middlewareReq, result)
|
|
return true, resp, err
|
|
}
|
|
|
|
func handleResponseNewGVK(
|
|
config *restclient.Config,
|
|
negotiatedSerializer runtime.NegotiatedSerializer,
|
|
resp *http.Response,
|
|
middlewareReq *request,
|
|
result *mutationResult,
|
|
) (*http.Response, error) {
|
|
// defer these status codes to client-go
|
|
switch {
|
|
case resp.StatusCode == http.StatusSwitchingProtocols,
|
|
resp.StatusCode < http.StatusOK || resp.StatusCode > http.StatusPartialContent:
|
|
return resp, nil
|
|
}
|
|
|
|
// always make sure we close the body, even if reading from it fails
|
|
defer resp.Body.Close()
|
|
respData, err := io.ReadAll(resp.Body)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to read response body: %w", err)
|
|
}
|
|
|
|
serializerInfo, err := getSerializerInfo(config, negotiatedSerializer, resp, middlewareReq)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
fixedRespData, err := maybeRestoreGVK(serializerInfo.Serializer, respData, result)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to restore GVK for %#v: %w", middlewareReq, err)
|
|
}
|
|
|
|
fixedRespData, err = maybeMutateResponse(serializerInfo.Serializer, fixedRespData, middlewareReq, result)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to mutate response for %#v: %w", middlewareReq, err)
|
|
}
|
|
|
|
newResp := &http.Response{}
|
|
*newResp = *resp
|
|
|
|
newResp.Body = io.NopCloser(bytes.NewBuffer(fixedRespData))
|
|
return newResp, nil
|
|
}
|
|
|
|
func maybeMutateResponse(serializer runtime.Serializer, fixedRespData []byte, middlewareReq *request, result *mutationResult) ([]byte, error) {
|
|
if len(middlewareReq.respFuncs) == 0 {
|
|
return fixedRespData, nil
|
|
}
|
|
|
|
decodedObj, err := runtime.Decode(serializer, fixedRespData)
|
|
if err != nil {
|
|
return fixedRespData, nil // if we cannot decode it, it is not for us - let client-go figure out what to do
|
|
}
|
|
|
|
if decodedObj.GetObjectKind().GroupVersionKind() != result.origGVK {
|
|
return fixedRespData, nil
|
|
}
|
|
|
|
var mutated bool
|
|
|
|
switch middlewareReq.Verb() {
|
|
case VerbList:
|
|
if err := meta.EachListItem(decodedObj, func(listObj runtime.Object) error {
|
|
obj, ok := listObj.(Object)
|
|
if !ok {
|
|
return fmt.Errorf("middleware request for %#v has invalid object semantics: %T", middlewareReq, decodedObj)
|
|
}
|
|
|
|
singleMutated, err := middlewareReq.mutateResponse(obj)
|
|
if err != nil {
|
|
return fmt.Errorf("response mutation failed for %#v: %w", middlewareReq, err)
|
|
}
|
|
|
|
mutated = mutated || singleMutated
|
|
|
|
return nil
|
|
}); err != nil {
|
|
return nil, fmt.Errorf("failed to iterate over list for %#v: %T", middlewareReq, decodedObj)
|
|
}
|
|
|
|
default:
|
|
obj, ok := decodedObj.(Object)
|
|
if !ok {
|
|
return nil, fmt.Errorf("middleware request for %#v has invalid object semantics: %T", middlewareReq, decodedObj)
|
|
}
|
|
|
|
mutated, err = middlewareReq.mutateResponse(obj)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("response mutation failed for %#v: %w", middlewareReq, err)
|
|
}
|
|
}
|
|
|
|
if !mutated {
|
|
return fixedRespData, nil
|
|
}
|
|
|
|
newData, err := runtime.Encode(serializer, decodedObj)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("new body encode failed: %w", err)
|
|
}
|
|
|
|
// only log if we mutated the response; we only need to log the unmutated response since client-go
|
|
// will log the mutated response for us
|
|
glogBody("unmutated response", fixedRespData)
|
|
|
|
return newData, nil
|
|
}
|