diff --git a/src/objectstore/r2.ts b/src/objectstore/r2.ts index 126c599..56359fd 100644 --- a/src/objectstore/r2.ts +++ b/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 { 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 { const hash = createHash("sha256"); await new Promise((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); diff --git a/test/r2_objectstore.test.ts b/test/r2_objectstore.test.ts index b12df59..e09ea4d 100644 --- a/test/r2_objectstore.test.ts +++ b/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(); const requests: Request[] = []; +const requestBodies: Array = []; 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 { 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",