fix(objectstore): send fixed-length body in putFile so SigV4 signature verifies on strict S3 backends

src/objectstore/r2.ts

@@ -1,6 +1,5 @@
 import { createHash, createHmac } from "node:crypto";
-import { createReadStream } from "node:fs";
-import { Readable } from "node:stream";
+import { createReadStream, openAsBlob } from "node:fs";
 import type { GetOptions, ObjectStore, PutResult } from "./interface";
 import { dsError } from "../util/ds_error.ts";
 
@@ -50,10 +49,6 @@ async function readResponseText(res: Response): Promise<string> {
   return XML_DECODER.decode(await readResponseBytes(res));
 }
 
-function fileStreamBody(path: string): BodyInit {
-  return Readable.toWeb(createReadStream(path)) as unknown as BodyInit;
-}
-
 async function sha256FileHex(path: string): Promise<string> {
   const hash = createHash("sha256");
   await new Promise<void>((resolve, reject) => {
@@ -241,7 +236,8 @@ export class R2ObjectStore implements ObjectStore {
       if (opts.contentType) headers["content-type"] = opts.contentType;
       const res = await this.request("PUT", this.objectPath(key), {
         headers,
-        body: fileStreamBody(path),
+        // fixed-length Blob, not a stream: streams make fetch chunk the body and drop the signed content-length
+        body: await openAsBlob(path),
         payloadHash,
       });
       if (!res.ok) this.wrapStatus("PUT", key, res);

test/r2_objectstore.test.ts

@@ -1,4 +1,8 @@
 import { afterEach, beforeEach, describe, expect, test } from "bun:test";
+import { createHash } from "node:crypto";
+import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
 import { R2ObjectStore } from "../src/objectstore/r2";
 
 type Entry = {
@@ -11,6 +15,7 @@ const TEXT_ENCODER = new TextEncoder();
 const originalFetch = globalThis.fetch;
 const entries = new Map<string, Entry>();
 const requests: Request[] = [];
+const requestBodies: Array<BodyInit | null | undefined> = [];
 
 function xmlEscape(value: string): string {
   return value
@@ -33,6 +38,7 @@ function etagFor(key: string): string {
 async function fakeFetch(input: RequestInfo | URL, init?: RequestInit): Promise<Response> {
   const req = new Request(input, init);
   requests.push(req);
+  requestBodies.push(init?.body);
   expect(req.headers.get("authorization")).toStartWith("AWS4-HMAC-SHA256 ");
   expect(req.headers.get("x-amz-date")).not.toBeNull();
   expect(req.headers.get("x-amz-content-sha256")).not.toBeNull();
@@ -96,6 +102,7 @@ describe("R2ObjectStore", () => {
   beforeEach(() => {
     entries.clear();
     requests.length = 0;
+    requestBodies.length = 0;
     globalThis.fetch = fakeFetch;
   });
 
@@ -140,6 +147,37 @@ describe("R2ObjectStore", () => {
     expect(requests[0]?.url).toBe("http://127.0.0.1:9000/bucket/streams/a");
   });
 
+  test("putFile sends a fixed-length body so the SigV4 signature covers content-length", async () => {
+    const store = new R2ObjectStore({
+      accountId: "acct",
+      bucket: "bucket",
+      accessKeyId: "key",
+      secretAccessKey: "secret",
+    });
+
+    const dir = mkdtempSync(join(tmpdir(), "r2-putfile-"));
+    const path = join(dir, "payload.bin");
+    const content = new Uint8Array(4096).map((_, i) => (i * 31 + 7) & 0xff);
+    writeFileSync(path, content);
+    try {
+      await store.putFile("streams/file", path, content.byteLength, { contentType: "application/octet-stream" });
+
+      const body = requestBodies[0];
+      expect(body).toBeInstanceOf(Blob);
+      expect(body).not.toBeInstanceOf(ReadableStream);
+      expect((body as Blob).size).toBe(content.byteLength);
+
+      const req = requests[0]!;
+      expect(req.headers.get("content-length")).toBe(String(content.byteLength));
+      const signedHeaders = req.headers.get("authorization")?.match(/SignedHeaders=([^,]+),/)?.[1]?.split(";") ?? [];
+      expect(signedHeaders).toContain("content-length");
+      expect(req.headers.get("x-amz-content-sha256")).toBe(createHash("sha256").update(content).digest("hex"));
+      expect(entries.get("streams/file")?.data).toEqual(content);
+    } finally {
+      rmSync(dir, { recursive: true, force: true });
+    }
+  });
+
   test("get handles missing objects", async () => {
     const store = new R2ObjectStore({
       accountId: "acct",