fix(agent): default S3 RequestChecksumCalculation to WhenRequired

docs/concepts.md

@@ -449,7 +449,7 @@ Two surfaces, one agent. The operator image doubles as a snapshot agent: `main.g
 An `EtcdSnapshot` is a one-shot record. The controller resolves `spec.clusterRef`, builds a Job (owned by the `EtcdSnapshot`, with `ttlSecondsAfterFinished` so it self-GCs, `automountServiceAccountToken: false`, and a restricted security context), and tracks `status.phase` through `Pending → Started → Complete | Failed`. The Job's Pod runs the snapshot agent, which:
 
 1. dials the cluster's client Service — TLS material (server CA + operator-client cert) is mounted from the same Secrets the operator dials with, and when `status.authEnabled` is latched the agent authenticates as `root` using the password from `spec.auth.rootCredentialsSecretRef`;
-2. streams `clientv3` `Maintenance.Snapshot` to a local file (PVC destination) or a temp file it then multipart-uploads to S3, hashing (sha256) and counting bytes as it goes;
+2. streams `clientv3` `Maintenance.Snapshot` to a local file (PVC destination) or directly to S3 via the transfer manager — no local staging, multipart above the manager's 5 MiB part size — hashing (sha256) and counting bytes as it goes;
 3. prints a marker line — `snapshot uploaded: uri="..." size=N sha256=<hex>` — that the controller scans out of the Pod log (via an uncached `APIReader` to find the Pod and the typed Clientset to read its log) to populate `status.artifact{uri,sizeBytes,checksum}`.
 
 Why parse a log line rather than have the agent write status? The agent has no Kubernetes API access by design (`automountServiceAccountToken: false`), so the controller — which does — is the one that records the result. The marker is the agent→controller channel. Terminal phases are sticky: a `Complete`/`Failed` snapshot is a historical record and never re-runs.

docs/operations.md

@@ -267,6 +267,8 @@ For a PVC destination use `destination.pvc.{claimName,subPath}` instead of `s3`.
 > **Object names must be unique.** The stored object is keyed by the `EtcdSnapshot` *name* (`<key-prefix>/<name>.db` for S3, `<name>.db` on a PVC). The agent **refuses to overwrite an existing snapshot it did not write** — for **both** S3 and PVC destinations — so a snapshot whose key/path already exists fails rather than silently clobbering an earlier snapshot. Give each snapshot a distinct name (the date-stamped CronJob pattern above does this) or a distinct destination `key`/`subPath`. (A *retry* of the same `EtcdSnapshot` is exempt: each snapshot is stamped with the snapshot's UID — S3 object metadata, or a sibling `<name>.db.uid` file on a PVC — so the agent recognizes its own prior write and the retry is idempotent.) The CRD's "immutability" is about the object: it does not version snapshots, so reusing a name after deleting the object replaces it.
 >
 > **S3 credentials need `s3:ListBucket` (or equivalent), not just `s3:GetObject`/`PutObject`.** The overwrite guard does a `HeadObject` on the target key; with `GetObject` but no `ListBucket`, S3 (and some MinIO/Ceph policies) returns **403 AccessDenied** for a *missing* key instead of 404. The agent cannot distinguish that from a real permission problem, so it **fails closed** (refuses the snapshot) rather than risk an overwrite. Grant the snapshot credentials `ListBucket` on the bucket so a HEAD on a missing key returns 404.
+>
+> **`InvalidArgument: x-amz-content-sha256 must be UNSIGNED-PAYLOAD, STREAMING-AWS4-HMAC-SHA256-PAYLOAD or a valid sha256 value`.** Some non-AWS, S3-compatible backends (Ceph RGW confirmed; some MinIO / Cloudflare R2 versions) reject the flexible-checksum trailer that recent `aws-sdk-go-v2` attaches to uploads by default. The agent pins its S3 request-checksum policy to *when-required* on both the client and the multipart transfer manager, so it does not send that trailer and uploads succeed on those backends. If you still hit this error, your agent image predates that fix — upgrade the operator image.
 
 If a snapshot ends up `Failed`, inspect the agent Job's Pod logs (the Job is `<snapshot-name>-snapshot` and is GC'd a few minutes after finishing via `ttlSecondsAfterFinished`):
 

internal/agent/agent.go

@@ -148,10 +148,38 @@ func (d destination) localPath(name string) string {
 	return base + "/" + name + ".db"
 }
 
+// s3RequestChecksumCalculation is the S3 request-checksum policy applied to every
+// S3 client the agent builds (s3Client). It has to be set in TWO independent
+// places, because the snapshot upload touches two SDK layers that each carry their
+// own copy of this knob and do NOT share it:
+//
+//   - the s3.Client option (s3Client, below) — covers the single-part PutObject
+//     the transfer manager issues for a sub-5-MiB body (a small or freshly
+//     bootstrapped cluster), plus HeadObject;
+//   - the transfer manager's own Uploader.RequestChecksumCalculation
+//     (newSnapshotUploader in snapshot.go) — covers the multipart parts the
+//     manager issues for any body over its 5-MiB part size, i.e. every snapshot
+//     of a non-trivial cluster. manager.NewUploader hardcodes this field to
+//     WhenSupported and never reads the client option, so setting only the client
+//     option leaves the multipart path broken.
+//
+// Both sites are load-bearing and MUST agree; deriving them from this one
+// constant is what keeps them from drifting. Delete either and the RGW failure
+// returns — for the multipart case, invisibly.
+//
+// Why WhenRequired: since early 2025 aws-sdk-go-v2 defaults to WhenSupported,
+// which stamps a CRC32 on every PutObject/UploadPart — over HTTPS that rides as a
+// STREAMING-UNSIGNED-PAYLOAD-TRAILER. Ceph RGW (the confirmed case) rejects the
+// accompanying header with "InvalidArgument: x-amz-content-sha256 must be
+// UNSIGNED-PAYLOAD, STREAMING-AWS4-HMAC-SHA256-PAYLOAD or a valid sha256 value".
+// WhenRequired attaches a checksum only when the operation mandates one (none for
+// a plain PutObject or UploadPart) and is equally accepted by real AWS S3.
+const s3RequestChecksumCalculation = aws.RequestChecksumCalculationWhenRequired
+
 // s3Client builds an S3 client honoring a custom endpoint, region and
 // path-style addressing. Credentials come from AWS_ACCESS_KEY_ID /
 // AWS_SECRET_ACCESS_KEY in the environment (loaded by the default config
-// chain).
+// chain). Used by BOTH the upload (snapshot) and download (restore) paths.
 func (d destination) s3Client(ctx context.Context) (*s3.Client, error) {
 	region := d.s3Region
 	if region == "" {
@@ -166,6 +194,21 @@ func (d destination) s3Client(ctx context.Context) (*s3.Client, error) {
 			o.BaseEndpoint = aws.String(d.s3Endpoint)
 		}
 		o.UsePathStyle = d.s3PathStyle
+		// One of the two sites that must pin this; see
+		// s3RequestChecksumCalculation for the full rationale and
+		// newSnapshotUploader for the other. Pinned unconditionally because
+		// LoadDefaultConfig would otherwise take it from
+		// AWS_REQUEST_CHECKSUM_CALCULATION, and the agent's environment is a closed
+		// list built by the controller — nothing can set that var on the Pod.
+		o.RequestChecksumCalculation = s3RequestChecksumCalculation
+		// ResponseChecksumValidation is deliberately left at the SDK default
+		// (WhenSupported): it validates only a checksum the backend actually
+		// returns and skips composite (multipart) ones, so it cannot break an
+		// S3-compatible backend. It is not an integrity guarantee for restore
+		// either — WhenRequired above means the agent stores no checksum, so for
+		// objects it wrote there is nothing to validate and the setting is inert.
+		// Left alone because there is no reason to change it, not because it
+		// protects anything.
 	}), nil
 }
 

internal/agent/agent_test.go

@@ -10,7 +10,13 @@ You may obtain a copy of the License at
 
 package agent
 
-import "testing"
+import (
+	"context"
+	"os"
+	"testing"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+)
 
 func TestObjectKey(t *testing.T) {
 	cases := []struct {
@@ -31,6 +37,63 @@ func TestObjectKey(t *testing.T) {
 	}
 }
 
+// hermeticAWSEnv isolates the aws-sdk-go-v2 default config chain from the host:
+// LoadDefaultConfig otherwise reads ~/.aws/config and AWS_PROFILE, so a runner
+// whose AWS_PROFILE points at a missing profile would fail these tests for
+// reasons unrelated to the assertion. It also pins static credentials and a
+// region so signing never reaches the EC2/ECS metadata endpoint.
+func hermeticAWSEnv(t *testing.T) {
+	t.Helper()
+	t.Setenv("AWS_CONFIG_FILE", os.DevNull)
+	t.Setenv("AWS_SHARED_CREDENTIALS_FILE", os.DevNull)
+	t.Setenv("AWS_PROFILE", "")
+	t.Setenv("AWS_ACCESS_KEY_ID", "test")
+	t.Setenv("AWS_SECRET_ACCESS_KEY", "test")
+	t.Setenv("AWS_REGION", "us-east-1")
+}
+
+// TestS3ClientChecksumWhenRequired is a unit check on the s3.Client OPTION only.
+// It does NOT exercise the path that actually broke: the transfer manager carries
+// its own RequestChecksumCalculation and overrides this on the multipart upload.
+// The end-to-end contract — that a real (>5 MiB) multipart snapshot upload emits
+// no checksum trailer — lives in TestUploadS3StreamMultipartNoChecksumTrailer
+// (snapshot_test.go). Keep both: this pins the client/HeadObject/single-part
+// knob, that one pins multipart.
+func TestS3ClientChecksumWhenRequired(t *testing.T) {
+	hermeticAWSEnv(t)
+	d := destination{
+		kind:        "s3",
+		s3Endpoint:  "https://s3.example.internal",
+		s3PathStyle: true,
+	}
+	c, err := d.s3Client(context.Background())
+	if err != nil {
+		t.Fatalf("s3Client: %v", err)
+	}
+	o := c.Options()
+	// Non-AWS S3-compatible backends (Ceph RGW, some MinIO/R2) reject the
+	// aws-sdk-go-v2 default (WhenSupported) flexible checksum, so the agent must
+	// only add a checksum when the operation requires one.
+	if o.RequestChecksumCalculation != aws.RequestChecksumCalculationWhenRequired {
+		t.Errorf("RequestChecksumCalculation = %v, want WhenRequired (%v)",
+			o.RequestChecksumCalculation, aws.RequestChecksumCalculationWhenRequired)
+	}
+	// The response side stays at the SDK default (WhenSupported): it validates only
+	// a checksum the backend actually returns and skips composite (multipart) ones,
+	// so it cannot break an S3-compatible backend and needs no pinning. This asserts
+	// we do NOT pin it, so a future "symmetry" change has to justify itself.
+	if o.ResponseChecksumValidation != aws.ResponseChecksumValidationWhenSupported {
+		t.Errorf("ResponseChecksumValidation = %v, want the SDK default WhenSupported (%v)",
+			o.ResponseChecksumValidation, aws.ResponseChecksumValidationWhenSupported)
+	}
+	if !o.UsePathStyle {
+		t.Error("UsePathStyle = false, want true")
+	}
+	if o.BaseEndpoint == nil || *o.BaseEndpoint != d.s3Endpoint {
+		t.Errorf("BaseEndpoint = %v, want %q", o.BaseEndpoint, d.s3Endpoint)
+	}
+}
+
 func TestURI(t *testing.T) {
 	s3 := destination{kind: "s3", s3Bucket: "bucket", s3Key: "pre"}
 	if got, want := s3.uri("snap"), "s3://bucket/pre/snap.db"; got != want {

internal/agent/snapshot.go

@@ -237,13 +237,24 @@ func uploadS3Stream(ctx context.Context, dest destination, key string, body io.R
 	if uid != "" {
 		metadata = map[string]string{snapshotUIDMetaKey: uid} // stamp ownership so a retry recognizes its own object
 	}
-	return uploadStreamHashed(ctx, manager.NewUploader(client), &s3.PutObjectInput{
+	return uploadStreamHashed(ctx, newSnapshotUploader(client), &s3.PutObjectInput{
 		Bucket:   aws.String(dest.s3Bucket),
 		Key:      aws.String(key),
 		Metadata: metadata,
 	}, body)
 }
 
+// newSnapshotUploader builds the S3 transfer manager for snapshot uploads.
+//
+// manager.Uploader carries its OWN RequestChecksumCalculation and never consults
+// the s3.Client option, so the multipart path needs the same pin the client got.
+// See s3RequestChecksumCalculation for why, and why both sites are load-bearing.
+func newSnapshotUploader(client *s3.Client) *manager.Uploader {
+	return manager.NewUploader(client, func(u *manager.Uploader) {
+		u.RequestChecksumCalculation = s3RequestChecksumCalculation
+	})
+}
+
 // s3Uploader abstracts manager.Uploader.Upload so uploadStreamHashed is testable
 // without a live S3 endpoint.
 type s3Uploader interface {

internal/agent/snapshot_test.go

@@ -11,15 +11,20 @@ You may obtain a copy of the License at
 package agent
 
 import (
+	"bytes"
 	"context"
 	"crypto/sha256"
 	"encoding/hex"
+	"encoding/pem"
 	"errors"
 	"fmt"
 	"io"
+	"net/http"
+	"net/http/httptest"
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
@@ -81,6 +86,175 @@ func TestUploadStreamHashed_UploadError(t *testing.T) {
 	}
 }
 
+// capturedReq records the checksum-related surface of one request the fake S3
+// endpoint received, so a test can assert no flexible-checksum trailer was sent.
+type capturedReq struct{ method, sha, trailer, algo, partNumber string }
+
+// startS3CaptureServer stands up a TLS httptest server that speaks just enough of
+// the S3 protocol for uploadS3Stream to run against — HeadObject, single-part
+// PutObject, and the multipart initiate/part/complete trio — trusts its cert
+// through the SDK default config chain (AWS_CA_BUNDLE, which LoadDefaultConfig
+// wires into the real client, so s3Client and newSnapshotUploader run exactly as
+// in production), and records every request's checksum headers. The returned func
+// snapshots the captured requests.
+func startS3CaptureServer(t *testing.T) (endpoint string, captured func() []capturedReq) {
+	t.Helper()
+
+	var mu sync.Mutex
+	var reqs []capturedReq
+
+	srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = io.Copy(io.Discard, r.Body)
+		q := r.URL.Query()
+		mu.Lock()
+		reqs = append(reqs, capturedReq{
+			method:     r.Method,
+			sha:        r.Header.Get("X-Amz-Content-Sha256"),
+			trailer:    r.Header.Get("X-Amz-Trailer"),
+			algo:       r.Header.Get("X-Amz-Sdk-Checksum-Algorithm"),
+			partNumber: q.Get("partNumber"),
+		})
+		mu.Unlock()
+
+		switch {
+		case r.Method == http.MethodHead:
+			// ensureObjectAbsent's HeadObject: report absent so the upload proceeds.
+			w.WriteHeader(http.StatusNotFound)
+		case r.Method == http.MethodPost && q.Has("uploads"):
+			_, _ = io.WriteString(w, `<?xml version="1.0" encoding="UTF-8"?>`+
+				`<InitiateMultipartUploadResult><Bucket>b</Bucket><Key>k</Key>`+
+				`<UploadId>test-upload-id</UploadId></InitiateMultipartUploadResult>`)
+		case r.Method == http.MethodPut && q.Has("partNumber"):
+			w.Header().Set("ETag", `"etag-`+q.Get("partNumber")+`"`)
+			w.WriteHeader(http.StatusOK)
+		case r.Method == http.MethodPost && q.Get("uploadId") != "":
+			_, _ = io.WriteString(w, `<?xml version="1.0" encoding="UTF-8"?>`+
+				`<CompleteMultipartUploadResult><Bucket>b</Bucket><Key>k</Key>`+
+				`<ETag>"final-etag"</ETag></CompleteMultipartUploadResult>`)
+		case r.Method == http.MethodPut:
+			// single-part PutObject (body under the manager's 5 MiB part size)
+			w.Header().Set("ETag", `"single-etag"`)
+			w.WriteHeader(http.StatusOK)
+		default:
+			w.WriteHeader(http.StatusOK)
+		}
+	}))
+	t.Cleanup(srv.Close)
+
+	caFile := filepath.Join(t.TempDir(), "ca.pem")
+	certPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: srv.Certificate().Raw})
+	if err := os.WriteFile(caFile, certPEM, 0o600); err != nil {
+		t.Fatalf("write CA bundle: %v", err)
+	}
+	t.Setenv("AWS_CA_BUNDLE", caFile)
+
+	return srv.URL, func() []capturedReq {
+		mu.Lock()
+		defer mu.Unlock()
+		out := make([]capturedReq, len(reqs))
+		copy(out, reqs)
+		return out
+	}
+}
+
+// assertNoChecksumTrailer fails if a request carries any of the flexible-checksum
+// surface that Ceph RGW rejects: the streaming-trailer content-sha256, an
+// x-amz-trailer, or an x-amz-sdk-checksum-algorithm.
+func assertNoChecksumTrailer(t *testing.T, rq capturedReq) {
+	t.Helper()
+	if rq.sha == "STREAMING-UNSIGNED-PAYLOAD-TRAILER" {
+		t.Errorf("%s carried x-amz-content-sha256=STREAMING-UNSIGNED-PAYLOAD-TRAILER — the checksum trailer Ceph RGW rejects", rq.method)
+	}
+	if rq.trailer != "" {
+		t.Errorf("%s carried x-amz-trailer=%q, want none", rq.method, rq.trailer)
+	}
+	if rq.algo != "" {
+		t.Errorf("%s carried x-amz-sdk-checksum-algorithm=%q, want none", rq.method, rq.algo)
+	}
+}
+
+// TestUploadS3StreamMultipartNoChecksumTrailer is the guard for the actual
+// failure mode. A snapshot of a non-trivial cluster exceeds the transfer manager's
+// 5 MiB part size, so the upload takes the MULTIPART branch, where manager.Uploader's
+// own RequestChecksumCalculation (NOT the s3.Client option) decides whether a CRC32
+// trailer is stamped on each UploadPart. Without newSnapshotUploader pinning it
+// to WhenRequired, every part rides `x-amz-content-sha256:
+// STREAMING-UNSIGNED-PAYLOAD-TRAILER` + `x-amz-trailer: x-amz-checksum-crc32` —
+// exactly the header Ceph RGW rejects.
+//
+// It fails if the uploader option is dropped; a check on s3.Options alone (see
+// agent_test.go) would stay green while this real path stayed broken.
+func TestUploadS3StreamMultipartNoChecksumTrailer(t *testing.T) {
+	hermeticAWSEnv(t)
+	endpoint, captured := startS3CaptureServer(t)
+
+	const bodySize = 12 << 20 // 12 MiB > 5 MiB part size ⇒ three parts ⇒ multipart branch
+	dest := destination{kind: "s3", s3Endpoint: endpoint, s3Bucket: "b", s3PathStyle: true}
+
+	size, sum, err := uploadS3Stream(context.Background(), dest, "k", bytes.NewReader(make([]byte, bodySize)), "test-uid")
+	if err != nil {
+		t.Fatalf("uploadS3Stream: %v", err)
+	}
+	if size != bodySize {
+		t.Errorf("streamed size = %d, want %d", size, bodySize)
+	}
+	if sum == "" {
+		t.Error("empty sha256 digest")
+	}
+
+	var sawUploadPart bool
+	for _, rq := range captured() {
+		if rq.method == http.MethodPut && rq.partNumber != "" {
+			sawUploadPart = true
+		}
+		assertNoChecksumTrailer(t, rq)
+	}
+	if !sawUploadPart {
+		t.Fatal("no UploadPart (PUT with partNumber) seen — the upload did not take the multipart branch, so this test is not exercising the path it guards")
+	}
+}
+
+// TestUploadS3StreamSinglePartNoChecksumTrailer guards the OTHER upload branch.
+// A snapshot of a small or freshly bootstrapped cluster is well under the transfer
+// manager's 5 MiB part size, so the manager issues a single-part PutObject rather
+// than multipart. There the s3.Client option (not the uploader's) is the only
+// thing keeping the CRC32 trailer off — so this pins the client-side setting
+// end-to-end. A struct-field check on s3.Options (agent_test.go) stays green even
+// when the real request carries the trailer — that gap is how a client-only pin
+// can look correct while every upload is still broken on the wire.
+func TestUploadS3StreamSinglePartNoChecksumTrailer(t *testing.T) {
+	hermeticAWSEnv(t)
+	endpoint, captured := startS3CaptureServer(t)
+
+	const bodySize = 1 << 20 // 1 MiB < 5 MiB part size ⇒ single-part PutObject branch
+	dest := destination{kind: "s3", s3Endpoint: endpoint, s3Bucket: "b", s3PathStyle: true}
+
+	size, sum, err := uploadS3Stream(context.Background(), dest, "k", bytes.NewReader(make([]byte, bodySize)), "test-uid")
+	if err != nil {
+		t.Fatalf("uploadS3Stream: %v", err)
+	}
+	if size != bodySize {
+		t.Errorf("streamed size = %d, want %d", size, bodySize)
+	}
+	if sum == "" {
+		t.Error("empty sha256 digest")
+	}
+
+	var putObjects int
+	for _, rq := range captured() {
+		if rq.method == http.MethodPut && rq.partNumber == "" {
+			putObjects++
+		}
+		if rq.partNumber != "" {
+			t.Errorf("saw a multipart UploadPart (partNumber=%q); a %d-byte body must take the single-part branch this test guards", rq.partNumber, bodySize)
+		}
+		assertNoChecksumTrailer(t, rq)
+	}
+	if putObjects != 1 {
+		t.Fatalf("saw %d single-part PutObject PUTs, want exactly 1 — the body no longer takes the single-part branch, so this test is not exercising the path it guards", putObjects)
+	}
+}
+
 type fakeHead struct {
 	out *s3.HeadObjectOutput
 	err error