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