chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
/**
* Minimal XRPC client over fetch, covering only what the git remote helper
* needs: sessions, record reads and CAS writes, blob transfer, and the
* com.atproto.space endpoints behind private repositories.
*/
import { createDpopProof, importDpopKey } from './dpop.js';
import { refreshOauthSession } from './oauth.js';
import {
encapsulate,
formatWantMacHeader,
MAC_HEADER,
parseMacHeader,
RECORD_MAC_TYPE,
verifyRecord,
WANT_MAC_HEADER,
} from './space-mac.js';
import { createSpaceSigHeaders } from './space-signature.js';
import { XrpcError } from './xrpc-error.js';
export { XrpcError };
/**
* @param {Response} res
* @returns {Promise<never>}
*/
async function throwXrpcError(res) {
/** @type {{error?: string, message?: string}} */
let body = {};
try {
body = /** @type {{error?: string, message?: string}} */ (await res.json());
} catch {
// Non-JSON error body
}
throw new XrpcError(
res.status,
body.error ?? 'Unknown',
body.message ?? `request failed with status ${res.status}`,
);
}
/**
* How space reads authenticate. The hosted account's own session, or a space
* credential presented under the Atproto-Space-Credential scheme, signed for
* the account the read is addressed to.
* @typedef {{kind: 'session'} |
* {kind: 'credential', credential: string, key: import('./space-signature.js').SpaceSigKey}} SpaceReadAuth
*/
export class XrpcClient {
/** @param {string} service - PDS base URL, e.g. https://pds.example.com */
constructor(service) {
this.service = service.replace(/\/+$/, '');
/** @type {string|null} */
this.accessJwt = null;
/** @type {{identifier: string, password: string}|null} */
this.credentials = null;
/** @type {string|null} */
this.did = null;
/** @type {SpaceReadAuth|null} */
this.spaceAuth = null;
/**
* The account whose repo this client reads in a space, which a space
* credential's signature names as its audience.
* @type {string|null}
*/
this.readsRepo = null;
/**
* Looks up an account's `#atproto_kem` key, which a read with a space
* credential asks for a MAC against.
* @type {((did: string) => Promise<string>)|null}
*/
this.resolveKemKey = null;
/** @type {{session: import('./oauth.js').OauthSession, key: import('./dpop.js').DpopKey, onUpdate: (session: import('./oauth.js').OauthSession) => void}|null} */
this.oauth = null;
}
/**
* Authenticate this client with a stored OAuth session instead of a
* password session. Every authenticated request then carries the DPoP-bound
* access token with a per-request proof.
* @param {import('./oauth.js').OauthSession} session
* @param {(session: import('./oauth.js').OauthSession) => void} [onUpdate] -
* called with the replacement session whenever a refresh rotates the pair
* @returns {Promise<void>}
*/
async setOauthSession(session, onUpdate = () => {}) {
this.oauth = {
session,
key: await importDpopKey(session.privateJwk),
onUpdate,
};
this.did = session.did;
}
/**
* Refresh the OAuth pair when the access token is at or past expiry. The
* refresh token rotates, so the caller's store learns the new pair through
* onUpdate; failing that refresh means the session is gone for good.
* @returns {Promise<void>}
*/
async ensureFreshAccessToken() {
const oauth = this.oauth;
if (!oauth || oauth.session.expiresAt - Date.now() > 30_000) return;
try {
oauth.session = await refreshOauthSession(oauth.session, oauth.key);
} catch (err) {
throw new XrpcError(
401,
'SessionExpired',
`stored OAuth session could not be refreshed (${err instanceof Error ? err.message : err}); run git-remote-atproto login`,
);
}
oauth.onUpdate(oauth.session);
}
/** @returns {Record<string, string>} */
authHeaders() {
return this.accessJwt ? { Authorization: `Bearer ${this.accessJwt}` } : {};
}
/**
* Fetch with this client's account auth: the OAuth access token under the
* DPoP scheme with a per-request proof, or the password session's Bearer
* token, or nothing. A 401 gets one recovery and a retry: OAuth mode
* refreshes the token pair, and a password session created by this client
* signs in again with the kept credentials. The password path is what
* keeps a long-lived process (the git-ci runner) writing after its access
* token expires.
* @param {string} url
* @param {{method?: string, headers?: Record<string, string>, body?: BodyInit}} [init]
* @returns {Promise<Response>}
*/
async authedFetch(url, init = {}) {
const { method = 'GET', headers = {}, body } = init;
if (!this.oauth) {
const attempt = () =>
fetch(url, {
method,
headers: { ...headers, ...this.authHeaders() },
body,
});
let res = await attempt();
if (res.status === 401 && this.credentials) {
await res.arrayBuffer();
await this.createSession(
this.credentials.identifier,
this.credentials.password,
);
res = await attempt();
}
return res;
}
await this.ensureFreshAccessToken();
const attempt = async () => {
const oauth = /** @type {NonNullable<typeof this.oauth>} */ (this.oauth);
const token = oauth.session.accessToken;
return fetch(url, {
method,
headers: {
...headers,
Authorization: `DPoP ${token}`,
DPoP: await createDpopProof({
key: oauth.key,
method,
url,
credential: token,
}),
},
body,
});
};
let res = await attempt();
if (res.status === 401) {
await res.arrayBuffer();
this.oauth.session.expiresAt = 0;
await this.ensureFreshAccessToken();
res = await attempt();
}
return res;
}
/**
* Headers for a space read presenting a space credential, signed for the
* repo the URL names or else the repo this client reads.
* @param {string} url
* @returns {Promise<Record<string, string>>}
*/
async credentialHeaders(url) {
const spaceAuth = this.spaceAuth;
if (spaceAuth?.kind !== 'credential') {
throw new Error('no space credential is set');
}
const audience = new URL(url).searchParams.get('repo') ?? this.readsRepo;
if (!audience) throw new Error('no repo to address the read to');
return createSpaceSigHeaders(spaceAuth.key, {
authorization: `Atproto-Space-Credential ${spaceAuth.credential}`,
audience,
});
}
/**
* Fetch a space read endpoint with whichever read auth is set: a space
* credential, or the account's own session.
* @param {string} url
* @param {Record<string, string>} [headers]
* @returns {Promise<Response>}
*/
spaceFetch(url, headers = {}) {
if (this.spaceAuth?.kind === 'credential') {
return (async () =>
fetch(url, {
headers: { ...headers, ...(await this.credentialHeaders(url)) },
}))();
}
return this.authedFetch(url, { headers });
}
/**
* @param {string} identifier
* @param {string} password
* @returns {Promise<{did: string, accessJwt: string}>}
*/
async createSession(identifier, password) {
const res = await fetch(
`${this.service}/xrpc/com.atproto.server.createSession`,
{
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ identifier, password }),
},
);
if (!res.ok) await throwXrpcError(res);
const session = /** @type {{did: string, accessJwt: string}} */ (
await res.json()
);
this.accessJwt = session.accessJwt;
this.credentials = { identifier, password };
this.did = session.did;
return session;
}
/**
* @param {string} handle
* @returns {Promise<string>} the DID
*/
async resolveHandle(handle) {
const res = await fetch(
`${this.service}/xrpc/com.atproto.identity.resolveHandle?handle=${encodeURIComponent(handle)}`,
);
if (!res.ok) await throwXrpcError(res);
const body = /** @type {{did: string}} */ (await res.json());
return body.did;
}
/**
* Whether a space exists, asked with no credential. A space that exists
* answers with its config to its owner and with a refusal to anyone else;
* only a space that does not exist answers not found. That is enough to tell a private repository's name
* from a public one that has not been pushed yet.
* @param {string} space - space AT-URI
* @returns {Promise<boolean>}
*/
async spaceExists(space) {
const params = new URLSearchParams({ space });
const res = await fetch(
`${this.service}/xrpc/com.atproto.space.getSpace?${params}`,
);
// A server with no spaces at all has no such method, which is the same
// answer as no such space.
return res.status !== 404 && res.status !== 501;
}
/**
* @param {string} repo - DID of the repo owner
* @param {string} collection
* @param {string} rkey
* @returns {Promise<{uri: string, cid: string, value: unknown}|null>} null when the record does not exist
*/
async getRecord(repo, collection, rkey) {
const params = new URLSearchParams({ repo, collection, rkey });
const res = await fetch(
`${this.service}/xrpc/com.atproto.repo.getRecord?${params}`,
);
if (res.status === 404) return null;
if (!res.ok) await throwXrpcError(res);
return /** @type {{uri: string, cid: string, value: unknown}} */ (
await res.json()
);
}
/**
* One page of an account's records in a collection.
* @param {string} repo - DID of the repo owner
* @param {string} collection
* @param {number} [limit]
* @returns {Promise<Array<{uri: string, cid: string, value: unknown}>>}
*/
async listRecords(repo, collection, limit = 100) {
const params = new URLSearchParams({
repo,
collection,
limit: String(limit),
});
const res = await fetch(
`${this.service}/xrpc/com.atproto.repo.listRecords?${params}`,
);
if (!res.ok) await throwXrpcError(res);
const body =
/** @type {{records?: Array<{uri: string, cid: string, value: unknown}>}} */ (
await res.json()
);
return body.records ?? [];
}
/**
* Write a record with compare-and-swap semantics. The swap has three
* states, and null is not the permissive one:
*
* - a cid: the stored record must be that version
* - null: no record may exist yet
* - undefined: no condition, so the write overwrites whatever is there
* @param {string} repo
* @param {string} collection
* @param {string} rkey
* @param {unknown} record
* @param {string|null} [swapRecord]
* @returns {Promise<{uri: string, cid: string}>}
*/
async putRecord(repo, collection, rkey, record, swapRecord) {
const res = await this.authedFetch(
`${this.service}/xrpc/com.atproto.repo.putRecord`,
{
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ repo, collection, rkey, record, swapRecord }),
},
);
if (!res.ok) await throwXrpcError(res);
return /** @type {{uri: string, cid: string}} */ (await res.json());
}
/**
* Create a record with a server-generated record key.
* @param {string} repo
* @param {string} collection
* @param {unknown} record
* @returns {Promise<{uri: string, cid: string}>}
*/
async createRecord(repo, collection, record) {
const res = await this.authedFetch(
`${this.service}/xrpc/com.atproto.repo.createRecord`,
{
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ repo, collection, record }),
},
);
if (!res.ok) await throwXrpcError(res);
return /** @type {{uri: string, cid: string}} */ (await res.json());
}
/**
* @param {Uint8Array} bytes
* @param {string} [mimeType] - recorded as the blob's media type
* @returns {Promise<import('./record.js').BlobObject>}
*/
async uploadBlob(bytes, mimeType = 'application/octet-stream') {
const res = await this.authedFetch(
`${this.service}/xrpc/com.atproto.repo.uploadBlob`,
{
method: 'POST',
headers: { 'Content-Type': mimeType },
body: /** @type {BodyInit} */ (bytes),
},
);
if (!res.ok) await throwXrpcError(res);
const body = /** @type {{blob: import('./record.js').BlobObject}} */ (
await res.json()
);
return body.blob;
}
/**
* @param {string} did
* @param {string} cid
* @returns {Promise<Uint8Array>}
*/
async getBlob(did, cid) {
const params = new URLSearchParams({ did, cid });
const res = await fetch(
`${this.service}/xrpc/com.atproto.sync.getBlob?${params}`,
);
if (!res.ok) await throwXrpcError(res);
return new Uint8Array(await res.arrayBuffer());
}
/**
* @param {string} space - space AT-URI
* @param {string} collection
* @param {string} rkey
* @returns {Promise<{uri: string, cid: string, value: unknown}|null>} null when the record does not exist
*/
async getSpaceRecord(space, collection, rkey) {
const params = new URLSearchParams({ space, collection, rkey });
if (this.readsRepo) params.set('repo', this.readsRepo);
const url = `${this.service}/xrpc/com.atproto.space.getRecord?${params}`;
if (this.spaceAuth?.kind !== 'credential') {
const res = await this.spaceFetch(url);
if (res.status === 404) return null;
if (!res.ok) await throwXrpcError(res);
return /** @type {{uri: string, cid: string, value: unknown}} */ (
await res.json()
);
}
// With a credential the record comes under a MAC from the author's KEM
// key, which is what shows the host served the author's record.
const repo = this.readsRepo;
if (!repo || !this.resolveKemKey) {
throw new Error('no repo to address the read to');
}
const { keyShare, sharedSecret } = await encapsulate(
await this.resolveKemKey(repo),
);
const res = await this.spaceFetch(url, {
[WANT_MAC_HEADER]: formatWantMacHeader(RECORD_MAC_TYPE, keyShare),
});
if (res.status === 404) return null;
if (!res.ok) await throwXrpcError(res);
const record = /** @type {{uri: string, cid: string, value: unknown}} */ (
await res.json()
);
const mac = parseMacHeader(res.headers.get(MAC_HEADER), RECORD_MAC_TYPE);
const uri = `${space}/${repo}/${collection}/${rkey}`;
if (!(await verifyRecord(record, { uri, sharedSecret, mac }))) {
throw new Error(`the record at ${uri} failed MAC verification`);
}
return record;
}
/**
* Write a space record with compare-and-swap semantics. Pass null to
* require that the record does not exist yet. The swap is a pds.js
* extension; a server without it ignores the field.
* @param {string} space - space AT-URI
* @param {string} repo - DID of the writer, which must be the session's
* @param {string} collection
* @param {string} rkey
* @param {unknown} record
* @param {string|null} swapRecord
* @returns {Promise<{uri: string, cid: string}>}
*/
async putSpaceRecord(space, repo, collection, rkey, record, swapRecord) {
const res = await this.authedFetch(
`${this.service}/xrpc/com.atproto.space.putRecord`,
{
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
space,
repo,
collection,
rkey,
record,
swapRecord,
}),
},
);
if (!res.ok) await throwXrpcError(res);
return /** @type {{uri: string, cid: string}} */ (await res.json());
}
/**
* @param {string} space - space AT-URI
* @param {string} cid
* @returns {Promise<Uint8Array>}
*/
async getSpaceBlob(space, cid) {
const params = new URLSearchParams({ space, cid });
const url = `${this.service}/xrpc/com.atproto.space.getBlob?${params}`;
const res = await this.spaceFetch(url);
if (!res.ok) await throwXrpcError(res);
return new Uint8Array(await res.arrayBuffer());
}
/**
* Put a pull request's work on the repository's canonical branch, as this
* session's account. The server refuses any account but the repository's
* owner.
* @param {{name: string, did: string, ref: string, mode?: 'merge'|'squash', handle?: string, subject?: string, body?: string}} body
* @returns {Promise<{sha: string, ref: string, mode: string, objects: number}>}
*/
async mergePull(body) {
const res = await this.authedFetch(
`${this.service}/xrpc/dev.pdsjs.git.mergePull`,
{
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(body ?? {}),
},
);
if (!res.ok) await throwXrpcError(res);
return /** @type {{sha: string, ref: string, mode: string, objects: number}} */ (
await res.json()
);
}
/**
* The space's writer set from its authority: every member repo that holds
* at least one record.
*
* Read with the space credential rather than a session. The writer set names
* a private space's membership, so the authority holds it to a credential and
* admits no session, not even its own.
*
* @param {string} space - space AT-URI
* @returns {Promise<Array<{did: string, repoRev: string}>>}
*/
async listSpaceRepos(space) {
/** @type {Array<{did: string, repoRev: string}>} */
const repos = [];
/** @type {string|undefined} */
let cursor;
do {
const params = new URLSearchParams({ space, limit: '100' });
if (cursor) params.set('cursor', cursor);
const res = await this.spaceFetch(
`${this.service}/xrpc/com.atproto.space.listRepos?${params}`,
);
if (!res.ok) await throwXrpcError(res);
const body =
/** @type {{repos: Array<{did: string, repoRev: string}>, cursor?: string}} */ (
await res.json()
);
repos.push(...body.repos);
cursor = body.cursor;
} while (cursor);
return repos;
}
/**
* Ask the session's own PDS for a token proving the user delegated a read
* of this space to the caller.
* @param {string} space - space AT-URI
* @returns {Promise<string>}
*/
async getDelegationToken(space) {
const params = new URLSearchParams({ space });
const res = await this.authedFetch(
`${this.service}/xrpc/com.atproto.space.getDelegationToken?${params}`,
);
if (!res.ok) await throwXrpcError(res);
const body = /** @type {{token: string}} */ (await res.json());
return body.token;
}
/**
* Exchange a delegation token at the space authority for a credential bound
* to `key`.
* @param {string} space - space AT-URI
* @param {string} delegationToken
* @param {import('./space-signature.js').SpaceSigKey} key
* @returns {Promise<string>}
*/
async getSpaceCredential(space, delegationToken, key) {
const res = await fetch(
`${this.service}/xrpc/com.atproto.space.getSpaceCredential`,
{
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
space,
credentialKid: key.did,
delegationToken,
}),
},
);
if (!res.ok) await throwXrpcError(res);
const body = /** @type {{credential: string}} */ (await res.json());
return body.credential;
}
}