feat: add support for aws ecr tokens (#2650)

pkg/extensions/sync/ecr_credential_helper.go

@@ -0,0 +1,142 @@
+//go:build sync
+// +build sync
+
+package sync
+
+import (
+	"context"
+	"encoding/base64"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/service/ecr"
+	syncconf "zotregistry.dev/zot/pkg/extensions/config/sync"
+	"zotregistry.dev/zot/pkg/log"
+)
+
+// ECR tokens are valid for 12 hours. The ExpiryWindow variable is set to 1 hour,
+// meaning if the remaining validity of the token is less than 1 hour, it will be considered expired.
+const ExpiryWindow int = 1
+
+type ECRCredential struct {
+	username string
+	password string
+	expiry   time.Time
+	account  string
+	region   string
+}
+
+type ECRCredentialsHelper struct {
+	credentials map[string]ECRCredential
+	log         log.Logger
+}
+
+func NewECRCredentialHelper(log log.Logger) CredentialHelper {
+	return &ECRCredentialsHelper{
+		credentials: make(map[string]ECRCredential),
+		log:         log,
+	}
+}
+
+// extractAccountAndRegion extracts the account ID and region from the given ECR URL
+// Example URL format: account.dkr.ecr.region.amazonaws.com
+func extractAccountAndRegion(url string) (string, string, error) {
+	parts := strings.Split(url, ".")
+	if len(parts) < 6 {
+		return "", "", fmt.Errorf("invalid URL format: %s", url)
+	}
+
+	accountID := parts[0] // First part is the account ID
+	region := parts[3]    // Fourth part is the region
+
+	return accountID, region, nil
+}
+
+func getECRCredentials(remoteAddress string) (ECRCredential, error) {
+	// Extract account ID and region from the URL
+	accountID, region, err := extractAccountAndRegion(remoteAddress)
+	if err != nil {
+		return ECRCredential{}, fmt.Errorf("failed to extract account and region from URL %s: %w", remoteAddress, err)
+	}
+
+	// Load the AWS config for the specific region
+	cfg, err := config.LoadDefaultConfig(context.TODO(), config.WithRegion(region))
+	if err != nil {
+		return ECRCredential{}, fmt.Errorf("unable to load AWS config for region %s: %w", region, err)
+	}
+
+	// Create an ECR client
+	ecrClient := ecr.NewFromConfig(cfg)
+
+	// Fetch the ECR authorization token
+	ecrAuth, err := ecrClient.GetAuthorizationToken(context.TODO(), &ecr.GetAuthorizationTokenInput{
+		RegistryIds: []string{accountID}, // Filter by the account ID
+	})
+	if err != nil {
+		return ECRCredential{}, fmt.Errorf("unable to get ECR authorization token for account %s: %w", accountID, err)
+	}
+
+	// Decode the base64-encoded ECR token
+	authToken := *ecrAuth.AuthorizationData[0].AuthorizationToken
+	decodedToken, err := base64.StdEncoding.DecodeString(authToken)
+	if err != nil {
+		return ECRCredential{}, fmt.Errorf("unable to decode ECR token: %w", err)
+	}
+
+	// Split the decoded token into username and password (username is "AWS")
+	tokenParts := strings.Split(string(decodedToken), ":")
+	if len(tokenParts) != 2 {
+		return ECRCredential{}, fmt.Errorf("invalid token format received from ECR")
+	}
+
+	expiry := *ecrAuth.AuthorizationData[0].ExpiresAt
+	username := tokenParts[0]
+	password := tokenParts[1]
+
+	return ECRCredential{username: username, password: password, expiry: expiry, account: accountID, region: region}, nil
+
+}
+
+// GetECRCredentials retrieves the ECR credentials (username and password) from AWS ECR
+func (credHelper *ECRCredentialsHelper) GetCredentials(urls []string) (syncconf.CredentialsFile, error) {
+	ecrCredentials := make(syncconf.CredentialsFile)
+
+	for _, url := range urls {
+		remoteAddress := StripRegistryTransport(url)
+		ecrCred, err := getECRCredentials(remoteAddress)
+		if err != nil {
+			return syncconf.CredentialsFile{}, fmt.Errorf("failed to get ECR credentials for URL %s: %w", url, err)
+		}
+		// Store the credentials in the map using the base URL as the key
+		ecrCredentials[remoteAddress] = syncconf.Credentials{
+			Username: ecrCred.username,
+			Password: ecrCred.password,
+		}
+		credHelper.credentials[remoteAddress] = ecrCred
+	}
+	return ecrCredentials, nil
+}
+
+func (credHelper *ECRCredentialsHelper) IsCredentialsValid(remoteAddress string) bool {
+	expiry := credHelper.credentials[remoteAddress].expiry
+	expiryDuration := time.Duration(ExpiryWindow) * time.Hour
+	if time.Until(expiry) <= expiryDuration {
+		credHelper.log.Info().Str("url", remoteAddress).Msg("The credentials are close to expiring")
+		return false
+	}
+
+	credHelper.log.Info().Str("url", remoteAddress).Msg("The credentials are valid")
+	return true
+}
+
+func (credHelper *ECRCredentialsHelper) RefreshCredentials(remoteAddress string) (syncconf.Credentials, error) {
+	credHelper.log.Info().Str("url", remoteAddress).Msg("Refreshing the ECR credentials")
+	ecrCred, err := getECRCredentials(remoteAddress)
+	if err != nil {
+		return syncconf.Credentials{}, fmt.Errorf("failed to get ECR credentials for URL %s: %w", remoteAddress, err)
+	}
+	return syncconf.Credentials{Username: ecrCred.username, Password: ecrCred.password}, nil
+
+}

pkg/extensions/sync/remote.go

@@ -6,6 +6,7 @@
 import (
 	"context"
 	"fmt"
+	"net/url"
 	"strings"
 
 	"github.com/containers/image/v5/docker"
@@ -43,20 +44,51 @@
 	return registry
 }
 
+func (registry *RemoteRegistry) SetUpstreamAuthConfig(username, password string) {
+	registry.context.DockerAuthConfig = &types.DockerAuthConfig{
+		Username: username,
+		Password: password,
+	}
+}
+
 func (registry *RemoteRegistry) GetContext() *types.SystemContext {
 	return registry.context
 }
 
 func (registry *RemoteRegistry) GetRepositories(ctx context.Context) ([]string, error) {
 	var catalog catalog
 
-	_, _, _, err := registry.client.MakeGetRequest(ctx, &catalog, "application/json", //nolint: dogsled
+	_, header, _, err := registry.client.MakeGetRequest(ctx, &catalog, "application/json", "", //nolint: dogsled
 		constants.RoutePrefix, constants.ExtCatalogPrefix)
 	if err != nil {
 		return []string{}, err
 	}
 
-	return catalog.Repositories, nil
+	var repos []string
+
+	repos = append(repos, catalog.Repositories...)
+
+	link := header.Get("Link")
+	for link != "" {
+		linkURLPart, _, _ := strings.Cut(link, ";")
+
+		linkURL, err := url.Parse(strings.Trim(linkURLPart, "<>"))
+		if err != nil {
+			return catalog.Repositories, err
+		}
+
+		_, header, _, err := registry.client.MakeGetRequest(ctx, &catalog, "application/json",
+			linkURL.RawQuery, constants.RoutePrefix, constants.ExtCatalogPrefix) //nolint: dogsled
+		if err != nil {
+			return repos, err
+		}
+
+		repos = append(repos, catalog.Repositories...)
+
+		link = header.Get("Link")
+	}
+
+	return repos, nil
 }
 
 func (registry *RemoteRegistry) GetDockerRemoteRepo(repo string) string {

pkg/extensions/sync/service.go

@@ -27,19 +27,20 @@
 )
 
 type BaseService struct {
-	config          syncconf.RegistryConfig
-	credentials     syncconf.CredentialsFile
-	clusterConfig   *config.ClusterConfig
-	remote          Remote
-	destination     Destination
-	retryOptions    *retry.RetryOptions
-	contentManager  ContentManager
-	storeController storage.StoreController
-	metaDB          mTypes.MetaDB
-	repositories    []string
-	references      references.References
-	client          *client.Client
-	log             log.Logger
+	config           syncconf.RegistryConfig
+	credentials      syncconf.CredentialsFile
+	credentialHelper CredentialHelper
+	clusterConfig    *config.ClusterConfig
+	remote           Remote
+	destination      Destination
+	retryOptions     *retry.RetryOptions
+	contentManager   ContentManager
+	storeController  storage.StoreController
+	metaDB           mTypes.MetaDB
+	repositories     []string
+	references       references.References
+	client           *client.Client
+	log              log.Logger
 }
 
 func New(
@@ -60,16 +61,35 @@
 	var err error
 
 	var credentialsFile syncconf.CredentialsFile
-	if credentialsFilepath != "" {
-		credentialsFile, err = getFileCredentials(credentialsFilepath)
-		if err != nil {
-			log.Error().Str("errortype", common.TypeOf(err)).Str("path", credentialsFilepath).
-				Err(err).Msg("couldn't get registry credentials from configured path")
+	if service.config.CredentialHelper == "" {
+		// Only load credentials from file if CredentialHelper is not set
+		if credentialsFilepath != "" {
+			log.Info().Msgf("Using file-based credentials because CredentialHelper is not set")
+			credentialsFile, err = getFileCredentials(credentialsFilepath)
+			if err != nil {
+				log.Error().Str("errortype", common.TypeOf(err)).Str("path", credentialsFilepath).
+					Err(err).Msg("couldn't get registry credentials from configured path")
+			}
+			service.credentialHelper = nil
+			service.credentials = credentialsFile
+		}
+	} else {
+		log.Info().Msgf("Using credentials helper, because CredentialHelper is set to %s", service.config.CredentialHelper)
+		switch service.config.CredentialHelper {
+		case "ecr":
+			// Logic to fetch credentials for  ECR
+			log.Info().Msg("Fetch the credentials using AWS ECR Auth Token.")
+			service.credentialHelper = NewECRCredentialHelper(log)
+			creds, err := service.credentialHelper.GetCredentials(service.config.URLs)
+			if err != nil {
+				log.Error().Err(err).Msg("Failed to retrieve credentials using ECR credentails helper.")
+			}
+			service.credentials = creds
+		default:
+			log.Warn().Msgf("Unsupported CredentialHelper: %s", service.config.CredentialHelper)
 		}
 	}
 
-	service.credentials = credentialsFile
-
 	// load the cluster config into the object
 	// can be nil if the user did not configure cluster config
 	service.clusterConfig = clusterConfig
@@ -102,7 +122,6 @@
 
 	service.retryOptions = retryOptions
 	service.storeController = storeController
-
 	// try to set next client.
 	if err := service.SetNextAvailableClient(); err != nil {
 		// if it's a ping issue, it will be retried
@@ -126,9 +145,45 @@
 	return service, nil
 }
 
+// refreshRegistryTemporaryCredentials refreshes the temporary credentials for the registry if necessary.
+// It checks whether a CredentialHelper is configured and if the current credentials have expired.
+// If the credentials are expired, it attempts to refresh them and updates the service configuration.
+func (service *BaseService) refreshRegistryTemporaryCredentials() error {
+
+	// If a CredentialHelper is configured, proceed to refresh the credentials if they are invalid or expired.
+	if service.config.CredentialHelper != "" {
+		// Strip the transport protocol (e.g., https:// or http://) from the remote address.
+		remoteAddress := StripRegistryTransport(service.client.GetHostname())
+
+		if !service.credentialHelper.IsCredentialsValid(remoteAddress) {
+			// Attempt to refresh the credentials using the CredentialHelper.
+			credentials, err := service.credentialHelper.RefreshCredentials(remoteAddress)
+			if err != nil {
+				service.log.Error().
+					Err(err).
+					Str("url", remoteAddress).
+					Msg("Failed to refresh the credentials")
+				return err
+			}
+			service.log.Info().
+				Str("url", remoteAddress).
+				Msg("Refreshing the upstream remote registry credentials")
+
+			// Update the service's credentials map with the new set of credentials.
+			service.credentials[remoteAddress] = credentials
+
+			// Set the upstream authentication context using the refreshed credentials.
+			service.remote.SetUpstreamAuthConfig(credentials.Username, credentials.Password)
+		}
+	}
+
+	// Return nil to indicate the operation completed successfully.
+	return nil
+}
+
 func (service *BaseService) SetNextAvailableClient() error {
 	if service.client != nil && service.client.Ping() {
-		return nil
+		return service.refreshRegistryTemporaryCredentials()
 	}
 
 	found := false
@@ -182,7 +237,6 @@
 
 	return nil
 }
-
 func (service *BaseService) GetRetryOptions() *retry.Options {
 	return service.retryOptions
 }

pkg/extensions/sync/sync.go

@@ -13,6 +13,7 @@ import (
 	"github.com/containers/image/v5/types"
 	"github.com/opencontainers/go-digest"
 
+	syncconf "zotregistry.dev/zot/pkg/extensions/config/sync"
 	"zotregistry.dev/zot/pkg/log"
 	"zotregistry.dev/zot/pkg/scheduler"
 )
@@ -48,6 +49,22 @@ type Registry interface {
 	GetContext() *types.SystemContext
 }
 
+// The CredentialHelper interface should be implemented by registries that use temporary tokens.
+// This interface defines methods to:
+// - Check if the credentials for a registry are still valid.
+// - Retrieve credentials for the specified registry URLs.
+// - Refresh credentials for a given registry URL.
+type CredentialHelper interface {
+	// Validates whether the credentials for the specified registry URL have expired.
+	IsCredentialsValid(url string) bool
+
+	// Retrieves credentials for the provided list of registry URLs.
+	GetCredentials(urls []string) (syncconf.CredentialsFile, error)
+
+	// Refreshes credentials for the specified registry URL.
+	RefreshCredentials(url string) (syncconf.Credentials, error)
+}
+
 /*
 Temporary oci layout, sync first pulls an image to this oci layout (using oci:// transport)
 then moves them into ImageStore.
@@ -68,6 +85,9 @@ type Remote interface {
 	// In the case of public dockerhub images 'library' namespace is added to the repo names of images
 	// eg: alpine -> library/alpine
 	GetDockerRemoteRepo(repo string) string
+	// SetUpstreamAuthConfig sets the upstream credentials used when the credential helper is set.
+	// This method refreshes the authentication configuration with the provided username and password.
+	SetUpstreamAuthConfig(username, password string)
 }
 
 // Local registry.