ContainerImage.Pinniped/internal/kubeclient/roundtrip.go

388 lines
12 KiB
Go
Raw Normal View History

// Copyright 2021 the Pinniped contributors. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
package kubeclient
import (
"bytes"
"fmt"
"io/ioutil"
"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/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 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
}
})
}
}
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 := ioutil.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 := ioutil.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 = ioutil.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
}