To avoid any network use unless necessary, delay establishing authorization
challenges with the upstream until any proxied data is found not to be local. Implement auth challenges behind an interface and add to unit tests. Also, remove a non-sensical unit test. Signed-off-by: Richard Scothern <richard.scothern@docker.com>
This commit is contained in:
parent
881ef1096f
commit
7d16fee7a4
8 changed files with 156 additions and 44 deletions
|
@ -8,6 +8,7 @@ import (
|
|||
)
|
||||
|
||||
const tokenURL = "https://auth.docker.io/token"
|
||||
const challengeHeader = "Docker-Distribution-Api-Version"
|
||||
|
||||
type userpass struct {
|
||||
username string
|
||||
|
@ -24,12 +25,8 @@ func (c credentials) Basic(u *url.URL) (string, string) {
|
|||
return up.username, up.password
|
||||
}
|
||||
|
||||
// ConfigureAuth authorizes with the upstream registry
|
||||
func ConfigureAuth(remoteURL, username, password string, cm auth.ChallengeManager) (auth.CredentialStore, error) {
|
||||
if err := ping(cm, remoteURL+"/v2/", "Docker-Distribution-Api-Version"); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// ConfigureAuth stores credentials for challenge responses
|
||||
func configureAuth(username, password string) (auth.CredentialStore, error) {
|
||||
creds := map[string]userpass{
|
||||
tokenURL: {
|
||||
username: username,
|
||||
|
|
|
@ -22,6 +22,7 @@ type proxyBlobStore struct {
|
|||
remoteStore distribution.BlobService
|
||||
scheduler *scheduler.TTLExpirationScheduler
|
||||
repositoryName reference.Named
|
||||
authChallenger authChallenger
|
||||
}
|
||||
|
||||
var _ distribution.BlobStore = &proxyBlobStore{}
|
||||
|
@ -121,6 +122,10 @@ func (pbs *proxyBlobStore) ServeBlob(ctx context.Context, w http.ResponseWriter,
|
|||
return nil
|
||||
}
|
||||
|
||||
if err := pbs.authChallenger.tryEstablishChallenges(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
mu.Lock()
|
||||
_, ok := inflight[dgst]
|
||||
if ok {
|
||||
|
@ -162,6 +167,10 @@ func (pbs *proxyBlobStore) Stat(ctx context.Context, dgst digest.Digest) (distri
|
|||
return distribution.Descriptor{}, err
|
||||
}
|
||||
|
||||
if err := pbs.authChallenger.tryEstablishChallenges(ctx); err != nil {
|
||||
return distribution.Descriptor{}, err
|
||||
}
|
||||
|
||||
return pbs.remoteStore.Stat(ctx, dgst)
|
||||
}
|
||||
|
||||
|
|
|
@ -168,6 +168,7 @@ func makeTestEnv(t *testing.T, name string) *testEnv {
|
|||
remoteStore: truthBlobs,
|
||||
localStore: localBlobs,
|
||||
scheduler: s,
|
||||
authChallenger: &mockChallenger{},
|
||||
}
|
||||
|
||||
te := &testEnv{
|
||||
|
@ -242,6 +243,11 @@ func TestProxyStoreStat(t *testing.T) {
|
|||
if (*remoteStats)["stat"] != remoteBlobCount {
|
||||
t.Errorf("Unexpected remote stat count")
|
||||
}
|
||||
|
||||
if te.store.authChallenger.(*mockChallenger).count != len(te.inRemote) {
|
||||
t.Fatalf("Unexpected auth challenge count, got %#v", te.store.authChallenger)
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func TestProxyStoreServeHighConcurrency(t *testing.T) {
|
||||
|
|
|
@ -19,6 +19,7 @@ type proxyManifestStore struct {
|
|||
remoteManifests distribution.ManifestService
|
||||
repositoryName reference.Named
|
||||
scheduler *scheduler.TTLExpirationScheduler
|
||||
authChallenger authChallenger
|
||||
}
|
||||
|
||||
var _ distribution.ManifestService = &proxyManifestStore{}
|
||||
|
@ -31,7 +32,9 @@ func (pms proxyManifestStore) Exists(ctx context.Context, dgst digest.Digest) (b
|
|||
if exists {
|
||||
return true, nil
|
||||
}
|
||||
|
||||
if err := pms.authChallenger.tryEstablishChallenges(ctx); err != nil {
|
||||
return false, err
|
||||
}
|
||||
return pms.remoteManifests.Exists(ctx, dgst)
|
||||
}
|
||||
|
||||
|
@ -41,6 +44,10 @@ func (pms proxyManifestStore) Get(ctx context.Context, dgst digest.Digest, optio
|
|||
var fromRemote bool
|
||||
manifest, err := pms.localManifests.Get(ctx, dgst, options...)
|
||||
if err != nil {
|
||||
if err := pms.authChallenger.tryEstablishChallenges(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
manifest, err = pms.remoteManifests.Get(ctx, dgst, options...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
|
|
@ -2,6 +2,7 @@ package proxy
|
|||
|
||||
import (
|
||||
"io"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"github.com/docker/distribution"
|
||||
|
@ -64,6 +65,20 @@ func (sm statsManifest) Put(ctx context.Context, manifest distribution.Manifest,
|
|||
}
|
||||
*/
|
||||
|
||||
type mockChallenger struct {
|
||||
sync.Mutex
|
||||
count int
|
||||
}
|
||||
|
||||
// Called for remote operations only
|
||||
func (mc *mockChallenger) tryEstablishChallenges(context.Context) error {
|
||||
mc.Lock()
|
||||
defer mc.Unlock()
|
||||
|
||||
mc.count++
|
||||
return nil
|
||||
}
|
||||
|
||||
func newManifestStoreTestEnv(t *testing.T, name, tag string) *manifestStoreTestEnv {
|
||||
nameRef, err := reference.ParseNamed(name)
|
||||
if err != nil {
|
||||
|
@ -120,6 +135,7 @@ func newManifestStoreTestEnv(t *testing.T, name, tag string) *manifestStoreTestE
|
|||
remoteManifests: truthManifests,
|
||||
scheduler: s,
|
||||
repositoryName: nameRef,
|
||||
authChallenger: &mockChallenger{},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
@ -198,6 +214,10 @@ func TestProxyManifests(t *testing.T) {
|
|||
t.Errorf("Unexpected exists count : \n%v \n%v", localStats, remoteStats)
|
||||
}
|
||||
|
||||
if env.manifests.authChallenger.(*mockChallenger).count != 1 {
|
||||
t.Fatalf("Expected 1 auth challenge, got %#v", env.manifests.authChallenger)
|
||||
}
|
||||
|
||||
// Get - should succeed and pull manifest into local
|
||||
_, err = env.manifests.Get(ctx, env.manifestDigest)
|
||||
if err != nil {
|
||||
|
@ -212,6 +232,10 @@ func TestProxyManifests(t *testing.T) {
|
|||
t.Errorf("Expected local put")
|
||||
}
|
||||
|
||||
if env.manifests.authChallenger.(*mockChallenger).count != 2 {
|
||||
t.Fatalf("Expected 2 auth challenges, got %#v", env.manifests.authChallenger)
|
||||
}
|
||||
|
||||
// Stat - should only go to local
|
||||
exists, err = env.manifests.Exists(ctx, env.manifestDigest)
|
||||
if err != nil {
|
||||
|
@ -225,17 +249,18 @@ func TestProxyManifests(t *testing.T) {
|
|||
t.Errorf("Unexpected exists count")
|
||||
}
|
||||
|
||||
// Get - should get from remote, to test freshness
|
||||
if env.manifests.authChallenger.(*mockChallenger).count != 2 {
|
||||
t.Fatalf("Expected 2 auth challenges, got %#v", env.manifests.authChallenger)
|
||||
}
|
||||
|
||||
// Get proxied - won't require another authchallenge
|
||||
_, err = env.manifests.Get(ctx, env.manifestDigest)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if (*remoteStats)["get"] != 2 && (*remoteStats)["exists"] != 1 && (*localStats)["put"] != 1 {
|
||||
t.Errorf("Unexpected get count")
|
||||
if env.manifests.authChallenger.(*mockChallenger).count != 2 {
|
||||
t.Fatalf("Expected 2 auth challenges, got %#v", env.manifests.authChallenger)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProxyTagService(t *testing.T) {
|
||||
|
||||
}
|
||||
|
|
|
@ -1,10 +1,11 @@
|
|||
package proxy
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"sync"
|
||||
|
||||
"fmt"
|
||||
"github.com/docker/distribution"
|
||||
"github.com/docker/distribution/configuration"
|
||||
"github.com/docker/distribution/context"
|
||||
|
@ -19,13 +20,10 @@ import (
|
|||
|
||||
// proxyingRegistry fetches content from a remote registry and caches it locally
|
||||
type proxyingRegistry struct {
|
||||
embedded distribution.Namespace // provides local registry functionality
|
||||
|
||||
scheduler *scheduler.TTLExpirationScheduler
|
||||
|
||||
remoteURL string
|
||||
credentialStore auth.CredentialStore
|
||||
challengeManager auth.ChallengeManager
|
||||
embedded distribution.Namespace // provides local registry functionality
|
||||
scheduler *scheduler.TTLExpirationScheduler
|
||||
remoteURL string
|
||||
authChallenger authChallenger
|
||||
}
|
||||
|
||||
// NewRegistryPullThroughCache creates a registry acting as a pull through cache
|
||||
|
@ -93,18 +91,20 @@ func NewRegistryPullThroughCache(ctx context.Context, registry distribution.Name
|
|||
return nil, err
|
||||
}
|
||||
|
||||
challengeManager := auth.NewSimpleChallengeManager()
|
||||
cs, err := ConfigureAuth(config.RemoteURL, config.Username, config.Password, challengeManager)
|
||||
cs, err := configureAuth(config.Username, config.Password)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &proxyingRegistry{
|
||||
embedded: registry,
|
||||
scheduler: s,
|
||||
challengeManager: challengeManager,
|
||||
credentialStore: cs,
|
||||
remoteURL: config.RemoteURL,
|
||||
embedded: registry,
|
||||
scheduler: s,
|
||||
remoteURL: config.RemoteURL,
|
||||
authChallenger: &remoteAuthChallenger{
|
||||
remoteURL: config.RemoteURL,
|
||||
challengeManager: auth.NewSimpleChallengeManager(),
|
||||
credentialStore: cs,
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
|
||||
|
@ -117,8 +117,13 @@ func (pr *proxyingRegistry) Repositories(ctx context.Context, repos []string, la
|
|||
}
|
||||
|
||||
func (pr *proxyingRegistry) Repository(ctx context.Context, name reference.Named) (distribution.Repository, error) {
|
||||
hcm, ok := pr.authChallenger.(*remoteAuthChallenger)
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("unexpected challenge manager type %T", pr.authChallenger)
|
||||
}
|
||||
|
||||
tr := transport.NewTransport(http.DefaultTransport,
|
||||
auth.NewAuthorizer(pr.challengeManager, auth.NewTokenHandler(http.DefaultTransport, pr.credentialStore, name.Name(), "pull")))
|
||||
auth.NewAuthorizer(hcm.challengeManager, auth.NewTokenHandler(http.DefaultTransport, hcm.credentialStore, name.Name(), "pull")))
|
||||
|
||||
localRepo, err := pr.embedded.Repository(ctx, name)
|
||||
if err != nil {
|
||||
|
@ -145,6 +150,7 @@ func (pr *proxyingRegistry) Repository(ctx context.Context, name reference.Named
|
|||
remoteStore: remoteRepo.Blobs(ctx),
|
||||
scheduler: pr.scheduler,
|
||||
repositoryName: name,
|
||||
authChallenger: pr.authChallenger,
|
||||
},
|
||||
manifests: &proxyManifestStore{
|
||||
repositoryName: name,
|
||||
|
@ -152,15 +158,53 @@ func (pr *proxyingRegistry) Repository(ctx context.Context, name reference.Named
|
|||
remoteManifests: remoteManifests,
|
||||
ctx: ctx,
|
||||
scheduler: pr.scheduler,
|
||||
authChallenger: pr.authChallenger,
|
||||
},
|
||||
name: name,
|
||||
tags: &proxyTagService{
|
||||
localTags: localRepo.Tags(ctx),
|
||||
remoteTags: remoteRepo.Tags(ctx),
|
||||
localTags: localRepo.Tags(ctx),
|
||||
remoteTags: remoteRepo.Tags(ctx),
|
||||
authChallenger: pr.authChallenger,
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
|
||||
// authChallenger encapsulates a request to the upstream to establish credential challenges
|
||||
type authChallenger interface {
|
||||
tryEstablishChallenges(context.Context) error
|
||||
}
|
||||
|
||||
type remoteAuthChallenger struct {
|
||||
remoteURL string
|
||||
sync.Mutex
|
||||
challengeManager auth.ChallengeManager
|
||||
credentialStore auth.CredentialStore
|
||||
}
|
||||
|
||||
// tryEstablishChallenges will attempt to get a challenge types for the upstream if none currently exist
|
||||
func (hcm *remoteAuthChallenger) tryEstablishChallenges(ctx context.Context) error {
|
||||
hcm.Lock()
|
||||
defer hcm.Unlock()
|
||||
|
||||
remoteURL := hcm.remoteURL + "/v2/"
|
||||
challenges, err := hcm.challengeManager.GetChallenges(remoteURL)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if len(challenges) > 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
// establish challenge type with upstream
|
||||
if err := ping(hcm.challengeManager, remoteURL, challengeHeader); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
context.GetLogger(ctx).Infof("Challenge established with upstream : %s %s", remoteURL, hcm.challengeManager)
|
||||
return nil
|
||||
}
|
||||
|
||||
// proxiedRepository uses proxying blob and manifest services to serve content
|
||||
// locally, or pulling it through from a remote and caching it locally if it doesn't
|
||||
// already exist
|
||||
|
|
|
@ -7,8 +7,9 @@ import (
|
|||
|
||||
// proxyTagService supports local and remote lookup of tags.
|
||||
type proxyTagService struct {
|
||||
localTags distribution.TagService
|
||||
remoteTags distribution.TagService
|
||||
localTags distribution.TagService
|
||||
remoteTags distribution.TagService
|
||||
authChallenger authChallenger
|
||||
}
|
||||
|
||||
var _ distribution.TagService = proxyTagService{}
|
||||
|
@ -17,16 +18,19 @@ var _ distribution.TagService = proxyTagService{}
|
|||
// tag service first and then caching it locally. If the remote is unavailable
|
||||
// the local association is returned
|
||||
func (pt proxyTagService) Get(ctx context.Context, tag string) (distribution.Descriptor, error) {
|
||||
desc, err := pt.remoteTags.Get(ctx, tag)
|
||||
err := pt.authChallenger.tryEstablishChallenges(ctx)
|
||||
if err == nil {
|
||||
err := pt.localTags.Tag(ctx, tag, desc)
|
||||
if err != nil {
|
||||
return distribution.Descriptor{}, err
|
||||
desc, err := pt.remoteTags.Get(ctx, tag)
|
||||
if err == nil {
|
||||
err := pt.localTags.Tag(ctx, tag, desc)
|
||||
if err != nil {
|
||||
return distribution.Descriptor{}, err
|
||||
}
|
||||
return desc, nil
|
||||
}
|
||||
return desc, nil
|
||||
}
|
||||
|
||||
desc, err = pt.localTags.Get(ctx, tag)
|
||||
desc, err := pt.localTags.Get(ctx, tag)
|
||||
if err != nil {
|
||||
return distribution.Descriptor{}, err
|
||||
}
|
||||
|
@ -46,9 +50,12 @@ func (pt proxyTagService) Untag(ctx context.Context, tag string) error {
|
|||
}
|
||||
|
||||
func (pt proxyTagService) All(ctx context.Context) ([]string, error) {
|
||||
tags, err := pt.remoteTags.All(ctx)
|
||||
err := pt.authChallenger.tryEstablishChallenges(ctx)
|
||||
if err == nil {
|
||||
return tags, err
|
||||
tags, err := pt.remoteTags.All(ctx)
|
||||
if err == nil {
|
||||
return tags, err
|
||||
}
|
||||
}
|
||||
return pt.localTags.All(ctx)
|
||||
}
|
||||
|
|
|
@ -69,8 +69,9 @@ func testProxyTagService(local, remote map[string]distribution.Descriptor) *prox
|
|||
remote = make(map[string]distribution.Descriptor)
|
||||
}
|
||||
return &proxyTagService{
|
||||
localTags: &mockTagStore{mapping: local},
|
||||
remoteTags: &mockTagStore{mapping: remote},
|
||||
localTags: &mockTagStore{mapping: local},
|
||||
remoteTags: &mockTagStore{mapping: remote},
|
||||
authChallenger: &mockChallenger{},
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -87,6 +88,10 @@ func TestGet(t *testing.T) {
|
|||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if proxyTags.authChallenger.(*mockChallenger).count != 1 {
|
||||
t.Fatalf("Expected 1 auth challenge call, got %#v", proxyTags.authChallenger)
|
||||
}
|
||||
|
||||
if d != remoteDesc {
|
||||
t.Fatal("unable to get put tag")
|
||||
}
|
||||
|
@ -112,6 +117,10 @@ func TestGet(t *testing.T) {
|
|||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if proxyTags.authChallenger.(*mockChallenger).count != 2 {
|
||||
t.Fatalf("Expected 2 auth challenge calls, got %#v", proxyTags.authChallenger)
|
||||
}
|
||||
|
||||
if d != newRemoteDesc {
|
||||
t.Fatal("unable to get put tag")
|
||||
}
|
||||
|
@ -142,7 +151,11 @@ func TestGet(t *testing.T) {
|
|||
t.Fatal("untagged tag should be pulled through")
|
||||
}
|
||||
|
||||
// Add another tag. Ensure both tags appear in enumerate
|
||||
if proxyTags.authChallenger.(*mockChallenger).count != 3 {
|
||||
t.Fatalf("Expected 3 auth challenge calls, got %#v", proxyTags.authChallenger)
|
||||
}
|
||||
|
||||
// Add another tag. Ensure both tags appear in 'All'
|
||||
err = proxyTags.remoteTags.Tag(ctx, "funtag", distribution.Descriptor{Size: 42})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
|
@ -161,4 +174,8 @@ func TestGet(t *testing.T) {
|
|||
if all[0] != "funtag" && all[1] != "remote" {
|
||||
t.Fatalf("Unexpected tags returned from All() : %v ", all)
|
||||
}
|
||||
|
||||
if proxyTags.authChallenger.(*mockChallenger).count != 4 {
|
||||
t.Fatalf("Expected 4 auth challenge calls, got %#v", proxyTags.authChallenger)
|
||||
}
|
||||
}
|
||||
|
|
Loading…
Reference in a new issue