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 } //nolint:gosec // target host is selected from trusted cluster membership config (not client-controlled input) 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 } //nolint:gosec // outbound request is restricted to the trusted cluster member selected above 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 }