zot/pkg/api/proxy.go

package api

import (
	"context"
	"fmt"
	"io"
	"net"
	"net/http"

	"github.com/gorilla/mux"

	"zotregistry.dev/zot/v2/pkg/api/constants"
	"zotregistry.dev/zot/v2/pkg/cluster"
	"zotregistry.dev/zot/v2/pkg/common"
)

// ClusterProxy wraps an http.HandlerFunc which requires proxying between zot instances to ensure
// that a given repository only has a single writer and reader for dist-spec operations in a scale-out cluster.
// based on the hash value of the repository name, the request will either be handled locally
// or proxied to another zot member in the cluster to get the data before sending a response to the client.
func ClusterProxy(ctrlr *Controller) func(http.HandlerFunc) http.HandlerFunc {
	return func(next http.HandlerFunc) http.HandlerFunc {
		return http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
			// Get cluster config safely
			clusterConfig := ctrlr.Config.CopyClusterConfig()
			logger := ctrlr.Log

			// if no cluster or single-node cluster, handle locally.
			if !clusterConfig.IsClustered() {
				next.ServeHTTP(response, request)

				return
			}

			// since the handler has been wrapped, it should be possible to get the name
			// of the repository from the mux.
			vars := mux.Vars(request)
			name, ok := vars["name"]

			if !ok || name == "" {
				response.WriteHeader(http.StatusNotFound)

				return
			}

			// the target member is the only one which should do read/write for the dist-spec APIs
			// for the given repository.
			targetMemberIndex, targetMember := cluster.ComputeTargetMember(clusterConfig.HashKey, clusterConfig.Members, name)
			logger.Debug().Str(constants.RepositoryLogKey, name).
				Msg(fmt.Sprintf("target member socket: %s index: %d", targetMember, targetMemberIndex))

			// if the target member is the same as the local member, the current member should handle the request.
			// since the instances have the same config, a quick index lookup is sufficient
			if targetMemberIndex == clusterConfig.Proxy.LocalMemberClusterSocketIndex {
				logger.Debug().Str(constants.RepositoryLogKey, name).Msg("handling the request locally")
				next.ServeHTTP(response, request)

				return
			}

			// if the header contains a hop-count, return an error response as there should be no multi-hop
			if request.Header.Get(constants.ScaleOutHopCountHeader) != "" {
				logger.Fatal().Str("url", request.URL.String()).
					Msg("failed to process request - cannot proxy an already proxied request")

				return
			}

			logger.Debug().Str(constants.RepositoryLogKey, name).Msg("proxying the request")

			proxyResponse, err := proxyHTTPRequest(request.Context(), request, targetMember, ctrlr)
			if err != nil {
				logger.Error().Err(err).Str(constants.RepositoryLogKey, name).Msg("failed to proxy the request")
				http.Error(response, err.Error(), http.StatusInternalServerError)

				return
			}
			defer proxyResponse.Body.Close()

			copyHeader(response.Header(), proxyResponse.Header)
			response.WriteHeader(proxyResponse.StatusCode)
			_, _ = io.Copy(response, proxyResponse.Body)
		})
	}
}

// gets all the server sockets of a target member - IP:Port.
// for IPv6, the socket is [IPv6]:Port.
// if the input is an IP address, returns the same targetMember in an array.
// if the input is a host name, performs a lookup and returns the server sockets.
func getTargetMemberServerSockets(targetMemberSocket string) ([]string, error) {
	targetHost, targetPort, err := net.SplitHostPort(targetMemberSocket)
	if err != nil {
		return []string{}, err
	}

	addr := net.ParseIP(targetHost)
	if addr != nil {
		// this is an IP address, return as is
		return []string{targetMemberSocket}, nil
	}
	// this is a hostname - try to resolve to an IP
	resolvedAddrs, err := common.GetIPFromHostName(targetHost)
	if err != nil {
		return []string{}, err
	}

	targetSockets := make([]string, len(resolvedAddrs))
	for idx, resolvedAddr := range resolvedAddrs {
		targetSockets[idx] = net.JoinHostPort(resolvedAddr, targetPort)
	}

	return targetSockets, nil
}

// proxy the request to the target member and return a pointer to the response or an error.
func proxyHTTPRequest(ctx context.Context, req *http.Request,
	targetMember string, ctrlr *Controller,
) (*http.Response, error) {
	cloneURL := *req.URL

	// Get HTTP TLS config safely
	httpTLSConfig := ctrlr.Config.CopyTLSConfig()

	proxyQueryScheme := constants.SchemeHTTP
	if httpTLSConfig != nil {
		proxyQueryScheme = constants.SchemeHTTPS
	}

	cloneURL.Scheme = proxyQueryScheme
	cloneURL.Host = targetMember

	requestBody := io.Reader(http.NoBody)
	if req.Body != nil {
		requestBody = req.Body
	}

	fwdRequest, err := http.NewRequestWithContext(ctx, req.Method, cloneURL.String(), requestBody)
	if err != nil {
		return nil, err
	}

	copyHeader(fwdRequest.Header, req.Header)

	// Preserve ContentLength from original request, including explicit zero-length
	// bodies, so empty requests are not forwarded as unknown-length chunked bodies.
	if req.ContentLength >= 0 {
		fwdRequest.ContentLength = req.ContentLength

		if req.ContentLength == 0 {
			fwdRequest.Body = http.NoBody
		}
	}

	// always set hop count to 1 for now.
	// the handler wrapper above will terminate the process if it sees a request that
	// already has a hop count but is due for proxying.
	fwdRequest.Header.Set(constants.ScaleOutHopCountHeader, "1")

	clientOpts := common.HTTPClientOptions{
		TLSEnabled: httpTLSConfig != nil,
		VerifyTLS:  httpTLSConfig != nil, // for now, always verify TLS when TLS mode is enabled
		Host:       targetMember,
	}

	// Get cluster config safely
	clusterConfig := ctrlr.Config.CopyClusterConfig()
	tlsConfig := clusterConfig.TLS

	if tlsConfig != nil {
		clientOpts.CertOptions.ClientCertFile = tlsConfig.Cert
		clientOpts.CertOptions.ClientKeyFile = tlsConfig.Key
		clientOpts.CertOptions.RootCaCertFile = tlsConfig.CACert
	}

	httpClient, err := common.CreateHTTPClient(&clientOpts)
	if err != nil {
		return nil, err
	}

	resp, err := httpClient.Do(fwdRequest)
	if err != nil {
		return nil, err
	}

	return resp, nil
}

func copyHeader(dst, src http.Header) {
	for k, vv := range src {
		for _, v := range vv {
			dst.Add(k, v)
		}
	}
}

// GetLocalMemberClusterSocket identifies and returns the cluster socket and index.
// This is the socket which the scale out cluster members will use for
// proxying and communication among each other.
// Returns index, socket, error.
// Returns an empty string and index value -1 if the cluster socket is not found.
func GetLocalMemberClusterSocket(members []string, localSockets []string) (int, string, error) {
	for memberIdx, member := range members {
		// for each member, get the full list of sockets, including DNS resolution
		memberSockets, err := getTargetMemberServerSockets(member)
		if err != nil {
			return -1, "", err
		}

		// for each member socket that we have, compare all the local sockets with
		// it to see if there is any match.
		for _, memberSocket := range memberSockets {
			for _, localSocket := range localSockets {
				// this checks if the sockets are equal at a host port level
				areSocketsEqual, err := common.AreSocketsEqual(memberSocket, localSocket)
				if err != nil {
					return -1, "", err
				}

				if areSocketsEqual {
					return memberIdx, member, nil
				}
			}
		}
	}

	return -1, "", nil
}