chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
// blobs-s3 fetch target tests - SigV4 signing, pagination, error mapping
import { createHash, createHmac } from 'node:crypto';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import {
createS3FetchBackupTarget,
PART_SIZE,
} from '../packages/blobs-s3/src/fetch-target.js';
const CONFIG = {
endpoint: 'https://s3.amazonaws.com',
bucket: 'examplebucket',
region: 'us-east-1',
accessKeyId: 'AKIAIOSFODNN7EXAMPLE',
secretAccessKey: 'wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY',
};
// The fixed moment AWS's documented signing examples use.
const DOC_TIME = new Date('2013-05-24T00:00:00Z');
const EMPTY_HASH =
'e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855';
/**
* Reference SigV4 over node:crypto — an implementation independent of the
* WebCrypto one under test. Its HMAC chain and canonical layout reproduce the
* worked example in AWS's own documentation (GetObject, 20130524,
* signature f0e8bdb8...), so agreement here is agreement with AWS.
* @param {{method: string, path: string, query?: string, payloadHash?: string}} req
* @returns {string} The Authorization header value
*/
function referenceAuthorization({
method,
path,
query = '',
payloadHash = EMPTY_HASH,
}) {
const amzDate = '20130524T000000Z';
const dateStamp = '20130524';
const canonicalRequest = [
method,
path,
query,
'host:s3.amazonaws.com',
`x-amz-content-sha256:${payloadHash}`,
`x-amz-date:${amzDate}`,
'',
'host;x-amz-content-sha256;x-amz-date',
payloadHash,
].join('\n');
const scope = `${dateStamp}/${CONFIG.region}/s3/aws4_request`;
const stringToSign = [
'AWS4-HMAC-SHA256',
amzDate,
scope,
createHash('sha256').update(canonicalRequest).digest('hex'),
].join('\n');
let key = createHmac('sha256', `AWS4${CONFIG.secretAccessKey}`)
.update(dateStamp)
.digest();
for (const step of [CONFIG.region, 's3', 'aws4_request']) {
key = createHmac('sha256', key).update(step).digest();
}
const signature = createHmac('sha256', key)
.update(stringToSign)
.digest('hex');
return `AWS4-HMAC-SHA256 Credential=${CONFIG.accessKeyId}/${scope}, SignedHeaders=host;x-amz-content-sha256;x-amz-date, Signature=${signature}`;
}
/**
* A fetch stub that records every call and replays canned responses.
* @param {...Response} responses
*/
function createFetchStub(...responses) {
/** @type {Array<{url: string, init: RequestInit}>} */
const calls = [];
const fetchImpl = async (
/** @type {string} */ url,
/** @type {RequestInit} */ init,
) => {
calls.push({ url, init });
return responses.shift() ?? new Response('', { status: 200 });
};
return {
calls,
fetchImpl: /** @type {typeof fetch} */ (/** @type {unknown} */ (fetchImpl)),
};
}
/**
* A byte stream delivered in chunks, the way a blob store hands one over.
* @param {Uint8Array} bytes
* @param {number} [chunkSize]
* @returns {ReadableStream<Uint8Array>}
*/
function streamOf(bytes, chunkSize = 64 * 1024) {
let at = 0;
return new ReadableStream({
pull(controller) {
if (at >= bytes.byteLength) {
controller.close();
return;
}
const end = Math.min(at + chunkSize, bytes.byteLength);
controller.enqueue(bytes.subarray(at, end));
at = end;
},
});
}
/**
* What UploadPart answers: an empty body and the part's ETag.
* @param {string} etag
* @returns {Response}
*/
function partResponse(etag) {
return new Response('', { status: 200, headers: { ETag: etag } });
}
/**
* @param {{init: RequestInit}} call
* @returns {Uint8Array}
*/
function bodyOf(call) {
return /** @type {Uint8Array} */ (call.init.body);
}
beforeEach(() => {
vi.useFakeTimers();
vi.setSystemTime(DOC_TIME);
});
afterEach(() => {
vi.useRealTimers();
});
describe('createS3FetchBackupTarget', () => {
it('signs a GET the way AWS documents', async () => {
const { calls, fetchImpl } = createFetchStub(
new Response(new Uint8Array([1, 2, 3]), { status: 200 }),
);
const target = createS3FetchBackupTarget({ ...CONFIG, fetchImpl });
const data = await target.get('test.txt');
expect(data).toEqual(new Uint8Array([1, 2, 3]));
const { url, init } = calls[0];
expect(url).toBe('https://s3.amazonaws.com/examplebucket/test.txt');
const headers = /** @type {Record<string, string>} */ (init.headers);
expect(headers['x-amz-date']).toBe('20130524T000000Z');
expect(headers['x-amz-content-sha256']).toBe(EMPTY_HASH);
expect(headers.Authorization).toBe(
referenceAuthorization({
method: 'GET',
path: '/examplebucket/test.txt',
}),
);
});
it('signs the payload hash into a PUT', async () => {
const { calls, fetchImpl } = createFetchStub();
const target = createS3FetchBackupTarget({ ...CONFIG, fetchImpl });
const body = new TextEncoder().encode('snapshot bytes');
await target.put(
'snapshots/2013/repo.car',
body,
'application/vnd.ipld.car',
);
const { url, init } = calls[0];
expect(url).toBe(
'https://s3.amazonaws.com/examplebucket/snapshots/2013/repo.car',
);
expect(init.method).toBe('PUT');
expect(init.body).toBe(body);
const headers = /** @type {Record<string, string>} */ (init.headers);
const bodyHash = createHash('sha256').update(body).digest('hex');
expect(headers['x-amz-content-sha256']).toBe(bodyHash);
expect(headers['Content-Type']).toBe('application/vnd.ipld.car');
expect(headers.Authorization).toBe(
referenceAuthorization({
method: 'PUT',
path: '/examplebucket/snapshots/2013/repo.car',
payloadHash: bodyHash,
}),
);
});
it('returns null for a missing object and throws on other errors', async () => {
const { fetchImpl } = createFetchStub(
new Response('', { status: 404 }),
new Response('<Error>AccessDenied</Error>', { status: 403 }),
);
const target = createS3FetchBackupTarget({ ...CONFIG, fetchImpl });
expect(await target.get('missing')).toBeNull();
await expect(target.get('forbidden')).rejects.toThrow(/403.*AccessDenied/s);
});
it('follows list pagination and decodes XML entities', async () => {
const page1 =
'<ListBucketResult><IsTruncated>true</IsTruncated>' +
'<Contents><Key>blobs/a</Key></Contents>' +
'<Contents><Key>blobs/b&c</Key></Contents>' +
'<NextContinuationToken>tok=1</NextContinuationToken></ListBucketResult>';
const page2 =
'<ListBucketResult><IsTruncated>false</IsTruncated>' +
'<Contents><Key>blobs/d</Key></Contents></ListBucketResult>';
const { calls, fetchImpl } = createFetchStub(
new Response(page1, { status: 200 }),
new Response(page2, { status: 200 }),
);
const target = createS3FetchBackupTarget({ ...CONFIG, fetchImpl });
const keys = await target.list('blobs/');
expect(keys).toEqual(['blobs/a', 'blobs/b&c', 'blobs/d']);
expect(calls[0].url).toBe(
'https://s3.amazonaws.com/examplebucket?list-type=2&prefix=blobs%2F',
);
// The token is carried on the second request, uri-encoded, and the query
// stays sorted so it matches what was signed.
expect(calls[1].url).toBe(
'https://s3.amazonaws.com/examplebucket?continuation-token=tok%3D1&list-type=2&prefix=blobs%2F',
);
});
it('sends a stream that fits in one part as a plain PUT', async () => {
const { calls, fetchImpl } = createFetchStub();
const target = createS3FetchBackupTarget({ ...CONFIG, fetchImpl });
await target.putStream?.(
'blobs/small',
streamOf(new Uint8Array(10).fill(7)),
'image/png',
);
expect(calls).toHaveLength(1);
expect(calls[0].url).toBe(
'https://s3.amazonaws.com/examplebucket/blobs/small',
);
expect(calls[0].init.method).toBe('PUT');
expect(bodyOf(calls[0]).byteLength).toBe(10);
});
it('sends a stream past one part as a multipart upload', async () => {
// Two full parts and a short third, so the last-part case is covered and
// the chunking is not aligned to the part size.
const total = PART_SIZE * 2 + 1000;
const { calls, fetchImpl } = createFetchStub(
new Response(
'<InitiateMultipartUploadResult><UploadId>up=1</UploadId></InitiateMultipartUploadResult>',
),
partResponse('"e1"'),
partResponse('"e2"'),
partResponse('"e3"'),
new Response(
'<CompleteMultipartUploadResult><ETag>"final"</ETag></CompleteMultipartUploadResult>',
),
);
const target = createS3FetchBackupTarget({ ...CONFIG, fetchImpl });
await target.putStream?.(
'blobs/video',
streamOf(new Uint8Array(total).fill(3), 100_000),
'video/mp4',
);
expect(calls.map((c) => `${c.init.method} ${c.url}`)).toEqual([
'POST https://s3.amazonaws.com/examplebucket/blobs/video?uploads=',
'PUT https://s3.amazonaws.com/examplebucket/blobs/video?partNumber=1&uploadId=up%3D1',
'PUT https://s3.amazonaws.com/examplebucket/blobs/video?partNumber=2&uploadId=up%3D1',
'PUT https://s3.amazonaws.com/examplebucket/blobs/video?partNumber=3&uploadId=up%3D1',
'POST https://s3.amazonaws.com/examplebucket/blobs/video?uploadId=up%3D1',
]);
expect(bodyOf(calls[1]).byteLength).toBe(PART_SIZE);
expect(bodyOf(calls[2]).byteLength).toBe(PART_SIZE);
expect(bodyOf(calls[3]).byteLength).toBe(1000);
expect(new TextDecoder().decode(bodyOf(calls[4]))).toBe(
'<CompleteMultipartUpload>' +
'<Part><PartNumber>1</PartNumber><ETag>"e1"</ETag></Part>' +
'<Part><PartNumber>2</PartNumber><ETag>"e2"</ETag></Part>' +
'<Part><PartNumber>3</PartNumber><ETag>"e3"</ETag></Part>' +
'</CompleteMultipartUpload>',
);
// Every part carries its own payload hash, signed like any other request.
const headers = /** @type {Record<string, string>} */ (
calls[1].init.headers
);
expect(headers.Authorization).toBe(
referenceAuthorization({
method: 'PUT',
path: '/examplebucket/blobs/video',
query: 'partNumber=1&uploadId=up%3D1',
payloadHash: createHash('sha256')
.update(bodyOf(calls[1]))
.digest('hex'),
}),
);
});
it('takes a buffer past one part through the same path', async () => {
const { calls, fetchImpl } = createFetchStub(
new Response(
'<InitiateMultipartUploadResult><UploadId>u2</UploadId></InitiateMultipartUploadResult>',
),
partResponse('"e1"'),
partResponse('"e2"'),
new Response(
'<CompleteMultipartUploadResult></CompleteMultipartUploadResult>',
),
);
const target = createS3FetchBackupTarget({ ...CONFIG, fetchImpl });
await target.put('snapshots/2013/repo.car', new Uint8Array(PART_SIZE + 5));
expect(calls).toHaveLength(4);
expect(bodyOf(calls[1]).byteLength).toBe(PART_SIZE);
expect(bodyOf(calls[2]).byteLength).toBe(5);
});
it('aborts the upload when a part fails', async () => {
const { calls, fetchImpl } = createFetchStub(
new Response(
'<InitiateMultipartUploadResult><UploadId>u3</UploadId></InitiateMultipartUploadResult>',
),
partResponse('"e1"'),
new Response('too big', { status: 413 }),
);
const target = createS3FetchBackupTarget({ ...CONFIG, fetchImpl });
await expect(
target.putStream?.(
'blobs/video',
streamOf(new Uint8Array(PART_SIZE * 2)),
),
).rejects.toThrow(/put part 2 of blobs\/video failed: HTTP 413/);
const abort = calls[calls.length - 1];
expect(abort.init.method).toBe('DELETE');
expect(abort.url).toBe(
'https://s3.amazonaws.com/examplebucket/blobs/video?uploadId=u3',
);
});
it('reports a failure the target hides in a 200 response', async () => {
const { fetchImpl } = createFetchStub(
new Response(
'<InitiateMultipartUploadResult><UploadId>u4</UploadId></InitiateMultipartUploadResult>',
),
partResponse('"e1"'),
partResponse('"e2"'),
new Response('<Error><Code>InternalError</Code></Error>', {
status: 200,
}),
);
const target = createS3FetchBackupTarget({ ...CONFIG, fetchImpl });
await expect(
target.putStream?.(
'blobs/video',
streamOf(new Uint8Array(PART_SIZE * 2)),
),
).rejects.toThrow(/complete upload blobs\/video failed.*InternalError/s);
});
it('refuses an upload the target will not name', async () => {
const { fetchImpl } = createFetchStub(
new Response(
'<InitiateMultipartUploadResult></InitiateMultipartUploadResult>',
),
);
const target = createS3FetchBackupTarget({ ...CONFIG, fetchImpl });
await expect(
target.putStream?.(
'blobs/video',
streamOf(new Uint8Array(PART_SIZE * 2)),
),
).rejects.toThrow(/no upload id/);
});
it('treats deleting an absent key as success', async () => {
const { fetchImpl } = createFetchStub(
new Response('', { status: 404 }),
new Response('nope', { status: 500 }),
);
const target = createS3FetchBackupTarget({ ...CONFIG, fetchImpl });
await expect(target.delete('already-gone')).resolves.toBeUndefined();
await expect(target.delete('broken')).rejects.toThrow(/500/);
});
});