chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
// @pdsjs/spaces/handlers/read - com.atproto.space read and sync endpoints.
//
// These are the repo-host role: an application syncing a space pulls each
// member's repo from its PDS through here.
//
// Two ways in: the hosted account's own session, or a space credential issued
// by the space's authority. The credential is what lets a syncer read a repo it
// does not own. It is signed by the authority, so it verifies without
// contacting them.
//
// A read that returns a commit or a record is MACed for the syncer that asked,
// under a secret derived from its Atproto-Space-Want-Mac key share and this
// account's KEM key. A credential read must ask. The account's own session may.
import { cborDecode, cborDecodeJson, findBlobRefs } from '@pdsjs/core/repo';
import { RepoCommit } from '../commit.js';
import { decapsulate } from '../kem.js';
import {
COMMIT_MAC_TYPE,
computeCommitMac,
computeRecordMac,
formatMacHeader,
MAC_HEADER,
parseWantMacHeader,
RECORD_MAC_TYPE,
SpaceMacError,
WANT_MAC_HEADER,
} from '../mac.js';
import { serializeRepo } from '../star.js';
import { parseSpaceUri } from '../uri.js';
import { createReadAuthorizer } from './auth.js';
/**
* @param {string} error
* @param {string} message
* @param {number} [status]
* @returns {Response}
*/
function errorResponse(error, message, status = 400) {
return Response.json({ error, message }, { status });
}
/**
* @param {URL} url
* @returns {{space: string, repo: string|null}|Response}
*/
function readTarget(url) {
const space = url.searchParams.get('space');
if (!space) return errorResponse('InvalidRequest', 'space is required');
try {
parseSpaceUri(space);
} catch {
return errorResponse('InvalidSpaceUri', `Not a space uri: ${space}`);
}
return { space, repo: url.searchParams.get('repo') };
}
/**
* @param {URL} url
* @param {string} name
* @param {number} def
* @param {number} max
* @returns {number|Response}
*/
function readLimit(url, name, def, max) {
const raw = url.searchParams.get(name);
if (raw === null) return def;
const value = Number(raw);
if (!Number.isInteger(value) || value < 1 || value > max) {
return errorResponse(
'InvalidRequest',
`${name} must be an integer between 1 and ${max}`,
);
}
return value;
}
/**
* The storage sort key a listRecords cursor names. A cursor is a record's
* AT-URI, as the reference gives it. Every row here shares `prefix`, so a
* cursor outside it puts either every row or no row past it.
*
* @param {string|null} cursor
* @param {string} prefix - `<space>/<did>/`
* @param {boolean} reverse
* @returns {string|null|false} false when no row is past the cursor
*/
function storageCursor(cursor, prefix, reverse) {
if (!cursor) return null;
if (cursor.startsWith(prefix)) return cursor.slice(prefix.length);
return cursor < prefix === reverse ? null : false;
}
/**
* atproto's JSON representation of a byte string.
* @param {Uint8Array} bytes
* @returns {{$bytes: string}}
*/
function toJsonBytes(bytes) {
let binary = '';
for (const b of bytes) binary += String.fromCharCode(b);
return { $bytes: btoa(binary).replace(/=+$/, '') };
}
/**
* @param {import('../mac.js').Commit} commit
* @returns {Object}
*/
function toJsonCommit(commit) {
return {
ver: commit.ver,
repo: commit.repo,
space: commit.space,
rev: commit.rev,
data: toJsonBytes(commit.data),
};
}
/**
* A commit over the repo's current state. Undefined when the repo has never
* been written to.
*
* @param {import('@pdsjs/core/ports').SpaceStoragePort} spaceStorage
* @param {string} space
* @param {string} repo
* @returns {Promise<import('../mac.js').Commit|undefined>}
*/
async function buildCommit(spaceStorage, space, repo) {
const state = await spaceStorage.getSpaceRepo(space);
if (!state?.setHash || !state.rev) return undefined;
return RepoCommit.fromState(state.setHash).toCommit({
space,
repo,
rev: state.rev,
});
}
/**
* Append a value to a response header that may already hold others.
* @param {Headers} headers
* @param {string} name
* @param {string} value
*/
function appendHeader(headers, name, value) {
const current = headers.get(name);
headers.set(name, current ? `${current}, ${value}` : value);
}
/**
* Every blob CID a space's records reference, paired with the rev of the commit
* that wrote the referencing record.
*
* Read from the records themselves rather than from a link table. A space repo
* on a personal server is small, and a table that can drift is the one thing a
* blob's reachability must not rest on.
*
* @param {import('@pdsjs/core/ports').SpaceStoragePort} spaceStorage
* @param {string} space
* @returns {Promise<Map<string, string>>} blob CID -> earliest referencing rev
*/
async function spaceBlobRefs(spaceStorage, space) {
/** @type {Map<string, string>} */
const refs = new Map();
const paths = await spaceStorage.listAllSpaceRecords(space);
for (const path of paths) {
const row = await spaceStorage.getSpaceRecord(
space,
path.collection,
path.rkey,
);
if (!row) continue;
for (const cid of findBlobRefs(cborDecode(row.value))) {
const seen = refs.get(cid);
if (seen === undefined || row.repoRev < seen) refs.set(cid, row.repoRev);
}
}
return refs;
}
/**
* @param {Object} ctx
* @param {import('@pdsjs/core/ports').SpaceStoragePort} ctx.spaceStorage
* @param {import('@pdsjs/core/ports').BlobPort} ctx.blobs
* @param {() => Promise<string|null>} ctx.getDid
* @param {() => Promise<{privateKey: Uint8Array, didKey: string}|null>} ctx.getKemKey
* @param {import('@pdsjs/core/ports').DidResolverPort} ctx.resolveDid
* @param {import('@pdsjs/core/ports').SignatureVerifierPort} ctx.verifier
* @returns {import('@pdsjs/core/pds').Routes}
*/
export function createReadRoutes(ctx) {
const { spaceStorage, blobs, getDid, getKemKey, resolveDid, verifier } = ctx;
const authorizeRead = createReadAuthorizer({
spaceStorage,
getDid,
resolveDid,
verifier,
});
/**
* The repo a request targets. `repo` is optional on the wire; this PDS hosts
* exactly one account, so anything else is a miss rather than a proxy.
* @param {string|null} repo
* @returns {Promise<string|Response>}
*/
async function resolveRepo(repo) {
const did = await getDid();
if (!did) return errorResponse('RepoNotFound', 'Server not initialised');
if (repo && repo !== did) {
return errorResponse('RepoNotFound', `Not hosted here: ${repo}`, 404);
}
return did;
}
/**
* The secret to MAC this response under, undefined when the account's own
* session reads without asking for one, or the refusal to return.
*
* @param {Request} request
* @param {{did: string}|null} auth
* @param {import('../mac.js').MacType} type
* @returns {Promise<Uint8Array|undefined|Response>}
*/
async function macSecret(request, auth, type) {
const header = request.headers.get(WANT_MAC_HEADER);
const hosted = await getDid();
const viaSession = Boolean(auth && auth.did === hosted);
if (header === null && viaSession) return undefined;
/** @type {string} */
let keyShare;
try {
keyShare = parseWantMacHeader(header, type);
} catch (err) {
if (err instanceof SpaceMacError) {
return errorResponse('BadSpaceMac', err.message);
}
throw err;
}
const kemKey = await getKemKey();
if (!kemKey) {
return errorResponse(
'InternalServerError',
'This account has no KEM key yet',
500,
);
}
try {
return await decapsulate(kemKey.privateKey, keyShare);
} catch (err) {
if (err instanceof SpaceMacError) {
return errorResponse('BadSpaceMac', err.message);
}
throw err;
}
}
/**
* A response that may carry a MAC. It varies on the request header, and a
* browser app may read the MAC from a cross-origin response.
*
* @param {BodyInit|null} body
* @param {Record<string, string>} headers
* @param {string|null} mac - the Atproto-Space-Mac value, when there is one
* @returns {Response}
*/
function macResponse(body, headers, mac) {
const response = new Response(body, { headers });
appendHeader(response.headers, 'Vary', WANT_MAC_HEADER);
if (mac) {
response.headers.set(MAC_HEADER, mac);
appendHeader(
response.headers,
'Access-Control-Expose-Headers',
MAC_HEADER,
);
}
return response;
}
/**
* @param {unknown} body
* @returns {[string, Record<string, string>]}
*/
function json(body) {
return [JSON.stringify(body), { 'content-type': 'application/json' }];
}
/** @type {import('@pdsjs/core/pds').Route} */
const latestCommitRoute = {
auth: 'optional',
handler: async (request, url, auth) => {
const target = readTarget(url);
if (target instanceof Response) return target;
const denied = await authorizeRead(request, target.space, auth);
if (denied) return denied;
const did = await resolveRepo(target.repo);
if (did instanceof Response) return did;
const secret = await macSecret(request, auth, COMMIT_MAC_TYPE);
if (secret instanceof Response) return secret;
const commit = await buildCommit(spaceStorage, target.space, did);
if (!commit) return macResponse(...json({}), null);
const mac = secret
? formatMacHeader(
COMMIT_MAC_TYPE,
await computeCommitMac(secret, commit),
)
: null;
return macResponse(...json({ commit: toJsonCommit(commit) }), mac);
},
};
return {
'/xrpc/com.atproto.space.getRecord': {
auth: 'optional',
handler: async (request, url, auth) => {
const target = readTarget(url);
if (target instanceof Response) return target;
const denied = await authorizeRead(request, target.space, auth);
if (denied) return denied;
const did = await resolveRepo(target.repo);
if (did instanceof Response) return did;
const collection = url.searchParams.get('collection');
const rkey = url.searchParams.get('rkey');
if (!collection || !rkey) {
return errorResponse(
'InvalidRequest',
'collection and rkey are required',
);
}
const secret = await macSecret(request, auth, RECORD_MAC_TYPE);
if (secret instanceof Response) return secret;
const row = await spaceStorage.getSpaceRecord(
target.space,
collection,
rkey,
);
if (!row) {
return errorResponse(
'RecordNotFound',
`Record not found: ${collection}/${rkey}`,
404,
);
}
const uri = `${target.space}/${did}/${collection}/${rkey}`;
const mac = secret
? formatMacHeader(
RECORD_MAC_TYPE,
await computeRecordMac(secret, uri, row.cid),
)
: null;
return macResponse(
...json({ uri, cid: row.cid, value: cborDecodeJson(row.value) }),
mac,
);
},
},
'/xrpc/com.atproto.space.listRecords': {
auth: 'optional',
handler: async (request, url, auth) => {
const target = readTarget(url);
if (target instanceof Response) return target;
const denied = await authorizeRead(request, target.space, auth);
if (denied) return denied;
const did = await resolveRepo(target.repo);
if (did instanceof Response) return did;
const limit = readLimit(url, 'limit', 50, 1000);
if (limit instanceof Response) return limit;
const prefix = `${target.space}/${did}/`;
const reverse = url.searchParams.get('reverse') === 'true';
const after = storageCursor(
url.searchParams.get('cursor'),
prefix,
reverse,
);
if (after === false) return Response.json({ records: [] });
const { records, cursor } = await spaceStorage.listSpaceRecords(
target.space,
{
collection: url.searchParams.get('collection'),
cursor: after,
limit,
reverse,
excludeValues: url.searchParams.get('excludeValues') === 'true',
},
);
return Response.json({
...(cursor ? { cursor: prefix + cursor } : {}),
records: records.map((r) => ({
collection: r.collection,
rkey: r.rkey,
cid: r.cid,
...(r.value === undefined
? {}
: { value: cborDecodeJson(r.value) }),
})),
});
},
},
'/xrpc/com.atproto.space.listRepoOps': {
auth: 'optional',
handler: async (request, url, auth) => {
const target = readTarget(url);
if (target instanceof Response) return target;
const denied = await authorizeRead(request, target.space, auth);
if (denied) return denied;
const did = await resolveRepo(target.repo);
if (did instanceof Response) return did;
const limit = readLimit(url, 'limit', 100, 1000);
if (limit instanceof Response) return limit;
const excludeValues = url.searchParams.get('excludeValues') === 'true';
const secret = await macSecret(request, auth, COMMIT_MAC_TYPE);
if (secret instanceof Response) return secret;
const ops = await spaceStorage.listSpaceOps(
target.space,
url.searchParams.get('since'),
limit,
);
// Inline each op's record value so a syncer advances without a
// getRecord round-trip per write. A value is omitted for deletes, when
// excludeValues is set, and when a later op superseded it — the stored
// record is the current one, so it is only this op's value if the CIDs
// still agree.
const entries = [];
for (const op of ops) {
/** @type {Record<string, unknown>} */
const entry = {
rev: op.rev,
collection: op.collection,
rkey: op.rkey,
cid: op.cid,
prev: op.prev,
};
if (!excludeValues && op.cid) {
const row = await spaceStorage.getSpaceRecord(
target.space,
op.collection,
op.rkey,
);
if (row && row.cid === op.cid) {
entry.value = cborDecodeJson(row.value);
}
}
entries.push(entry);
}
/** @type {Record<string, unknown>} */
const body = { ops: entries };
/** @type {string|null} */
let mac = null;
// The commit describes the head, so it is only meaningful once the
// response has reached it. A full page may have more behind it. A
// page without a commit has nothing to MAC: its ops are checked once
// the final page's commit is compared against the syncer's set hash.
if (ops.length < limit) {
const commit = await buildCommit(spaceStorage, target.space, did);
if (commit) {
body.commit = toJsonCommit(commit);
if (secret) {
mac = formatMacHeader(
COMMIT_MAC_TYPE,
await computeCommitMac(secret, commit),
);
}
}
} else {
body.cursor = ops[ops.length - 1].rev;
}
return macResponse(...json(body), mac);
},
},
'/xrpc/com.atproto.space.getLatestCommit': latestCommitRoute,
// The same handler under the name this method shipped as before the draft
// renamed it. Both are in use, and a client that knows only one reads the
// whole feature as missing.
'/xrpc/com.atproto.space.getRepoState': latestCommitRoute,
// Blobs are uploaded through com.atproto.repo.uploadBlob and referenced
// from a space record, so a client writing blob-bearing records needs a
// blob permission alongside its space permission.
'/xrpc/com.atproto.space.listBlobs': {
auth: 'optional',
handler: async (request, url, auth) => {
const target = readTarget(url);
if (target instanceof Response) return target;
const denied = await authorizeRead(request, target.space, auth);
if (denied) return denied;
const did = await resolveRepo(target.repo);
if (did instanceof Response) return did;
const limit = readLimit(url, 'limit', 500, 1000);
if (limit instanceof Response) return limit;
const since = url.searchParams.get('since');
const cursor = url.searchParams.get('cursor');
const refs = await spaceBlobRefs(spaceStorage, target.space);
const cids = [...refs]
.filter(([, rev]) => (since ? rev > since : true))
.map(([cid]) => cid)
.filter((cid) => (cursor ? cid > cursor : true))
.sort();
const page = cids.slice(0, limit);
return Response.json({
cids: page,
...(page.length < limit ? {} : { cursor: page[page.length - 1] }),
});
},
},
// Scoped to the space: a credential admits its holder to that space, not to
// the account's whole blob store. The reference serves any blob the account
// holds once the space read passes.
'/xrpc/com.atproto.space.getBlob': {
auth: 'optional',
handler: async (request, url, auth) => {
const target = readTarget(url);
if (target instanceof Response) return target;
const denied = await authorizeRead(request, target.space, auth);
if (denied) return denied;
const did = await resolveRepo(target.repo);
if (did instanceof Response) return did;
const cid = url.searchParams.get('cid');
if (!cid) return errorResponse('InvalidRequest', 'cid is required');
const refs = await spaceBlobRefs(spaceStorage, target.space);
if (!refs.has(cid)) {
return errorResponse(
'BlobNotFound',
`No record in ${target.space} references ${cid}`,
404,
);
}
const blob = await blobs.get(did, cid);
if (!blob) {
return errorResponse('BlobNotFound', 'Blob not found', 404);
}
return new Response(/** @type {BodyInit} */ (blob.data), {
headers: {
'Content-Type': blob.mimeType,
'X-Content-Type-Options': 'nosniff',
// Force download and block execution so a hostile blob cannot run
// in the PDS origin.
'Content-Disposition': `attachment; filename="${cid}"`,
'Content-Security-Policy': "default-src 'none'; sandbox",
},
});
},
},
'/xrpc/com.atproto.space.getRepo': {
auth: 'optional',
handler: async (request, url, auth) => {
const target = readTarget(url);
if (target instanceof Response) return target;
const denied = await authorizeRead(request, target.space, auth);
if (denied) return denied;
const did = await resolveRepo(target.repo);
if (did instanceof Response) return did;
const secret = await macSecret(request, auth, COMMIT_MAC_TYPE);
if (secret instanceof Response) return secret;
const commit = await buildCommit(spaceStorage, target.space, did);
if (!commit) {
return errorResponse(
'RepoNotFound',
`Could not find repo for space: ${target.space}`,
404,
);
}
const paths = await spaceStorage.listAllSpaceRecords(target.space);
/** @type {import('../star.js').SerializedRecord[]} */
const records = [];
for (const p of paths) {
const row = await spaceStorage.getSpaceRecord(
target.space,
p.collection,
p.rkey,
);
if (row) {
records.push({
collection: p.collection,
rkey: p.rkey,
bytes: row.value,
});
}
}
const star = serializeRepo(commit, records);
// Copy into a plain ArrayBuffer: a Uint8Array view is not a BodyInit
// per the DOM lib types, and `.buffer` widens to ArrayBufferLike, which
// includes SharedArrayBuffer.
const body = new ArrayBuffer(star.byteLength);
new Uint8Array(body).set(star);
const mac = secret
? formatMacHeader(
COMMIT_MAC_TYPE,
await computeCommitMac(secret, commit),
)
: null;
return macResponse(
body,
{ 'content-type': 'application/octet-stream' },
mac,
);
},
},
};
}