Merge pull request #9593 from liggitt/registry-deferred-auth-error

Merged by openshift-bot

Author: OpenShift Bot

Godeps/_workspace/src/github.com/docker/distribution/registry/storage/linkedblobstore.go

@@ -21,6 +21,19 @@ import (
 	imageapi "github.com/openshift/origin/pkg/image/api"
 )
 
+type deferredErrors map[string]error
+
+func (d deferredErrors) Add(namespace string, name string, err error) {
+	d[namespace+"/"+name] = err
+}
+func (d deferredErrors) Get(namespace string, name string) (error, bool) {
+	err, exists := d[namespace+"/"+name]
+	return err, exists
+}
+func (d deferredErrors) Empty() bool {
+	return len(d) == 0
+}
+
 // DefaultRegistryClient is exposed for testing the registry with fake client.
 var DefaultRegistryClient = NewRegistryClient(clientcmd.NewConfig().BindToFile())
 
@@ -61,6 +74,27 @@ func UserClientFrom(ctx context.Context) (client.Interface, bool) {
 	return userClient, ok
 }
 
+const authPerformedKey = "openshift.auth.performed"
+
+func WithAuthPerformed(parent context.Context) context.Context {
+	return context.WithValue(parent, authPerformedKey, true)
+}
+
+func AuthPerformed(ctx context.Context) bool {
+	authPerformed, ok := ctx.Value(authPerformedKey).(bool)
+	return ok && authPerformed
+}
+
+const deferredErrorsKey = "openshift.auth.deferredErrors"
+
+func WithDeferredErrors(parent context.Context, errs deferredErrors) context.Context {
+	return context.WithValue(parent, deferredErrorsKey, errs)
+}
+func DeferredErrorsFrom(ctx context.Context) (deferredErrors, bool) {
+	errs, ok := ctx.Value(deferredErrorsKey).(deferredErrors)
+	return errs, ok
+}
+
 type AccessController struct {
 	realm  string
 	config restclient.Config
@@ -160,6 +194,11 @@ func (ac *AccessController) Authorized(ctx context.Context, accessRecords ...reg
 		}
 	}
 
+	// pushChecks remembers which ns/name pairs had push access checks done
+	pushChecks := map[string]bool{}
+	// possibleCrossMountErrors holds errors which may be related to cross mount errors
+	possibleCrossMountErrors := deferredErrors{}
+
 	verifiedPrune := false
 
 	// Validate all requested accessRecords
@@ -178,6 +217,7 @@ func (ac *AccessController) Authorized(ctx context.Context, accessRecords ...reg
 			switch access.Action {
 			case "push":
 				verb = "update"
+				pushChecks[imageStreamNS+"/"+imageStreamName] = true
 			case "pull":
 				verb = "get"
 			case "*":
@@ -197,7 +237,11 @@ func (ac *AccessController) Authorized(ctx context.Context, accessRecords ...reg
 				verifiedPrune = true
 			default:
 				if err := verifyImageStreamAccess(ctx, imageStreamNS, imageStreamName, verb, osClient); err != nil {
-					return nil, ac.wrapErr(err)
+					if access.Action == "pull" {
+						possibleCrossMountErrors.Add(imageStreamNS, imageStreamName, ac.wrapErr(err))
+					} else {
+						return nil, ac.wrapErr(err)
+					}
 				}
 			}
 
@@ -219,6 +263,26 @@ func (ac *AccessController) Authorized(ctx context.Context, accessRecords ...reg
 		}
 	}
 
+	// deal with any possible cross-mount errors
+	for namespaceAndName, err := range possibleCrossMountErrors {
+		// If we have no push requests, this can't be a cross-mount request, so error
+		if len(pushChecks) == 0 {
+			return nil, err
+		}
+		// If we also requested a push to this ns/name, this isn't a cross-mount request, so error
+		if pushChecks[namespaceAndName] {
+			return nil, err
+		}
+	}
+
+	// Conditionally add auth errors we want to handle later to the context
+	if !possibleCrossMountErrors.Empty() {
+		context.GetLogger(ctx).Debugf("Origin auth: deferring errors: %#v", possibleCrossMountErrors)
+		ctx = WithDeferredErrors(ctx, possibleCrossMountErrors)
+	}
+	// Always add a marker to the context so we know auth was run
+	ctx = WithAuthPerformed(ctx)
+
 	return WithUserClient(ctx, osClient), nil
 }
 

pkg/dockerregistry/server/auth.go

@@ -95,6 +95,7 @@ func TestAccessController(t *testing.T) {
 		openshiftResponses []response
 		expectedError      error
 		expectedChallenge  bool
+		expectedRepoErr    string
 		expectedActions    []string
 	}{
 		"no token": {
@@ -203,7 +204,7 @@ func TestAccessController(t *testing.T) {
 				{Resource: auth.Resource{Type: "repository", Name: "foo/aaa"}, Action: "pull"},
 				{Resource: auth.Resource{Type: "repository", Name: "bar/bbb"}, Action: "push"},
 				{Resource: auth.Resource{Type: "admin"}, Action: "prune"},
-				{Resource: auth.Resource{Type: "repository", Name: "baz/ccc"}, Action: "pull"},
+				{Resource: auth.Resource{Type: "repository", Name: "baz/ccc"}, Action: "push"},
 			},
 			basicToken: "b3BlbnNoaWZ0OmF3ZXNvbWU=",
 			openshiftResponses: []response{
@@ -221,6 +222,30 @@ func TestAccessController(t *testing.T) {
 				"POST /oapi/v1/namespaces/baz/localsubjectaccessreviews",
 			},
 		},
+		"deferred cross-mount error": {
+			// cross-mount push requests check pull/push access on the target repo and pull access on the source repo.
+			// we expect the access check failure for fromrepo/bbb to be added to the context as a deferred error,
+			// which our blobstore will look for and prevent a cross mount from.
+			access: []auth.Access{
+				{Resource: auth.Resource{Type: "repository", Name: "pushrepo/aaa"}, Action: "pull"},
+				{Resource: auth.Resource{Type: "repository", Name: "pushrepo/aaa"}, Action: "push"},
+				{Resource: auth.Resource{Type: "repository", Name: "fromrepo/bbb"}, Action: "pull"},
+			},
+			basicToken: "b3BlbnNoaWZ0OmF3ZXNvbWU=",
+			openshiftResponses: []response{
+				{200, runtime.EncodeOrDie(kapi.Codecs.LegacyCodec(registered.GroupOrDie(kapi.GroupName).GroupVersions[0]), &api.SubjectAccessReviewResponse{Namespace: "pushrepo", Allowed: true, Reason: "authorized!"})},
+				{200, runtime.EncodeOrDie(kapi.Codecs.LegacyCodec(registered.GroupOrDie(kapi.GroupName).GroupVersions[0]), &api.SubjectAccessReviewResponse{Namespace: "pushrepo", Allowed: true, Reason: "authorized!"})},
+				{200, runtime.EncodeOrDie(kapi.Codecs.LegacyCodec(registered.GroupOrDie(kapi.GroupName).GroupVersions[0]), &api.SubjectAccessReviewResponse{Namespace: "fromrepo", Allowed: false, Reason: "no!"})},
+			},
+			expectedError:     nil,
+			expectedChallenge: false,
+			expectedRepoErr:   "fromrepo/bbb",
+			expectedActions: []string{
+				"POST /oapi/v1/namespaces/pushrepo/localsubjectaccessreviews",
+				"POST /oapi/v1/namespaces/pushrepo/localsubjectaccessreviews",
+				"POST /oapi/v1/namespaces/fromrepo/localsubjectaccessreviews",
+			},
+		},
 		"valid openshift token": {
 			access: []auth.Access{{
 				Resource: auth.Resource{
@@ -309,6 +334,22 @@ func TestAccessController(t *testing.T) {
 				t.Errorf("%s: expected auth context but got nil", k)
 				continue
 			}
+			if !AuthPerformed(authCtx) {
+				t.Errorf("%s: expected AuthPerformed to be true", k)
+				continue
+			}
+			deferredErrors, hasDeferred := DeferredErrorsFrom(authCtx)
+			if len(test.expectedRepoErr) > 0 {
+				if !hasDeferred || deferredErrors[test.expectedRepoErr] == nil {
+					t.Errorf("%s: expected deferred error for repo %s, got none", k, test.expectedRepoErr)
+					continue
+				}
+			} else {
+				if hasDeferred && len(deferredErrors) > 0 {
+					t.Errorf("%s: didn't expect deferred errors, got %#v", k, deferredErrors)
+					continue
+				}
+			}
 		} else {
 			_, isChallenge := err.(auth.Challenge)
 			if test.expectedChallenge != isChallenge {

pkg/dockerregistry/server/auth_test.go

@@ -0,0 +1,129 @@
+package server
+
+import (
+	"fmt"
+	"net/http"
+
+	"github.com/docker/distribution"
+	"github.com/docker/distribution/context"
+	"github.com/docker/distribution/digest"
+	"github.com/docker/distribution/registry/storage"
+)
+
+// errorBlobStore wraps a distribution.BlobStore for a particular repo.
+// before delegating, it ensures auth completed and there were no errors relevant to the repo.
+type errorBlobStore struct {
+	store distribution.BlobStore
+	repo  *repository
+}
+
+var _ distribution.BlobStore = &errorBlobStore{}
+
+func (r *errorBlobStore) Stat(ctx context.Context, dgst digest.Digest) (distribution.Descriptor, error) {
+	if err := r.repo.checkPendingErrors(ctx); err != nil {
+		return distribution.Descriptor{}, err
+	}
+	return r.store.Stat(ctx, dgst)
+}
+
+func (r *errorBlobStore) Get(ctx context.Context, dgst digest.Digest) ([]byte, error) {
+	if err := r.repo.checkPendingErrors(ctx); err != nil {
+		return nil, err
+	}
+	return r.store.Get(ctx, dgst)
+}
+
+func (r *errorBlobStore) Open(ctx context.Context, dgst digest.Digest) (distribution.ReadSeekCloser, error) {
+	if err := r.repo.checkPendingErrors(ctx); err != nil {
+		return nil, err
+	}
+	return r.store.Open(ctx, dgst)
+}
+
+func (r *errorBlobStore) Put(ctx context.Context, mediaType string, p []byte) (distribution.Descriptor, error) {
+	if err := r.repo.checkPendingErrors(ctx); err != nil {
+		return distribution.Descriptor{}, err
+	}
+	return r.store.Put(ctx, mediaType, p)
+}
+
+func (r *errorBlobStore) Create(ctx context.Context, options ...distribution.BlobCreateOption) (distribution.BlobWriter, error) {
+	if err := r.repo.checkPendingErrors(ctx); err != nil {
+		return nil, err
+	}
+
+	opts, err := effectiveCreateOptions(options)
+	if err != nil {
+		return nil, err
+	}
+	if err := checkPendingCrossMountErrors(ctx, opts); err != nil {
+		context.GetLogger(r.repo.ctx).Debugf("disabling cross-mount because of pending error: %v", err)
+		options = append(options, guardCreateOptions{DisableCrossMount: true})
+	} else {
+		options = append(options, guardCreateOptions{})
+	}
+
+	return r.store.Create(ctx, options...)
+}
+
+func (r *errorBlobStore) Resume(ctx context.Context, id string) (distribution.BlobWriter, error) {
+	if err := r.repo.checkPendingErrors(ctx); err != nil {
+		return nil, err
+	}
+	return r.store.Resume(ctx, id)
+}
+
+func (r *errorBlobStore) ServeBlob(ctx context.Context, w http.ResponseWriter, req *http.Request, dgst digest.Digest) error {
+	if err := r.repo.checkPendingErrors(ctx); err != nil {
+		return err
+	}
+	return r.store.ServeBlob(ctx, w, req, dgst)
+}
+
+func (r *errorBlobStore) Delete(ctx context.Context, dgst digest.Digest) error {
+	if err := r.repo.checkPendingErrors(ctx); err != nil {
+		return err
+	}
+	return r.store.Delete(ctx, dgst)
+}
+
+// Find out what the blob creation options are going to do by dry-running them
+func effectiveCreateOptions(options []distribution.BlobCreateOption) (*storage.CreateOptions, error) {
+	opts := &storage.CreateOptions{}
+	for _, createOptions := range options {
+		err := createOptions.Apply(opts)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return opts, nil
+}
+
+func checkPendingCrossMountErrors(ctx context.Context, opts *storage.CreateOptions) error {
+	if !opts.Mount.ShouldMount {
+		return nil
+	}
+	namespace, name, err := getNamespaceName(opts.Mount.From.Name())
+	if err != nil {
+		return err
+	}
+	return checkPendingErrors(context.GetLogger(ctx), ctx, namespace, name)
+}
+
+// guardCreateOptions ensures the expected options type is passed, and optionally disables cross mounting
+type guardCreateOptions struct {
+	DisableCrossMount bool
+}
+
+var _ distribution.BlobCreateOption = guardCreateOptions{}
+
+func (f guardCreateOptions) Apply(v interface{}) error {
+	opts, ok := v.(*storage.CreateOptions)
+	if !ok {
+		return fmt.Errorf("Unexpected create options: %#v", v)
+	}
+	if f.DisableCrossMount {
+		opts.Mount.ShouldMount = false
+	}
+	return nil
+}

pkg/dockerregistry/server/errorblobstore.go

@@ -0,0 +1,50 @@
+package server
+
+import (
+	"github.com/docker/distribution"
+	"github.com/docker/distribution/context"
+)
+
+// errorTagService wraps a distribution.TagService for a particular repo.
+// before delegating, it ensures auth completed and there were no errors relevant to the repo.
+type errorTagService struct {
+	tags distribution.TagService
+	repo *repository
+}
+
+var _ distribution.TagService = &errorTagService{}
+
+func (t *errorTagService) Get(ctx context.Context, tag string) (distribution.Descriptor, error) {
+	if err := t.repo.checkPendingErrors(ctx); err != nil {
+		return distribution.Descriptor{}, err
+	}
+	return t.tags.Get(ctx, tag)
+}
+
+func (t *errorTagService) Tag(ctx context.Context, tag string, desc distribution.Descriptor) error {
+	if err := t.repo.checkPendingErrors(ctx); err != nil {
+		return err
+	}
+	return t.tags.Tag(ctx, tag, desc)
+}
+
+func (t *errorTagService) Untag(ctx context.Context, tag string) error {
+	if err := t.repo.checkPendingErrors(ctx); err != nil {
+		return err
+	}
+	return t.tags.Untag(ctx, tag)
+}
+
+func (t *errorTagService) All(ctx context.Context) ([]string, error) {
+	if err := t.repo.checkPendingErrors(ctx); err != nil {
+		return nil, err
+	}
+	return t.tags.All(ctx)
+}
+
+func (t *errorTagService) Lookup(ctx context.Context, digest distribution.Descriptor) ([]string, error) {
+	if err := t.repo.checkPendingErrors(ctx); err != nil {
+		return nil, err
+	}
+	return t.tags.Lookup(ctx, digest)
+}