repository.go

package server

import (
	"fmt"
	"net/http"
	"strings"

	"github.com/docker/distribution"
	"github.com/docker/distribution/context"
	"github.com/docker/distribution/digest"
	registrystorage "github.com/docker/distribution/registry/storage"

	restclient "k8s.io/client-go/rest"

	imageapiv1 "github.com/openshift/origin/pkg/image/apis/image/v1"

	"github.com/openshift/image-registry/pkg/dockerregistry/server/audit"
	"github.com/openshift/image-registry/pkg/dockerregistry/server/cache"
	"github.com/openshift/image-registry/pkg/dockerregistry/server/metrics"
)

var (
	// secureTransport is the transport pool used for pullthrough to remote registries marked as
	// secure.
	secureTransport http.RoundTripper
	// insecureTransport is the transport pool that does not verify remote TLS certificates for use
	// during pullthrough against registries marked as insecure.
	insecureTransport http.RoundTripper
)

func init() {
	secureTransport = http.DefaultTransport
	var err error
	insecureTransport, err = restclient.TransportFor(&restclient.Config{TLSClientConfig: restclient.TLSClientConfig{Insecure: true}})
	if err != nil {
		panic(fmt.Sprintf("Unable to configure a default transport for importing insecure images: %v", err))
	}
}

// repository wraps a distribution.Repository and allows manifests to be served from the OpenShift image
// API.
type repository struct {
	distribution.Repository

	ctx        context.Context
	app        *App
	crossmount bool

	imageStream *imageStream

	// remoteBlobGetter is used to fetch blobs from remote registries if pullthrough is enabled.
	remoteBlobGetter BlobGetterService
}

// Repository returns a new repository middleware.
func (app *App) Repository(ctx context.Context, repo distribution.Repository, crossmount bool) (distribution.Repository, distribution.BlobDescriptorServiceFactory, error) {
	registryOSClient, err := app.registryClient.Client()
	if err != nil {
		return nil, nil, err
	}

	context.GetLogger(ctx).Infof("Using %q as Docker Registry URL", app.config.Server.Addr)

	nameParts := strings.SplitN(repo.Named().Name(), "/", 2)
	if len(nameParts) != 2 {
		return nil, nil, fmt.Errorf("invalid repository name %q: it must be of the format <project>/<name>", repo.Named().Name())
	}
	namespace, name := nameParts[0], nameParts[1]

	imageStreamGetter := &cachedImageStreamGetter{
		ctx:          ctx,
		namespace:    namespace,
		name:         name,
		isNamespacer: registryOSClient,
	}

	r := &repository{
		Repository: repo,

		ctx:        ctx,
		app:        app,
		crossmount: crossmount,

		imageStream: &imageStream{
			namespace:         nameParts[0],
			name:              nameParts[1],
			registryOSClient:  registryOSClient,
			cachedImages:      make(map[digest.Digest]*imageapiv1.Image),
			imageStreamGetter: imageStreamGetter,
			cache: &cache.RepoDigest{
				Cache: app.cache,
			},
		},
	}

	if app.config.Pullthrough.Enabled {
		r.remoteBlobGetter = NewBlobGetterService(
			r.imageStream.namespace,
			r.imageStream.name,
			imageStreamGetter.get,
			registryOSClient,
			r.imageStream.cache)
	}

	bdsf := blobDescriptorServiceFactoryFunc(r.BlobDescriptorService)

	return r, bdsf, nil
}

// Manifests returns r, which implements distribution.ManifestService.
func (r *repository) Manifests(ctx context.Context, options ...distribution.ManifestServiceOption) (distribution.ManifestService, error) {
	// we do a verification of our own
	// TODO: let upstream do the verification once they pass correct context object to their manifest handler
	opts := append(options, registrystorage.SkipLayerVerification())
	ms, err := r.Repository.Manifests(ctx, opts...)
	if err != nil {
		return nil, err
	}

	ms = &manifestService{
		manifests:     ms,
		blobStore:     r.Blobs(ctx),
		serverAddr:    r.app.config.Server.Addr,
		imageStream:   r.imageStream,
		acceptSchema2: r.app.config.Compatibility.AcceptSchema2,
	}

	if r.app.config.Pullthrough.Enabled {
		ms = &pullthroughManifestService{
			ManifestService: ms,
			imageStream:     r.imageStream,
		}
	}

	ms = newPendingErrorsManifestService(ms, r)

	if audit.LoggerExists(ctx) {
		ms = audit.NewManifestService(ctx, ms)
	}

	if r.app.config.Metrics.Enabled {
		ms = metrics.NewManifestService(ms, r.Named().Name())
	}

	return ms, nil
}

// Blobs returns a blob store which can delegate to remote repositories.
func (r *repository) Blobs(ctx context.Context) distribution.BlobStore {
	bs := r.Repository.Blobs(ctx)

	if r.app.quotaEnforcing.enforcementEnabled {
		bs = &quotaRestrictedBlobStore{
			BlobStore: bs,

			repo: r,
		}
	}

	if r.app.config.Pullthrough.Enabled {
		bs = &pullthroughBlobStore{
			BlobStore: bs,

			imageStream:      r.imageStream,
			remoteBlobGetter: r.remoteBlobGetter,
			writeLimiter:     r.app.writeLimiter,
			mirror:           r.app.config.Pullthrough.Mirror,
		}
	}

	bs = newPendingErrorsBlobStore(bs, r)

	if audit.LoggerExists(ctx) {
		bs = audit.NewBlobStore(ctx, bs)
	}

	if r.app.config.Metrics.Enabled {
		bs = metrics.NewBlobStore(bs, r.Named().Name())
	}

	return bs
}

// Tags returns a reference to this repository tag service.
func (r *repository) Tags(ctx context.Context) distribution.TagService {
	ts := r.Repository.Tags(ctx)

	ts = &tagService{
		TagService:         ts,
		imageStream:        r.imageStream,
		pullthroughEnabled: r.app.config.Pullthrough.Enabled,
	}

	ts = newPendingErrorsTagService(ts, r)

	if audit.LoggerExists(ctx) {
		ts = audit.NewTagService(ctx, ts)
	}

	if r.app.config.Metrics.Enabled {
		ts = metrics.NewTagService(ts, r.Named().Name())
	}

	return ts
}

func (r *repository) BlobDescriptorService(svc distribution.BlobDescriptorService) distribution.BlobDescriptorService {
	svc = &cache.RepositoryScopedBlobDescriptor{
		Repo:  r.Named().String(),
		Cache: r.app.cache,
		Svc:   svc,
	}
	svc = &blobDescriptorService{svc, r}
	svc = newPendingErrorsBlobDescriptorService(svc, r)
	return svc
}

func (r *repository) checkPendingErrors(ctx context.Context) error {
	return checkPendingErrors(ctx, context.GetLogger(r.ctx), r.imageStream.namespace, r.imageStream.name)
}

func checkPendingErrors(ctx context.Context, logger context.Logger, namespace, name string) error {
	if !authPerformed(ctx) {
		return fmt.Errorf("openshift.auth.completed missing from context")
	}

	deferredErrors, haveDeferredErrors := deferredErrorsFrom(ctx)
	if !haveDeferredErrors {
		return nil
	}

	repoErr, haveRepoErr := deferredErrors.Get(namespace, name)
	if !haveRepoErr {
		return nil
	}

	logger.Debugf("Origin auth: found deferred error for %s/%s: %v", namespace, name, repoErr)
	return repoErr
}