Merge pull request #1466 from RichardScothern/proxy-lazy-auth
Lazily evaluate auth challenges
This commit is contained in:
commit
ad1b181291
8 changed files with 172 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"
|
||||
|
@ -10,6 +11,7 @@ import (
|
|||
"github.com/docker/distribution/manifest"
|
||||
"github.com/docker/distribution/manifest/schema1"
|
||||
"github.com/docker/distribution/reference"
|
||||
"github.com/docker/distribution/registry/client/auth"
|
||||
"github.com/docker/distribution/registry/proxy/scheduler"
|
||||
"github.com/docker/distribution/registry/storage"
|
||||
"github.com/docker/distribution/registry/storage/cache/memory"
|
||||
|
@ -64,6 +66,28 @@ func (sm statsManifest) Put(ctx context.Context, manifest distribution.Manifest,
|
|||
}
|
||||
*/
|
||||
|
||||
type mockChallenger struct {
|
||||
sync.Mutex
|
||||
count int
|
||||
}
|
||||
|
||||
// Called for remote operations only
|
||||
func (m *mockChallenger) tryEstablishChallenges(context.Context) error {
|
||||
m.Lock()
|
||||
defer m.Unlock()
|
||||
|
||||
m.count++
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *mockChallenger) credentialStore() auth.CredentialStore {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *mockChallenger) challengeManager() auth.ChallengeManager {
|
||||
return nil
|
||||
}
|
||||
|
||||
func newManifestStoreTestEnv(t *testing.T, name, tag string) *manifestStoreTestEnv {
|
||||
nameRef, err := reference.ParseNamed(name)
|
||||
if err != nil {
|
||||
|
@ -120,6 +144,7 @@ func newManifestStoreTestEnv(t *testing.T, name, tag string) *manifestStoreTestE
|
|||
remoteManifests: truthManifests,
|
||||
scheduler: s,
|
||||
repositoryName: nameRef,
|
||||
authChallenger: &mockChallenger{},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
@ -198,6 +223,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 +241,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 +258,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"
|
||||
|
@ -20,12 +21,9 @@ 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
|
||||
authChallenger authChallenger
|
||||
}
|
||||
|
||||
// NewRegistryPullThroughCache creates a registry acting as a pull through cache
|
||||
|
@ -93,8 +91,7 @@ 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
|
||||
}
|
||||
|
@ -102,9 +99,12 @@ func NewRegistryPullThroughCache(ctx context.Context, registry distribution.Name
|
|||
return &proxyingRegistry{
|
||||
embedded: registry,
|
||||
scheduler: s,
|
||||
challengeManager: challengeManager,
|
||||
credentialStore: cs,
|
||||
remoteURL: config.RemoteURL,
|
||||
authChallenger: &remoteAuthChallenger{
|
||||
remoteURL: config.RemoteURL,
|
||||
cm: auth.NewSimpleChallengeManager(),
|
||||
cs: cs,
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
|
||||
|
@ -117,8 +117,10 @@ func (pr *proxyingRegistry) Repositories(ctx context.Context, repos []string, la
|
|||
}
|
||||
|
||||
func (pr *proxyingRegistry) Repository(ctx context.Context, name reference.Named) (distribution.Repository, error) {
|
||||
c := pr.authChallenger
|
||||
|
||||
tr := transport.NewTransport(http.DefaultTransport,
|
||||
auth.NewAuthorizer(pr.challengeManager, auth.NewTokenHandler(http.DefaultTransport, pr.credentialStore, name.Name(), "pull")))
|
||||
auth.NewAuthorizer(c.challengeManager(), auth.NewTokenHandler(http.DefaultTransport, c.credentialStore(), name.Name(), "pull")))
|
||||
|
||||
localRepo, err := pr.embedded.Repository(ctx, name)
|
||||
if err != nil {
|
||||
|
@ -145,6 +147,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 +155,63 @@ 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),
|
||||
authChallenger: pr.authChallenger,
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
|
||||
// authChallenger encapsulates a request to the upstream to establish credential challenges
|
||||
type authChallenger interface {
|
||||
tryEstablishChallenges(context.Context) error
|
||||
challengeManager() auth.ChallengeManager
|
||||
credentialStore() auth.CredentialStore
|
||||
}
|
||||
|
||||
type remoteAuthChallenger struct {
|
||||
remoteURL string
|
||||
sync.Mutex
|
||||
cm auth.ChallengeManager
|
||||
cs auth.CredentialStore
|
||||
}
|
||||
|
||||
func (r *remoteAuthChallenger) credentialStore() auth.CredentialStore {
|
||||
return r.cs
|
||||
}
|
||||
|
||||
func (r *remoteAuthChallenger) challengeManager() auth.ChallengeManager {
|
||||
return r.cm
|
||||
}
|
||||
|
||||
// tryEstablishChallenges will attempt to get a challenge type for the upstream if none currently exist
|
||||
func (r *remoteAuthChallenger) tryEstablishChallenges(ctx context.Context) error {
|
||||
r.Lock()
|
||||
defer r.Unlock()
|
||||
|
||||
remoteURL := r.remoteURL + "/v2/"
|
||||
challenges, err := r.cm.GetChallenges(remoteURL)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if len(challenges) > 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
// establish challenge type with upstream
|
||||
if err := ping(r.cm, remoteURL, challengeHeader); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
context.GetLogger(ctx).Infof("Challenge established with upstream : %s %s", remoteURL, r.cm)
|
||||
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
|
||||
|
|
|
@ -9,6 +9,7 @@ import (
|
|||
type proxyTagService struct {
|
||||
localTags distribution.TagService
|
||||
remoteTags distribution.TagService
|
||||
authChallenger authChallenger
|
||||
}
|
||||
|
||||
var _ distribution.TagService = proxyTagService{}
|
||||
|
@ -17,6 +18,8 @@ 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) {
|
||||
err := pt.authChallenger.tryEstablishChallenges(ctx)
|
||||
if err == nil {
|
||||
desc, err := pt.remoteTags.Get(ctx, tag)
|
||||
if err == nil {
|
||||
err := pt.localTags.Tag(ctx, tag, desc)
|
||||
|
@ -25,8 +28,9 @@ func (pt proxyTagService) Get(ctx context.Context, tag string) (distribution.Des
|
|||
}
|
||||
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,10 +50,13 @@ func (pt proxyTagService) Untag(ctx context.Context, tag string) error {
|
|||
}
|
||||
|
||||
func (pt proxyTagService) All(ctx context.Context) ([]string, error) {
|
||||
err := pt.authChallenger.tryEstablishChallenges(ctx)
|
||||
if err == nil {
|
||||
tags, err := pt.remoteTags.All(ctx)
|
||||
if err == nil {
|
||||
return tags, err
|
||||
}
|
||||
}
|
||||
return pt.localTags.All(ctx)
|
||||
}
|
||||
|
||||
|
|
|
@ -71,6 +71,7 @@ func testProxyTagService(local, remote map[string]distribution.Descriptor) *prox
|
|||
return &proxyTagService{
|
||||
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