chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
// @pdsjs/core/handlers/xrpc-blob - blob upload, download and listing.
//
// uploadBlob is com.atproto.repo.*, getBlob and listBlobs are
// com.atproto.sync.*. All three read or write blob storage rather than the repo,
// so they are together here.
import { parseByteRange } from '../http.js';
import {
blobCidFromDigest,
cidToString,
isSpaceRecordUri,
sniffMimeType,
} from '../repo.js';
import { ScopePermissions } from '../scope.js';
import { createSha256 } from '../sha256.js';
/**
* @typedef {Object} BlobContext
* @property {import('../ports.js').ActorStoragePort} actorStorage
* @property {import('../ports.js').BlobPort} blobs
* @property {number} blobUploadLimit - Largest accepted upload, in bytes
* @property {import('../ports.js').HasherFactory} [hasher] - SHA-256 factory; absent, the plain JavaScript one
* @property {boolean} readOnly
* @property {() => Promise<string|null>} getDid
* @property {() => Response} readOnlyError
*/
// sniffMimeType reads at most this many bytes of the content.
const SNIFF_LENGTH = 12;
/**
* A BlobWriter over the one-shot `put`, for a port without `createWriter`.
* It holds every chunk, so it costs what the buffered upload path always
* cost; the upload limit bounds it.
* @param {import('../ports.js').BlobPort} blobs
* @param {string} did
* @returns {import('../ports.js').BlobWriter}
*/
function createBufferedWriter(blobs, did) {
/** @type {Uint8Array[]} */
let chunks = [];
return {
async write(chunk) {
chunks.push(chunk);
},
async commit(cid, mimeType) {
let size = 0;
for (const chunk of chunks) size += chunk.length;
const data = new Uint8Array(size);
let at = 0;
for (const chunk of chunks) {
data.set(chunk, at);
at += chunk.length;
}
await blobs.put(did, cid, data, mimeType);
},
async abort() {
chunks = [];
},
};
}
/**
* @param {string} error
* @param {string} message
* @param {number} [status]
* @returns {Response}
*/
function errorResponse(error, message, status = 400) {
return Response.json({ error, message }, { status });
}
/**
* @param {BlobContext} ctx
* @returns {{ routes: import('../pds.js').Routes }}
*/
export function createBlobHandlers(ctx) {
const {
actorStorage,
blobs,
blobUploadLimit,
hasher = createSha256,
readOnly,
getDid,
readOnlyError,
} = ctx;
/**
* com.atproto.repo.uploadBlob
* @param {Request} request
* @param {URL} _url
* @param {import('../pds.js').Auth|null} auth
* @returns {Promise<Response>}
*/
async function handleUploadBlob(request, _url, auth) {
if (readOnly) return readOnlyError();
if (!auth) {
return errorResponse(
'AuthenticationRequired',
'Authentication required',
401,
);
}
const did = await getDid();
if (!did) {
return errorResponse('InvalidRequest', 'PDS not initialized');
}
const contentType =
request.headers.get('content-type') || 'application/octet-stream';
// Deliberately not gated on checkAccountActive: an incoming migration
// uploads its blobs while the account is still deactivated, before
// activateAccount. A blob with no record referencing it is inert and gets
// reaped as an orphan, and record writes remain gated, so accepting blobs
// in this window is safe and matches the reference PDS.
const tooLarge = () =>
errorResponse(
'PayloadTooLarge',
`Blob exceeds the maximum size of ${blobUploadLimit} bytes`,
413,
);
const declared = Number(request.headers.get('content-length'));
if (Number.isFinite(declared) && declared > blobUploadLimit) {
return tooLarge();
}
// The body streams through: each chunk feeds the hash and the writer, and
// only the sniffing prefix stays in memory. The size check runs per chunk
// because content-length is not required to be present or honest.
const hash = hasher();
const writer = blobs.createWriter
? await blobs.createWriter(did)
: createBufferedWriter(blobs, did);
let size = 0;
let head = new Uint8Array(0);
try {
if (request.body) {
const reader = request.body.getReader();
while (true) {
const { done, value } = await reader.read();
if (done) break;
const chunk =
value instanceof Uint8Array ? value : new Uint8Array(value);
size += chunk.length;
if (size > blobUploadLimit) {
await reader.cancel();
await writer.abort();
return tooLarge();
}
if (head.length < SNIFF_LENGTH) {
const take = Math.min(SNIFF_LENGTH - head.length, chunk.length);
const grown = new Uint8Array(head.length + take);
grown.set(head);
grown.set(chunk.subarray(0, take), head.length);
head = grown;
}
await hash.update(chunk);
await writer.write(chunk);
}
}
const mimeType = sniffMimeType(head) || contentType;
const permissions = new ScopePermissions(auth.scope);
if (!permissions.allowsBlob(mimeType)) {
await writer.abort();
return errorResponse(
'Forbidden',
`Missing required scope for blob upload of type ${mimeType}`,
403,
);
}
const cid = cidToString(blobCidFromDigest(await hash.digest()));
await actorStorage.putBlob(cid, mimeType, size);
await writer.commit(cid, mimeType);
return Response.json({
blob: {
$type: 'blob',
ref: { $link: cid },
mimeType,
size,
},
});
} catch (e) {
// Failed reads and failed staging writes end the same way: nothing
// stays behind under a temporary key.
try {
await writer.abort();
} catch {
// the original error is the one to surface
}
throw e;
}
}
/**
* The response headers every getBlob answer carries. Disposition and CSP
* force download and block execution so hostile blobs can't run in the PDS
* origin; browsers still play such a response inside a video or audio
* element, since disposition applies to navigation, not embedding.
* @param {string} cid
* @returns {Record<string, string>}
*/
function blobHeaders(cid) {
return {
'Cache-Control': 'public, max-age=31536000, immutable',
'X-Content-Type-Options': 'nosniff',
'Content-Disposition': `attachment; filename="${cid}"`,
'Content-Security-Policy': "default-src 'none'; sandbox",
'Accept-Ranges': 'bytes',
};
}
/**
* com.atproto.sync.getBlob
* @param {Request} request
* @param {URL} url
* @returns {Promise<Response>}
*/
async function handleGetBlob(request, url) {
const did = url.searchParams.get('did');
const cid = url.searchParams.get('cid');
if (!did || !cid) {
return errorResponse('InvalidRequest', 'Missing required parameters');
}
const myDid = await getDid();
if (did !== myDid) {
return errorResponse('InvalidRequest', 'DID does not match this PDS');
}
if (!cid.startsWith('b') || cid.length < 10) {
return errorResponse('InvalidRequest', 'Invalid CID format');
}
// A blob only space records reference is private data. This endpoint is
// unauthenticated, so such a blob is absent here; com.atproto.space.getBlob
// serves it to a session or a space credential. A blob with no references
// at all stays served: the public write flow uploads before it commits.
const uris = await actorStorage.listBlobRecordUris(cid);
if (uris.length > 0 && uris.every(isSpaceRecordUri)) {
return errorResponse('BlobNotFound', 'Blob not found', 404);
}
// Range service needs the blob's size before any bytes move, and the
// metadata row holds it. Without that row (an old mock, or bytes that
// arrived outside the upload paths) the whole object goes out buffered,
// as it always did.
const meta = actorStorage.getBlob ? await actorStorage.getBlob(cid) : null;
if (!meta) {
const result = await blobs.get(did, cid);
if (!result) {
return errorResponse('BlobNotFound', 'Blob not found', 404);
}
return new Response(/** @type {BodyInit} */ (result.data), {
headers: {
...blobHeaders(cid),
'Content-Type': result.mimeType,
'Content-Length': String(result.data.length),
},
});
}
const range = parseByteRange(request.headers.get('range'), meta.size);
if (range === 'unsatisfiable') {
return new Response(null, {
status: 416,
headers: {
'Content-Range': `bytes */${meta.size}`,
'Accept-Ranges': 'bytes',
},
});
}
/** @type {BodyInit} */
let body;
/** @type {string} */
let mimeType;
if (blobs.getStream) {
const found = await blobs.getStream(did, cid, range ?? undefined);
if (!found) {
return errorResponse('BlobNotFound', 'Blob not found', 404);
}
body = found.body;
mimeType = found.mimeType;
} else {
const result = await blobs.get(did, cid);
if (!result) {
return errorResponse('BlobNotFound', 'Blob not found', 404);
}
body = /** @type {BodyInit} */ (
range ? result.data.subarray(range.start, range.end + 1) : result.data
);
mimeType = result.mimeType;
}
const length = range ? range.end - range.start + 1 : meta.size;
/** @type {Record<string, string>} */
const headers = {
...blobHeaders(cid),
'Content-Type': mimeType,
'Content-Length': String(length),
};
if (range) {
headers['Content-Range'] =
`bytes ${range.start}-${range.end}/${meta.size}`;
}
return new Response(body, { status: range ? 206 : 200, headers });
}
/**
* com.atproto.sync.listBlobs
* @param {Request} _request
* @param {URL} url
* @returns {Promise<Response>}
*/
async function handleListBlobs(_request, url) {
const limit = Math.min(
parseInt(url.searchParams.get('limit') || '500', 10),
1000,
);
const cursor = url.searchParams.get('cursor');
const result = await actorStorage.listBlobs(cursor, limit);
// Space-only blobs are absent from the public listing the same way they
// are absent from getBlob. The cursor advances over the unfiltered page,
// so a filtered-out blob costs a page slot but never stalls paging.
/** @type {string[]} */
const cids = [];
for (const cid of result.cids) {
const uris = await actorStorage.listBlobRecordUris(cid);
if (uris.length > 0 && uris.every(isSpaceRecordUri)) continue;
cids.push(cid);
}
return Response.json({
cids,
// Match the reference PDS: omit cursor on a final page rather than
// emitting `cursor: null`, which strict lexicon validators reject.
...(result.cursor ? { cursor: result.cursor } : {}),
});
}
return {
routes: {
'/xrpc/com.atproto.repo.uploadBlob': {
method: 'POST',
auth: 'required',
// The Bluesky video service uploads a processed video here with the
// service auth token the account minted for this server.
serviceAuth: true,
handler: handleUploadBlob,
},
'/xrpc/com.atproto.sync.getBlob': {
handler: handleGetBlob,
},
'/xrpc/com.atproto.sync.listBlobs': {
handler: handleListBlobs,
},
},
};
}