Merge pull request #886 from dmcgowan/pusher

Add push object

cmd/dist/fetch.go

@@ -72,7 +72,11 @@ Most of this is experimental and there are few leaps to make this work.`,
 		resolved := make(chan struct{})
 		eg.Go(func() error {
 			ongoing.add(ref)
-			name, desc, fetcher, err := resolver.Resolve(ctx, ref)
+			name, desc, err := resolver.Resolve(ctx, ref)
+			if err != nil {
+				return err
+			}
+			fetcher, err := resolver.Fetcher(ctx, name)
 			if err != nil {
 				return err
 			}

cmd/dist/fetchobject.go

@@ -33,7 +33,11 @@ var fetchObjectCommand = cli.Command{
 		ctx = log.WithLogger(ctx, log.G(ctx).WithField("ref", ref))
 
 		log.G(ctx).Infof("resolving")
-		_, desc, fetcher, err := resolver.Resolve(ctx, ref)
+		name, desc, err := resolver.Resolve(ctx, ref)
+		if err != nil {
+			return err
+		}
+		fetcher, err := resolver.Fetcher(ctx, name)
 		if err != nil {
 			return err
 		}

cmd/dist/main.go

@@ -77,6 +77,7 @@ distribution tool
 		fetchObjectCommand,
 		applyCommand,
 		rootfsCommand,
+		pushObjectCommand,
 	}
 	app.Before = func(context *cli.Context) error {
 		timeout = context.GlobalDuration("timeout")

cmd/dist/pull.go

@@ -64,11 +64,16 @@ command. As part of this process, we do the following:
 		resolved := make(chan struct{})
 		eg.Go(func() error {
 			ongoing.add(ref)
-			name, desc, fetcher, err := resolver.Resolve(ctx, ref)
+			name, desc, err := resolver.Resolve(ctx, ref)
 			if err != nil {
 				log.G(ctx).WithError(err).Error("failed to resolve")
 				return err
 			}
+			fetcher, err := resolver.Fetcher(ctx, name)
+			if err != nil {
+				return err
+			}
+
 			log.G(ctx).WithField("image", name).Debug("fetching")
 			resolvedImageName = name
 			close(resolved)

cmd/dist/pushobject.go

@@ -0,0 +1,75 @@
+package main
+
+import (
+	"fmt"
+
+	"github.com/containerd/containerd/log"
+	digest "github.com/opencontainers/go-digest"
+	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
+	"github.com/urfave/cli"
+)
+
+var pushObjectCommand = cli.Command{
+	Name:        "push-object",
+	Usage:       "pushes an object to a remote",
+	ArgsUsage:   "[flags] <remote> <object> <type>",
+	Description: `Push objects by identifier to a remote.`,
+	Flags:       registryFlags,
+	Action: func(clicontext *cli.Context) error {
+		var (
+			ref    = clicontext.Args().Get(0)
+			object = clicontext.Args().Get(1)
+			media  = clicontext.Args().Get(2)
+		)
+		dgst, err := digest.Parse(object)
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel := appContext()
+		defer cancel()
+
+		resolver, err := getResolver(ctx, clicontext)
+		if err != nil {
+			return err
+		}
+
+		ctx = log.WithLogger(ctx, log.G(ctx).WithField("ref", ref))
+
+		log.G(ctx).Infof("resolving")
+		pusher, err := resolver.Pusher(ctx, ref)
+		if err != nil {
+			return err
+		}
+
+		cs, err := resolveContentStore(clicontext)
+		if err != nil {
+			return err
+		}
+
+		info, err := cs.Info(ctx, dgst)
+		if err != nil {
+			return err
+		}
+		desc := ocispec.Descriptor{
+			MediaType: media,
+			Digest:    dgst,
+			Size:      info.Size,
+		}
+
+		rc, err := cs.Reader(ctx, dgst)
+		if err != nil {
+			return err
+		}
+		defer rc.Close()
+
+		// TODO: Progress reader
+		if err = pusher.Push(ctx, desc, rc); err != nil {
+			return err
+		}
+
+		fmt.Printf("Pushed %s %s\n", desc.Digest, desc.MediaType)
+
+		return nil
+	},
+}

reference/reference.go

@@ -79,20 +79,20 @@ func Parse(s string) (Spec, error) {
 		return Spec{}, ErrHostnameRequired
 	}
 
-	parts := splitRe.Split(u.Path, 2)
+	var object string
-	if len(parts) < 2 {
-		return Spec{}, ErrObjectRequired
-	}
 
-	// This allows us to retain the @ to signify digests or shortend digests in
+	if idx := splitRe.FindStringIndex(u.Path); idx != nil {
-	// the object.
+		// This allows us to retain the @ to signify digests or shortend digests in
-	object := u.Path[len(parts[0]):]
+		// the object.
-	if object[:1] == ":" {
+		object = u.Path[idx[0]:]
-		object = object[1:]
+		if object[:1] == ":" {
+			object = object[1:]
+		}
+		u.Path = u.Path[:idx[0]]
 	}
 
 	return Spec{
-		Locator: path.Join(u.Host, parts[0]),
+		Locator: path.Join(u.Host, u.Path),
 		Object:  object,
 	}, nil
 }
@@ -119,6 +119,9 @@ func (r Spec) Digest() digest.Digest {
 
 // String returns the normalized string for the ref.
 func (r Spec) String() string {
+	if r.Object == "" {
+		return r.Locator
+	}
 	if r.Object[:1] == "@" {
 		return fmt.Sprintf("%v%v", r.Locator, r.Object)
 	}

reference/reference_test.go

@@ -78,9 +78,14 @@ func TestReferenceParser(t *testing.T) {
 			Err:   ErrHostnameRequired,
 		},
 		{
-			Name:  "ErrObjectRequired",
+			Name:       "ErrObjectRequired",
-			Input: "docker.io/library/redis?fooo=asdf",
+			Input:      "docker.io/library/redis?fooo=asdf",
-			Err:   ErrObjectRequired,
+			Hostname:   "docker.io",
+			Normalized: "docker.io/library/redis",
+			Expected: Spec{
+				Locator: "docker.io/library/redis",
+				Object:  "",
+			},
 		},
 		{
 			Name:     "Subdomain",

remotes/docker/fetcher.go

@@ -0,0 +1,78 @@
+package docker
+
+import (
+	"context"
+	"io"
+	"net/http"
+	"path"
+	"strings"
+
+	"github.com/Sirupsen/logrus"
+	"github.com/containerd/containerd/images"
+	"github.com/containerd/containerd/log"
+	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
+	"github.com/pkg/errors"
+)
+
+type dockerFetcher struct {
+	*dockerBase
+}
+
+func (r dockerFetcher) Fetch(ctx context.Context, desc ocispec.Descriptor) (io.ReadCloser, error) {
+	ctx = log.WithLogger(ctx, log.G(ctx).WithFields(
+		logrus.Fields{
+			"base":   r.base.String(),
+			"digest": desc.Digest,
+		},
+	))
+
+	paths, err := getV2URLPaths(desc)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, path := range paths {
+		u := r.url(path)
+
+		req, err := http.NewRequest(http.MethodGet, u, nil)
+		if err != nil {
+			return nil, err
+		}
+
+		req.Header.Set("Accept", strings.Join([]string{desc.MediaType, `*`}, ", "))
+		resp, err := r.doRequestWithRetries(ctx, req, nil)
+		if err != nil {
+			return nil, err
+		}
+
+		if resp.StatusCode > 299 {
+			if resp.StatusCode == http.StatusNotFound {
+				continue // try one of the other urls.
+			}
+			resp.Body.Close()
+			return nil, errors.Errorf("unexpected status code %v: %v", u, resp.Status)
+		}
+
+		return resp.Body, nil
+	}
+
+	return nil, errors.New("not found")
+}
+
+// getV2URLPaths generates the candidate urls paths for the object based on the
+// set of hints and the provided object id. URLs are returned in the order of
+// most to least likely succeed.
+func getV2URLPaths(desc ocispec.Descriptor) ([]string, error) {
+	var urls []string
+
+	switch desc.MediaType {
+	case images.MediaTypeDockerSchema2Manifest, images.MediaTypeDockerSchema2ManifestList,
+		ocispec.MediaTypeImageManifest, ocispec.MediaTypeImageIndex:
+		urls = append(urls, path.Join("manifests", desc.Digest.String()))
+	}
+
+	// always fallback to attempting to get the object out of the blobs store.
+	urls = append(urls, path.Join("blobs", desc.Digest.String()))
+
+	return urls, nil
+}

remotes/docker/pusher.go

@@ -0,0 +1,144 @@
+package docker
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"io/ioutil"
+	"net/http"
+	"path"
+	"strings"
+
+	"github.com/containerd/containerd/images"
+	"github.com/containerd/containerd/log"
+	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
+	"github.com/pkg/errors"
+)
+
+type dockerPusher struct {
+	*dockerBase
+	tag string
+}
+
+func (p dockerPusher) Push(ctx context.Context, desc ocispec.Descriptor, r io.Reader) error {
+	var (
+		isManifest bool
+		existCheck string
+	)
+
+	switch desc.MediaType {
+	case images.MediaTypeDockerSchema2Manifest, images.MediaTypeDockerSchema2ManifestList,
+		ocispec.MediaTypeImageManifest, ocispec.MediaTypeImageIndex:
+		isManifest = true
+		existCheck = path.Join("manifests", desc.Digest.String())
+	default:
+		existCheck = path.Join("blobs", desc.Digest.String())
+	}
+
+	req, err := http.NewRequest(http.MethodHead, p.url(existCheck), nil)
+	if err != nil {
+		return err
+	}
+
+	req.Header.Set("Accept", strings.Join([]string{desc.MediaType, `*`}, ", "))
+	resp, err := p.doRequestWithRetries(ctx, req, nil)
+	if err != nil {
+		if errors.Cause(err) != ErrInvalidAuthorization {
+			return err
+		}
+		log.G(ctx).WithError(err).Debugf("Unable to check existence, continuing with push")
+	} else {
+		if resp.StatusCode == http.StatusOK {
+			return nil
+		}
+		if resp.StatusCode != http.StatusNotFound {
+			// TODO: log error
+			return errors.Errorf("unexpected response: %s", resp.Status)
+		}
+	}
+
+	// TODO: Lookup related objects for cross repository push
+
+	if isManifest {
+		// Read all to use bytes.Reader for using GetBody
+		b, err := ioutil.ReadAll(r)
+		if err != nil {
+			return errors.Wrap(err, "failed to read manifest")
+		}
+		var putPath string
+		if p.tag != "" {
+			putPath = path.Join("manifests", p.tag)
+		} else {
+			putPath = path.Join("manifests", desc.Digest.String())
+		}
+
+		req, err := http.NewRequest(http.MethodPut, p.url(putPath), nil)
+		if err != nil {
+			return err
+		}
+		req.ContentLength = int64(len(b))
+		req.Body = ioutil.NopCloser(bytes.NewReader(b))
+		req.GetBody = func() (io.ReadCloser, error) {
+			return ioutil.NopCloser(bytes.NewReader(b)), nil
+		}
+		req.Header.Add("Content-Type", desc.MediaType)
+
+		resp, err := p.doRequestWithRetries(ctx, req, nil)
+		if err != nil {
+			return err
+		}
+		if resp.StatusCode != http.StatusCreated {
+			// TODO: log error
+			return errors.Errorf("unexpected response: %s", resp.Status)
+		}
+	} else {
+		// TODO: Do monolithic upload if size is small
+
+		// TODO: Turn multi-request blob uploader into ingester
+
+		// Start upload request
+		req, err := http.NewRequest(http.MethodPost, p.url("blobs", "uploads")+"/", nil)
+		if err != nil {
+			return err
+		}
+
+		resp, err := p.doRequestWithRetries(ctx, req, nil)
+		if err != nil {
+			return err
+		}
+		if resp.StatusCode != http.StatusAccepted {
+			// TODO: log error
+			return errors.Errorf("unexpected response: %s", resp.Status)
+		}
+
+		location := resp.Header.Get("Location")
+		// Support paths without host in location
+		if strings.HasPrefix(location, "/") {
+			u := p.base
+			u.Path = location
+			location = u.String()
+		}
+
+		// TODO: Support chunked upload
+		req, err = http.NewRequest(http.MethodPut, location, r)
+		if err != nil {
+			return err
+		}
+		q := req.URL.Query()
+		q.Add("digest", desc.Digest.String())
+		req.URL.RawQuery = q.Encode()
+		req.ContentLength = desc.Size
+
+		resp, err = p.doRequest(ctx, req)
+		if err != nil {
+			return err
+		}
+		if resp.StatusCode != http.StatusCreated {
+			// TODO: log error
+			return errors.Errorf("unexpected response: %s", resp.Status)
+		}
+
+	}
+
+	return nil
+}

remotes/docker/resolver.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"io"
 	"io/ioutil"
 	"net/http"
 	"net/textproto"
@@ -66,13 +65,169 @@ func NewResolver(options ResolverOptions) remotes.Resolver {
 
 var _ remotes.Resolver = &dockerResolver{}
 
-func (r *dockerResolver) Resolve(ctx context.Context, ref string) (string, ocispec.Descriptor, remotes.Fetcher, error) {
+func (r *dockerResolver) Resolve(ctx context.Context, ref string) (string, ocispec.Descriptor, error) {
 	refspec, err := reference.Parse(ref)
 	if err != nil {
-		return "", ocispec.Descriptor{}, nil, err
+		return "", ocispec.Descriptor{}, err
+	}
+
+	if refspec.Object == "" {
+		return "", ocispec.Descriptor{}, reference.ErrObjectRequired
+	}
+
+	base, err := r.base(refspec)
+	if err != nil {
+		return "", ocispec.Descriptor{}, err
+	}
+
+	fetcher := dockerFetcher{
+		dockerBase: base,
 	}
 
 	var (
+		urls []string
+		dgst = refspec.Digest()
+	)
+
+	if dgst != "" {
+		if err := dgst.Validate(); err != nil {
+			// need to fail here, since we can't actually resolve the invalid
+			// digest.
+			return "", ocispec.Descriptor{}, err
+		}
+
+		// turns out, we have a valid digest, make a url.
+		urls = append(urls, fetcher.url("manifests", dgst.String()))
+	} else {
+		urls = append(urls, fetcher.url("manifests", refspec.Object))
+	}
+
+	// fallback to blobs on not found.
+	urls = append(urls, fetcher.url("blobs", dgst.String()))
+
+	for _, u := range urls {
+		req, err := http.NewRequest(http.MethodHead, u, nil)
+		if err != nil {
+			return "", ocispec.Descriptor{}, err
+		}
+
+		// set headers for all the types we support for resolution.
+		req.Header.Set("Accept", strings.Join([]string{
+			images.MediaTypeDockerSchema2Manifest,
+			images.MediaTypeDockerSchema2ManifestList,
+			ocispec.MediaTypeImageManifest,
+			ocispec.MediaTypeImageIndex, "*"}, ", "))
+
+		log.G(ctx).Debug("resolving")
+		resp, err := fetcher.doRequestWithRetries(ctx, req, nil)
+		if err != nil {
+			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)
+		}
+
+		// 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 != "" {
+			if err := dgstHeader.Validate(); err != nil {
+				return "", ocispec.Descriptor{}, errors.Wrapf(err, "%q in header not a valid digest", dgstHeader)
+			}
+			dgst = dgstHeader
+		}
+
+		if dgst == "" {
+			return "", ocispec.Descriptor{}, errors.Errorf("could not resolve digest for %v", ref)
+		}
+
+		var (
+			size       int64
+			sizeHeader = resp.Header.Get("Content-Length")
+		)
+
+		size, err = strconv.ParseInt(sizeHeader, 10, 64)
+		if err != nil {
+
+			return "", ocispec.Descriptor{}, errors.Wrapf(err, "invalid size header: %q", sizeHeader)
+		}
+		if size < 0 {
+			return "", ocispec.Descriptor{}, errors.Errorf("%q in header not a valid size", sizeHeader)
+		}
+
+		desc := ocispec.Descriptor{
+			Digest:    dgst,
+			MediaType: resp.Header.Get("Content-Type"), // need to strip disposition?
+			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)
+}
+
+func (r *dockerResolver) Fetcher(ctx context.Context, ref string) (remotes.Fetcher, error) {
+	refspec, err := reference.Parse(ref)
+	if err != nil {
+		return nil, err
+	}
+
+	base, err := r.base(refspec)
+	if err != nil {
+		return nil, err
+	}
+
+	return dockerFetcher{
+		dockerBase: base,
+	}, nil
+}
+
+func (r *dockerResolver) Pusher(ctx context.Context, ref string) (remotes.Pusher, error) {
+	refspec, err := reference.Parse(ref)
+	if err != nil {
+		return nil, err
+	}
+
+	// Manifests can be pushed by digest like any other object, but the passed in
+	// reference cannot take a digest without the associated content. A tag is allowed
+	// and will be used to tag pushed manifests.
+	if refspec.Object != "" && strings.Contains(refspec.Object, "@") {
+		return nil, errors.New("cannot use digest reference for push locator")
+	}
+
+	base, err := r.base(refspec)
+	if err != nil {
+		return nil, err
+	}
+
+	return dockerPusher{
+		dockerBase: base,
+		tag:        refspec.Object,
+	}, nil
+}
+
+type dockerBase struct {
+	base  url.URL
+	token string
+
+	client   *http.Client
+	useBasic bool
+	username string
+	secret   string
+}
+
+func (r *dockerResolver) base(refspec reference.Spec) (*dockerBase, error) {
+	var (
+		err              error
 		base             url.URL
 		username, secret string
 	)
@@ -93,184 +248,55 @@ func (r *dockerResolver) Resolve(ctx context.Context, ref string) (string, ocisp
 	if r.credentials != nil {
 		username, secret, err = r.credentials(base.Host)
 		if err != nil {
-			return "", ocispec.Descriptor{}, nil, err
+			return nil, err
 		}
 	}
 
 	prefix := strings.TrimPrefix(refspec.Locator, host+"/")
 	base.Path = path.Join("/v2", prefix)
 
-	fetcher := &dockerFetcher{
+	return &dockerBase{
 		base:     base,
 		client:   r.client,
 		username: username,
 		secret:   secret,
-	}
+	}, nil
-
-	var (
-		urls []string
-		dgst = refspec.Digest()
-	)
-
-	if dgst != "" {
-		if err := dgst.Validate(); err != nil {
-			// need to fail here, since we can't actually resolve the invalid
-			// digest.
-			return "", ocispec.Descriptor{}, nil, err
-		}
-
-		// turns out, we have a valid digest, make a url.
-		urls = append(urls, fetcher.url("manifests", dgst.String()))
-	} else {
-		urls = append(urls, fetcher.url("manifests", refspec.Object))
-	}
-
-	// fallback to blobs on not found.
-	urls = append(urls, fetcher.url("blobs", dgst.String()))
-
-	for _, u := range urls {
-		req, err := http.NewRequest(http.MethodHead, u, nil)
-		if err != nil {
-			return "", ocispec.Descriptor{}, nil, err
-		}
-
-		// set headers for all the types we support for resolution.
-		req.Header.Set("Accept", strings.Join([]string{
-			images.MediaTypeDockerSchema2Manifest,
-			images.MediaTypeDockerSchema2ManifestList,
-			ocispec.MediaTypeImageManifest,
-			ocispec.MediaTypeImageIndex, "*"}, ", "))
-
-		log.G(ctx).Debug("resolving")
-		resp, err := fetcher.doRequest(ctx, req)
-		if err != nil {
-			return "", ocispec.Descriptor{}, nil, err
-		}
-		resp.Body.Close() // don't care about body contents.
-
-		if resp.StatusCode > 299 {
-			if resp.StatusCode == http.StatusNotFound {
-				continue
-			}
-			return "", ocispec.Descriptor{}, nil, errors.Errorf("unexpected status code %v: %v", u, resp.Status)
-		}
-
-		// 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 != "" {
-			if err := dgstHeader.Validate(); err != nil {
-				return "", ocispec.Descriptor{}, nil, errors.Wrapf(err, "%q in header not a valid digest", dgstHeader)
-			}
-			dgst = dgstHeader
-		}
-
-		if dgst == "" {
-			return "", ocispec.Descriptor{}, nil, errors.Errorf("could not resolve digest for %v", ref)
-		}
-
-		var (
-			size       int64
-			sizeHeader = resp.Header.Get("Content-Length")
-		)
-
-		size, err = strconv.ParseInt(sizeHeader, 10, 64)
-		if err != nil {
-
-			return "", ocispec.Descriptor{}, nil, errors.Wrapf(err, "invalid size header: %q", sizeHeader)
-		}
-		if size < 0 {
-			return "", ocispec.Descriptor{}, nil, errors.Errorf("%q in header not a valid size", sizeHeader)
-		}
-
-		desc := ocispec.Descriptor{
-			Digest:    dgst,
-			MediaType: resp.Header.Get("Content-Type"), // need to strip disposition?
-			Size:      size,
-		}
-
-		log.G(ctx).WithField("desc.digest", desc.Digest).Debug("resolved")
-		return ref, desc, fetcher, nil
-	}
-
-	return "", ocispec.Descriptor{}, nil, errors.Errorf("%v not found", ref)
 }
 
-type dockerFetcher struct {
+func (r *dockerBase) url(ps ...string) string {
-	base  url.URL
-	token string
-
-	client   *http.Client
-	useBasic bool
-	username string
-	secret   string
-}
-
-func (r *dockerFetcher) Fetch(ctx context.Context, desc ocispec.Descriptor) (io.ReadCloser, error) {
-	ctx = log.WithLogger(ctx, log.G(ctx).WithFields(
-		logrus.Fields{
-			"base":   r.base.String(),
-			"digest": desc.Digest,
-		},
-	))
-
-	paths, err := getV2URLPaths(desc)
-	if err != nil {
-		return nil, err
-	}
-
-	for _, path := range paths {
-		u := r.url(path)
-
-		req, err := http.NewRequest(http.MethodGet, u, nil)
-		if err != nil {
-			return nil, err
-		}
-
-		req.Header.Set("Accept", strings.Join([]string{desc.MediaType, `*`}, ", "))
-		resp, err := r.doRequest(ctx, req)
-		if err != nil {
-			return nil, err
-		}
-
-		if resp.StatusCode > 299 {
-			if resp.StatusCode == http.StatusNotFound {
-				continue // try one of the other urls.
-			}
-			resp.Body.Close()
-			return nil, errors.Errorf("unexpected status code %v: %v", u, resp.Status)
-		}
-
-		return resp.Body, nil
-	}
-
-	return nil, errors.New("not found")
-}
-
-func (r *dockerFetcher) url(ps ...string) string {
 	url := r.base
 	url.Path = path.Join(url.Path, path.Join(ps...))
 	return url.String()
 }
 
-func (r *dockerFetcher) doRequest(ctx context.Context, req *http.Request) (*http.Response, error) {
+func (r *dockerBase) authorize(req *http.Request) {
-	return r.doRequestWithRetries(ctx, req, nil)
+	if r.useBasic {
+		req.SetBasicAuth(r.username, r.secret)
+	} else if r.token != "" {
+		req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", r.token))
+	}
 }
 
-func (r *dockerFetcher) doRequestWithRetries(ctx context.Context, req *http.Request, responses []*http.Response) (*http.Response, error) {
+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).Debug("fetch content")
+	log.G(ctx).WithField("request.headers", req.Header).WithField("request.method", req.Method).Debug("Do request")
 	r.authorize(req)
 	resp, err := ctxhttp.Do(ctx, r.client, req)
 	if err != nil {
-		return nil, err
+		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")
+	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)
+	if err != nil {
+		return nil, err
+	}
 
 	responses = append(responses, resp)
 	req, err = r.retryRequest(ctx, req, responses)
@@ -283,15 +309,7 @@ func (r *dockerFetcher) doRequestWithRetries(ctx context.Context, req *http.Requ
 	return resp, err
 }
 
-func (r *dockerFetcher) authorize(req *http.Request) {
+func (r *dockerBase) retryRequest(ctx context.Context, req *http.Request, responses []*http.Response) (*http.Request, error) {
-	if r.useBasic {
-		req.SetBasicAuth(r.username, r.secret)
-	} else if r.token != "" {
-		req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", r.token))
-	}
-}
-
-func (r *dockerFetcher) retryRequest(ctx context.Context, req *http.Request, responses []*http.Response) (*http.Request, error) {
 	if len(responses) > 5 {
 		return nil, nil
 	}
@@ -300,19 +318,19 @@ func (r *dockerFetcher) retryRequest(ctx context.Context, req *http.Request, res
 		log.G(ctx).WithField("header", last.Header.Get("WWW-Authenticate")).Debug("Unauthorized")
 		for _, c := range parseAuthHeader(last.Header) {
 			if c.scheme == bearerAuth {
-				if errStr := c.parameters["error"]; errStr != "" {
+				if err := invalidAuthorization(c, responses); err != nil {
-					// TODO: handle expired case
+					r.token = ""
-					return nil, errors.Wrapf(ErrInvalidAuthorization, "server message: %s", errStr)
+					return nil, err
 				}
 				if err := r.setTokenAuth(ctx, c.parameters); err != nil {
 					return nil, err
 				}
-				return req, nil
+				return copyRequest(req)
 			} else if c.scheme == basicAuth {
 				if r.username != "" && r.secret != "" {
 					r.useBasic = true
 				}
-				return req, nil
+				return copyRequest(req)
 			}
 		}
 		return nil, nil
@@ -322,7 +340,7 @@ func (r *dockerFetcher) retryRequest(ctx context.Context, req *http.Request, res
 		if strings.Contains(req.URL.Path, "/manifests/") {
 			// TODO: copy request?
 			req.Method = http.MethodGet
-			return req, nil
+			return copyRequest(req)
 		}
 	}
 
@@ -330,6 +348,42 @@ func (r *dockerFetcher) retryRequest(ctx context.Context, req *http.Request, res
 	return nil, nil
 }
 
+func invalidAuthorization(c challenge, responses []*http.Response) error {
+	errStr := c.parameters["error"]
+	if errStr == "" {
+		return nil
+	}
+
+	n := len(responses)
+	if n == 1 || (n > 1 && !sameRequest(responses[n-2].Request, responses[n-1].Request)) {
+		return nil
+	}
+
+	return errors.Wrapf(ErrInvalidAuthorization, "server message: %s", errStr)
+}
+
+func sameRequest(r1, r2 *http.Request) bool {
+	if r1.Method != r2.Method {
+		return false
+	}
+	if *r1.URL != *r2.URL {
+		return false
+	}
+	return true
+}
+
+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
+		}
+	}
+	return &ireq, nil
+}
+
 func isManifestAccept(h http.Header) bool {
 	for _, ah := range h[textproto.CanonicalMIMEHeaderKey("Accept")] {
 		switch ah {
@@ -346,7 +400,7 @@ func isManifestAccept(h http.Header) bool {
 	return false
 }
 
-func (r *dockerFetcher) setTokenAuth(ctx context.Context, params map[string]string) error {
+func (r *dockerBase) setTokenAuth(ctx context.Context, params map[string]string) error {
 	realm, ok := params["realm"]
 	if !ok {
 		return errors.New("no realm specified for token auth challenge")
@@ -387,24 +441,6 @@ func (r *dockerFetcher) setTokenAuth(ctx context.Context, params map[string]stri
 	return nil
 }
 
-// getV2URLPaths generates the candidate urls paths for the object based on the
-// set of hints and the provided object id. URLs are returned in the order of
-// most to least likely succeed.
-func getV2URLPaths(desc ocispec.Descriptor) ([]string, error) {
-	var urls []string
-
-	switch desc.MediaType {
-	case images.MediaTypeDockerSchema2Manifest, images.MediaTypeDockerSchema2ManifestList,
-		ocispec.MediaTypeImageManifest, ocispec.MediaTypeImageIndex:
-		urls = append(urls, path.Join("manifests", desc.Digest.String()))
-	}
-
-	// always fallback to attempting to get the object out of the blobs store.
-	urls = append(urls, path.Join("blobs", desc.Digest.String()))
-
-	return urls, nil
-}
-
 type tokenOptions struct {
 	realm   string
 	service string
@@ -419,7 +455,7 @@ type postTokenResponse struct {
 	Scope        string    `json:"scope"`
 }
 
-func (r *dockerFetcher) fetchTokenWithOAuth(ctx context.Context, to tokenOptions) (string, error) {
+func (r *dockerBase) fetchTokenWithOAuth(ctx context.Context, to tokenOptions) (string, error) {
 	form := url.Values{}
 	form.Set("scope", strings.Join(to.scopes, " "))
 	form.Set("service", to.service)
@@ -473,7 +509,7 @@ type getTokenResponse struct {
 }
 
 // getToken fetches a token using a GET request
-func (r *dockerFetcher) getToken(ctx context.Context, to tokenOptions) (string, error) {
+func (r *dockerBase) getToken(ctx context.Context, to tokenOptions) (string, error) {
 	req, err := http.NewRequest("GET", to.realm, nil)
 	if err != nil {
 		return "", err

remotes/docker/resolver_test.go

@@ -166,7 +166,7 @@ func TestBadTokenResolver(t *testing.T) {
 	resolver := NewResolver(ro)
 	image := fmt.Sprintf("%s/doesntmatter:sometatg", base)
 
-	_, _, _, err := resolver.Resolve(ctx, image)
+	_, _, err := resolver.Resolve(ctx, image)
 	if err == nil {
 		t.Fatal("Expected error getting token with inssufficient scope")
 	}
@@ -261,7 +261,11 @@ func runBasicTest(t *testing.T, name string, sf func(h http.Handler) (string, Re
 	resolver := NewResolver(ro)
 	image := fmt.Sprintf("%s/%s:%s", base, name, tag)
 
-	_, d, f, err := resolver.Resolve(ctx, image)
+	_, d, err := resolver.Resolve(ctx, image)
+	if err != nil {
+		t.Fatal(err)
+	}
+	f, err := resolver.Fetcher(ctx, image)
 	if err != nil {
 		t.Fatal(err)
 	}

remotes/resolver.go

@@ -7,7 +7,7 @@ import (
 	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
 )
 
-// Resolver provides a remote based on a locator.
+// Resolver provides remotes based on a locator.
 type Resolver interface {
 	// Resolve attempts to resolve the reference into a name and descriptor.
 	//
@@ -20,7 +20,15 @@ type Resolver interface {
 	// While the name may differ from ref, it should itself be a valid ref.
 	//
 	// If the resolution fails, an error will be returned.
-	Resolve(ctx context.Context, ref string) (name string, desc ocispec.Descriptor, fetcher Fetcher, err error)
+	Resolve(ctx context.Context, ref string) (name string, desc ocispec.Descriptor, err error)
+
+	// Fetcher returns a new fetcher for the provided reference.
+	// All content fetched from the returned fetcher will be
+	// from the namespace referred to by ref.
+	Fetcher(ctx context.Context, ref string) (Fetcher, error)
+
+	// Pusher returns a new pusher for the provided reference
+	Pusher(ctx context.Context, ref string) (Pusher, error)
 }
 
 type Fetcher interface {
@@ -28,6 +36,12 @@ type Fetcher interface {
 	Fetch(ctx context.Context, desc ocispec.Descriptor) (io.ReadCloser, error)
 }
 
+type Pusher interface {
+	// Push pushes the resource identified by the descriptor using the
+	// passed in reader.
+	Push(ctx context.Context, d ocispec.Descriptor, r io.Reader) error
+}
+
 // FetcherFunc allows package users to implement a Fetcher with just a
 // function.
 type FetcherFunc func(ctx context.Context, desc ocispec.Descriptor) (io.ReadCloser, error)
@@ -35,3 +49,11 @@ type FetcherFunc func(ctx context.Context, desc ocispec.Descriptor) (io.ReadClos
 func (fn FetcherFunc) Fetch(ctx context.Context, desc ocispec.Descriptor) (io.ReadCloser, error) {
 	return fn(ctx, desc)
 }
+
+// PusherFunc allows package users to implement a Pusher with just a
+// function.
+type PusherFunc func(ctx context.Context, desc ocispec.Descriptor, r io.Reader) error
+
+func (fn PusherFunc) Pusher(ctx context.Context, desc ocispec.Descriptor, r io.Reader) error {
+	return fn(ctx, desc, r)
+}