From ee72a40ce1f8c0cd58de00e815241c0a60b4fe94 Mon Sep 17 00:00:00 2001 From: Rhythm Garg Date: Thu, 30 Jul 2026 10:35:09 +0000 Subject: [PATCH] fix(cloud-backups): checksum Azure uploads server-side (CRC64), at every payload size The S3 upload path opts into server-side integrity verification, so a completed upload means the stored bytes are the bytes we sent. The Azure path asked for nothing, so Azure received NO checksum at all and a "successful" upload proved only that bytes arrived -- while audit-rotate DROPs the source table once an upload verifies, and the default gate is existence + size, which a length-preserving corruption passes. That gap is why the docs told Azure users to run --verify-restore (a full re-download + decrypt + pg_restore -l) on every weekly run: several GB of billed egress and ~10 extra minutes per run to compensate for a missing header. Set TransactionalValidation on the upload so the SDK computes a CRC64 and Azure validates it server-side, rejecting the payload on mismatch. That alone is NOT sufficient, and the difference is easy to miss: azblob's UploadStream short-circuits to a single-shot Put Blob whenever the whole payload fit in block 0 and is smaller than BlockSize (blockblob/chunkwriting.go commitBlocks), and that path silently discards the option, because UploadStreamOptions.getUploadOptions does not copy TransactionalValidation. Every .sha256 sidecar (65 bytes) and every archive under the 10 MiB block size would therefore have uploaded unchecksummed while the docs claimed otherwise -- worse than the status quo, since it would also have removed the --verify-restore recommendation that was the only mitigation for exactly those payloads. So UploadStream now buffers at most one block to see which path applies and sends undersized payloads through blockblob.Upload, which does honour the option. Verified on the wire, not just structurally. A test asserts every payload request carries x-ms-content-crc64 across four sizes (65 B, BlockSize-1, BlockSize, BlockSize+1KiB) using a non-seekable pipe, like the real pg_dump | age source. Without the size split it fails on the two undersized cases (comp="" Put Blob, no checksum header) while an options-struct assertion stays green -- which is precisely why the test checks the wire. Memory is bounded and flat in archive size: ComputeCRC64 is applied per BLOCK, so a 4.5 GB archive costs the same as a 100 MB one. It is not free -- ReadAll's append growth re-buffers each block at roughly 2x, measuring ~+100-200 MiB RSS at BlockSize 10 MiB x Concurrency 3, well inside the cronjob's 1Gi limit but above a naive BlockSize*Concurrency estimate. The comment records the measured figure rather than the naive one. Also corrects the README: uploads are now checksum-verified on both backends at every size, so --verify-restore is about proving RESTORABILITY (a valid, decryptable dump, and that the passphrase in the secret actually works) rather than catching corruption in transit. A self-hosted S3-compatible endpoint still receives the SDK's default CRC32 but enforcement is up to that server, so the recommendation stands there. ReARM-Agent: 1420896f-adf5-4843-896f-d863cfcc6528 ReARM-Agentic-Session: 4477ca5a-8dfc-435b-804c-952d39345110 --- cloud-backups/README.md | 4 +- cloud-backups/internal/storage/azure.go | 70 ++++++++++++++- cloud-backups/internal/storage/azure_test.go | 93 ++++++++++++++++++++ 3 files changed, 162 insertions(+), 5 deletions(-) diff --git a/cloud-backups/README.md b/cloud-backups/README.md index 706975f..753481c 100644 --- a/cloud-backups/README.md +++ b/cloud-backups/README.md @@ -241,7 +241,9 @@ Relieves disk pressure from a large, **write-only** audit table (per-entity revi Design notes: - The rename holds an `ACCESS EXCLUSIVE` lock only for a catalog-only operation (sub-second even on large tables). `--lock-timeout` makes it **fail-safe**: on contention the rotate rolls back untouched and retries next run. - The `DROP` is the sole irreversible step and runs **only after a verified upload**. By default, verification is a **byte-integrity** gate: the upload completed (on real AWS, S3 verifies a SHA-256 of every part server-side and refuses on mismatch) **+** a `HeadObject` size match **+** a whole-object SHA-256 written as a `.sha256` sidecar. This does **not** by itself prove the archive is *restorable*. -- `--verify-restore` upgrades that to a **restorability** check: it re-downloads the object, **decrypts** it, runs **`pg_restore -l`** (proves it's a structurally valid, decryptable dump — catches a truncated/corrupt archive or a wrong/lost age passphrase before the source is dropped), and matches the SHA-256. Costs a full re-download — reserve for first/manual runs. **Recommended on Azure**, whose uploads are not server-side checksum-verified. +- `--verify-restore` upgrades that to a **restorability** check: it re-downloads the object, **decrypts** it, runs **`pg_restore -l`** (proves it's a structurally valid, decryptable dump — catches a truncated/corrupt archive or a wrong/lost age passphrase before the source is dropped), and matches the SHA-256. Costs a full re-download -- reserve for first/manual runs. + + Uploads are checksum-verified server-side on **both** backends, at every payload size: real AWS S3 gets a SHA-256 (per part on a multipart upload, whole-object on a small one) and Azure gets a CRC64 (per block, or whole-object for a payload below the block size). So on those two backends `--verify-restore` is about proving *restorability* -- a valid, decryptable dump -- not about catching corruption in transit. A self-hosted S3-compatible endpoint (MinIO/Ceph) still *receives* the SDK's default CRC32, but whether it validates one is up to that server, so prefer enabling `--verify-restore` there. - `--no-drop` runs everything except the drop (and exits non-zero so it can't be mistaken for a completed rotation). `--drop-pending` is the confirm step: it does **not** rotate — it verifies each already-backed-up leftover against its sidecar (+ `pg_restore -l`) and drops only the ones that verify, so you drop the exact object you inspected without re-dumping. - Object names are deterministic per archive (`{prefix}-{archive}.dump[.age]`, where `{archive}` carries a unique UTC timestamp): distinct and never overwritten across rotations, while a retry or recovery of the *same* archive reuses the key instead of accumulating duplicates. Point this at a **separate, permanent-retention bucket**, not the expiring DB-backup bucket. - Runs as the same DB role the application uses, so the fresh table is owned by (and writable by) the app. diff --git a/cloud-backups/internal/storage/azure.go b/cloud-backups/internal/storage/azure.go index 5afaad5..42267e0 100644 --- a/cloud-backups/internal/storage/azure.go +++ b/cloud-backups/internal/storage/azure.go @@ -1,15 +1,20 @@ package storage import ( + "bytes" "context" + "errors" "fmt" "io" "time" "github.com/Azure/azure-sdk-for-go/sdk/azcore/policy" + "github.com/Azure/azure-sdk-for-go/sdk/azcore/streaming" "github.com/Azure/azure-sdk-for-go/sdk/azidentity" "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob" + "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/blob" "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/bloberror" + "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/blockblob" ) const ( @@ -44,12 +49,69 @@ func newAzureProvider(ctx context.Context, cfg *Config) (*azureProvider, error) return &azureProvider{client: client, container: cfg.AzureContainer}, nil } +// Have the SDK compute a CRC64 and send it as x-ms-content-crc64; Azure validates it +// server-side and rejects the payload on mismatch. Without it Azure receives NO checksum +// at all, so a "successful" upload means only that bytes arrived, not that they arrived +// intact -- and this tool DROPs the source table once an upload verifies. The S3 path +// opts into the same protection (see newS3Provider/UploadStream); this brings Azure to +// parity, so the cheap existence/size gate can be trusted there instead of requiring a +// full re-download via --verify-restore. +// +// Cost: ComputeCRC64 hashes from a buffer rather than in-line, and it is applied per +// BLOCK, so peak memory stays flat in archive size (a 4.5 GB archive costs the same as a +// 100 MB one). It is not free though: io.ReadAll's append growth re-buffers each block at +// roughly 2x, which measured ~+100-200 MiB RSS at BlockSize 10 MiB x Concurrency 3 -- +// well inside the cronjob's 1Gi limit, but above a naive BlockSize*Concurrency estimate. +func azureValidation() blob.TransferValidationType { + return blob.TransferValidationTypeComputeCRC64() +} + +func azureUploadStreamOptions() *azblob.UploadStreamOptions { + return &azblob.UploadStreamOptions{ + BlockSize: AzureBlockSize, + Concurrency: AzureConcurrency, + TransactionalValidation: azureValidation(), + } +} + +// UploadStream streams the payload to a block blob, checksummed whatever its size. +// +// The size split is NOT an optimization -- it is required for the checksum to exist at +// all. azblob's UploadStream short-circuits to a single-shot Put Blob whenever the whole +// payload fit in block 0 and is smaller than BlockSize (blockblob/chunkwriting.go +// commitBlocks), and that path silently DROPS TransactionalValidation, because +// UploadStreamOptions.getUploadOptions does not copy it. So without this split every +// .sha256 sidecar and every archive under 10 MiB would upload with no server-side +// integrity check whatsoever -- exactly the case where a length-preserving corruption +// would survive the size gate and authorize an irreversible DROP. blockblob.Upload does +// honour the option, so undersized payloads go through it directly. func (p *azureProvider) UploadStream(ctx context.Context, remotePath string, reader io.Reader) error { - opts := &azblob.UploadStreamOptions{ - BlockSize: AzureBlockSize, - Concurrency: AzureConcurrency, + // Buffer at most one block to find out which path applies. A bytes.Buffer (rather + // than a fixed AzureBlockSize array) keeps a 65-byte sidecar from allocating 10 MiB. + var head bytes.Buffer + _, err := io.CopyN(&head, reader, AzureBlockSize) + switch { + case errors.Is(err, io.EOF): + // Fewer than BlockSize bytes available: the whole payload is in hand. + return p.uploadSingleBlock(ctx, remotePath, head.Bytes()) + case err != nil: + return fmt.Errorf("reading first block of %q: %w", remotePath, err) } - _, err := p.client.UploadStream(ctx, p.container, remotePath, reader, opts) + _, err = p.client.UploadStream(ctx, p.container, remotePath, + io.MultiReader(bytes.NewReader(head.Bytes()), reader), azureUploadStreamOptions()) + if err != nil && ctx.Err() != nil { + return fmt.Errorf("upload interrupted: %w", err) + } + return err +} + +// uploadSingleBlock puts a payload smaller than one block in a single request, keeping +// the CRC64 that UploadStream's own single-shot path would have discarded. +func (p *azureProvider) uploadSingleBlock(ctx context.Context, remotePath string, data []byte) error { + blobClient := p.client.ServiceClient().NewContainerClient(p.container).NewBlockBlobClient(remotePath) + _, err := blobClient.Upload(ctx, streaming.NopCloser(bytes.NewReader(data)), &blockblob.UploadOptions{ + TransactionalValidation: azureValidation(), + }) if err != nil && ctx.Err() != nil { return fmt.Errorf("upload interrupted: %w", err) } diff --git a/cloud-backups/internal/storage/azure_test.go b/cloud-backups/internal/storage/azure_test.go index 2be0086..f7cb71c 100644 --- a/cloud-backups/internal/storage/azure_test.go +++ b/cloud-backups/internal/storage/azure_test.go @@ -1,13 +1,106 @@ package storage import ( + "bytes" + "context" "errors" "fmt" + "io" + "net/http" + "net/http/httptest" + "sync" "testing" "github.com/Azure/azure-sdk-for-go/sdk/azcore" + "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob" ) +func TestAzureUploadStreamOptions(t *testing.T) { + opts := azureUploadStreamOptions() + if opts.TransactionalValidation == nil { + t.Error("TransactionalValidation is nil: blocks would be staged with no x-ms-content-crc64, so Azure cannot reject a corrupted block") + } + if opts.BlockSize != AzureBlockSize { + t.Errorf("BlockSize = %d, want %d", opts.BlockSize, AzureBlockSize) + } + if opts.Concurrency != AzureConcurrency { + t.Errorf("Concurrency = %d, want %d", opts.Concurrency, AzureConcurrency) + } +} + +// Every upload must carry a server-verifiable checksum REGARDLESS of size, because a +// verified upload is what authorizes an irreversible DROP. This asserts it on the wire +// rather than on the options struct: setting TransactionalValidation is necessary but NOT +// sufficient, since azblob's UploadStream silently drops it on the single-shot Put Blob +// path it takes for payloads below BlockSize. Without the size split in UploadStream this +// test fails for the small cases while the options-struct test above still passes. +func TestAzureUploadStreamChecksumsEverySize(t *testing.T) { + cases := []struct { + name string + size int + wantParts int // staged blocks; 0 => single-shot Put Blob + }{ + {"sha256 sidecar sized", 65, 0}, + {"just under one block", AzureBlockSize - 1, 0}, + {"exactly one block", AzureBlockSize, 1}, + {"spans two blocks", AzureBlockSize + 1024, 2}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + var ( + mu sync.Mutex + bodies int + unsigned []string + ) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + io.Copy(io.Discard, r.Body) + comp := r.URL.Query().Get("comp") + if comp == "block" || comp == "" { // a staged block, or a single-shot Put Blob + mu.Lock() + bodies++ + if r.Header.Get("x-ms-content-crc64") == "" { + unsigned = append(unsigned, fmt.Sprintf("comp=%q len=%s", comp, r.Header.Get("Content-Length"))) + } + mu.Unlock() + } + w.Header().Set("ETag", `"0x1"`) + w.Header().Set("Last-Modified", "Wed, 30 Jul 2026 00:00:00 GMT") + w.WriteHeader(http.StatusCreated) + })) + defer srv.Close() + + client, err := azblob.NewClientWithNoCredential(srv.URL, nil) + if err != nil { + t.Fatalf("client: %v", err) + } + p := &azureProvider{client: client, container: "archives"} + // A pipe, not a file: the real caller streams pg_dump | age, so the source is + // not seekable and its length is unknown up front. + pr, pw := io.Pipe() + go func() { + pw.Write(bytes.Repeat([]byte("a"), tc.size)) + pw.Close() + }() + if err := p.UploadStream(context.Background(), "audit_archive.dump.age", pr); err != nil { + t.Fatalf("UploadStream: %v", err) + } + + mu.Lock() + defer mu.Unlock() + if len(unsigned) > 0 { + t.Errorf("%d of %d payload request(s) carried NO x-ms-content-crc64: %v", len(unsigned), bodies, unsigned) + } + wantBodies := tc.wantParts + if wantBodies == 0 { + wantBodies = 1 // one single-shot Put Blob + } + if bodies != wantBodies { + t.Errorf("payload requests = %d, want %d", bodies, wantBodies) + } + }) + } +} + func TestMapAzureNotFound(t *testing.T) { cases := []struct { name string