POC for deferred auth errors

Author: Michal Minar

pkg/dockerregistry/server/auth.go

@@ -21,6 +21,24 @@ import (
 	imageapi "github.com/openshift/origin/pkg/image/api"
 )
 
+// crossRepoMountAuthError
+type crossRepoMountAuthError struct {
+	targetRepoName string
+	sourceRepoName string
+	err            error
+}
+
+func (e crossRepoMountAuthError) Error() string {
+	if e.targetRepoName != "" {
+		return fmt.Sprintf("unauthorized to read from source repository %q during cross-repo mount to %q: %v",
+			e.sourceRepoName, e.targetRepoName, e.err)
+	}
+	return fmt.Sprintf("unauthorized to read from source repository %q during cross-repo mount: %v",
+		e.sourceRepoName, e.targetRepoName, e.err)
+}
+
+type deferredErrors map[string]error
+
 // DefaultRegistryClient is exposed for testing the registry with fake client.
 var DefaultRegistryClient = NewRegistryClient(clientcmd.NewConfig().BindToFile())
 
@@ -61,6 +79,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 +199,11 @@ func (ac *AccessController) Authorized(ctx context.Context, accessRecords ...reg
 		}
 	}
 
+	// pushChecks remembers which ns/name pairs had push access checks done
+	pushChecks := map[string]struct{}{}
+	// possibleCrossMountErrors holds errors which may be related to cross mount errors
+	possibleCrossMountErrors := deferredErrors{}
+
 	verifiedPrune := false
 
 	// Validate all requested accessRecords
@@ -178,6 +222,7 @@ func (ac *AccessController) Authorized(ctx context.Context, accessRecords ...reg
 			switch access.Action {
 			case "push":
 				verb = "update"
+				pushChecks[access.Resource.Name] = struct{}{}
 			case "pull":
 				verb = "get"
 			case "*":
@@ -197,7 +242,14 @@ 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 verb == "get" {
+						possibleCrossMountErrors[access.Resource.Name] = &crossRepoMountAuthError{
+							sourceRepoName: access.Resource.Name,
+							err:            err,
+						}
+					} else {
+						return nil, ac.wrapErr(err)
+					}
 				}
 			}
 
@@ -219,6 +271,35 @@ 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 _, exists := pushChecks[namespaceAndName]; exists {
+			return nil, err
+		}
+		if len(pushChecks) > 1 {
+			context.GetLogger(ctx).Warn("cannot determine cross-repo mount target from multiple push checks")
+			continue
+		}
+		for target := range pushChecks {
+			crmErr := err.(*crossRepoMountAuthError)
+			crmErr.targetRepoName = target
+		}
+	}
+
+	// Conditionally add auth errors we want to handle later to the context
+	if len(possibleCrossMountErrors) != 0 {
+		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
 }
 
@@ -272,6 +353,9 @@ func verifyOpenShiftUser(ctx context.Context, client client.UsersInterface) erro
 	return nil
 }
 
+// verifyImageStreamAccess returns nil if the given client is granted access to an image stream identified by
+// <namespace>/<imageRepo>. Otherwise the access is denied. The user embedded in given client must be able to
+// <verb> (get/update) imagestreams/layers resource in the <namespace> to have the access granted.
 func verifyImageStreamAccess(ctx context.Context, namespace, imageRepo, verb string, client client.LocalSubjectAccessReviewsNamespacer) error {
 	sar := authorizationapi.LocalSubjectAccessReview{
 		Action: authorizationapi.AuthorizationAttributes{
@@ -299,6 +383,9 @@ func verifyImageStreamAccess(ctx context.Context, namespace, imageRepo, verb str
 	return nil
 }
 
+// verifyPruneAccess returns nil if the given client is granted access to images resource. Otherwise the
+// access is denied. The user embedded in given client must be able to delete images resource in a cluster to
+// have the access granted.
 func verifyPruneAccess(ctx context.Context, client client.SubjectAccessReviews) error {
 	sar := authorizationapi.SubjectAccessReview{
 		Action: authorizationapi.AuthorizationAttributes{

pkg/dockerregistry/server/auth_test.go

@@ -91,6 +91,8 @@ func TestAccessController(t *testing.T) {
 
 	tests := map[string]struct {
 		access             []auth.Access
+		uri                string
+		method             string
 		basicToken         string
 		openshiftResponses []response
 		expectedError      error
@@ -263,6 +265,92 @@ func TestAccessController(t *testing.T) {
 				"POST /oapi/v1/subjectaccessreviews",
 			},
 		},
+		"cross-repo mount": {
+			access: []auth.Access{
+				{
+					Resource: auth.Resource{
+						Type: "repository",
+						Name: "crossrepo/source",
+					},
+					Action: "pull",
+				},
+				{
+					Resource: auth.Resource{
+						Type: "repository",
+						Name: "foo/destination",
+					},
+					Action: "push",
+				},
+			},
+			uri:        "/v2/crossrepo/destination/blobs/uploads/?from=crossrepo/source&mount=sha256:da71393503ec9136cf62056c233f5d25b878e372c840170d91d65f8cdf94def2",
+			method:     "POST",
+			basicToken: "b3BlbnNoaWZ0OmF3ZXNvbWU=",
+			openshiftResponses: []response{
+				{200, runtime.EncodeOrDie(kapi.Codecs.LegacyCodec(registered.GroupOrDie(kapi.GroupName).GroupVersions[0]), &api.SubjectAccessReviewResponse{Namespace: "crossrepo", Allowed: true, Reason: "authorized!"})},
+			},
+			expectedError:     nil,
+			expectedChallenge: false,
+			expectedActions:   []string{"POST /oapi/v1/namespaces/foo/localsubjectaccessreviews"},
+		},
+		"cross-repo mount missing from attribute": {
+			access: []auth.Access{
+				{
+					Resource: auth.Resource{
+						Type: "repository",
+						Name: "crossrepo/source",
+					},
+					Action: "pull",
+				},
+				{
+					Resource: auth.Resource{
+						Type: "repository",
+						Name: "foo/destination",
+					},
+					Action: "push",
+				},
+			},
+			uri:        "/v2/foo/destination/blobs/uploads/?mount=sha256:da71393503ec9136cf62056c233f5d25b878e372c840170d91d65f8cdf94def2",
+			method:     "POST",
+			basicToken: "b3BlbnNoaWZ0OmF3ZXNvbWU=",
+			openshiftResponses: []response{
+				{200, runtime.EncodeOrDie(kapi.Codecs.LegacyCodec(registered.GroupOrDie(kapi.GroupName).GroupVersions[0]), &api.SubjectAccessReviewResponse{Namespace: "crossrepo", Allowed: false, Reason: "no!"})},
+				{200, runtime.EncodeOrDie(kapi.Codecs.LegacyCodec(registered.GroupOrDie(kapi.GroupName).GroupVersions[0]), &api.SubjectAccessReviewResponse{Namespace: "foo", Allowed: true, Reason: "authorized!"})},
+			},
+			expectedError:     ErrOpenShiftAccessDenied,
+			expectedChallenge: true,
+			expectedActions:   []string{"POST /oapi/v1/namespaces/crossrepo/localsubjectaccessreviews"},
+		},
+		"cross-repo mount with unexpected method": {
+			access: []auth.Access{
+				{
+					Resource: auth.Resource{
+						Type: "repository",
+						Name: "crossrepo/source",
+					},
+					Action: "pull",
+				},
+				{
+					Resource: auth.Resource{
+						Type: "repository",
+						Name: "foo/destination",
+					},
+					Action: "push",
+				},
+			},
+			uri:        "/v2/crossrepo/destination/blobs/uploads/?from=crossrepo/source&mount=sha256:da71393503ec9136cf62056c233f5d25b878e372c840170d91d65f8cdf94def2",
+			method:     "PUT",
+			basicToken: "b3BlbnNoaWZ0OmF3ZXNvbWU=",
+			openshiftResponses: []response{
+				{200, runtime.EncodeOrDie(kapi.Codecs.LegacyCodec(registered.GroupOrDie(kapi.GroupName).GroupVersions[0]), &api.SubjectAccessReviewResponse{Namespace: "crossrepo", Allowed: true, Reason: "authorized!"})},
+				{200, runtime.EncodeOrDie(kapi.Codecs.LegacyCodec(registered.GroupOrDie(kapi.GroupName).GroupVersions[0]), &api.SubjectAccessReviewResponse{Namespace: "foo", Allowed: false, Reason: "authorized!"})},
+			},
+			expectedError:     ErrOpenShiftAccessDenied,
+			expectedChallenge: true,
+			expectedActions: []string{
+				"POST /oapi/v1/namespaces/crossrepo/localsubjectaccessreviews",
+				"POST /oapi/v1/namespaces/foo/localsubjectaccessreviews",
+			},
+		},
 	}
 
 	for k, test := range tests {
@@ -274,7 +362,11 @@ func TestAccessController(t *testing.T) {
 		if len(test.basicToken) > 0 {
 			req.Header.Set("Authorization", fmt.Sprintf("Basic %s", test.basicToken))
 		}
-		ctx := context.WithValue(context.Background(), "http.request", req)
+		ctx := context.WithValues(context.Background(), map[string]interface{}{
+			"http.request":        req,
+			"http.request.uri":    test.uri,
+			"http.request.method": test.method,
+		})
 
 		server, actions := simulateOpenShiftMaster(test.openshiftResponses)
 		DefaultRegistryClient = NewRegistryClient(&clientcmd.Config{

pkg/dockerregistry/server/errorblobstore.go

@@ -0,0 +1,99 @@
+package server
+
+import (
+	"fmt"
+	"net/http"
+	"strings"
+
+	"github.com/docker/distribution"
+	"github.com/docker/distribution/context"
+	"github.com/docker/distribution/digest"
+	"github.com/docker/distribution/reference"
+	"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) {
+	context.GetLogger(ctx).Infof("(*errorBlobStore).Stat: starting")
+	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) {
+	context.GetLogger(ctx).Infof("(*errorBlobStore).Get: starting")
+	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) {
+	context.GetLogger(ctx).Infof("(*errorBlobStore).Open: starting")
+	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) {
+	context.GetLogger(ctx).Infof("(*errorBlobStore).Create: starting with %d options", len(options))
+	if err := r.repo.checkPendingErrors(ctx); err != nil {
+		return nil, err
+	}
+
+	options = append(options, storage.WithRepositoryMiddlewareWrapper(
+		func(ctx context.Context, repo distribution.Repository, name reference.Named) (distribution.Repository, error) {
+			context.GetLogger(r.repo.ctx).Infof("(*errorBlobStore).Create: called middleware wrapper function")
+			nameParts := strings.SplitN(name.Name(), "/", 2)
+			if len(nameParts) != 2 {
+				return nil, fmt.Errorf("invalid repository name %q: it must be of the format <project>/<name>", repo.Named().Name())
+			}
+			middleware := *r.repo
+			middleware.Repository = repo
+			middleware.namespace = nameParts[0]
+			middleware.name = nameParts[1]
+			context.GetLogger(r.repo.ctx).Infof("(*errorBlobStore).Create: returning new middleware for repository=%s", middleware.Name())
+			return &middleware, nil
+		}))
+
+	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 {
+	context.GetLogger(ctx).Infof("(*errorBlobStore).ServeBlob: starting")
+	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)
+}