fix: s3 checksum

Author: kamushadenes

internal/sync/images.go

@@ -4,13 +4,16 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"path/filepath"
 	"slices"
 	"strings"
 	"time"
 
 	"github.com/Altinity/docker-sync/config"
 	"github.com/Altinity/docker-sync/internal/telemetry"
 	"github.com/Altinity/docker-sync/structs"
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/s3"
 	"github.com/cenkalti/backoff/v4"
 	"github.com/google/go-containerregistry/pkg/name"
 	"github.com/google/go-containerregistry/pkg/v1/remote"
@@ -162,7 +165,35 @@ func SyncImage(ctx context.Context, image *structs.Image) error {
 
 	for _, dst := range image.Targets {
 		if strings.HasPrefix(dst, "r2:") || strings.HasPrefix(dst, "s3:") {
-			// Comparison is performed during push
+			var s3Session *s3.S3
+			var bucket *string
+			var err error
+
+			if strings.HasPrefix(dst, "r2:") {
+				s3Session, bucket, err = getR2Session(dst)
+			}
+			if strings.HasPrefix(dst, "s3:") {
+				s3Session, bucket, err = getS3Session(dst)
+			}
+			if err != nil {
+				return err
+			}
+
+			s3Lister, err := s3Session.ListObjectsV2(&s3.ListObjectsV2Input{
+				Bucket: bucket,
+				Prefix: aws.String(filepath.Join("v2", image.GetSourceRepository(), "manifests")),
+			})
+			if err != nil {
+				return err
+			}
+
+			for _, obj := range s3Lister.Contents {
+				fname := filepath.Base(*obj.Key)
+				if !strings.HasPrefix(fname, "sha256:") {
+					dstTags = append(dstTags, fname)
+				}
+			}
+
 			continue
 		}
 

internal/sync/s3.go

@@ -3,7 +3,7 @@ package sync
 import (
 	"bytes"
 	"context"
-	"crypto/sha256"
+	"crypto/md5"
 	"fmt"
 	"io"
 	"net/http"
@@ -71,7 +71,7 @@ func pushS3WithSession(ctx context.Context, s3Session *s3.S3, bucket *string, im
 		"v2",
 		acl,
 		aws.String("application/json"),
-		bytes.NewReader([]byte{}), // No content is needed, we just need to return a 200.
+		bytes.NewReader([]byte("{}")), // No content is needed, we just need to return a 200.
 	); err != nil {
 		return err
 	}
@@ -168,19 +168,20 @@ func pushS3WithSession(ctx context.Context, s3Session *s3.S3, bucket *string, im
 		ctx,
 		s3Session,
 		bucket,
-		filepath.Join(baseDir, "manifests", tag),
+		filepath.Join(baseDir, "manifests", desc.Digest.String()),
 		acl,
 		mediaType,
 		bytes.NewReader(manifest),
 	); err != nil {
 		return err
 	}
 
+	// Tag is added last so it can be used to check for duplication.
 	if err := syncObject(
 		ctx,
 		s3Session,
 		bucket,
-		filepath.Join(baseDir, "manifests", desc.Digest.String()),
+		manifestKey(image, tag),
 		acl,
 		mediaType,
 		bytes.NewReader(manifest),
@@ -191,15 +192,19 @@ func pushS3WithSession(ctx context.Context, s3Session *s3.S3, bucket *string, im
 	return nil
 }
 
-func syncObject(ctx context.Context, s3Session *s3.S3, bucket *string, key string, acl *string, contentType *string, r io.ReadSeeker) error {
-	// FIXME: we are reading the object every time to calculate the digest. This is inefficient.
-	h := sha256.New()
-	if _, err := io.Copy(h, r); err != nil {
-		return err
+func s3ObjectExists(s3Session *s3.S3, bucket *string, key string) (bool, error) {
+	_, err := s3Session.HeadObject(&s3.HeadObjectInput{
+		Bucket: bucket,
+		Key:    &key,
+	})
+	if err != nil {
+		return false, err
 	}
-	calculatedDigest := fmt.Sprintf("sha256:%x", h.Sum(nil))
-	r.Seek(0, io.SeekStart)
 
+	return true, nil
+}
+
+func syncObject(ctx context.Context, s3Session *s3.S3, bucket *string, key string, acl *string, contentType *string, r io.ReadSeeker) error {
 	head, err := s3Session.HeadObject(&s3.HeadObjectInput{
 		Bucket: bucket,
 		Key:    &key,
@@ -210,13 +215,32 @@ func syncObject(ctx context.Context, s3Session *s3.S3, bucket *string, key strin
 		}
 	}
 
-	headMetadataDigest, digestPresent := head.Metadata["X-Calculated-Digest"]
+	// We store the digest as metadata so we can compare with the ETag without having to download the object.
+	headMetadataDigestPtr, digestPresent := head.Metadata["X-Calculated-Digest"]
+	var headMetadataDigest string
+	if digestPresent {
+		headMetadataDigest = *headMetadataDigestPtr
+	}
+
+	var etag string
+	if head != nil && head.ETag != nil {
+		etag = strings.ReplaceAll(*head.ETag, `"`, "")
+	}
 
 	if head == nil ||
 		head.ContentType == nil ||
 		*head.ContentType != *contentType ||
 		!digestPresent ||
-		*headMetadataDigest != calculatedDigest {
+		headMetadataDigest != etag {
+
+		r.Seek(0, io.SeekStart)
+		h := md5.New()
+		if _, err := io.Copy(h, r); err != nil {
+			return err
+		}
+		calculatedDigest := fmt.Sprintf("%x", h.Sum(nil))
+		r.Seek(0, io.SeekStart)
+
 		log.Info().
 			Str("bucket", *bucket).
 			Str("key", key).
@@ -239,3 +263,7 @@ func syncObject(ctx context.Context, s3Session *s3.S3, bucket *string, key strin
 
 	return nil
 }
+
+func manifestKey(image *structs.Image, tag string) string {
+	return filepath.Join("v2", image.GetSourceRepository(), "manifests", tag)
+}