github/containerd
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 2 Wiki Activity

Add push object

Browse Source 
Split resolver to only return a name with separate methods
for getting a fetcher and pusher. Add implementation for
push.

Signed-off-by: Derek McGowan <derek@mcgstyle.net>
This commit is contained in:
Derek McGowan
2017-05-16 13:08:06 -07:00
parent 8f3b89c79d
commit 735b0e515e
12 changed files with 590 additions and 209 deletions
Download Patch File Download Diff File Expand all files Collapse all files

78
remotes/docker/fetcher.go Normal file
View File

@@ -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
}

144
remotes/docker/pusher.go Normal file
View File

@@ -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
}

414
remotes/docker/resolver.go
View File

@@ -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,
}
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)
}, nil
}
type dockerFetcher struct {
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 {
func (r *dockerBase) 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) {
return r.doRequestWithRetries(ctx, req, nil)
func (r *dockerBase) authorize(req *http.Request) {
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) {
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) {
func (r *dockerBase) 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 != "" {
// TODO: handle expired case
return nil, errors.Wrapf(ErrInvalidAuthorization, "server message: %s", errStr)
if err := invalidAuthorization(c, responses); err != nil {
r.token = ""
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

8
remotes/docker/resolver_test.go
View File

@@ -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)
}