diff --git a/auth/cache.go b/auth/cache.go
new file mode 100644
index 00000000..6f3fcb04
--- /dev/null
+++ b/auth/cache.go
@@ -0,0 +1,46 @@
+/*
+Copyright 2023 The Flux authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package auth
+
+import (
+	"sync"
+	"time"
+)
+
+var once sync.Once
+var cache Store
+
+// InitCache intializes the pacakge cache with the provided cache object.
+// Consumers that want automatic caching when using `GetRegistryAuthenticator()`
+// or `GetGitCredentials()` must call this before. It should only be called once,
+// all subsequent calls will be a no-op.
+func InitCache(s Store) {
+	once.Do(func() {
+		cache = s
+	})
+}
+
+// GetCache returns a handle to the package level cache.
+func GetCache() Store {
+	return cache
+}
+
+// Store is a general purpose key value store.
+type Store interface {
+	Set(key string, val interface{}, ttl time.Duration) error
+	Get(key string) (interface{}, bool)
+}
diff --git a/auth/internal/testutils/dummy_store.go b/auth/internal/testutils/dummy_store.go
new file mode 100644
index 00000000..e5aa656c
--- /dev/null
+++ b/auth/internal/testutils/dummy_store.go
@@ -0,0 +1,45 @@
+/*
+Copyright 2023 The Flux authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package testutils
+
+import (
+	"time"
+
+	"github.com/fluxcd/pkg/auth"
+)
+
+type dummyCache struct {
+	items map[string]interface{}
+}
+
+func NewDummyCache() auth.Store {
+	return &dummyCache{
+		items: make(map[string]interface{}),
+	}
+}
+
+var _ auth.Store = &dummyCache{}
+
+func (c *dummyCache) Set(key string, item interface{}, _ time.Duration) error {
+	c.items[key] = item
+	return nil
+}
+
+func (c *dummyCache) Get(key string) (interface{}, bool) {
+	item, ok := c.items[key]
+	return item, ok
+}
diff --git a/auth/options.go b/auth/options.go
new file mode 100644
index 00000000..68fe5240
--- /dev/null
+++ b/auth/options.go
@@ -0,0 +1,57 @@
+/*
+Copyright 2023 The Flux authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package auth
+
+import (
+	"github.com/fluxcd/pkg/auth/aws"
+	"github.com/fluxcd/pkg/auth/azure"
+	"github.com/fluxcd/pkg/auth/gcp"
+	"github.com/fluxcd/pkg/auth/github"
+	corev1 "k8s.io/api/core/v1"
+)
+
+const (
+	ProviderAWS    = "aws"
+	ProviderAzure  = "azure"
+	ProviderGCP    = "gcp"
+	ProviderGitHub = "github"
+)
+
+// AuthOptions contains options that can be used for authentication.
+type AuthOptions struct {
+	// Secret contains information that can be used to obtain the required
+	// set of credentials.
+	Secret *corev1.Secret
+
+	// ProviderOptions specifies the options to configure various authentication
+	// providers.
+	ProviderOptions ProviderOptions
+
+	// CacheKey is the key to use for caching the authentication credentials.
+	// Consumers must make sure to call `InitCache()` in order for caching to
+	// be enabled.
+	CacheKey string
+}
+
+// ProviderOptions contains options to configure various authentication
+// providers.
+type ProviderOptions struct {
+	AwsOpts    []aws.ProviderOptFunc
+	AzureOpts  []azure.ProviderOptFunc
+	GcpOpts    []gcp.ProviderOptFunc
+	GitHubOpts []github.ProviderOptFunc
+}
diff --git a/auth/registry/authenticator.go b/auth/registry/authenticator.go
new file mode 100644
index 00000000..7ce6b628
--- /dev/null
+++ b/auth/registry/authenticator.go
@@ -0,0 +1,88 @@
+/*
+Copyright 2023 The Flux authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package registry
+
+import (
+	"context"
+	"time"
+
+	"github.com/fluxcd/pkg/auth"
+	"github.com/fluxcd/pkg/auth/aws"
+	"github.com/fluxcd/pkg/auth/azure"
+	"github.com/fluxcd/pkg/auth/gcp"
+	"github.com/google/go-containerregistry/pkg/authn"
+)
+
+// GetAuthenticator returns an authenticator that can provide credentials to
+// access the provided registry.
+// If caching is enabled and authOpts.CacheKey is not blank, the authentication
+// config is cached according to the ttl advertised by the registry provider.
+func GetAuthenticator(ctx context.Context, registry string, provider string, authOpts *auth.AuthOptions) (authn.Authenticator, error) {
+	var authConfig authn.AuthConfig
+
+	cache := auth.GetCache()
+	if cache != nil && authOpts != nil && authOpts.CacheKey != "" {
+		val, found := cache.Get(authOpts.CacheKey)
+		if found {
+			authConfig = val.(authn.AuthConfig)
+			return authn.FromConfig(authConfig), nil
+		}
+	}
+
+	var err error
+	var expiresIn time.Duration
+	switch provider {
+	case auth.ProviderAWS:
+		var opts []aws.ProviderOptFunc
+		if authOpts != nil {
+			opts = authOpts.ProviderOptions.AwsOpts
+		}
+		awsProvider := aws.NewProvider(opts...)
+		authConfig, expiresIn, err = awsProvider.GetECRAuthConfig(ctx, registry)
+	case auth.ProviderAzure:
+		var opts []azure.ProviderOptFunc
+		scopeOpt := azure.GetScopeProiderOption(registry)
+		if scopeOpt != nil {
+			opts = append(opts, scopeOpt)
+		}
+		if authOpts != nil {
+			opts = authOpts.ProviderOptions.AzureOpts
+		}
+
+		azureProvider := azure.NewProvider(opts...)
+		authConfig, expiresIn, err = azureProvider.GetACRAuthConfig(ctx, registry)
+	case auth.ProviderGCP:
+		var opts []gcp.ProviderOptFunc
+		if authOpts != nil {
+			opts = authOpts.ProviderOptions.GcpOpts
+		}
+		gcpProvider := gcp.NewProvider(opts...)
+		authConfig, expiresIn, err = gcpProvider.GetGARAuthConfig(ctx)
+	default:
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	if cache != nil && authOpts != nil && authOpts.CacheKey != "" {
+		if err := cache.Set(authOpts.CacheKey, authConfig, expiresIn); err != nil {
+			return nil, err
+		}
+	}
+	return authn.FromConfig(authConfig), nil
+}
diff --git a/auth/registry/authenticator_test.go b/auth/registry/authenticator_test.go
new file mode 100644
index 00000000..e46ea8bf
--- /dev/null
+++ b/auth/registry/authenticator_test.go
@@ -0,0 +1,231 @@
+/*
+Copyright 2023 The Flux authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package registry
+
+import (
+	"context"
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	"crypto/rand"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	awssdk "github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/fluxcd/pkg/auth"
+	"github.com/fluxcd/pkg/auth/aws"
+	"github.com/fluxcd/pkg/auth/azure"
+	"github.com/fluxcd/pkg/auth/gcp"
+	"github.com/fluxcd/pkg/auth/internal/testutils"
+	"github.com/golang-jwt/jwt/v5"
+	"github.com/google/go-containerregistry/pkg/authn"
+
+	. "github.com/onsi/gomega"
+)
+
+func TestGetAuthenticator(t *testing.T) {
+	g := NewWithT(t)
+	expiresAt := time.Now().UTC().Add(time.Hour)
+	token := jwt.NewWithClaims(jwt.SigningMethodES256, jwt.RegisteredClaims{
+		Issuer:    "auth.microsoft.com",
+		Subject:   "fluxcd",
+		ExpiresAt: jwt.NewNumericDate(expiresAt),
+	})
+
+	pk, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+	g.Expect(err).ToNot(HaveOccurred())
+	tokenStr, err := token.SignedString(pk)
+	g.Expect(err).ToNot(HaveOccurred())
+
+	var s auth.Store
+	s = testutils.NewDummyCache()
+	auth.InitCache(s)
+
+	tests := []struct {
+		name           string
+		authOpts       *auth.AuthOptions
+		provider       string
+		responseBody   string
+		beforeFunc     func(authOpts *auth.AuthOptions, serverURL string, registry *string)
+		afterFunc      func(t *WithT, cache auth.Store, authConfig authn.AuthConfig)
+		expectCacheHit bool
+		wantAuthConfig *authn.AuthConfig
+	}{
+		{
+			name:     "get authenticator from gcp",
+			provider: auth.ProviderGCP,
+			authOpts: &auth.AuthOptions{
+				CacheKey: "gcp-123",
+			},
+			responseBody: `{
+	"access_token": "access-token",
+	"expires_in": 10,
+	"token_type": "Bearer"
+}`,
+			beforeFunc: func(authOpts *auth.AuthOptions, serverURL string, registry *string) {
+				authOpts.ProviderOptions.GcpOpts = []gcp.ProviderOptFunc{gcp.WithTokenURL(serverURL), gcp.WithEmailURL(serverURL)}
+			},
+			wantAuthConfig: &authn.AuthConfig{
+				Username: gcp.DefaultGARUsername,
+				Password: "access-token",
+			},
+			afterFunc: func(t *WithT, cache auth.Store, authConfig authn.AuthConfig) {
+				val, ok := cache.Get("gcp-123")
+				t.Expect(ok).To(BeTrue())
+				ac := val.(authn.AuthConfig)
+				t.Expect(ac).To(Equal(authConfig))
+			},
+		},
+		{
+			name:     "get authenticator from cache",
+			provider: auth.ProviderGCP,
+			authOpts: &auth.AuthOptions{
+				CacheKey: "gcp-123",
+			},
+			expectCacheHit: true,
+			wantAuthConfig: &authn.AuthConfig{
+				Username: gcp.DefaultGARUsername,
+				Password: "access-token",
+			},
+		},
+		{
+			name:     "get authenticator from gcp if cache key is absent",
+			provider: auth.ProviderGCP,
+			responseBody: `{
+	"access_token": "access-token",
+	"expires_in": 10,
+	"token_type": "Bearer"
+}`,
+			authOpts: &auth.AuthOptions{},
+			beforeFunc: func(authOpts *auth.AuthOptions, serverURL string, _ *string) {
+				authOpts.ProviderOptions.GcpOpts = []gcp.ProviderOptFunc{gcp.WithTokenURL(serverURL), gcp.WithEmailURL(serverURL)}
+			},
+			wantAuthConfig: &authn.AuthConfig{
+				Username: gcp.DefaultGARUsername,
+				Password: "access-token",
+			},
+		},
+		{
+			name:     "get authenticator from aws",
+			provider: auth.ProviderAWS,
+			authOpts: &auth.AuthOptions{
+				CacheKey: "aws-123",
+			},
+			responseBody: fmt.Sprintf(`{
+	"authorizationData": [
+		{
+			"authorizationToken": "c29tZS1rZXk6c29tZS1zZWNyZXQ=",
+			"expiresAt": %d
+		}
+	]
+}`, expiresAt.Unix()),
+			beforeFunc: func(authOpts *auth.AuthOptions, serverURL string, registry *string) {
+				cfg := awssdk.NewConfig()
+				cfg.EndpointResolverWithOptions = awssdk.EndpointResolverWithOptionsFunc(
+					func(service, region string, options ...interface{}) (awssdk.Endpoint, error) {
+						return awssdk.Endpoint{URL: serverURL}, nil
+					})
+				cfg.Credentials = credentials.NewStaticCredentialsProvider("x", "y", "z")
+				authOpts.ProviderOptions.AwsOpts = []aws.ProviderOptFunc{aws.WithConfig(*cfg)}
+				*registry = "0123.dkr.ecr.us-east-1.amazonaws.com"
+			},
+			wantAuthConfig: &authn.AuthConfig{
+				Username: "some-key",
+				Password: "some-secret",
+			},
+			afterFunc: func(t *WithT, cache auth.Store, authConfig authn.AuthConfig) {
+				val, ok := cache.Get("aws-123")
+				t.Expect(ok).To(BeTrue())
+				ac := val.(authn.AuthConfig)
+				t.Expect(ac).To(Equal(authConfig))
+			},
+		},
+		{
+			name:     "get authenticator from azure",
+			provider: auth.ProviderAzure,
+			authOpts: &auth.AuthOptions{
+				CacheKey: "azure-123",
+			},
+			responseBody: fmt.Sprintf(`{"refresh_token": "%s"}`, tokenStr),
+			beforeFunc: func(authOpts *auth.AuthOptions, serverURL string, registry *string) {
+				authOpts.ProviderOptions.AzureOpts = []azure.ProviderOptFunc{
+					azure.WithCredential(&azure.FakeTokenCredential{Token: "foo"}),
+				}
+				*registry = serverURL
+			},
+			wantAuthConfig: &authn.AuthConfig{
+				Username: "00000000-0000-0000-0000-000000000000",
+				Password: tokenStr,
+			},
+			afterFunc: func(t *WithT, cache auth.Store, authConfig authn.AuthConfig) {
+				val, ok := cache.Get("azure-123")
+				t.Expect(ok).To(BeTrue())
+				ac := val.(authn.AuthConfig)
+				t.Expect(ac).To(Equal(authConfig))
+			},
+		},
+		{
+			name:           "unknown provider",
+			provider:       "generic",
+			wantAuthConfig: nil,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			g := NewWithT(t)
+
+			var count int
+			handler := func(w http.ResponseWriter, r *http.Request) {
+				count += 1
+				w.WriteHeader(http.StatusOK)
+				w.Write([]byte(tt.responseBody))
+			}
+			srv := httptest.NewServer(http.HandlerFunc(handler))
+			t.Cleanup(func() {
+				srv.Close()
+			})
+
+			var registry string
+			if tt.beforeFunc != nil {
+				tt.beforeFunc(tt.authOpts, srv.URL, &registry)
+			}
+
+			authenticator, err := GetAuthenticator(context.TODO(), registry, tt.provider, tt.authOpts)
+			g.Expect(err).ToNot(HaveOccurred())
+			if tt.wantAuthConfig == nil {
+				g.Expect(authenticator).To(BeNil())
+				return
+			}
+			ac, err := authenticator.Authorization()
+			g.Expect(err).ToNot(HaveOccurred())
+
+			g.Expect(*ac).To(Equal(*tt.wantAuthConfig))
+			if tt.afterFunc != nil {
+				tt.afterFunc(g, s, *ac)
+			}
+			if tt.expectCacheHit {
+				g.Expect(count).To(Equal(0))
+			} else {
+				g.Expect(count).To(BeNumerically(">", 0))
+			}
+		})
+	}
+}