Update resolver to handle endpoint configuration

Adds support for registry mirrors
Adds support for multiple pull endpoints
Adds capabilities to limit trust in public mirrors
Fixes user agent header missing


Signed-off-by: Derek McGowan <derek@mcgstyle.net>

remotes/docker/resolver.go

@@ -18,9 +18,10 @@ package docker
 
 import (
 	"context"
+	"fmt"
 	"io"
+	"io/ioutil"
 	"net/http"
-	"net/url"
 	"path"
 	"strings"
 
@@ -46,6 +47,19 @@ var (
 	// ErrInvalidAuthorization is used when credentials are passed to a server but
 	// those credentials are rejected.
 	ErrInvalidAuthorization = errors.New("authorization failed")
+
+	// MaxManifestSize represents the largest size accepted from a registry
+	// during resolution. Larger manifests may be accepted using a
+	// resolution method other than the registry.
+	//
+	// NOTE: The max supported layers by some runtimes is 128 and individual
+	// layers will not contribute more than 256 bytes, making a
+	// reasonable limit for a large image manifests of 32K bytes.
+	// 4M bytes represents a much larger upper bound for images which may
+	// contain large annotations or be non-images. A proper manifest
+	// design puts large metadata in subobjects, as is consistent the
+	// intent of the manifest design.
+	MaxManifestSize int64 = 4 * 1048 * 1048
 )
 
 // Authorizer is used to authorize HTTP requests based on 401 HTTP responses.
@@ -72,31 +86,38 @@ type Authorizer interface {
 
 // ResolverOptions are used to configured a new Docker register resolver
 type ResolverOptions struct {
-	// Authorizer is used to authorize registry requests
-	Authorizer Authorizer
-
-	// Credentials provides username and secret given a host.
-	// If username is empty but a secret is given, that secret
-	// is interpreted as a long lived token.
-	// Deprecated: use Authorizer
-	Credentials func(string) (string, string, error)
-
-	// Host provides the hostname given a namespace.
-	Host func(string) (string, error)
+	// Hosts returns registry host configurations for a namespace.
+	Hosts RegistryHosts
 
 	// Headers are the HTTP request header fields sent by the resolver
 	Headers http.Header
 
-	// PlainHTTP specifies to use plain http and not https
-	PlainHTTP bool
-
-	// Client is the http client to used when making registry requests
-	Client *http.Client
-
 	// Tracker is used to track uploads to the registry. This is used
 	// since the registry does not have upload tracking and the existing
 	// mechanism for getting blob upload status is expensive.
 	Tracker StatusTracker
+
+	// Authorizer is used to authorize registry requests
+	// Deprecated: use Hosts
+	Authorizer Authorizer
+
+	// Credentials provides username and secret given a host.
+	// If username is empty but a secret is given, that secret
+	// is interpreted as a long lived token.
+	// Deprecated: use Hosts
+	Credentials func(string) (string, string, error)
+
+	// Host provides the hostname given a namespace.
+	// Deprecated: use Hosts
+	Host func(string) (string, error)
+
+	// PlainHTTP specifies to use plain http and not https
+	// Deprecated: use Hosts
+	PlainHTTP bool
+
+	// Client is the http client to used when making registry requests
+	// Deprecated: use Hosts
+	Client *http.Client
 }
 
 // DefaultHost is the default host function.
@@ -108,13 +129,10 @@ func DefaultHost(ns string) (string, error) {
 }
 
 type dockerResolver struct {
-	auth      Authorizer
-	host      func(string) (string, error)
-	headers   http.Header
-	uagent    string
-	plainHTTP bool
-	client    *http.Client
-	tracker   StatusTracker
+	hosts         RegistryHosts
+	header        http.Header
+	resolveHeader http.Header
+	tracker       StatusTracker
 }
 
 // NewResolver returns a new resolver to a Docker registry
@@ -122,39 +140,56 @@ func NewResolver(options ResolverOptions) remotes.Resolver {
 	if options.Tracker == nil {
 		options.Tracker = NewInMemoryTracker()
 	}
-	if options.Host == nil {
-		options.Host = DefaultHost
-	}
+
 	if options.Headers == nil {
 		options.Headers = make(http.Header)
 	}
+	if _, ok := options.Headers["User-Agent"]; !ok {
+		options.Headers.Set("User-Agent", "containerd/"+version.Version)
+	}
+
+	resolveHeader := http.Header{}
 	if _, ok := options.Headers["Accept"]; !ok {
 		// set headers for all the types we support for resolution.
-		options.Headers.Set("Accept", strings.Join([]string{
+		resolveHeader.Set("Accept", strings.Join([]string{
 			images.MediaTypeDockerSchema2Manifest,
 			images.MediaTypeDockerSchema2ManifestList,
 			ocispec.MediaTypeImageManifest,
 			ocispec.MediaTypeImageIndex, "*"}, ", "))
-	}
-	ua := options.Headers.Get("User-Agent")
-	if ua != "" {
-		options.Headers.Del("User-Agent")
 	} else {
-		ua = "containerd/" + version.Version
+		resolveHeader["Accept"] = options.Headers["Accept"]
+		delete(options.Headers, "Accept")
 	}
 
-	if options.Authorizer == nil {
-		options.Authorizer = NewAuthorizer(options.Client, options.Credentials)
-		options.Authorizer.(*dockerAuthorizer).ua = ua
+	if options.Hosts == nil {
+		opts := []RegistryOpt{}
+		if options.Host != nil {
+			opts = append(opts, WithHostTranslator(options.Host))
+		}
+
+		if options.Authorizer == nil {
+			options.Authorizer = NewDockerAuthorizer(
+				WithAuthClient(options.Client),
+				WithAuthHeader(options.Headers),
+				WithAuthCreds(options.Credentials))
+		}
+		opts = append(opts, WithAuthorizer(options.Authorizer))
+
+		if options.Client != nil {
+			opts = append(opts, WithClient(options.Client))
+		}
+		if options.PlainHTTP {
+			opts = append(opts, WithPlainHTTP(MatchAllHosts))
+		} else {
+			opts = append(opts, WithPlainHTTP(MatchLocalhost))
+		}
+		options.Hosts = ConfigureDefaultRegistries(opts...)
 	}
 	return &dockerResolver{
-		auth:      options.Authorizer,
-		host:      options.Host,
-		headers:   options.Headers,
-		uagent:    ua,
-		plainHTTP: options.PlainHTTP,
-		client:    options.Client,
-		tracker:   options.Tracker,
+		hosts:         options.Hosts,
+		header:        options.Headers,
+		resolveHeader: resolveHeader,
+		tracker:       options.Tracker,
 	}
 }
 
@@ -201,13 +236,11 @@ func (r *dockerResolver) Resolve(ctx context.Context, ref string) (string, ocisp
 		return "", ocispec.Descriptor{}, err
 	}
 
-	fetcher := dockerFetcher{
-		dockerBase: base,
-	}
-
 	var (
-		urls []string
-		dgst = refspec.Digest()
+		lastErr error
+		paths   [][]string
+		dgst    = refspec.Digest()
+		caps    = HostCapabilityPull
 	)
 
 	if dgst != "" {
@@ -218,100 +251,130 @@ func (r *dockerResolver) Resolve(ctx context.Context, ref string) (string, ocisp
 		}
 
 		// turns out, we have a valid digest, make a url.
-		urls = append(urls, fetcher.url("manifests", dgst.String()))
+		paths = append(paths, []string{"manifests", dgst.String()})
 
 		// fallback to blobs on not found.
-		urls = append(urls, fetcher.url("blobs", dgst.String()))
+		paths = append(paths, []string{"blobs", dgst.String()})
 	} else {
-		urls = append(urls, fetcher.url("manifests", refspec.Object))
+		// Add
+		paths = append(paths, []string{"manifests", refspec.Object})
+		caps |= HostCapabilityResolve
+	}
+
+	hosts := base.filterHosts(caps)
+	if len(hosts) == 0 {
+		return "", ocispec.Descriptor{}, errors.Wrap(errdefs.ErrNotFound, "no resolve hosts")
 	}
 
 	ctx, err = contextWithRepositoryScope(ctx, refspec, false)
 	if err != nil {
 		return "", ocispec.Descriptor{}, err
 	}
-	for _, u := range urls {
-		req, err := http.NewRequest(http.MethodHead, u, nil)
-		if err != nil {
-			return "", ocispec.Descriptor{}, err
-		}
 
-		req.Header = r.headers
+	for _, u := range paths {
+		for _, host := range hosts {
+			ctx := log.WithLogger(ctx, log.G(ctx).WithField("host", host.Host))
 
-		log.G(ctx).Debug("resolving")
-		resp, err := fetcher.doRequestWithRetries(ctx, req, nil)
-		if err != nil {
-			if errors.Cause(err) == ErrInvalidAuthorization {
-				err = errors.Wrapf(err, "pull access denied, repository does not exist or may require authorization")
+			req := base.request(host, http.MethodHead, u...)
+			for key, value := range r.resolveHeader {
+				req.header[key] = append(req.header[key], value...)
 			}
-			return "", ocispec.Descriptor{}, err
-		}
-		resp.Body.Close() // don't care about body contents.
 
-		if resp.StatusCode > 299 {
-			if resp.StatusCode == http.StatusNotFound {
+			log.G(ctx).Debug("resolving")
+			resp, err := req.doWithRetries(ctx, nil)
+			if err != nil {
+				if errors.Cause(err) == ErrInvalidAuthorization {
+					err = errors.Wrapf(err, "pull access denied, repository does not exist or may require authorization")
+				}
+				return "", ocispec.Descriptor{}, err
+			}
+			resp.Body.Close() // don't care about body contents.
+
+			if resp.StatusCode > 299 {
+				if resp.StatusCode == http.StatusNotFound {
+					continue
+				}
+				return "", ocispec.Descriptor{}, errors.Errorf("unexpected status code %v: %v", u, resp.Status)
+			}
+			size := resp.ContentLength
+			contentType := getManifestMediaType(resp)
+
+			// if no digest was provided, then only a resolve
+			// trusted registry was contacted, in this case use
+			// the digest header (or content from GET)
+			if dgst == "" {
+				// this is the only point at which we trust the registry. we use the
+				// content headers to assemble a descriptor for the name. when this becomes
+				// more robust, we mostly get this information from a secure trust store.
+				dgstHeader := digest.Digest(resp.Header.Get("Docker-Content-Digest"))
+
+				if dgstHeader != "" && size != -1 {
+					if err := dgstHeader.Validate(); err != nil {
+						return "", ocispec.Descriptor{}, errors.Wrapf(err, "%q in header not a valid digest", dgstHeader)
+					}
+					dgst = dgstHeader
+				}
+			}
+			if dgst == "" || size == -1 {
+				log.G(ctx).Debug("no Docker-Content-Digest header, fetching manifest instead")
+
+				req = base.request(host, http.MethodGet, u...)
+				for key, value := range r.resolveHeader {
+					req.header[key] = append(req.header[key], value...)
+				}
+
+				resp, err := req.doWithRetries(ctx, nil)
+				if err != nil {
+					return "", ocispec.Descriptor{}, err
+				}
+				defer resp.Body.Close()
+
+				bodyReader := countingReader{reader: resp.Body}
+
+				contentType = getManifestMediaType(resp)
+				if dgst == "" {
+					if contentType == images.MediaTypeDockerSchema1Manifest {
+						b, err := schema1.ReadStripSignature(&bodyReader)
+						if err != nil {
+							return "", ocispec.Descriptor{}, err
+						}
+
+						dgst = digest.FromBytes(b)
+					} else {
+						dgst, err = digest.FromReader(&bodyReader)
+						if err != nil {
+							return "", ocispec.Descriptor{}, err
+						}
+					}
+				} else if _, err := io.Copy(ioutil.Discard, &bodyReader); err != nil {
+					return "", ocispec.Descriptor{}, err
+				}
+				size = bodyReader.bytesRead
+			}
+			// Prevent resolving to excessively large manifests
+			if size > MaxManifestSize {
+				if lastErr == nil {
+					lastErr = errors.Wrapf(errdefs.ErrNotFound, "rejecting %d byte manifest for %s", size, ref)
+				}
 				continue
 			}
-			return "", ocispec.Descriptor{}, errors.Errorf("unexpected status code %v: %v", u, resp.Status)
+
+			desc := ocispec.Descriptor{
+				Digest:    dgst,
+				MediaType: contentType,
+				Size:      size,
+			}
+
+			log.G(ctx).WithField("desc.digest", desc.Digest).Debug("resolved")
+			return ref, desc, nil
 		}
-		size := resp.ContentLength
-
-		// this is the only point at which we trust the registry. we use the
-		// content headers to assemble a descriptor for the name. when this becomes
-		// more robust, we mostly get this information from a secure trust store.
-		dgstHeader := digest.Digest(resp.Header.Get("Docker-Content-Digest"))
-		contentType := getManifestMediaType(resp)
-
-		if dgstHeader != "" && size != -1 {
-			if err := dgstHeader.Validate(); err != nil {
-				return "", ocispec.Descriptor{}, errors.Wrapf(err, "%q in header not a valid digest", dgstHeader)
-			}
-			dgst = dgstHeader
-		} else {
-			log.G(ctx).Debug("no Docker-Content-Digest header, fetching manifest instead")
-
-			req, err := http.NewRequest(http.MethodGet, u, nil)
-			if err != nil {
-				return "", ocispec.Descriptor{}, err
-			}
-			req.Header = r.headers
-
-			resp, err := fetcher.doRequestWithRetries(ctx, req, nil)
-			if err != nil {
-				return "", ocispec.Descriptor{}, err
-			}
-			defer resp.Body.Close()
-
-			bodyReader := countingReader{reader: resp.Body}
-
-			contentType = getManifestMediaType(resp)
-			if contentType == images.MediaTypeDockerSchema1Manifest {
-				b, err := schema1.ReadStripSignature(&bodyReader)
-				if err != nil {
-					return "", ocispec.Descriptor{}, err
-				}
-
-				dgst = digest.FromBytes(b)
-			} else {
-				dgst, err = digest.FromReader(&bodyReader)
-				if err != nil {
-					return "", ocispec.Descriptor{}, err
-				}
-			}
-			size = bodyReader.bytesRead
-		}
-
-		desc := ocispec.Descriptor{
-			Digest:    dgst,
-			MediaType: contentType,
-			Size:      size,
-		}
-
-		log.G(ctx).WithField("desc.digest", desc.Digest).Debug("resolved")
-		return ref, desc, nil
 	}
 
-	return "", ocispec.Descriptor{}, errors.Errorf("%v not found", ref)
+	if lastErr == nil {
+		lastErr = errors.Wrap(errdefs.ErrNotFound, ref)
+	}
+
+	return "", ocispec.Descriptor{}, lastErr
 }
 
 func (r *dockerResolver) Fetcher(ctx context.Context, ref string) (remotes.Fetcher, error) {
@@ -356,56 +419,58 @@ func (r *dockerResolver) Pusher(ctx context.Context, ref string) (remotes.Pusher
 }
 
 type dockerBase struct {
-	refspec reference.Spec
-	base    url.URL
-	uagent  string
-
-	client *http.Client
-	auth   Authorizer
+	refspec   reference.Spec
+	namespace string
+	hosts     []RegistryHost
+	header    http.Header
 }
 
 func (r *dockerResolver) base(refspec reference.Spec) (*dockerBase, error) {
-	var (
-		err  error
-		base url.URL
-	)
-
 	host := refspec.Hostname()
-	base.Host = host
-	if r.host != nil {
-		base.Host, err = r.host(host)
-		if err != nil {
-			return nil, err
-		}
+	hosts, err := r.hosts(host)
+	if err != nil {
+		return nil, err
 	}
-
-	base.Scheme = "https"
-	if r.plainHTTP || strings.HasPrefix(base.Host, "localhost:") {
-		base.Scheme = "http"
-	}
-
-	prefix := strings.TrimPrefix(refspec.Locator, host+"/")
-	base.Path = path.Join("/v2", prefix)
-
 	return &dockerBase{
-		refspec: refspec,
-		base:    base,
-		uagent:  r.uagent,
-		client:  r.client,
-		auth:    r.auth,
+		refspec:   refspec,
+		namespace: strings.TrimPrefix(refspec.Locator, host+"/"),
+		hosts:     hosts,
+		header:    r.header,
 	}, nil
 }
 
-func (r *dockerBase) url(ps ...string) string {
-	url := r.base
-	url.Path = path.Join(url.Path, path.Join(ps...))
-	return url.String()
+func (r *dockerBase) filterHosts(caps HostCapabilities) (hosts []RegistryHost) {
+	for _, host := range r.hosts {
+		if host.Capabilities.Has(caps) {
+			hosts = append(hosts, host)
+		}
+	}
+	return
 }
 
-func (r *dockerBase) authorize(ctx context.Context, req *http.Request) error {
+func (r *dockerBase) request(host RegistryHost, method string, ps ...string) *request {
+	header := http.Header{}
+	for key, value := range r.header {
+		header[key] = append(header[key], value...)
+	}
+	parts := append([]string{"/", host.Path, r.namespace}, ps...)
+	p := path.Join(parts...)
+	// Join strips trailing slash, re-add ending "/" if included
+	if len(parts) > 0 && strings.HasSuffix(parts[len(parts)-1], "/") {
+		p = p + "/"
+	}
+	return &request{
+		method: method,
+		path:   p,
+		header: header,
+		host:   host,
+	}
+}
+
+func (r *request) authorize(ctx context.Context, req *http.Request) error {
 	// Check if has header for host
-	if r.auth != nil {
-		if err := r.auth.Authorize(ctx, req); err != nil {
+	if r.host.Authorizer != nil {
+		if err := r.host.Authorizer.Authorize(ctx, req); err != nil {
 			return err
 		}
 	}
@@ -413,83 +478,132 @@ func (r *dockerBase) authorize(ctx context.Context, req *http.Request) error {
 	return nil
 }
 
-func (r *dockerBase) doRequest(ctx context.Context, req *http.Request) (*http.Response, error) {
-	ctx = log.WithLogger(ctx, log.G(ctx).WithField("url", req.URL.String()))
-	log.G(ctx).WithField("request.headers", req.Header).WithField("request.method", req.Method).Debug("do request")
-	req.Header.Set("User-Agent", r.uagent)
+type request struct {
+	method string
+	path   string
+	header http.Header
+	host   RegistryHost
+	body   func() (io.ReadCloser, error)
+	size   int64
+}
+
+func (r *request) do(ctx context.Context) (*http.Response, error) {
+	u := r.host.Scheme + "://" + r.host.Host + r.path
+	req, err := http.NewRequest(r.method, u, nil)
+	if err != nil {
+		return nil, err
+	}
+	req.Header = r.header
+	if r.body != nil {
+		req.GetBody = r.body
+		if r.size > 0 {
+			req.ContentLength = r.size
+		}
+	}
+
+	ctx = log.WithLogger(ctx, log.G(ctx).WithField("url", u))
+	log.G(ctx).WithFields(requestFields(req)).Debug("do request")
 	if err := r.authorize(ctx, req); err != nil {
 		return nil, errors.Wrap(err, "failed to authorize")
 	}
-	resp, err := ctxhttp.Do(ctx, r.client, req)
+	resp, err := ctxhttp.Do(ctx, r.host.Client, req)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to do request")
 	}
-	log.G(ctx).WithFields(logrus.Fields{
-		"status":           resp.Status,
-		"response.headers": resp.Header,
-	}).Debug("fetch response received")
+	log.G(ctx).WithFields(responseFields(resp)).Debug("fetch response received")
 	return resp, nil
 }
 
-func (r *dockerBase) doRequestWithRetries(ctx context.Context, req *http.Request, responses []*http.Response) (*http.Response, error) {
-	resp, err := r.doRequest(ctx, req)
+func (r *request) doWithRetries(ctx context.Context, responses []*http.Response) (*http.Response, error) {
+	resp, err := r.do(ctx)
 	if err != nil {
 		return nil, err
 	}
 
 	responses = append(responses, resp)
-	req, err = r.retryRequest(ctx, req, responses)
+	retry, err := r.retryRequest(ctx, responses)
 	if err != nil {
 		resp.Body.Close()
 		return nil, err
 	}
-	if req != nil {
+	if retry {
 		resp.Body.Close()
-		return r.doRequestWithRetries(ctx, req, responses)
+		return r.doWithRetries(ctx, responses)
 	}
 	return resp, err
 }
 
-func (r *dockerBase) retryRequest(ctx context.Context, req *http.Request, responses []*http.Response) (*http.Request, error) {
+func (r *request) retryRequest(ctx context.Context, responses []*http.Response) (bool, error) {
 	if len(responses) > 5 {
-		return nil, nil
+		return false, nil
 	}
 	last := responses[len(responses)-1]
 	switch last.StatusCode {
 	case http.StatusUnauthorized:
 		log.G(ctx).WithField("header", last.Header.Get("WWW-Authenticate")).Debug("Unauthorized")
-		if r.auth != nil {
-			if err := r.auth.AddResponses(ctx, responses); err == nil {
-				return copyRequest(req)
+		if r.host.Authorizer != nil {
+			if err := r.host.Authorizer.AddResponses(ctx, responses); err == nil {
+				return true, nil
 			} else if !errdefs.IsNotImplemented(err) {
-				return nil, err
+				return false, err
 			}
 		}
-		return nil, nil
+
+		return false, nil
 	case http.StatusMethodNotAllowed:
 		// Support registries which have not properly implemented the HEAD method for
 		// manifests endpoint
-		if req.Method == http.MethodHead && strings.Contains(req.URL.Path, "/manifests/") {
-			// TODO: copy request?
-			req.Method = http.MethodGet
-			return copyRequest(req)
+		if r.method == http.MethodHead && strings.Contains(r.path, "/manifests/") {
+			r.method = http.MethodGet
+			return true, nil
 		}
 	case http.StatusRequestTimeout, http.StatusTooManyRequests:
-		return copyRequest(req)
+		return true, nil
 	}
 
 	// TODO: Handle 50x errors accounting for attempt history
-	return nil, nil
+	return false, nil
 }
 
-func copyRequest(req *http.Request) (*http.Request, error) {
-	ireq := *req
-	if ireq.GetBody != nil {
-		var err error
-		ireq.Body, err = ireq.GetBody()
-		if err != nil {
-			return nil, err
+func (r *request) String() string {
+	return r.host.Scheme + "://" + r.host.Host + r.path
+}
+
+func requestFields(req *http.Request) logrus.Fields {
+	fields := map[string]interface{}{
+		"request.method": req.Method,
+	}
+	for k, vals := range req.Header {
+		k = strings.ToLower(k)
+		if k == "authorization" {
+			continue
+		}
+		for i, v := range vals {
+			field := "request.header." + k
+			if i > 0 {
+				field = fmt.Sprintf("%s.%d", field, i)
+			}
+			fields[field] = v
 		}
 	}
-	return &ireq, nil
+
+	return logrus.Fields(fields)
+}
+
+func responseFields(resp *http.Response) logrus.Fields {
+	fields := map[string]interface{}{
+		"response.status": resp.Status,
+	}
+	for k, vals := range resp.Header {
+		k = strings.ToLower(k)
+		for i, v := range vals {
+			field := "response.header." + k
+			if i > 0 {
+				field = fmt.Sprintf("%s.%d", field, i)
+			}
+			fields[field] = v
+		}
+	}
+
+	return logrus.Fields(fields)
 }