chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
// The com.atproto.space and com.atproto.memberspace calls this app makes.
//
// Two roles answer, and they are rarely the same server. A repo host serves the
// records one account wrote, so a read is addressed to that account's own PDS.
// An authority serves the space itself: its config, its members, and the list
// of repos that hold data in it. Every function here names which one it asks.
import { verifyCommit } from '@pdsjs/spaces/commit';
import {
COMMIT_MAC_TYPE,
MAC_HEADER,
parseMacHeader,
RECORD_MAC_TYPE,
} from '@pdsjs/spaces/mac';
import { verifyRecord } from '@pdsjs/spaces/star';
import { parseSpaceUri } from '@pdsjs/spaces/uri';
import {
credentialFetch,
repoHost,
spaceCredential,
wantMac,
} from '#/lib/credential.js';
import { authedFetch } from '#/lib/oauth.js';
import { policyUnion } from '#/lib/policy.js';
/** An XRPC method that answered with an error. */
export class XrpcError extends Error {
/**
* @param {string} name - the error name the method returned
* @param {string} message
* @param {number} status
*/
constructor(name, message, status) {
super(message);
this.name = name;
this.status = status;
}
}
/**
* @param {Response} response
* @returns {Promise<any>}
*/
async function body(response) {
if (response.ok) return response.json();
const failure = await response.json().catch(() => ({}));
throw new XrpcError(
failure.error || 'RequestFailed',
failure.message || `The server answered ${response.status}.`,
response.status,
);
}
/**
* @param {string} host
* @param {string} method
* @param {Record<string, string|number|boolean|null|undefined>} [params]
* @returns {string}
*/
function endpoint(host, method, params = {}) {
const url = new URL(`${host}/xrpc/${method}`);
for (const [key, value] of Object.entries(params)) {
if (value !== null && value !== undefined && value !== '') {
url.searchParams.set(key, String(value));
}
}
return url.toString();
}
/** The reads a host answers with a MAC when they come with a credential. */
const MAC_TYPES =
/** @type {Record<string, import('@pdsjs/spaces/mac').MacType>} */ ({
'com.atproto.space.getRecord': RECORD_MAC_TYPE,
'com.atproto.space.getLatestCommit': COMMIT_MAC_TYPE,
'com.atproto.space.listRepoOps': COMMIT_MAC_TYPE,
});
/**
* @param {string} b64 - base64, padded or not
* @returns {Uint8Array}
*/
function fromBase64(b64) {
return Uint8Array.from(
atob(b64.padEnd(Math.ceil(b64.length / 4) * 4, '=')),
(c) => c.charCodeAt(0),
);
}
/**
* Whether an answer is the author's, by the MAC it carries: a record must be
* the one asked for under a MAC over its uri and cid, a commit must be for
* this repo and space under a MAC over its data. An answer with no commit in
* it carries no MAC.
*
* @param {Response} response
* @param {Object} read
* @param {string} read.method
* @param {string} read.space
* @param {string} read.repo
* @param {Record<string, string|number|boolean|null|undefined>} read.params
* @param {Uint8Array} read.sharedSecret
* @returns {Promise<boolean>}
*/
async function macHolds(
response,
{ method, space, repo, params, sharedSecret },
) {
const answer = await response.clone().json();
const type = MAC_TYPES[method];
if (type === COMMIT_MAC_TYPE && !answer.commit) return true;
let mac;
try {
mac = parseMacHeader(response.headers.get(MAC_HEADER), type);
} catch {
return false;
}
if (type === RECORD_MAC_TYPE) {
const uri = `${space}/${repo}/${params.collection}/${params.rkey}`;
return verifyRecord(answer, { uri, sharedSecret, mac });
}
const { commit } = answer;
return verifyCommit(
{ ...commit, data: fromBase64(commit.data?.$bytes ?? '') },
{ space, repo, sharedSecret, mac },
);
}
/**
* Read from a repo host. The signed-in account's own host answers to its
* session; anybody else's answers only to a credential for the space, and a
* commit or record read that way is checked against the MAC it comes with.
*
* @param {import('#/lib/oauth.js').Session} session
* @param {Object} target
* @param {string} target.space
* @param {string} [target.repo] - whose repo, default the signed-in account
* @param {string} method
* @param {Record<string, string|number|boolean|null|undefined>} [params]
* @returns {Promise<Response>}
*/
async function readFromHost(session, target, method, params = {}) {
const repo = target.repo || session.did;
if (repo === session.did) {
return authedFetch(
session,
endpoint(session.pds, method, { space: target.space, ...params }),
);
}
const host = await repoHost(repo);
const credential = await spaceCredential(session, target.space);
const url = endpoint(host, method, { space: target.space, repo, ...params });
const type = MAC_TYPES[method];
if (!type) return credentialFetch(credential, url);
const { headers, sharedSecret } = await wantMac(repo, type);
const response = await credentialFetch(credential, url, { headers });
if (
response.ok &&
!(await macHolds(response, {
method,
space: target.space,
repo,
params,
sharedSecret,
}))
) {
throw new XrpcError(
'BadSpaceMac',
`The answer from ${host} does not carry its author's MAC.`,
response.status,
);
}
return response;
}
/**
* The PDS that hosts a space's authority.
* @param {string} space
* @returns {Promise<string>}
*/
async function authorityHost(space) {
return repoHost(parseSpaceUri(space).spaceDid);
}
// ---- the authority's view of a space ----
/**
* Every space the signed-in account holds a repo in. This is not "spaces I am a
* member of": a PDS tracks the spaces its own user has written to.
*
* @param {import('#/lib/oauth.js').Session} session
* @param {{type?: string, did?: string, cursor?: string, limit?: number}} [options]
* @returns {Promise<{spaces: Array<{uri: string, isOwner?: boolean}>, cursor?: string}>}
*/
export async function listSpaces(session, options = {}) {
return body(
await authedFetch(
session,
endpoint(session.pds, 'com.atproto.space.listSpaces', {
type: options.type,
did: options.did,
cursor: options.cursor,
limit: options.limit ?? 100,
}),
),
);
}
/**
* Ask the authority that governs a space.
*
* The signed-in account's own server answers to its session. Anybody else's is
* given a space credential, presented the way a repo host is given one: hosts
* differ on whether these methods need authentication at all, and a credential
* satisfies the ones that do without changing the answer from the ones that
* do not.
*
* @param {import('#/lib/oauth.js').Session} session
* @param {string} space
* @param {string} method
* @returns {Promise<Response>}
*/
/**
* @typedef {Object} Source
* @property {string} method - the XRPC method that answered
* @property {string} host - the server it was addressed to
* @property {'session'|'credential'|'anonymous'} auth - what was presented
* @property {string} [after] - a method tried first that the host did not serve
*/
/**
* @param {import('#/lib/oauth.js').Session} session
* @param {string} space - the space at-uri
* @param {string} method - the XRPC method to ask for
*/
async function askAuthority(session, space, method) {
const { spaceDid } = parseSpaceUri(space);
const host = await authorityHost(space);
const url = endpoint(host, method, { space });
/** @param {Response} response @param {Source['auth']} auth */
const answered = (response, auth) => ({
response,
source: { method, host, auth },
});
if (spaceDid === session.did) {
return answered(await authedFetch(session, url), 'session');
}
try {
const credential = await spaceCredential(session, space);
return answered(await credentialFetch(credential, url), 'credential');
} catch {
// No credential to be had, which is what a space this account has not been
// admitted to looks like. A host that serves the method openly still
// answers, so the anonymous read is worth making before reporting a
// failure.
return answered(await fetch(url), 'anonymous');
}
}
/**
* A space's policy, from the authority that governs it. The authority answers
* only its owner's session, so the page asks this only of a space this account
* hosts.
*
* @param {import('#/lib/oauth.js').Session} session
* @param {string} space
* @returns {Promise<any>}
*/
export async function getSpace(session, space) {
const asked = await askAuthority(
session,
space,
'com.atproto.space.getSpace',
);
return { value: await body(asked.response), source: asked.source };
}
/**
* The accounts an authority admits to a space. Only the authority's own account
* may read this, so it is offered for a space this user hosts.
*
* @param {import('#/lib/oauth.js').Session} session
* @param {string} space
* @returns {Promise<{members: Array<{did: string, read: boolean, write: boolean}>, cursor?: string}>}
*/
export async function listMembers(session, space) {
return body(
await authedFetch(
session,
endpoint(session.pds, 'com.atproto.memberspace.listMembers', {
space,
limit: 1000,
}),
),
);
}
/**
* The writer set: the repos the authority has heard from. It is the sync
* boundary rather than an access list, so it enumerates writers and never
* readers.
*
* A credential, even for a space this account is the authority of. The
* reference holds this method to `spaceCredentialAuth`, which admits no
* session at all, and a host that serves it openly accepts a credential just
* the same. The anonymous read stays as a last resort for a host that wants
* nothing, which is what an older pds.js is.
*
* @param {import('#/lib/oauth.js').Session} session
* @param {string} space
* @returns {Promise<{repos: Array<{did: string, rev: string, data: {$bytes: string}}>, cursor?: string}>}
*/
export async function listRepos(session, space) {
const url = endpoint(
await authorityHost(space),
'com.atproto.space.listRepos',
{ space },
);
try {
return body(
await credentialFetch(await spaceCredential(session, space), url),
);
} catch (error) {
if (error instanceof XrpcError) throw error;
return body(await fetch(url));
}
}
/**
* Where a read came from, in the words a page uses to say so.
* @param {Source} source
* @returns {string}
*/
export function sourceAuthLabel(source) {
if (source.auth === 'session') return 'your own session';
if (source.auth === 'credential') return 'a space credential';
return 'no credentials at all';
}
// ---- one repo's records ----
/**
* @param {import('#/lib/oauth.js').Session} session
* @param {{space: string, repo?: string}} target
* @param {{collection?: string, cursor?: string, limit?: number, reverse?: boolean, excludeValues?: boolean}} [options]
* @returns {Promise<{records: Array<{collection: string, rkey: string, cid: string, value?: any}>, cursor?: string}>}
*/
export async function listRecords(session, target, options = {}) {
return body(
await readFromHost(session, target, 'com.atproto.space.listRecords', {
collection: options.collection,
cursor: options.cursor,
limit: options.limit ?? 100,
reverse: options.reverse ? 'true' : undefined,
excludeValues: options.excludeValues ? 'true' : undefined,
}),
);
}
// A page beyond this many records is a repo nobody browses by hand, and the
// shape of it is already clear from what came back.
const INDEX_PAGE_LIMIT = 20;
/**
* Every record path in one repo's slice of a space, without the values.
*
* No endpoint reports which collections a space holds, so the page counts them
* from the paths themselves. Values are excluded because a heading only needs
* to know a record exists.
*
* @param {import('#/lib/oauth.js').Session} session
* @param {{space: string, repo?: string}} target
* @returns {Promise<{records: Array<{collection: string, rkey: string, cid: string}>, complete: boolean}>}
*/
export async function listAllRecords(session, target) {
/** @type {Array<{collection: string, rkey: string, cid: string}>} */
const records = [];
let cursor;
for (let page = 0; page < INDEX_PAGE_LIMIT; page += 1) {
const answer = await listRecords(session, target, {
cursor,
limit: 1000,
excludeValues: true,
});
records.push(...answer.records);
cursor = answer.cursor;
if (!cursor) return { records, complete: true };
}
return { records, complete: false };
}
/**
* @param {import('#/lib/oauth.js').Session} session
* @param {{space: string, repo?: string}} target
* @param {string} collection
* @param {string} rkey
* @returns {Promise<{uri: string, cid: string, value: any}>}
*/
export async function getRecord(session, target, collection, rkey) {
return body(
await readFromHost(session, target, 'com.atproto.space.getRecord', {
collection,
rkey,
}),
);
}
/**
* The write log, newest last. Each entry carries the record's current value
* when no later write superseded it.
*
* @param {import('#/lib/oauth.js').Session} session
* @param {{space: string, repo?: string}} target
* @param {{since?: string, limit?: number}} [options]
* @returns {Promise<{ops: Array<{rev: string, collection: string, rkey: string, cid: string|null, prev: string|null, value?: any}>, cursor?: string, commit?: any}>}
*/
export async function listRepoOps(session, target, options = {}) {
return body(
await readFromHost(session, target, 'com.atproto.space.listRepoOps', {
since: options.since,
limit: options.limit ?? 200,
excludeValues: 'true',
}),
);
}
/**
* The commit over a repo's current state in a space, or null for a repo that
* has never been written to.
*
* @param {import('#/lib/oauth.js').Session} session
* @param {{space: string, repo?: string}} target
* @returns {Promise<any|null>}
*/
export async function getLatestCommit(session, target) {
const response = await readFromHost(
session,
target,
'com.atproto.space.getLatestCommit',
);
const answer = await body(response);
return answer.commit ?? null;
}
/**
* @param {import('#/lib/oauth.js').Session} session
* @param {{space: string, repo?: string}} target
* @returns {Promise<{cids: string[], cursor?: string}>}
*/
export async function listBlobs(session, target) {
return body(
await readFromHost(session, target, 'com.atproto.space.listBlobs', {
limit: 1000,
}),
);
}
/**
* One blob a space's records reference. The PDS serves it with a
* content-disposition that stops it from being displayed in its own origin, so
* this reads the bytes and hands back an object URL for this page to show.
*
* @param {import('#/lib/oauth.js').Session} session
* @param {{space: string, repo?: string}} target
* @param {string} cid
* @returns {Promise<{url: string, type: string, size: number}>}
*/
export async function getBlob(session, target, cid) {
const response = await readFromHost(
session,
target,
'com.atproto.space.getBlob',
{ cid },
);
if (!response.ok) await body(response);
const blob = await response.blob();
return { url: URL.createObjectURL(blob), type: blob.type, size: blob.size };
}
// ---- writes, all of them against the signed-in account's own repo ----
/**
* @param {import('#/lib/oauth.js').Session} session
* @param {string} method
* @param {Record<string, unknown>} input
* @returns {Promise<any>}
*/
async function post(session, method, input) {
return body(
await authedFetch(session, `${session.pds}/xrpc/${method}`, {
method: 'POST',
headers: { 'content-type': 'application/json' },
body: JSON.stringify(input),
}),
);
}
/**
* A record write, addressed at the signed-in account's own repo. `repo` is what
* a space write names, and the host refuses one that names anybody else, so
* every caller here sends the same value.
*
* @param {import('#/lib/oauth.js').Session} session
* @param {string} method
* @param {Record<string, unknown>} input
* @returns {Promise<any>}
*/
function write(session, method, input) {
return post(session, method, { repo: session.did, ...input });
}
/**
* @param {import('#/lib/oauth.js').Session} session
* @param {{space: string, collection: string, rkey?: string, record: unknown}} input
* @returns {Promise<{uri: string, cid: string}>}
*/
export function createRecord(session, input) {
return write(session, 'com.atproto.space.createRecord', input);
}
/**
* @param {import('#/lib/oauth.js').Session} session
* @param {{space: string, collection: string, rkey: string, record: unknown}} input
* @returns {Promise<{uri: string, cid: string}>}
*/
export function putRecord(session, input) {
return write(session, 'com.atproto.space.putRecord', input);
}
/**
* @param {import('#/lib/oauth.js').Session} session
* @param {{space: string, collection: string, rkey: string}} input
* @returns {Promise<any>}
*/
export function deleteRecord(session, input) {
return write(session, 'com.atproto.space.deleteRecord', input);
}
/**
* Create a space governed by the memberspace policy: its members alone read
* and write, with any app.
*
* @param {import('#/lib/oauth.js').Session} session
* @param {{type: string, skey?: string}} input
* @returns {Promise<{uri: string}>}
*/
export function createSpace(session, { type, skey }) {
return post(session, 'com.atproto.space.createSpace', {
spaceType: type,
...(skey ? { skey } : {}),
policy: policyUnion('memberspace'),
});
}
/**
* Replace a space's policy. A policy of another kind discards the state the
* previous one held, such as the member list.
*
* @param {import('#/lib/oauth.js').Session} session
* @param {{space: string, policy: unknown}} input
* @returns {Promise<any>}
*/
export function updateSpace(session, input) {
return post(session, 'com.atproto.space.updateSpace', input);
}
/**
* Delete a space at the authority that governs it. The records stay: they
* belong to the members who wrote them, and the proposal has each repo host
* keep its own.
*
* @param {import('#/lib/oauth.js').Session} session
* @param {{space: string}} input
* @returns {Promise<any>}
*/
export function deleteSpace(session, input) {
return post(session, 'com.atproto.space.deleteSpace', input);
}
/**
* @param {import('#/lib/oauth.js').Session} session
* @param {{space: string, did: string, read: boolean, write: boolean}} input
* @returns {Promise<any>}
*/
export function putMember(session, input) {
return post(session, 'com.atproto.memberspace.putMember', input);
}
/**
* @param {import('#/lib/oauth.js').Session} session
* @param {{space: string, did: string}} input
* @returns {Promise<any>}
*/
export function removeMember(session, input) {
return post(session, 'com.atproto.memberspace.removeMember', input);
}