chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
// @pdsjs/blobs-s3/fetch-target - S3-compatible backup target over plain fetch
//
// Signature Version 4 implemented directly on WebCrypto, so this runs
// anywhere fetch does — Cloudflare Workers included, where the AWS SDK does
// not go. Path-style addressing, the four basic operations, and multipart
// upload: all a backup target needs, and the part every S3-compatible
// provider (Bunny, Backblaze B2, MinIO, R2, AWS itself) implements.
/**
* @typedef {import('@pdsjs/core/ports').BackupTargetPort} BackupTargetPort
*/
const encoder = new TextEncoder();
/** SHA-256 of an empty body, the constant AWS documents. */
const EMPTY_PAYLOAD_HASH =
'e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855';
/**
* Bytes per part of a multipart upload. Above S3's 5 MiB minimum, which
* every part except the last one must meet, and 10,000 parts at this size
* carry 80 GB. An upload holds the part it is sending and the one behind
* it, never the whole object.
*/
export const PART_SIZE = 8 * 1024 * 1024;
/**
* @param {Uint8Array} bytes
* @returns {string}
*/
function hex(bytes) {
let out = '';
for (const b of bytes) out += b.toString(16).padStart(2, '0');
return out;
}
/**
* @param {Uint8Array} data
* @returns {Promise<string>}
*/
async function sha256Hex(data) {
const digest = await crypto.subtle.digest(
'SHA-256',
/** @type {BufferSource} */ (data),
);
return hex(new Uint8Array(digest));
}
/**
* @param {Uint8Array} key
* @param {string} data
* @returns {Promise<Uint8Array>}
*/
async function hmac(key, data) {
const cryptoKey = await crypto.subtle.importKey(
'raw',
/** @type {BufferSource} */ (key),
{ name: 'HMAC', hash: 'SHA-256' },
false,
['sign'],
);
const mac = await crypto.subtle.sign(
'HMAC',
cryptoKey,
/** @type {BufferSource} */ (encoder.encode(data)),
);
return new Uint8Array(mac);
}
/**
* AWS uri-encoding: RFC 3986 strict — encodeURIComponent plus the characters
* it leaves bare that AWS requires encoded.
* @param {string} value
* @returns {string}
*/
function uriEncode(value) {
return encodeURIComponent(value).replace(
/[!'()*]/g,
(c) => `%${c.charCodeAt(0).toString(16).toUpperCase()}`,
);
}
/**
* A path for the canonical request and the URL: each segment encoded, the
* slashes kept.
* @param {string} path
* @returns {string}
*/
function encodePath(path) {
return path.split('/').map(uriEncode).join('/');
}
/** Decode the five XML entities a ListObjectsV2 key can carry. */
/**
* @param {string} value
* @returns {string}
*/
function decodeXmlEntities(value) {
return value
.replaceAll('<', '<')
.replaceAll('>', '>')
.replaceAll('"', '"')
.replaceAll(''', "'")
.replaceAll('&', '&');
}
/**
* @param {Uint8Array[]} chunks
* @param {number} size - Total length of the chunks
* @returns {Uint8Array}
*/
function concat(chunks, size) {
if (chunks.length === 1) return chunks[0];
const out = new Uint8Array(size);
let at = 0;
for (const chunk of chunks) {
out.set(chunk, at);
at += chunk.byteLength;
}
return out;
}
/**
* Cut a byte stream into parts. The last part is whatever is left, which is
* the one place S3 accepts a part under the minimum size.
* @param {ReadableStream<Uint8Array>} stream
* @returns {AsyncGenerator<Uint8Array>}
*/
async function* streamParts(stream) {
const reader = stream.getReader();
/** @type {Uint8Array[]} */
let pending = [];
let size = 0;
try {
while (true) {
const { done, value } = await reader.read();
if (done) break;
if (!value.byteLength) continue;
pending.push(value);
size += value.byteLength;
while (size >= PART_SIZE) {
const joined = concat(pending, size);
yield joined.subarray(0, PART_SIZE);
const remainder = joined.subarray(PART_SIZE);
pending = remainder.byteLength ? [remainder] : [];
size = remainder.byteLength;
}
}
} finally {
reader.releaseLock();
}
if (size) yield concat(pending, size);
}
/**
* Cut a buffer into parts, as views rather than copies.
* @param {Uint8Array} data
* @returns {AsyncGenerator<Uint8Array>}
*/
async function* bufferParts(data) {
for (let at = 0; at < data.byteLength; at += PART_SIZE) {
yield data.subarray(at, Math.min(at + PART_SIZE, data.byteLength));
}
}
/**
* Backups in any S3-compatible bucket, over nothing but fetch and WebCrypto —
* the target to use on Cloudflare Workers, and the simplest one anywhere
* else. Uses path-style addressing (`endpoint/bucket/key`), which every
* compatible provider accepts. Give backups their own bucket: core deletes
* anything under `snapshots/` and `blobs/` that its retention policy no
* longer wants.
*
* @param {Object} options
* @param {string} options.endpoint - Base URL of the S3 API, e.g.
* `https://ny.storage.bunnycdn.com` or `https://s3.us-west-004.backblazeb2.com`
* @param {string} options.bucket - Bucket (or storage zone) name
* @param {string} options.region - Signing region; many providers accept
* anything here, AWS and B2 want the real one
* @param {string} options.accessKeyId
* @param {string} options.secretAccessKey
* @param {typeof fetch} [options.fetchImpl] - Injectable for tests
* @returns {BackupTargetPort}
*/
export function createS3FetchBackupTarget({
endpoint,
bucket,
region,
accessKeyId,
secretAccessKey,
fetchImpl,
}) {
const url = new URL(endpoint);
const basePath = url.pathname.replace(/\/+$/, '');
const doFetch = fetchImpl ?? fetch;
/**
* Sign and send one request.
* @param {string} method
* @param {string} key - Object key, or '' for bucket-level requests
* @param {Object} [options]
* @param {Record<string, string>} [options.query]
* @param {Uint8Array} [options.body]
* @param {string} [options.contentType]
* @returns {Promise<Response>}
*/
async function send(method, key, { query = {}, body, contentType } = {}) {
const now = new Date();
const amzDate = `${now.toISOString().slice(0, 19).replace(/[-:]/g, '')}Z`;
const dateStamp = amzDate.slice(0, 8);
const payloadHash = body ? await sha256Hex(body) : EMPTY_PAYLOAD_HASH;
const canonicalUri = encodePath(
key ? `${basePath}/${bucket}/${key}` : `${basePath}/${bucket}`,
);
const canonicalQuery = Object.keys(query)
.sort()
.map((k) => `${uriEncode(k)}=${uriEncode(query[k])}`)
.join('&');
const signedHeaders = 'host;x-amz-content-sha256;x-amz-date';
const canonicalRequest = [
method,
canonicalUri,
canonicalQuery,
`host:${url.host}`,
`x-amz-content-sha256:${payloadHash}`,
`x-amz-date:${amzDate}`,
'',
signedHeaders,
payloadHash,
].join('\n');
const scope = `${dateStamp}/${region}/s3/aws4_request`;
const stringToSign = [
'AWS4-HMAC-SHA256',
amzDate,
scope,
await sha256Hex(encoder.encode(canonicalRequest)),
].join('\n');
let signingKey = await hmac(
encoder.encode(`AWS4${secretAccessKey}`),
dateStamp,
);
for (const step of [region, 's3', 'aws4_request']) {
signingKey = await hmac(signingKey, step);
}
const signature = hex(await hmac(signingKey, stringToSign));
/** @type {Record<string, string>} */
const headers = {
Authorization: `AWS4-HMAC-SHA256 Credential=${accessKeyId}/${scope}, SignedHeaders=${signedHeaders}, Signature=${signature}`,
'x-amz-content-sha256': payloadHash,
'x-amz-date': amzDate,
};
if (contentType) headers['Content-Type'] = contentType;
const target = `${url.origin}${canonicalUri}${
canonicalQuery ? `?${canonicalQuery}` : ''
}`;
return doFetch(target, {
method,
headers,
body: /** @type {BodyInit|undefined} */ (body),
});
}
/**
* The error for a response that failed, for the caller to throw. Reading
* the detail out of the body is what makes this asynchronous, so callers
* write `throw await failure(...)`. A `return` of a rejected promise from
* inside a `try` is not caught by that same try.
* @param {Response} response
* @param {string} doing
* @returns {Promise<Error>}
*/
async function failure(response, doing) {
const detail = (await response.text().catch(() => '')).slice(0, 200);
return new Error(
`Backup target ${doing} failed: HTTP ${response.status}${
detail ? ` — ${detail}` : ''
}`,
);
}
/**
* One object in one request.
* @param {string} key
* @param {Uint8Array} data
* @param {string} [contentType]
* @returns {Promise<void>}
*/
async function putBytes(key, data, contentType) {
const response = await send('PUT', key, {
body: data,
contentType: contentType || 'application/octet-stream',
});
if (!response.ok) throw await failure(response, `put ${key}`);
// Some providers only release the connection once the body is read.
await response.arrayBuffer().catch(() => undefined);
}
/**
* One object as a multipart upload, or as a plain PUT when it fits in a
* single part. Providers answer HTTP 413 to a body past their single
* request limit, which every video-sized blob is.
* @param {string} key
* @param {AsyncGenerator<Uint8Array>} parts
* @param {string} [contentType]
* @returns {Promise<void>}
*/
async function putParts(key, parts, contentType) {
const first = await parts.next();
const head = first.done ? new Uint8Array(0) : first.value;
let next = await parts.next();
if (next.done) return putBytes(key, head, contentType);
const created = await send('POST', key, {
query: { uploads: '' },
contentType: contentType || 'application/octet-stream',
});
if (!created.ok) throw await failure(created, `start upload ${key}`);
const uploadId = decodeXmlEntities(
/<UploadId>([^<]*)<\/UploadId>/.exec(await created.text())?.[1] ?? '',
);
if (!uploadId) {
throw new Error(`Backup target start upload ${key} failed: no upload id`);
}
/** @type {string[]} */
const etags = [];
try {
let part = head;
while (true) {
const number = etags.length + 1;
const response = await send('PUT', key, {
query: { partNumber: String(number), uploadId },
body: part,
});
if (!response.ok)
throw await failure(response, `put part ${number} of ${key}`);
await response.arrayBuffer().catch(() => undefined);
const etag = response.headers.get('etag');
if (!etag) {
throw new Error(
`Backup target put part ${number} of ${key} failed: no etag`,
);
}
etags.push(etag);
if (next.done) break;
part = next.value;
next = await parts.next();
}
const body = [
'<CompleteMultipartUpload>',
...etags.map(
(etag, index) =>
`<Part><PartNumber>${index + 1}</PartNumber><ETag>${etag}</ETag></Part>`,
),
'</CompleteMultipartUpload>',
].join('');
const completed = await send('POST', key, {
query: { uploadId },
body: encoder.encode(body),
contentType: 'application/xml',
});
if (!completed.ok)
throw await failure(completed, `complete upload ${key}`);
// S3 answers 200 and then reports the failure in the body.
const text = await completed.text();
if (text.includes('<Error')) {
throw new Error(
`Backup target complete upload ${key} failed: ${text.slice(0, 200)}`,
);
}
} catch (err) {
// The bucket charges for the parts of an upload nobody completes.
await send('DELETE', key, { query: { uploadId } })
.then((response) => response.arrayBuffer())
.catch(() => undefined);
throw err;
}
}
return {
/**
* @param {string} key
* @param {Uint8Array} data
* @param {string} [contentType]
*/
async put(key, data, contentType) {
if (data.byteLength <= PART_SIZE) {
return putBytes(key, data, contentType);
}
return putParts(key, bufferParts(data), contentType);
},
/**
* @param {string} key
* @param {ReadableStream<Uint8Array>} body
* @param {string} [contentType]
*/
async putStream(key, body, contentType) {
return putParts(key, streamParts(body), contentType);
},
/** @param {string} key */
async get(key) {
const response = await send('GET', key);
if (response.status === 404) return null;
if (!response.ok) throw await failure(response, `get ${key}`);
return new Uint8Array(await response.arrayBuffer());
},
/** @param {string} prefix */
async list(prefix) {
/** @type {string[]} */
const keys = [];
/** @type {string|undefined} */
let continuationToken;
do {
/** @type {Record<string, string>} */
const query = { 'list-type': '2', prefix };
if (continuationToken) query['continuation-token'] = continuationToken;
const response = await send('GET', '', { query });
if (!response.ok) throw await failure(response, `list ${prefix}`);
const xml = await response.text();
for (const match of xml.matchAll(/<Key>([^<]*)<\/Key>/g)) {
keys.push(decodeXmlEntities(match[1]));
}
continuationToken = xml.includes('<IsTruncated>true</IsTruncated>')
? decodeXmlEntities(
/<NextContinuationToken>([^<]*)<\/NextContinuationToken>/.exec(
xml,
)?.[1] ?? '',
) || undefined
: undefined;
} while (continuationToken);
return keys;
},
/** @param {string} key */
async delete(key) {
const response = await send('DELETE', key);
// Gone already is the outcome we wanted.
if (!response.ok && response.status !== 404) {
throw await failure(response, `delete ${key}`);
}
await response.arrayBuffer().catch(() => undefined);
},
};
}