diff --git a/cmd/kosli/root.go b/cmd/kosli/root.go
index 2f67671c9..8cf78a77f 100644
--- a/cmd/kosli/root.go
+++ b/cmd/kosli/root.go
@@ -119,6 +119,8 @@ Paths the list already matches stay excluded whatever is later added there, so k
 	// the server is the authority on which types are actually accepted
 	validEnvTypesList = "K8S, ECS, S3, lambda, server, docker, azure-apps, cloud-run, logical"
 
+	validS3FingerprintSources = "content, metadata"
+
 	// single source of truth for the service account privilege list shown in
 	// flag help texts; the server is the authority on which privileges are
 	// actually accepted
@@ -267,8 +269,9 @@ Paths the list already matches stay excluded whatever is later added there, so k
 	awsSecretKeyFlag                = "The AWS secret access key."
 	awsRegionFlag                   = "The AWS region."
 	bucketNameFlag                  = "The name of the S3 bucket."
-	downloadConcurrencyFlag         = "[optional] The number of S3 objects to download at the same time when fingerprinting the bucket. Each object in flight may hold up to 40 MB of download buffers in memory, on top of the disk the --download-budget allows."
-	downloadBudgetFlag              = "[optional] The maximum total size of the S3 objects downloading at the same time, which caps the temporary disk the snapshot uses. A bare number is megabytes; add K, M, G or T (optionally followed by B) to choose the unit, e.g. 512M or 8G. An object larger than the budget still downloads, on its own. Objects are downloaded to the OS temporary directory."
+	downloadConcurrencyFlag         = "[optional] The number of S3 objects to fetch at the same time when fingerprinting the bucket. When downloading, the default, each object in flight may hold up to 40 MB of download buffers in memory, on top of the disk the --download-budget allows. With --fingerprint-source metadata it bounds the checksum reads instead, which hold neither, and defaults to 32."
+	downloadBudgetFlag              = "[optional] The maximum total size of the S3 objects downloading at the same time, which caps the temporary disk the snapshot uses. A bare number is megabytes; add K, M, G or T (optionally followed by B) to choose the unit, e.g. 512M or 8G. An object larger than the budget still downloads, on its own. Objects are downloaded to the OS temporary directory. Has no effect with --fingerprint-source metadata, which reads stored checksums and uses no temporary disk."
+	s3FingerprintSourceFlag         = "[defaulted] Where each object's SHA256 comes from when fingerprinting the bucket. Valid sources are: [" + validS3FingerprintSources + "]. 'content' downloads every contributing object and hashes it. 'metadata' reads the SHA256 checksum S3 stores for each object instead, which skips the download but requires every contributing object to have been uploaded with a full-object SHA256 checksum. Both produce the same fingerprint and need the same permissions."
 	bucketPathsFlag                 = "[optional] The comma separated list of file and/or directory paths in the S3 bucket to include when fingerprinting. Paths match by literal prefix. Cannot be used together with --exclude or --exclude-regex."
 	bucketPathsRegexFlag            = "[optional] The comma separated list of Go regular expressions matched against object keys in the S3 bucket to include when fingerprinting. Cannot be used together with --exclude or --exclude-regex."
 	excludeBucketPathsFlag          = "[optional] The comma separated list of file and/or directory paths in the S3 bucket to exclude when fingerprinting. Paths match by literal prefix. Cannot be used together with --include or --include-regex."
diff --git a/cmd/kosli/snapshotS3.go b/cmd/kosli/snapshotS3.go
index 619ce7ac5..52a104b07 100644
--- a/cmd/kosli/snapshotS3.go
+++ b/cmd/kosli/snapshotS3.go
@@ -16,9 +16,14 @@ const snapshotS3ShortDesc = `Report a snapshot of the content of an AWS S3 bucke
 const snapshotS3LongDesc = snapshotS3ShortDesc + awsAuthDesc + `
 You can report the entire bucket content, or filter some of the content using ^--include^ / ^--exclude^ (literal prefix match) or ^--include-regex^ / ^--exclude-regex^ (Go regular expressions matched against the full object key).
 In all cases, the content is reported as one artifact. If you wish to report separate files/dirs within the same bucket as separate artifacts, you need to run the command twice.
-Object keys are never used as local file names: each object is downloaded to a temporary file, hashed and removed, and the fingerprint is computed from the keys and the content digests, so any key S3 accepts can be fingerprinted on any operating system.
+Object keys are never used as local file names: by default each object is downloaded to a temporary file, hashed and removed, and the fingerprint is computed from the keys and the content digests, so any key S3 accepts can be fingerprinted on any operating system.
 Keys that cannot form a directory tree are rejected and fail the snapshot, naming every key involved: a key containing a ^..^ segment, two keys that resolve to the same path (such as ^a//b^ and ^a/b^), or an object whose key is also a prefix of other objects (such as ^a^ beside ^a/b^). A legitimate key of that shape can be left out with ^--exclude-regex^ (anchor and escape it, since the pattern is a regular expression matched against the whole key); when ^--include^ or ^--include-regex^ is set, exclude filters are ignored, so narrow the include filter instead.
 
+By default each object's SHA256 comes from downloading the object and hashing it. ^--fingerprint-source metadata^ reads the SHA256 checksum S3 stores for the object instead, which skips the download, the temporary disk and the hashing. Everything else, like the keys, the ^.kosli_ignore^ rules, the way digests combine into the fingerprint, is the same in both modes, so the fingerprint is identical and a snapshot matches the artifact you attested either way. Two conditions apply:
+- Every contributing object must carry a full-object SHA256 checksum. S3 only stores one when the upload asked for it, for example ^aws s3api put-object --checksum-algorithm SHA256^. Objects without one fail the snapshot, all named in one run.
+- A multipart upload gets a composite SHA256, which hashes the checksums of the parts rather than the object content, so it cannot serve as the object's fingerprint. Such an object can be collapsed into a single part in place with ^aws s3api copy-object --checksum-algorithm SHA256 --copy-source yourBucket/yourKey --bucket yourBucket --key yourKey^.
+A root ^.kosli_ignore^ is still downloaded in this mode, because its rules decide which objects contribute; the objects it excludes are never fetched and need no checksum. Reading a checksum does not need fewer permissions than downloading: AWS requires ^s3:GetObject^ for both, and an SSE-KMS encrypted object additionally needs ^kms:GenerateDataKey^ and ^kms:Decrypt^ either way.
+
 ` + kosliIgnoreDescNoExclude
 
 const snapshotS3Example = `
@@ -69,14 +74,28 @@ kosli snapshot s3 yourEnvironmentName \
 	--exclude-regex '.*\.png$' \
 	--api-token yourAPIToken \
 	--org yourOrgName
+
+# report contents of an AWS S3 bucket without downloading the objects,
+# using the SHA256 checksums S3 stores for them:
+kosli snapshot s3 yourEnvironmentName \
+	--bucket yourBucketName \
+	--fingerprint-source metadata \
+	--api-token yourAPIToken \
+	--org yourOrgName
 `
 
+const (
+	fingerprintSourceContent  = "content"
+	fingerprintSourceMetadata = "metadata"
+)
+
 type snapshotS3Options struct {
 	bucket              string
 	includePaths        []string
 	includeRegex        []string
 	excludePaths        []string
 	excludeRegex        []string
+	fingerprintSource   string
 	downloadConcurrency int
 	downloadBudget      string
 	downloadLimits      aws.DownloadLimits
@@ -112,7 +131,14 @@ func newSnapshotS3Cmd(out io.Writer) *cobra.Command {
 				}
 			}
 
-			return o.resolveDownloadLimits()
+			if o.fingerprintSource != fingerprintSourceContent && o.fingerprintSource != fingerprintSourceMetadata {
+				return ErrorBeforePrintingUsage(cmd, fmt.Sprintf(
+					"%s is not a valid fingerprint source. Valid sources are: [%s]",
+					o.fingerprintSource, validS3FingerprintSources))
+			}
+
+			// Changed covers env vars and config too: bindFlags applies them with Flags().Set.
+			return o.resolveDownloadLimits(cmd.Flags().Changed("download-concurrency"))
 		},
 		RunE: func(cmd *cobra.Command, args []string) error {
 			return o.run(args)
@@ -124,6 +150,7 @@ func newSnapshotS3Cmd(out io.Writer) *cobra.Command {
 	cmd.Flags().StringSliceVar(&o.includeRegex, "include-regex", []string{}, bucketPathsRegexFlag)
 	cmd.Flags().StringSliceVarP(&o.excludePaths, "exclude", "x", []string{}, excludeBucketPathsFlag)
 	cmd.Flags().StringSliceVar(&o.excludeRegex, "exclude-regex", []string{}, excludeBucketPathsRegexFlag)
+	cmd.Flags().StringVar(&o.fingerprintSource, "fingerprint-source", fingerprintSourceContent, s3FingerprintSourceFlag)
 	cmd.Flags().IntVar(&o.downloadConcurrency, "download-concurrency", aws.DefaultDownloadLimits.Concurrency, downloadConcurrencyFlag)
 	cmd.Flags().StringVar(&o.downloadBudget, "download-budget", defaultDownloadBudget, downloadBudgetFlag)
 	addAWSAuthFlags(cmd, o.awsStaticCreds)
@@ -149,7 +176,12 @@ func (o *snapshotS3Options) run(args []string) error {
 		return err
 	}
 
-	s3Data, err := o.awsStaticCreds.GetS3Data(o.bucket, o.includePaths, o.includeRegex, o.excludePaths, o.excludeRegex, o.downloadLimits, logger)
+	harvest := o.awsStaticCreds.GetS3Data
+	if o.fingerprintSource == fingerprintSourceMetadata {
+		harvest = o.awsStaticCreds.GetS3DataFromMetadata
+	}
+
+	s3Data, err := harvest(o.bucket, o.includePaths, o.includeRegex, o.excludePaths, o.excludeRegex, o.downloadLimits, logger)
 	if err != nil {
 		return err
 	}
@@ -175,7 +207,7 @@ func (o *snapshotS3Options) run(args []string) error {
 // spells it; a test keeps the two equal.
 const defaultDownloadBudget = "512M"
 
-func (o *snapshotS3Options) resolveDownloadLimits() error {
+func (o *snapshotS3Options) resolveDownloadLimits(concurrencySet bool) error {
 	if o.downloadConcurrency < 1 {
 		return fmt.Errorf("--download-concurrency must be at least 1, got %d", o.downloadConcurrency)
 	}
@@ -183,6 +215,10 @@ func (o *snapshotS3Options) resolveDownloadLimits() error {
 	if err != nil {
 		return fmt.Errorf("invalid --download-budget: %w", err)
 	}
-	o.downloadLimits = aws.DownloadLimits{Concurrency: o.downloadConcurrency, BytesInFlight: budget}
+	concurrency := o.downloadConcurrency
+	if o.fingerprintSource == fingerprintSourceMetadata && !concurrencySet {
+		concurrency = aws.DefaultMetadataConcurrency
+	}
+	o.downloadLimits = aws.DownloadLimits{Concurrency: concurrency, BytesInFlight: budget}
 	return nil
 }
diff --git a/cmd/kosli/snapshotS3Limits_test.go b/cmd/kosli/snapshotS3Limits_test.go
new file mode 100644
index 000000000..3d9747fea
--- /dev/null
+++ b/cmd/kosli/snapshotS3Limits_test.go
@@ -0,0 +1,46 @@
+package main
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/kosli-dev/cli/internal/aws"
+	"github.com/stretchr/testify/require"
+)
+
+func TestResolveDownloadLimitsPicksTheConcurrencyForTheSource(t *testing.T) {
+	for _, tc := range []struct {
+		name           string
+		source         string
+		concurrency    int
+		concurrencySet bool
+		want           int
+	}{
+		{name: "content mode takes the download default", source: fingerprintSourceContent,
+			concurrency: aws.DefaultDownloadLimits.Concurrency, want: aws.DefaultDownloadLimits.Concurrency},
+		{name: "metadata mode takes its own default", source: fingerprintSourceMetadata,
+			concurrency: aws.DefaultDownloadLimits.Concurrency, want: aws.DefaultMetadataConcurrency},
+		{name: "an explicit value wins in metadata mode", source: fingerprintSourceMetadata,
+			concurrency: 4, concurrencySet: true, want: 4},
+		{name: "an explicit value that equals the download default still wins", source: fingerprintSourceMetadata,
+			concurrency: aws.DefaultDownloadLimits.Concurrency, concurrencySet: true, want: aws.DefaultDownloadLimits.Concurrency},
+		{name: "an explicit value wins in content mode", source: fingerprintSourceContent,
+			concurrency: 4, concurrencySet: true, want: 4},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			o := &snapshotS3Options{fingerprintSource: tc.source, downloadConcurrency: tc.concurrency, downloadBudget: defaultDownloadBudget}
+			require.NoError(t, o.resolveDownloadLimits(tc.concurrencySet))
+			require.Equal(t, tc.want, o.downloadLimits.Concurrency)
+		})
+	}
+}
+
+func TestDefaultMetadataConcurrencyIsWiderThanTheDownloadDefault(t *testing.T) {
+	require.Greater(t, aws.DefaultMetadataConcurrency, aws.DefaultDownloadLimits.Concurrency,
+		"a source that holds no buffers should not be throttled below the download default")
+}
+
+// cobra shows only the flag's own default, so the help text spells out the metadata one.
+func TestDownloadConcurrencyHelpStatesTheMetadataDefault(t *testing.T) {
+	require.Contains(t, downloadConcurrencyFlag, fmt.Sprintf("defaults to %d", aws.DefaultMetadataConcurrency))
+}
diff --git a/cmd/kosli/snapshotS3_test.go b/cmd/kosli/snapshotS3_test.go
index 4932700f0..f430e420d 100644
--- a/cmd/kosli/snapshotS3_test.go
+++ b/cmd/kosli/snapshotS3_test.go
@@ -1,9 +1,12 @@
 package main
 
 import (
+	"crypto/sha256"
+	"encoding/base64"
 	"fmt"
 	"testing"
 
+	s3Types "github.com/aws/aws-sdk-go-v2/service/s3/types"
 	"github.com/kosli-dev/cli/internal/aws"
 	"github.com/stretchr/testify/suite"
 )
@@ -33,12 +36,18 @@ func (suite *SnapshotS3TestSuite) SetupTest() {
 	// Inject a fake S3 client so tests run without AWS credentials.
 	// The fake is seeded with the objects the test cases filter on.
 	bucketName := suite.bucketName
+	objects := map[string][]byte{
+		"README.md":                  []byte("# kosli cli public\n"),
+		"dummy/dummy_2/template.yml": []byte("key: value\n"),
+	}
+	// Only README.md has a stored checksum, so the metadata cases can cover both outcomes.
+	readmeSum := sha256.Sum256(objects["README.md"])
 	aws.NewS3ClientFunc = func(_ *aws.AWSStaticCreds) (aws.S3API, error) {
 		return &aws.FakeS3Client{
-			Bucket: bucketName,
-			Objects: map[string][]byte{
-				"README.md":                  []byte("# kosli cli public\n"),
-				"dummy/dummy_2/template.yml": []byte("key: value\n"),
+			Bucket:  bucketName,
+			Objects: objects,
+			Checksums: map[string]aws.FakeS3Checksum{
+				"README.md": {SHA256: base64.StdEncoding.EncodeToString(readmeSum[:]), Type: s3Types.ChecksumTypeFullObject},
 			},
 		}, nil
 	}
@@ -141,6 +150,28 @@ func (suite *SnapshotS3TestSuite) TestSnapshotS3Cmd() {
 			cmd:       fmt.Sprintf(`snapshot s3 %s %s --bucket %s --download-budget 0`, suite.envName, suite.defaultKosliArguments, suite.bucketName),
 			golden:    "Error: invalid --download-budget: size \"0\" must be at least 1 byte\n",
 		},
+		{
+			name:   "--fingerprint-source metadata fingerprints from the stored checksum",
+			cmd:    fmt.Sprintf(`snapshot s3 %s %s --bucket %s --include README.md --fingerprint-source metadata`, suite.envName, suite.defaultKosliArguments, suite.bucketName),
+			golden: "bucket kosli-cli-public was reported to environment snapshot-s3-env\n",
+		},
+		{
+			name:   "--fingerprint-source content is the default behaviour",
+			cmd:    fmt.Sprintf(`snapshot s3 %s %s --bucket %s --fingerprint-source content`, suite.envName, suite.defaultKosliArguments, suite.bucketName),
+			golden: "bucket kosli-cli-public was reported to environment snapshot-s3-env\n",
+		},
+		{
+			wantError: true,
+			name:      "--fingerprint-source rejects an unknown value",
+			cmd:       fmt.Sprintf(`snapshot s3 %s %s --bucket %s --fingerprint-source etag`, suite.envName, suite.defaultKosliArguments, suite.bucketName),
+			golden:    "Error: etag is not a valid fingerprint source. Valid sources are: [content, metadata]\nUsage: kosli snapshot s3 ENVIRONMENT-NAME [flags]\n",
+		},
+		{
+			wantError: true,
+			name:      "--fingerprint-source metadata fails on an object with no stored checksum",
+			cmd:       fmt.Sprintf(`snapshot s3 %s %s --bucket %s --include dummy --fingerprint-source metadata`, suite.envName, suite.defaultKosliArguments, suite.bucketName),
+			golden:    "Error: object key [dummy/dummy_2/template.yml] has no SHA256 checksum, so its fingerprint cannot be read from S3 metadata. Upload it with one: aws s3api put-object --bucket kosli-cli-public --key dummy/dummy_2/template.yml --body <file> --checksum-algorithm SHA256; or fingerprint by downloading the objects instead\n",
+		},
 	}
 
 	for _, t := range tests {
diff --git a/cmd/kosli/testdata/empty-flag-audit-coverage.json b/cmd/kosli/testdata/empty-flag-audit-coverage.json
index 01381dd12..5228b8d52 100644
--- a/cmd/kosli/testdata/empty-flag-audit-coverage.json
+++ b/cmd/kosli/testdata/empty-flag-audit-coverage.json
@@ -891,6 +891,7 @@
   "dry-run": "bool",
   "exclude": "stringSlice",
   "exclude-regex": "stringSlice",
+  "fingerprint-source": "string",
   "include": "stringSlice",
   "include-regex": "stringSlice"
  },
diff --git a/internal/aws/aws.go b/internal/aws/aws.go
index c2f9d7164..bb37dbf70 100644
--- a/internal/aws/aws.go
+++ b/internal/aws/aws.go
@@ -151,17 +151,26 @@ type S3DownloadAPI interface {
 	DownloadObject(ctx context.Context, params *transfermanager.DownloadObjectInput, optFns ...func(*transfermanager.Options)) (*transfermanager.DownloadObjectOutput, error)
 }
 
+// S3HeadAPI reads an object's metadata, including the checksum S3 stores for
+// it, which is returned only when the request sets ChecksumMode to
+// ChecksumModeEnabled. The real *s3.Client satisfies this implicitly.
+type S3HeadAPI interface {
+	HeadObject(ctx context.Context, params *s3.HeadObjectInput, optFns ...func(*s3.Options)) (*s3.HeadObjectOutput, error)
+}
+
 // S3API is the combined S3 surface that GetS3Data depends on.
 type S3API interface {
 	S3ListAPI
 	S3DownloadAPI
+	S3HeadAPI
 }
 
-// s3Client combines the two real AWS clients that back S3API: *s3.Client for
-// listing and *transfermanager.Client for downloading.
+// s3Client combines the real AWS clients that back S3API: *s3.Client for
+// listing and metadata, and *transfermanager.Client for downloading.
 type s3Client struct {
 	S3ListAPI
 	S3DownloadAPI
+	S3HeadAPI
 }
 
 // defaultNewS3Client creates a real S3 client from credentials.
@@ -172,9 +181,13 @@ func defaultNewS3Client(creds *AWSStaticCreds) (S3API, error) {
 	}
 	// Five parts per object is the SDK's default, pinned so the connection count,
 	// objects in flight times parts, cannot move with an SDK upgrade.
-	return &s3Client{S3ListAPI: client, S3DownloadAPI: transfermanager.New(client, func(o *transfermanager.Options) {
-		o.Concurrency = 5
-	})}, nil
+	return &s3Client{
+		S3ListAPI: client,
+		S3DownloadAPI: transfermanager.New(client, func(o *transfermanager.Options) {
+			o.Concurrency = 5
+		}),
+		S3HeadAPI: client,
+	}, nil
 }
 
 // NewS3ClientFunc is the factory used by GetS3Data to create an S3API client.
@@ -369,7 +382,7 @@ func processOneLambdaFunc(lastModified, codeSha256, functionName string, package
 	lambdaData.Digests = map[string]string{functionName: codeSha256}
 
 	if packageType == types.PackageTypeZip {
-		lambdaData.Digests[functionName], err = decodeLambdaFingerprint(codeSha256)
+		lambdaData.Digests[functionName], err = decodeBase64Sha256(codeSha256)
 		if err != nil {
 			return lambdaData, err
 		}
@@ -384,8 +397,9 @@ func formatLambdaLastModified(lastModified string) (time.Time, error) {
 	return time.Parse(layout, lastModified)
 }
 
-// decodeLambdaFingerprint decodes a base64 lambda function fingerprint
-func decodeLambdaFingerprint(fingerprint string) (string, error) {
+// decodeBase64Sha256 converts a Base64 SHA256 digest, the form AWS reports
+// stored digests in, into the hex form Kosli fingerprints use.
+func decodeBase64Sha256(fingerprint string) (string, error) {
 	sha256base64, err := base64.StdEncoding.DecodeString(fingerprint)
 	if err != nil {
 		return "", err
@@ -454,8 +468,15 @@ func (staticCreds *AWSStaticCreds) GetS3Data(bucket string, includePaths, includ
 	return getS3DataFromClient(client, bucket, includePaths, includeRegex, excludePaths, excludeRegex, limits, logger)
 }
 
-// getS3DataFromClient harvests bucket content using the provided S3API client.
+// getS3DataFromClient harvests bucket content using the provided S3API client,
+// downloading and hashing each object.
 func getS3DataFromClient(client S3API, bucket string, includePaths, includeRegex, excludePaths, excludeRegex []string, limits DownloadLimits, logger *logger.Logger) ([]*S3Data, error) {
+	return getS3DataWithSource(client, downloadDigests(client, bucket, logger), bucket, includePaths, includeRegex, excludePaths, excludeRegex, limits, logger)
+}
+
+// getS3DataWithSource lists and filters the bucket, then fingerprints what is
+// left with digests from source.
+func getS3DataWithSource(client S3API, source s3DigestSource, bucket string, includePaths, includeRegex, excludePaths, excludeRegex []string, limits DownloadLimits, logger *logger.Logger) ([]*S3Data, error) {
 	s3Data := []*S3Data{}
 
 	includeRegexCompiled, err := compilePathRegex(includeRegex)
@@ -485,7 +506,7 @@ func getS3DataFromClient(client S3API, bucket string, includePaths, includeRegex
 		return s3Data, fmt.Errorf("bucket [%s] reported no modification time for any matching object", bucket)
 	}
 
-	artifactName, sha256, err := fingerprintS3Objects(client, bucket, objects, limits, logger)
+	artifactName, sha256, err := fingerprintS3Tree(client, source, bucket, objects, limits, logger)
 	if err != nil {
 		return s3Data, err
 	}
@@ -519,6 +540,12 @@ type DownloadLimits struct {
 // Lambda's default /tmp, and part buffers near 320 MiB of memory.
 var DefaultDownloadLimits = DownloadLimits{Concurrency: 8, BytesInFlight: 512 << 20}
 
+// DefaultMetadataConcurrency is how many checksum reads run at once when
+// fingerprinting from S3 metadata. A HeadObject holds no part buffers or temp
+// disk, so it is sized against S3's request rate instead: 32 at a 30 ms round
+// trip is about 1,000 reads a second, well under the 5,500 S3 allows per prefix.
+const DefaultMetadataConcurrency = 32
+
 // listMatchingS3Objects lists the bucket, dropping folder markers and keys the
 // filters exclude, in the order S3 returns them.
 func listMatchingS3Objects(client S3ListAPI, bucket string, includePaths []string, includeRegex []*regexp.Regexp,
@@ -568,17 +595,42 @@ func listMatchingS3Objects(client S3ListAPI, bucket string, includePaths []strin
 	return objects, nil
 }
 
-// fingerprintS3Objects fingerprints the objects as the directory their keys
-// describe, without ever using a key as a local file name. Each object is
-// downloaded to an anonymous temp file, hashed and removed; the fingerprint is
-// then computed from the (key, sha256) pairs by digest.VirtualDirSha256, which
-// reproduces what digest.DirSha256 gives the same tree on disk. A single object
-// is fingerprinted as that file and named after it, as before.
-//
-// A root .kosli_ignore is downloaded first so its rules can be applied, and
-// objects the rules exclude are not downloaded at all. The remaining objects
-// download in parallel within limits; the first failure cancels the rest.
+// s3DigestSource supplies each object's content sha256 to the fingerprint
+// pipeline. It is the only part that varies by source, so two sources cannot
+// fingerprint the same bucket differently.
+type s3DigestSource struct {
+	// sha256 returns the hex digest of an object's content. tempDir is scratch
+	// space the pipeline removes afterwards.
+	sha256 func(ctx context.Context, tempDir string, object s3Object) (string, error)
+	// usesDisk means an object's listed size occupies temp disk while sha256
+	// runs, so it counts against DownloadLimits.BytesInFlight.
+	usesDisk bool
+}
+
+// downloadDigests is the content-mode source: download, hash, remove.
+func downloadDigests(downloader S3DownloadAPI, bucket string, logger *logger.Logger) s3DigestSource {
+	return s3DigestSource{
+		sha256: func(ctx context.Context, tempDir string, object s3Object) (string, error) {
+			return downloadAndHashS3Object(ctx, downloader, tempDir, bucket, object.key, nil, logger)
+		},
+		usesDisk: true,
+	}
+}
+
+// fingerprintS3Objects fingerprints the objects with the download source.
 func fingerprintS3Objects(downloader S3DownloadAPI, bucket string, objects []s3Object, limits DownloadLimits, logger *logger.Logger) (string, string, error) {
+	return fingerprintS3Tree(downloader, downloadDigests(downloader, bucket, logger), bucket, objects, limits, logger)
+}
+
+// fingerprintS3Tree fingerprints the objects as the directory their keys
+// describe, without using a key as a local file name: digest.VirtualDirSha256
+// combines the (key, sha256) pairs from source exactly as digest.DirSha256
+// would on disk. A single object is fingerprinted as that file and named
+// after it.
+//
+// The root .kosli_ignore is always downloaded, whatever the source, because its
+// rules decide which other objects are fetched at all.
+func fingerprintS3Tree(downloader S3DownloadAPI, source s3DigestSource, bucket string, objects []s3Object, limits DownloadLimits, logger *logger.Logger) (string, string, error) {
 	keys := make([]string, len(objects))
 	for i, object := range objects {
 		keys[i] = object.key
@@ -607,7 +659,12 @@ func fingerprintS3Objects(downloader S3DownloadAPI, bucket string, objects []s3O
 	// One object is fingerprinted as that file and named after it, as it was
 	// when the objects were laid out on disk.
 	if file, ok := digest.SingleVirtualFile(files); ok {
-		sha256, err := downloadAndHashS3Object(context.TODO(), downloader, tempDir, bucket, objects[0].key, nil, logger)
+		fetch := source
+		// The ignore file is downloaded even when alone, so no source needs its checksum.
+		if file.Path == digest.IgnoreFileName {
+			fetch = downloadDigests(downloader, bucket, logger)
+		}
+		sha256, err := fetch.sha256(context.TODO(), tempDir, objects[0])
 		if err != nil {
 			return "", "", err
 		}
@@ -656,9 +713,9 @@ func fingerprintS3Objects(downloader S3DownloadAPI, bucket string, objects []s3O
 		files[i].Sha256 = sha256
 	}
 
-	// Each download writes its own slot, so the manifest stays in listing order
-	// however the downloads interleave.
-	if err := downloadS3ObjectsInParallel(downloader, tempDir, bucket, objects, toDownload, files, limits, logger); err != nil {
+	// Each fetch writes its own slot, so the manifest stays in listing order
+	// however the fetches interleave.
+	if err := fetchS3DigestsInParallel(source, tempDir, objects, toDownload, files, limits, logger); err != nil {
 		return "", "", err
 	}
 
@@ -678,11 +735,13 @@ func ignoreRuleError(err error) error {
 	return err
 }
 
-// downloadS3ObjectsInParallel fetches the objects at indexes and writes each
-// digest into files at the same index. A fixed worker pool bounds downloads and
-// goroutines alike, a weighted semaphore bounds their listed bytes, and the
-// first error cancels the context so nothing further starts.
-func downloadS3ObjectsInParallel(downloader S3DownloadAPI, tempDir, bucket string, objects []s3Object, indexes []int,
+// fetchS3DigestsInParallel reads the digests of the objects at indexes from
+// source and writes each into files at the same index. A fixed worker pool
+// bounds fetches and goroutines alike, a weighted semaphore bounds the listed
+// bytes of sources that use the disk, and the first transport error cancels the
+// context so nothing further starts. unusableChecksumErrors are collected
+// and reported together instead.
+func fetchS3DigestsInParallel(source s3DigestSource, tempDir string, objects []s3Object, indexes []int,
 	files []digest.VirtualFile, limits DownloadLimits, logger *logger.Logger) error {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -696,6 +755,10 @@ func downloadS3ObjectsInParallel(downloader S3DownloadAPI, tempDir, bucket strin
 		}
 		cancel()
 	}
+	var (
+		unusableMu sync.Mutex
+		unusable   []error
+	)
 
 	work := make(chan int)
 	var wg sync.WaitGroup
@@ -706,17 +769,29 @@ func downloadS3ObjectsInParallel(downloader S3DownloadAPI, tempDir, bucket strin
 			for i := range work {
 				object := objects[i]
 				// An object larger than the budget takes all of it and so runs alone.
-				weight := max(min(object.size, limits.BytesInFlight), 1)
-				if err := budget.Acquire(ctx, weight); err != nil {
-					return // cancelled while waiting
+				var weight int64
+				if source.usesDisk {
+					weight = max(min(object.size, limits.BytesInFlight), 1)
+					if err := budget.Acquire(ctx, weight); err != nil {
+						return // cancelled while waiting
+					}
 				}
-				sha256, err := downloadAndHashS3Object(ctx, downloader, tempDir, bucket, object.key, nil, logger)
-				budget.Release(weight)
-				if err != nil {
+				sha256, err := source.sha256(ctx, tempDir, object)
+				if source.usesDisk {
+					budget.Release(weight)
+				}
+				var unusableErr unusableChecksumError
+				switch {
+				case err == nil:
+					files[i].Sha256 = sha256
+				case errors.As(err, &unusableErr):
+					unusableMu.Lock()
+					unusable = append(unusable, err)
+					unusableMu.Unlock()
+				default:
 					fail(err)
 					return
 				}
-				files[i].Sha256 = sha256
 			}
 		}()
 	}
@@ -736,8 +811,11 @@ feed:
 	case err := <-firstErr:
 		return err
 	default:
-		return nil
 	}
+	if len(unusable) > 0 {
+		return combineUnusableChecksumErrors(unusable)
+	}
+	return nil
 }
 
 // downloadAndHashS3Object fetches one object into a fresh temp file, lets
diff --git a/internal/aws/aws_test.go b/internal/aws/aws_test.go
index afc3bacaa..bab7dfef6 100644
--- a/internal/aws/aws_test.go
+++ b/internal/aws/aws_test.go
@@ -68,9 +68,9 @@ func (suite *AWSTestSuite) TestDecodeLambdaFingerprint() {
 		},
 	} {
 		suite.Run(t.name, func() {
-			got, err := decodeLambdaFingerprint(t.base64Fingerprint)
+			got, err := decodeBase64Sha256(t.base64Fingerprint)
 			require.False(suite.T(), (err != nil) != t.wantErr,
-				"decodeLambdaFingerprint() error = %v, wantErr %v", err, t.wantErr)
+				"decodeBase64Sha256() error = %v, wantErr %v", err, t.wantErr)
 			if !t.wantErr {
 				require.Equal(suite.T(), t.wantFingerprint, got)
 			}
diff --git a/internal/aws/fake_s3.go b/internal/aws/fake_s3.go
index 0fc388170..f250642e0 100644
--- a/internal/aws/fake_s3.go
+++ b/internal/aws/fake_s3.go
@@ -17,6 +17,13 @@ import (
 // entry in FakeS3Client.LastModified.
 var fakeS3LastModified = time.Date(2024, 1, 15, 10, 30, 0, 0, time.UTC)
 
+// FakeS3Checksum is the additional checksum S3 has stored for an object.
+type FakeS3Checksum struct {
+	// SHA256 is Base64, as S3 returns it; a composite one ends in a "-N" part count.
+	SHA256 string
+	Type   s3Types.ChecksumType
+}
+
 // FakeS3Client is an in-memory implementation of S3API for testing.
 // It simulates continuation-token pagination and returns errors for unknown
 // buckets and missing objects.
@@ -33,6 +40,9 @@ type FakeS3Client struct {
 	// NoLastModified lists keys whose listing entry carries no LastModified at
 	// all, as some S3-compatible stores return.
 	NoLastModified map[string]bool
+	// Checksums maps object key to its stored checksum. A key with no entry has
+	// none, like an object uploaded without --checksum-algorithm.
+	Checksums map[string]FakeS3Checksum
 	// PageSize controls how many objects are returned per ListObjectsV2 call.
 	// Defaults to 1000 (matching the AWS default) if zero.
 	PageSize int
@@ -42,6 +52,9 @@ type FakeS3Client struct {
 	// DownloadObjectErr, if set, is returned by DownloadObject for any object.
 	// Useful for testing error propagation.
 	DownloadObjectErr error
+	// HeadObjectErr, if set, is returned by HeadObject for any object.
+	// Useful for testing error propagation.
+	HeadObjectErr error
 }
 
 func (f *FakeS3Client) pageSize() int {
@@ -141,6 +154,40 @@ func (f *FakeS3Client) ListObjectsV2(_ context.Context, params *s3.ListObjectsV2
 	return out, nil
 }
 
+func (f *FakeS3Client) HeadObject(_ context.Context, params *s3.HeadObjectInput, _ ...func(*s3.Options)) (*s3.HeadObjectOutput, error) {
+	if params.Bucket == nil || params.Key == nil {
+		return nil, fmt.Errorf("missing required fields: Bucket and Key")
+	}
+	if *params.Bucket != f.Bucket {
+		// Real S3 returns *types.NoSuchBucket.
+		return nil, fmt.Errorf("bucket not found: %s", *params.Bucket)
+	}
+	if f.HeadObjectErr != nil {
+		return nil, f.HeadObjectErr
+	}
+	content, ok := f.Objects[*params.Key]
+	if !ok {
+		// Real S3 returns *types.NotFound for HeadObject.
+		return nil, fmt.Errorf("object not found: %s", *params.Key)
+	}
+
+	out := &s3.HeadObjectOutput{
+		ContentLength: aws.Int64(int64(len(content))),
+		LastModified:  aws.Time(f.lastModified(*params.Key)),
+	}
+
+	// Withhold the checksum unless asked, as S3 does, so a caller that forgets
+	// ChecksumMode fails here too.
+	if params.ChecksumMode != s3Types.ChecksumModeEnabled {
+		return out, nil
+	}
+	if checksum, ok := f.Checksums[*params.Key]; ok {
+		out.ChecksumSHA256 = aws.String(checksum.SHA256)
+		out.ChecksumType = checksum.Type
+	}
+	return out, nil
+}
+
 func (f *FakeS3Client) DownloadObject(_ context.Context, params *transfermanager.DownloadObjectInput, _ ...func(*transfermanager.Options)) (*transfermanager.DownloadObjectOutput, error) {
 	if params.Bucket == nil || params.Key == nil {
 		return nil, fmt.Errorf("missing required fields: Bucket and Key")
diff --git a/internal/aws/s3_contract_test.go b/internal/aws/s3_contract_test.go
index d01e03e2d..2dbc6377b 100644
--- a/internal/aws/s3_contract_test.go
+++ b/internal/aws/s3_contract_test.go
@@ -2,6 +2,8 @@ package aws
 
 import (
 	"context"
+	"crypto/sha256"
+	"encoding/base64"
 	"errors"
 	"os"
 	"path/filepath"
@@ -11,6 +13,7 @@ import (
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
+	s3Types "github.com/aws/aws-sdk-go-v2/service/s3/types"
 	"github.com/kosli-dev/cli/internal/testHelpers"
 	"github.com/stretchr/testify/require"
 )
@@ -18,6 +21,12 @@ import (
 // errInjected is the error tests inject into FakeS3Client to exercise error paths.
 var errInjected = errors.New("injected error")
 
+// base64Sha256 is content's SHA256 in the Base64 form S3 reports checksums in.
+func base64Sha256(content []byte) string {
+	sum := sha256.Sum256(content)
+	return base64.StdEncoding.EncodeToString(sum[:])
+}
+
 // runS3ContractTests exercises the S3API contract. It verifies the behaviours
 // we depend on — object listing, continuation-token pagination, object
 // download, and error responses for missing buckets and keys.
@@ -27,7 +36,10 @@ var errInjected = errors.New("injected error")
 //
 // bucket must name a bucket the client can see, holding at least two objects.
 // existingKey must name an object in that bucket with a non-empty body.
-func runS3ContractTests(t *testing.T, client S3API, bucket, existingKey string) {
+// sha256ChecksumKey names an object with a stored SHA256 checksum, or is empty
+// to skip those sub-tests: kosli-cli-public has none, and adding one would move
+// the fingerprints TestGetS3Data pins.
+func runS3ContractTests(t *testing.T, client S3API, bucket, existingKey, sha256ChecksumKey string) {
 	t.Helper()
 
 	t.Run("ListObjectsV2 returns objects with keys and modification times", func(t *testing.T) {
@@ -132,6 +144,55 @@ func runS3ContractTests(t *testing.T, client S3API, bucket, existingKey string)
 		})
 		require.Error(t, err)
 	})
+
+	t.Run("HeadObject returns object metadata", func(t *testing.T) {
+		out, err := client.HeadObject(context.TODO(), &s3.HeadObjectInput{
+			Bucket: aws.String(bucket),
+			Key:    aws.String(existingKey),
+		})
+		require.NoError(t, err)
+		require.NotNil(t, out)
+		require.NotNil(t, out.ContentLength, "ContentLength should be present")
+		require.NotNil(t, out.LastModified, "LastModified should be present")
+	})
+
+	t.Run("HeadObject errors for a missing key", func(t *testing.T) {
+		_, err := client.HeadObject(context.TODO(), &s3.HeadObjectInput{
+			Bucket: aws.String(bucket),
+			Key:    aws.String("nonexistent-key-that-should-not-exist"),
+		})
+		require.Error(t, err)
+	})
+
+	t.Run("HeadObject omits the checksum unless ChecksumMode is enabled", func(t *testing.T) {
+		if sha256ChecksumKey == "" {
+			t.Skip("no object with an SHA256 checksum available in this bucket")
+		}
+		out, err := client.HeadObject(context.TODO(), &s3.HeadObjectInput{
+			Bucket: aws.String(bucket),
+			Key:    aws.String(sha256ChecksumKey),
+		})
+		require.NoError(t, err)
+		require.Nil(t, out.ChecksumSHA256,
+			"ChecksumSHA256 should be absent when ChecksumMode is not enabled")
+	})
+
+	t.Run("HeadObject returns the stored SHA256 when ChecksumMode is enabled", func(t *testing.T) {
+		if sha256ChecksumKey == "" {
+			t.Skip("no object with an SHA256 checksum available in this bucket")
+		}
+		out, err := client.HeadObject(context.TODO(), &s3.HeadObjectInput{
+			Bucket:       aws.String(bucket),
+			Key:          aws.String(sha256ChecksumKey),
+			ChecksumMode: s3Types.ChecksumModeEnabled,
+		})
+		require.NoError(t, err)
+		require.NotNil(t, out.ChecksumSHA256, "ChecksumSHA256 should be present")
+		require.NotEmpty(t, *out.ChecksumSHA256)
+		require.NotContains(t, *out.ChecksumSHA256, "-",
+			"a single-part upload should carry a full-object checksum")
+		require.Equal(t, s3Types.ChecksumTypeFullObject, out.ChecksumType)
+	})
 }
 
 func TestS3Contract_Fake(t *testing.T) {
@@ -142,11 +203,17 @@ func TestS3Contract_Fake(t *testing.T) {
 			"README.md":                  []byte("# readme\n"),
 			"dummy/dummy_2/template.yml": []byte("key: value\n"),
 		},
+		Checksums: map[string]FakeS3Checksum{
+			"README.md": {
+				SHA256: base64Sha256([]byte("# readme\n")),
+				Type:   s3Types.ChecksumTypeFullObject,
+			},
+		},
 		// One object per page so the pagination contract is genuinely exercised.
 		PageSize: 1,
 	}
 
-	runS3ContractTests(t, client, bucket, "README.md")
+	runS3ContractTests(t, client, bucket, "README.md", "README.md")
 
 	// Error injection is a fake-specific mechanism with no real-API equivalent.
 	// These tests verify the fake itself, not the contract.
@@ -172,6 +239,16 @@ func TestS3Contract_Fake(t *testing.T) {
 		require.Error(t, err)
 	})
 
+	t.Run("HeadObject returns error when HeadObjectErr is injected", func(t *testing.T) {
+		client.HeadObjectErr = errInjected
+		defer func() { client.HeadObjectErr = nil }()
+		_, err := client.HeadObject(context.TODO(), &s3.HeadObjectInput{
+			Bucket: aws.String(bucket),
+			Key:    aws.String("README.md"),
+		})
+		require.Error(t, err)
+	})
+
 	// The fake rejects listing inputs outside the contract. Real S3 accepts
 	// these without erroring, so they cannot live in runS3ContractTests — they
 	// exist so a future caller fails loudly instead of hanging or panicking.
@@ -203,5 +280,5 @@ func TestS3Contract_RealAWS(t *testing.T) {
 	client, err := defaultNewS3Client(creds)
 	require.NoError(t, err)
 
-	runS3ContractTests(t, client, "kosli-cli-public", "README.md")
+	runS3ContractTests(t, client, "kosli-cli-public", "README.md", "")
 }
diff --git a/internal/aws/s3_metadata.go b/internal/aws/s3_metadata.go
new file mode 100644
index 000000000..4b3f22d82
--- /dev/null
+++ b/internal/aws/s3_metadata.go
@@ -0,0 +1,133 @@
+package aws
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sort"
+	"strings"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	s3Types "github.com/aws/aws-sdk-go-v2/service/s3/types"
+	"github.com/kosli-dev/cli/internal/logger"
+)
+
+// GetS3DataFromMetadata returns what GetS3Data returns, but takes each
+// object's SHA256 from the checksum S3 stores for it instead of downloading
+// the object. Every contributing object needs a full-object SHA256 checksum.
+// Reading one needs s3:GetObject, the same permission as a download.
+func (staticCreds *AWSStaticCreds) GetS3DataFromMetadata(bucket string, includePaths, includeRegex, excludePaths, excludeRegex []string, limits DownloadLimits, logger *logger.Logger) ([]*S3Data, error) {
+	client, err := NewS3ClientFunc(staticCreds)
+	if err != nil {
+		return []*S3Data{}, err
+	}
+	return getS3DataFromMetadataClient(client, bucket, includePaths, includeRegex, excludePaths, excludeRegex, limits, logger)
+}
+
+// getS3DataFromMetadataClient harvests bucket content using the provided client,
+// reading digests from stored checksums.
+func getS3DataFromMetadataClient(client S3API, bucket string, includePaths, includeRegex, excludePaths, excludeRegex []string, limits DownloadLimits, logger *logger.Logger) ([]*S3Data, error) {
+	return getS3DataWithSource(client, metadataDigests(client, bucket, logger), bucket, includePaths, includeRegex, excludePaths, excludeRegex, limits, logger)
+}
+
+// metadataDigests is the source that reads each object's stored SHA256 with a
+// HeadObject.
+func metadataDigests(client S3HeadAPI, bucket string, logger *logger.Logger) s3DigestSource {
+	return s3DigestSource{
+		sha256: func(ctx context.Context, _ string, object s3Object) (string, error) {
+			out, err := client.HeadObject(ctx, &s3.HeadObjectInput{
+				Bucket:       aws.String(bucket),
+				Key:          aws.String(object.key),
+				ChecksumMode: s3Types.ChecksumModeEnabled,
+			})
+			if err != nil {
+				return "", fmt.Errorf("failed to read the checksum of object key [%s]: %w. This needs the "+
+					"s3:GetObject permission, the same one downloading the object needs; an SSE-KMS object "+
+					"also needs kms:GenerateDataKey and kms:Decrypt", object.key, err)
+			}
+			sha256, err := objectChecksumSha256(bucket, object.key, out)
+			if err != nil {
+				return "", err
+			}
+			logger.Debug("object key [%s] -- stored checksum digest: %s", object.key, sha256)
+			return sha256, nil
+		},
+	}
+}
+
+// unusableChecksumError is a problem with one object's checksum rather than
+// the connection, so the fan-out reports every such object in one run instead
+// of stopping at the first.
+type unusableChecksumError struct {
+	msg string
+}
+
+func (e unusableChecksumError) Error() string { return e.msg }
+
+// objectChecksumSha256 converts one HeadObject result into the hex SHA256 of
+// the object's content, or explains why it cannot.
+func objectChecksumSha256(bucket, key string, out *s3.HeadObjectOutput) (string, error) {
+	if out.ChecksumSHA256 == nil || *out.ChecksumSHA256 == "" {
+		return "", unusableChecksumError{fmt.Sprintf("object key [%s] has no SHA256 checksum, so its fingerprint "+
+			"cannot be read from S3 metadata. Upload it with one: aws s3api put-object --bucket %s --key %s "+
+			"--body <file> --checksum-algorithm SHA256; or fingerprint by downloading the objects instead",
+			key, bucket, key)}
+	}
+
+	// A composite checksum hashes the part checksums, not the object. S3 may mark
+	// it with the COMPOSITE type, the "-N" suffix or both, so either is enough.
+	checksum := *out.ChecksumSHA256
+	if out.ChecksumType == s3Types.ChecksumTypeComposite || hasPartCountSuffix(checksum) {
+		return "", unusableChecksumError{fmt.Sprintf("object key [%s] has a multipart (composite) SHA256 checksum "+
+			"[%s], which hashes the part checksums rather than the object content. Collapse it into a single "+
+			"part in place: aws s3api copy-object --checksum-algorithm SHA256 --copy-source %s/%s --bucket %s "+
+			"--key %s; or fingerprint by downloading the objects instead",
+			key, checksum, bucket, key, bucket, key)}
+	}
+
+	sha256, err := decodeBase64Sha256(checksum)
+	if err != nil {
+		return "", unusableChecksumError{fmt.Sprintf("object key [%s] has an SHA256 checksum that cannot be decoded [%s]: %v",
+			key, checksum, err)}
+	}
+	return sha256, nil
+}
+
+// hasPartCountSuffix reports whether checksum ends in a composite's "-N" part
+// count. Any other "-" is left to fail decoding, rather than be reported as a
+// composite that copy-object would not fix.
+func hasPartCountSuffix(checksum string) bool {
+	dash := strings.LastIndex(checksum, "-")
+	if dash <= 0 || dash == len(checksum)-1 {
+		return false
+	}
+	for _, r := range checksum[dash+1:] {
+		if r < '0' || r > '9' {
+			return false
+		}
+	}
+	return true
+}
+
+// combineUnusableChecksumErrors joins the per-object errors into one, listing
+// at most maxReportedS3KeyProblems of them.
+func combineUnusableChecksumErrors(errs []error) error {
+	// The workers finish in any order; sorting keeps the message stable.
+	messages := make([]string, 0, len(errs))
+	for _, err := range errs {
+		messages = append(messages, err.Error())
+	}
+	sort.Strings(messages)
+
+	if len(messages) == 1 {
+		return errors.New(messages[0])
+	}
+	shown, suffix := messages, ""
+	if len(shown) > maxReportedS3KeyProblems {
+		shown = shown[:maxReportedS3KeyProblems]
+		suffix = fmt.Sprintf("\n(and %d more)", len(messages)-maxReportedS3KeyProblems)
+	}
+	return fmt.Errorf("%d objects cannot be fingerprinted from S3 metadata:\n%s%s",
+		len(messages), strings.Join(shown, "\n"), suffix)
+}
diff --git a/internal/aws/s3_metadata_test.go b/internal/aws/s3_metadata_test.go
new file mode 100644
index 000000000..0656ee3d3
--- /dev/null
+++ b/internal/aws/s3_metadata_test.go
@@ -0,0 +1,372 @@
+package aws
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	s3Types "github.com/aws/aws-sdk-go-v2/service/s3/types"
+	"github.com/kosli-dev/cli/internal/logger"
+	"github.com/stretchr/testify/require"
+	"github.com/stretchr/testify/suite"
+)
+
+type S3MetadataTestSuite struct {
+	suite.Suite
+}
+
+// fullObjectChecksums is what S3 stores for objects uploaded single-part with
+// --checksum-algorithm SHA256.
+func fullObjectChecksums(objects map[string][]byte) map[string]FakeS3Checksum {
+	checksums := map[string]FakeS3Checksum{}
+	for key, content := range objects {
+		if strings.HasSuffix(key, "/") {
+			continue // folder markers carry no checksum
+		}
+		checksums[key] = FakeS3Checksum{SHA256: base64Sha256(content), Type: s3Types.ChecksumTypeFullObject}
+	}
+	return checksums
+}
+
+func checksummedBucket(objects map[string][]byte) *FakeS3Client {
+	return &FakeS3Client{Bucket: fakeS3TestBucketName, Objects: objects, Checksums: fullObjectChecksums(objects)}
+}
+
+func snapshotMetadata(t *testing.T, client S3API, excludePaths []string) ([]*S3Data, error) {
+	t.Helper()
+	return getS3DataFromMetadataClient(client, fakeS3TestBucketName, nil, nil, excludePaths, nil,
+		DefaultDownloadLimits, logger.NewStandardLogger())
+}
+
+// The property the feature rests on: stored checksums fingerprint a bucket
+// exactly as downloading does, artifact name and timestamp included.
+func (suite *S3MetadataTestSuite) TestMatchesContentMode() {
+	for _, t := range []struct {
+		name         string
+		objects      map[string][]byte
+		excludePaths []string
+	}{
+		{name: "a single object", objects: map[string][]byte{"README.md": []byte(fakeReadmeBody)}},
+		{name: "a single nested object", objects: map[string][]byte{"dummy/dummy_2/template.yml": []byte(fakeTemplateBody)}},
+		{name: "two objects at the root", objects: map[string][]byte{"README.md": []byte(fakeReadmeBody), "notes.txt": []byte(fakeNotesBody)}},
+		{
+			name: "objects nested under prefixes with folder markers",
+			objects: map[string][]byte{
+				"dir/": nil, "dir/sub/": nil, "dir/sub/x.yml": []byte("x"), "dir/y.txt": []byte("y"), "README.md": []byte("r"),
+			},
+		},
+		{
+			// '.' sorts before '/', so a flat key sort would diverge from the tree walk.
+			name:    "a prefix sharing a name prefix with a sibling object",
+			objects: map[string][]byte{"a.txt": []byte("1"), "a/z": []byte("2"), "a/b/c": []byte("3"), "b": []byte("4")},
+		},
+		{
+			name:    "unusual key shapes",
+			objects: map[string][]byte{"/lead.txt": []byte("u\n"), "a//b": []byte("o\n"), "./c.txt": []byte("t\n"), `d\e.txt`: []byte("n\n")},
+		},
+		{
+			name: "a root .kosli_ignore whose rules exclude objects",
+			objects: map[string][]byte{
+				".kosli_ignore": []byte("logs\n*.tmp\n"), "app.js": []byte("app"), "lib/util.js": []byte("util"),
+				"logs/a.log": []byte("a"), "logs/deep/b.log": []byte("b"), "scratch.tmp": []byte("tmp"),
+			},
+		},
+		{name: "a lone .kosli_ignore", objects: map[string][]byte{".kosli_ignore": []byte("logs\n")}},
+		{name: "a nested .kosli_ignore is an ordinary object", objects: map[string][]byte{"README.md": []byte("r"), "vendor/.kosli_ignore": []byte("README.md\n")}},
+		{
+			name:         "filters apply before either source",
+			objects:      map[string][]byte{"README.md": []byte("r"), "filtered/out.txt": []byte("o"), "keep/in.txt": []byte("i")},
+			excludePaths: []string{"filtered/"},
+		},
+	} {
+		suite.Run(t.name, func() {
+			content, err := getS3DataFromClient(checksummedBucket(t.objects), fakeS3TestBucketName, nil, nil, t.excludePaths, nil,
+				DefaultDownloadLimits, logger.NewStandardLogger())
+			require.NoError(suite.T(), err)
+
+			metadata, err := snapshotMetadata(suite.T(), checksummedBucket(t.objects), t.excludePaths)
+			require.NoError(suite.T(), err)
+
+			require.Equal(suite.T(), content, metadata, "metadata mode must produce exactly what content mode produces")
+		})
+	}
+}
+
+// Existing snapshots on the server carry these fingerprints, so metadata mode
+// must reproduce them from checksums alone.
+func (suite *S3MetadataTestSuite) TestPinnedFingerprints() {
+	for _, t := range []struct {
+		name             string
+		objects          map[string][]byte
+		wantArtifactName string
+		wantFingerprint  string
+	}{
+		{
+			name:             "unusual key shapes fold as filepath.Join folded them",
+			objects:          map[string][]byte{"/lead.txt": []byte("u\n"), "a//b": []byte("o\n"), "./c.txt": []byte("t\n"), `d\e.txt`: []byte("n\n")},
+			wantArtifactName: fakeS3TestBucketName,
+			wantFingerprint:  "27e2d8aa07677b7818b8cf101b45c928aa8fbf7d17a8c8efc84469e24a106ec3",
+		},
+		{
+			name:             "a dot sorts before a slash",
+			objects:          map[string][]byte{"a.txt": []byte("1"), "a/z": []byte("2"), "a/b/c": []byte("3"), "b": []byte("4")},
+			wantArtifactName: fakeS3TestBucketName,
+			wantFingerprint:  "aaddb8f3e299e12316d42fa9ac36d4ed7d38c34e7ec002239269c86a033bb9fd",
+		},
+		{
+			name:             "nested prefixes with folder markers",
+			objects:          map[string][]byte{"dir/": nil, "dir/sub/": nil, "dir/sub/x.yml": []byte("x"), "dir/y.txt": []byte("y"), "README.md": []byte("r")},
+			wantArtifactName: fakeS3TestBucketName,
+			wantFingerprint:  "c26910cdb177dde3c6493d18ba1e916b04715cdfc535e4552b5f9831590e933a",
+		},
+		{
+			name:             "a single object is the file, named by its base name",
+			objects:          map[string][]byte{"only/one/file.bin": []byte("solo")},
+			wantArtifactName: "file.bin",
+			wantFingerprint:  "5364f2f2fc4f54e9d47ad29cfb08ef430c8153394bf2a0dff5cbe77a0ffef861",
+		},
+	} {
+		suite.Run(t.name, func() {
+			data, err := snapshotMetadata(suite.T(), checksummedBucket(t.objects), nil)
+			require.NoError(suite.T(), err)
+			require.Len(suite.T(), data, 1)
+			require.Equal(suite.T(), map[string]string{t.wantArtifactName: t.wantFingerprint}, data[0].Digests)
+		})
+	}
+}
+
+// countingHeader records HeadObject calls and their peak overlap, and can delay
+// each one so the overlap is real.
+type countingHeader struct {
+	S3API
+	delay time.Duration
+
+	mu                  sync.Mutex
+	calls               map[string]int
+	inFlight            int
+	maxInFlight         int
+	checksumModeMissing int
+}
+
+func (h *countingHeader) HeadObject(ctx context.Context, params *s3.HeadObjectInput, optFns ...func(*s3.Options)) (*s3.HeadObjectOutput, error) {
+	h.mu.Lock()
+	if h.calls == nil {
+		h.calls = map[string]int{}
+	}
+	h.calls[*params.Key]++
+	if params.ChecksumMode != s3Types.ChecksumModeEnabled {
+		h.checksumModeMissing++
+	}
+	h.inFlight++
+	h.maxInFlight = max(h.maxInFlight, h.inFlight)
+	h.mu.Unlock()
+	defer func() {
+		h.mu.Lock()
+		h.inFlight--
+		h.mu.Unlock()
+	}()
+	time.Sleep(h.delay)
+	return h.S3API.HeadObject(ctx, params, optFns...)
+}
+
+func (suite *S3MetadataTestSuite) TestReadsMetadataNotContent() {
+	objects := map[string][]byte{
+		".kosli_ignore": []byte("logs\n"), "app.js": []byte("app"), "lib/util.js": []byte("util"),
+		"logs/a.log": []byte("a"), "lib/": nil, "filtered/out.txt": []byte("out"),
+	}
+	checksums := fullObjectChecksums(objects)
+	// Neither the ignore file (downloaded) nor an excluded object (never
+	// fetched) needs a stored checksum.
+	delete(checksums, ".kosli_ignore")
+	delete(checksums, "logs/a.log")
+	delete(checksums, "filtered/out.txt")
+	downloads := &recordingDownloader{S3API: &FakeS3Client{Bucket: fakeS3TestBucketName, Objects: objects, Checksums: checksums}}
+	client := &countingHeader{S3API: downloads}
+
+	data, err := getS3DataFromMetadataClient(client, fakeS3TestBucketName, nil, nil, []string{"filtered/"}, nil,
+		DefaultDownloadLimits, logger.NewStandardLogger())
+	require.NoError(suite.T(), err)
+	require.Len(suite.T(), data, 1)
+	require.Equal(suite.T(), []string{".kosli_ignore"}, downloads.downloadedKeys(), "only the ignore file may be downloaded")
+	require.Equal(suite.T(), map[string]int{"app.js": 1, "lib/util.js": 1}, client.calls, "exactly one HeadObject per contributing object")
+	require.Zero(suite.T(), client.checksumModeMissing, "every HeadObject must ask for the stored checksum")
+}
+
+// A lone ignore file takes the single-file branch, not the ignore pass, and must
+// still be downloaded rather than read through HeadObject.
+func (suite *S3MetadataTestSuite) TestALoneIgnoreFileNeedsNoChecksum() {
+	objects := map[string][]byte{".kosli_ignore": []byte("logs\n")}
+	downloads := &recordingDownloader{S3API: &FakeS3Client{Bucket: fakeS3TestBucketName, Objects: objects}}
+	client := &countingHeader{S3API: downloads}
+
+	metadata, err := getS3DataFromMetadataClient(client, fakeS3TestBucketName, nil, nil, nil, nil,
+		DefaultDownloadLimits, logger.NewStandardLogger())
+	require.NoError(suite.T(), err)
+	require.Equal(suite.T(), []string{".kosli_ignore"}, downloads.downloadedKeys())
+	require.Empty(suite.T(), client.calls, "the ignore file must not be read through HeadObject")
+
+	content, err := getS3DataFromClient(&FakeS3Client{Bucket: fakeS3TestBucketName, Objects: objects}, fakeS3TestBucketName,
+		nil, nil, nil, nil, DefaultDownloadLimits, logger.NewStandardLogger())
+	require.NoError(suite.T(), err)
+	require.Equal(suite.T(), content, metadata)
+}
+
+// A one-byte budget would serialise downloads; HEADs use no disk and must still overlap.
+func (suite *S3MetadataTestSuite) TestHeadsAreBoundedByConcurrencyNotBytes() {
+	objects := map[string][]byte{}
+	for i := 0; i < 40; i++ {
+		objects[fmt.Sprintf("dir%d/object-%03d.bin", i%4, i)] = []byte(fmt.Sprintf("%08d", i))
+	}
+	client := &countingHeader{S3API: checksummedBucket(objects), delay: 5 * time.Millisecond}
+
+	_, err := getS3DataFromMetadataClient(client, fakeS3TestBucketName, nil, nil, nil, nil,
+		DownloadLimits{Concurrency: 4, BytesInFlight: 1}, logger.NewStandardLogger())
+	require.NoError(suite.T(), err)
+	require.Len(suite.T(), client.calls, 40)
+	require.LessOrEqual(suite.T(), client.maxInFlight, 4)
+	require.GreaterOrEqual(suite.T(), client.maxInFlight, 2, "HEADs must actually overlap for the bound to be tested")
+}
+
+func (suite *S3MetadataTestSuite) TestErrors() {
+	readme := []byte(fakeReadmeBody)
+	for _, t := range []struct {
+		name       string
+		objects    map[string][]byte
+		checksums  map[string]FakeS3Checksum
+		listErr    error
+		headErr    error
+		wantErrMsg []string
+	}{
+		{
+			// Pinned in full because the SnapshotS3TestSuite golden repeats it.
+			name:    "an object with no stored checksum",
+			objects: map[string][]byte{"README.md": readme},
+			wantErrMsg: []string{"object key [README.md] has no SHA256 checksum, so its fingerprint cannot be read from S3 " +
+				"metadata. Upload it with one: aws s3api put-object --bucket " + fakeS3TestBucketName + " --key README.md " +
+				"--body <file> --checksum-algorithm SHA256; or fingerprint by downloading the objects instead"},
+		},
+		{
+			name:       "a composite multipart checksum",
+			objects:    map[string][]byte{"README.md": readme},
+			checksums:  map[string]FakeS3Checksum{"README.md": {SHA256: base64Sha256(readme) + "-4", Type: s3Types.ChecksumTypeComposite}},
+			wantErrMsg: []string{"multipart (composite) SHA256 checksum", "aws s3api copy-object --checksum-algorithm SHA256"},
+		},
+		{
+			name:       "a composite checksum is rejected on its type even without a suffix",
+			objects:    map[string][]byte{"README.md": readme},
+			checksums:  map[string]FakeS3Checksum{"README.md": {SHA256: base64Sha256(readme), Type: s3Types.ChecksumTypeComposite}},
+			wantErrMsg: []string{"multipart (composite) SHA256 checksum"},
+		},
+		{
+			name:       "a checksum that is not valid Base64",
+			objects:    map[string][]byte{"README.md": readme},
+			checksums:  map[string]FakeS3Checksum{"README.md": {SHA256: "not base64!", Type: s3Types.ChecksumTypeFullObject}},
+			wantErrMsg: []string{"object key [README.md]", "not base64!"},
+		},
+		{
+			name:       "a metadata request failure names the permission it needs",
+			objects:    map[string][]byte{"README.md": readme},
+			checksums:  fullObjectChecksums(map[string][]byte{"README.md": readme}),
+			headErr:    errInjected,
+			wantErrMsg: []string{"injected error", "s3:GetObject"},
+		},
+		{
+			name:       "a listing error propagates",
+			objects:    map[string][]byte{"README.md": readme},
+			checksums:  fullObjectChecksums(map[string][]byte{"README.md": readme}),
+			listErr:    errInjected,
+			wantErrMsg: []string{"injected error"},
+		},
+		{
+			name:       "an empty bucket keeps the content-mode message",
+			objects:    map[string][]byte{"dir/": nil},
+			wantErrMsg: []string{"no matching file or dirs in bucket: [" + fakeS3TestBucketName + "]"},
+		},
+	} {
+		suite.Run(t.name, func() {
+			client := &FakeS3Client{Bucket: fakeS3TestBucketName, Objects: t.objects, Checksums: t.checksums,
+				ListObjectsV2Err: t.listErr, HeadObjectErr: t.headErr}
+			_, err := snapshotMetadata(suite.T(), client, nil)
+			require.Error(suite.T(), err)
+			for _, want := range t.wantErrMsg {
+				require.Contains(suite.T(), err.Error(), want)
+			}
+		})
+	}
+}
+
+func (suite *S3MetadataTestSuite) TestCompositeDetectionKeysOffThePartCountSuffix() {
+	readme := []byte(fakeReadmeBody)
+	for _, t := range []struct {
+		name          string
+		checksum      FakeS3Checksum
+		wantComposite bool
+	}{
+		{name: "a part-count suffix with no type", checksum: FakeS3Checksum{SHA256: base64Sha256(readme) + "-4"}, wantComposite: true},
+		{name: "a COMPOSITE type with no suffix", checksum: FakeS3Checksum{SHA256: base64Sha256(readme), Type: s3Types.ChecksumTypeComposite}, wantComposite: true},
+		{name: "a dash that is not a suffix", checksum: FakeS3Checksum{SHA256: "S0tL-S0tLS0_LS0tLS0tLS0tLS0tLS0tLS0tLS0tLS0=", Type: s3Types.ChecksumTypeFullObject}},
+		{name: "a dash followed by more than digits", checksum: FakeS3Checksum{SHA256: base64Sha256(readme) + "-4a"}},
+		{name: "a trailing dash", checksum: FakeS3Checksum{SHA256: base64Sha256(readme) + "-"}},
+	} {
+		suite.Run(t.name, func() {
+			client := &FakeS3Client{Bucket: fakeS3TestBucketName, Objects: map[string][]byte{"README.md": readme},
+				Checksums: map[string]FakeS3Checksum{"README.md": t.checksum}}
+			_, err := snapshotMetadata(suite.T(), client, nil)
+			require.Error(suite.T(), err)
+			if t.wantComposite {
+				require.Contains(suite.T(), err.Error(), "multipart (composite)")
+				return
+			}
+			require.Contains(suite.T(), err.Error(), "cannot be decoded")
+			require.NotContains(suite.T(), err.Error(), "copy-object", "an undecodable checksum is not fixed by collapsing parts")
+		})
+	}
+}
+
+func (suite *S3MetadataTestSuite) TestReportsEveryUnusableObjectTogether() {
+	objects := map[string][]byte{}
+	for i := 0; i < 6; i++ {
+		objects[fmt.Sprintf("object-%d.txt", i)] = []byte(fmt.Sprintf("content %d", i))
+	}
+	checksums := fullObjectChecksums(objects)
+	delete(checksums, "object-0.txt")
+	delete(checksums, "object-2.txt")
+	checksums["object-4.txt"] = FakeS3Checksum{SHA256: checksums["object-4.txt"].SHA256 + "-3", Type: s3Types.ChecksumTypeComposite}
+
+	_, err := snapshotMetadata(suite.T(), &FakeS3Client{Bucket: fakeS3TestBucketName, Objects: objects, Checksums: checksums}, nil)
+	require.Error(suite.T(), err)
+	for _, key := range []string{"object-0.txt", "object-2.txt", "object-4.txt"} {
+		require.Contains(suite.T(), err.Error(), "["+key+"]")
+	}
+	for _, key := range []string{"object-1.txt", "object-3.txt", "object-5.txt"} {
+		require.NotContains(suite.T(), err.Error(), "["+key+"]", "objects that are fine must not be named")
+	}
+	require.Contains(suite.T(), err.Error(), "no SHA256 checksum")
+	require.Contains(suite.T(), err.Error(), "multipart (composite)")
+
+	suite.Run("the list is capped", func() {
+		many := map[string][]byte{}
+		for i := 0; i < 40; i++ {
+			many[fmt.Sprintf("object-%02d.txt", i)] = []byte("x")
+		}
+		_, err := snapshotMetadata(suite.T(), &FakeS3Client{Bucket: fakeS3TestBucketName, Objects: many}, nil)
+		require.Error(suite.T(), err)
+		require.Contains(suite.T(), err.Error(), "(and 30 more)")
+		require.Equal(suite.T(), maxReportedS3KeyProblems, strings.Count(err.Error(), "has no SHA256 checksum"))
+	})
+
+	suite.Run("a transport error still stops the run", func() {
+		client := &FakeS3Client{Bucket: fakeS3TestBucketName, Objects: objects, Checksums: checksums, HeadObjectErr: errInjected}
+		_, err := snapshotMetadata(suite.T(), client, nil)
+		require.ErrorIs(suite.T(), err, errInjected)
+		require.NotContains(suite.T(), err.Error(), "no SHA256 checksum", "a connection failure is not a per-object report")
+	})
+}
+
+func TestS3MetadataTestSuite(t *testing.T) {
+	suite.Run(t, new(S3MetadataTestSuite))
+}