chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
// Signers here use @pdsjs/core/crypto's sign(), which normalizes to low-S.
// atproto rejects high-S signatures as malleable and raw crypto.subtle.sign
// emits them about half the time, so a test signing the raw way would verify
// fine under WebCrypto while the reference implementation rejected it.
import { compressPublicKey, sign } from '@pdsjs/core/crypto';
import { publicKeyToDidKey } from '@pdsjs/core/plc';
import { beforeEach, describe, expect, it } from 'vitest';
import { ATPROTO_PDS_ID } from '../src/authority.js';
import {
createSpaceSigHeaders,
generateSpaceSigKey,
} from '../src/http-signature.js';
import {
createMemoryBlobContext,
createMemorySpaceStorage,
} from '../src/memory-storage.js';
import { createSpaceRoutes } from '../src/routes.js';
import { nextTid } from '../src/sequence.js';
import { createServiceAuth } from '../src/service-auth.js';
import { makeSpaceRow } from '../src/space-row.js';
import { createSpaceToken } from '../src/token.js';
import { createVerifier } from '../src/verifier.js';
const OWNER = 'did:plc:owner';
const MEMBER = 'did:plc:member';
const TYPE = 'com.example.forum';
const SPACE = `at://${OWNER}/space/${TYPE}/general`;
// Full management plus reads, so the manage tests exercise the handlers rather
// than the scope grammar (which has its own suite).
const SCOPE = `space:${TYPE}?authority=${OWNER}&manage=create&manage=update&manage=delete`;
async function makeIdentity() {
const keys = await crypto.subtle.generateKey(
{ name: 'ECDSA', namedCurve: 'P-256' },
true,
['sign', 'verify'],
);
const raw = new Uint8Array(
await crypto.subtle.exportKey('raw', keys.publicKey),
);
const didKey = publicKeyToDidKey(compressPublicKey(raw));
return {
multibase: didKey.slice('did:key:'.length),
signer: {
/** @param {Uint8Array} bytes */
sign: (bytes) => sign(keys.privateKey, bytes),
},
};
}
/** @type {any} */ let owner;
/** @type {any} */ let member;
/** @type {any} */ let didDocs;
/** @type {any} */ let routes;
/** @type {any} */ let store;
/** @type {any[]} */ let sent;
/** The application's key, which every credential here is bound to. */
/** @type {import('../src/http-signature.js').SpaceSigKey} */ let holder;
beforeEach(async () => {
owner = await makeIdentity();
member = await makeIdentity();
holder = await generateSpaceSigKey();
sent = [];
const doc = (/** @type {string} */ did, /** @type {any} */ id) => ({
id: did,
verificationMethod: [
{ id: '#atproto', type: 'Multikey', publicKeyMultibase: id.multibase },
],
service: [
{
id: ATPROTO_PDS_ID,
type: 'AtprotoPersonalDataServer',
serviceEndpoint: `https://${did.split(':').pop()}.example`,
},
],
});
didDocs = { [OWNER]: doc(OWNER, owner), [MEMBER]: doc(MEMBER, member) };
store = createMemorySpaceStorage();
routes = createSpaceRoutes({
spaceStorage: store,
...createMemoryBlobContext(),
getDid: async () => OWNER,
getKemKey: async () => null,
getSigner: async () => owner.signer,
resolveDid: async (/** @type {string} */ d) => didDocs[d] ?? null,
verifier: createVerifier(),
fetch: async (/** @type {any} */ url, /** @type {any} */ init) => {
sent.push({ url: String(url), init });
return new Response('{}', { status: 200 });
},
});
});
/** @param {string} path @param {any} body @param {any} [auth] @param {any} [headers] */
function post(path, body, auth = { did: OWNER, scope: SCOPE }, headers = {}) {
const url = new URL(`https://pds.example${path}`);
return routes[path].handler(
new Request(url, { method: 'POST', body: JSON.stringify(body), headers }),
url,
auth,
);
}
/** @param {string} path @param {Record<string,string>} params @param {any} [auth] */
function get(path, params = {}, auth = { did: OWNER, scope: SCOPE }) {
const url = new URL(`https://pds.example${path}`);
for (const [k, v] of Object.entries(params)) url.searchParams.set(k, v);
return routes[path].handler(new Request(url), url, auth);
}
/**
* Call an endpoint with a space credential, presented the only way a bound
* credential can be: the Atproto-Space-Credential scheme, signed together with the
* audience by the key the credential names. Every endpoint here answers for
* the hosted account, which is also the authority.
*
* @param {string} path
* @param {Object} opts
* @param {any} [opts.body] - a JSON body, which makes this a POST
* @param {Record<string,string>} [opts.params] - query parameters, for a GET
* @param {string} [opts.iss] - the issuing authority
* @param {string} [opts.sub] - the space the credential is for
* @param {any} [opts.signer] - the authority's signing key
* @param {string} [opts.audience] - who the request is addressed to
* @returns {Promise<Response>}
*/
async function withCredential(
path,
{ body, params, iss = OWNER, sub = SPACE, signer, audience = OWNER } = {},
) {
const credential = await createSpaceToken(
'credential',
{ iss, sub, keyId: holder.did },
signer ?? owner.signer,
);
const url = new URL(`https://pds.example${path}`);
for (const [k, v] of Object.entries(params ?? {})) {
url.searchParams.set(k, v);
}
const method = body === undefined ? 'GET' : 'POST';
return routes[path].handler(
new Request(url, {
method,
...(body === undefined ? {} : { body: JSON.stringify(body) }),
headers: await createSpaceSigHeaders(holder, {
authorization: `Atproto-Space-Credential ${credential}`,
audience,
}),
}),
url,
null,
);
}
/** Repo revs, as TIDs, in order. */
const REV = '3lkaaaaaaaa22';
const LATER_REV = '3lkbbbbbbbb22';
/** @param {number} ms - epoch ms */
function tidAt(ms) {
const chars = '234567abcdefghijklmnopqrstuvwxyz';
let micros = ms * 1000;
let out = '';
for (let i = 0; i < 11; i++) {
out = chars[micros % 32] + out;
micros = Math.floor(micros / 32);
}
return `${out}22`;
}
/** A stand-in commit data, the 32 bytes an LtHash digest is. */
const DATA_BYTES = new Uint8Array(32).fill(7);
/** The same, as notifyWrite carries it. */
const DATA = { $bytes: btoa(String.fromCharCode(...DATA_BYTES)) };
const MEMBERSPACE = { $type: 'com.atproto.memberspace.policy' };
/** The memberspace policy with its defaults filled in, as stored. */
const MEMBERSPACE_STORED = {
$type: 'com.atproto.memberspace.policy',
publicRead: false,
publicWrite: false,
publishPolicy: false,
publishMembers: false,
};
describe('createSpace', () => {
it('creates a space owned by the hosted account', async () => {
const res = await post('/xrpc/com.atproto.space.createSpace', {
spaceType: TYPE,
skey: 'general',
policy: MEMBERSPACE,
});
expect(res.status).toBe(200);
expect((await res.json()).uri).toBe(SPACE);
const row = await store.getSpace(SPACE);
expect(row.isOwner).toBe(true);
expect(row.policy).toEqual(MEMBERSPACE_STORED);
});
it('requires a space type', async () => {
const res = await post('/xrpc/com.atproto.space.createSpace', {
skey: 'general',
policy: MEMBERSPACE,
});
expect(res.status).toBe(400);
});
it('requires a policy', async () => {
const res = await post('/xrpc/com.atproto.space.createSpace', {
spaceType: TYPE,
});
expect(res.status).toBe(400);
expect((await res.json()).error).toBe('InvalidRequest');
});
it('leaves the owner able to mint a credential for it', async () => {
// The authority is admitted under every policy without being a member, so
// the owner reads the space it has just made.
await post('/xrpc/com.atproto.space.createSpace', {
spaceType: TYPE,
skey: 'general',
policy: MEMBERSPACE,
});
expect(await store.getMember(SPACE, OWNER)).toBeNull();
const url = new URL(
'https://pds.example/xrpc/com.atproto.space.getDelegationToken',
);
url.searchParams.set('space', SPACE);
const { token } = await (
await routes['/xrpc/com.atproto.space.getDelegationToken'].handler(
new Request(url),
url,
{ did: OWNER, scope: `space:${TYPE}?authority=${OWNER}` },
)
).json();
const credUrl = new URL(
'https://pds.example/xrpc/com.atproto.space.getSpaceCredential',
);
const res = await routes[
'/xrpc/com.atproto.space.getSpaceCredential'
].handler(
new Request(credUrl, {
method: 'POST',
body: JSON.stringify({
space: SPACE,
credentialKid: holder.did,
delegationToken: token,
}),
}),
credUrl,
null,
);
expect(res.status).toBe(200);
expect(typeof (await res.json()).credential).toBe('string');
});
it('generates a skey when none is given', async () => {
const { uri } = await (
await post('/xrpc/com.atproto.space.createSpace', {
spaceType: TYPE,
policy: MEMBERSPACE,
})
).json();
expect(uri.split('/').pop()).toHaveLength(13);
});
it('fills in a memberspace configuration', async () => {
await post('/xrpc/com.atproto.space.createSpace', {
spaceType: TYPE,
skey: 'general',
policy: {
$type: 'com.atproto.memberspace.policy',
publicRead: true,
allowedApps: ['https://app.example/client-metadata.json'],
unknownField: 1,
},
});
expect((await store.getSpace(SPACE)).policy).toEqual({
...MEMBERSPACE_STORED,
publicRead: true,
allowedApps: ['https://app.example/client-metadata.json'],
});
});
it('takes a personalspace and a managedspace', async () => {
await post('/xrpc/com.atproto.space.createSpace', {
spaceType: TYPE,
skey: 'mine',
policy: { $type: 'com.atproto.personalspace.policy' },
});
await post('/xrpc/com.atproto.space.createSpace', {
spaceType: TYPE,
skey: 'managed',
policy: {
$type: 'com.atproto.managedspace.policy',
managingApp: 'did:web:app.example#forum',
},
});
expect(
(await store.getSpace(`at://${OWNER}/space/${TYPE}/mine`)).policy,
).toEqual({ $type: 'com.atproto.personalspace.policy' });
expect(
(await store.getSpace(`at://${OWNER}/space/${TYPE}/managed`)).policy,
).toEqual({
$type: 'com.atproto.managedspace.policy',
managingApp: 'did:web:app.example#forum',
publishPolicy: false,
});
});
it('rejects a managedspace whose managing app is not a DID', async () => {
const res = await post('/xrpc/com.atproto.space.createSpace', {
spaceType: TYPE,
policy: {
$type: 'com.atproto.managedspace.policy',
managingApp: 'https://app.example',
},
});
expect(res.status).toBe(400);
expect((await res.json()).error).toBe('InvalidRequest');
});
it('rejects a policy it does not implement', async () => {
const res = await post('/xrpc/com.atproto.space.createSpace', {
spaceType: TYPE,
policy: { $type: 'com.example.inviteOnly.policy' },
});
expect((await res.json()).error).toBe('UnsupportedPolicy');
});
it('rejects a malformed field', async () => {
const res = await post('/xrpc/com.atproto.space.createSpace', {
spaceType: TYPE,
policy: { ...MEMBERSPACE, allowedApps: 'https://app.example' },
});
expect((await res.json()).error).toBe('InvalidRequest');
});
it('publishes the policy when it asks to be', async () => {
const policy = { ...MEMBERSPACE_STORED, publishPolicy: true };
await post('/xrpc/com.atproto.space.createSpace', {
spaceType: TYPE,
skey: 'general',
policy,
});
const record = await store.getSpaceRecord(
SPACE,
'com.atproto.space.policy',
'self',
);
expect(record).not.toBeNull();
// The commit puts the authority in the writer set, as any write does.
expect(await store.getSpaceWriter(SPACE, OWNER)).not.toBeNull();
});
it('refuses a caller who is not the hosted account', async () => {
const res = await post(
'/xrpc/com.atproto.space.createSpace',
{ spaceType: TYPE, policy: MEMBERSPACE },
{ did: MEMBER, scope: SCOPE },
);
expect(res.status).toBe(403);
expect((await res.json()).error).toBe('NotSpaceOwner');
});
it('rejects a type that is not an NSID', async () => {
const res = await post('/xrpc/com.atproto.space.createSpace', {
spaceType: 'nodots',
policy: MEMBERSPACE,
});
expect((await res.json()).error).toBe('InvalidType');
});
it('rejects a duplicate', async () => {
const body = { spaceType: TYPE, skey: 'general', policy: MEMBERSPACE };
await post('/xrpc/com.atproto.space.createSpace', body);
const res = await post('/xrpc/com.atproto.space.createSpace', body);
expect((await res.json()).error).toBe('SpaceAlreadyExists');
});
it('requires a manage=create grant', async () => {
const res = await post(
'/xrpc/com.atproto.space.createSpace',
{ spaceType: TYPE, policy: MEMBERSPACE },
{ did: OWNER, scope: `space:${TYPE}?authority=${OWNER}` },
);
expect(res.status).toBe(403);
expect((await res.json()).error).toBe('InvalidToken');
});
it('no longer serves the simplespace methods', () => {
expect(
Object.keys(routes).filter((path) => path.includes('simplespace')),
).toEqual([]);
});
});
describe('members', () => {
beforeEach(async () => {
await store.putSpace(makeSpaceRow(SPACE, { isOwner: true }));
});
/** @param {string} did */
const member = (did, read = true, write = true) => ({
space: SPACE,
did,
read,
write,
});
it('adds, lists and removes', async () => {
await post('/xrpc/com.atproto.memberspace.putMember', member(MEMBER));
let body = await (
await get('/xrpc/com.atproto.memberspace.listMembers', { space: SPACE })
).json();
expect(body.members).toEqual([{ did: MEMBER, read: true, write: true }]);
await post('/xrpc/com.atproto.memberspace.removeMember', {
space: SPACE,
did: MEMBER,
});
body = await (
await get('/xrpc/com.atproto.memberspace.listMembers', { space: SPACE })
).json();
expect(body.members).toEqual([]);
});
it('requires both rights', async () => {
const res = await post('/xrpc/com.atproto.memberspace.putMember', {
space: SPACE,
did: MEMBER,
});
expect(res.status).toBe(400);
});
it('drops a member from the writer set when it loses write', async () => {
await store.putSpaceWriter(SPACE, MEMBER, REV, DATA_BYTES, REV);
await post(
'/xrpc/com.atproto.memberspace.putMember',
member(MEMBER, true, false),
);
expect(await store.getSpaceWriter(SPACE, MEMBER)).toBeNull();
expect(await store.getMember(SPACE, MEMBER)).toEqual({
read: true,
write: false,
});
});
it('drops a removed member from the writer set', async () => {
await post('/xrpc/com.atproto.memberspace.putMember', member(MEMBER));
await store.putSpaceWriter(SPACE, MEMBER, REV, DATA_BYTES, REV);
await post('/xrpc/com.atproto.memberspace.removeMember', {
space: SPACE,
did: MEMBER,
});
expect(await store.getSpaceWriter(SPACE, MEMBER)).toBeNull();
});
it('publishes and withdraws member records when the policy asks', async () => {
await store.putSpace(
makeSpaceRow(SPACE, {
isOwner: true,
policy: { ...MEMBERSPACE_STORED, publishMembers: true },
}),
);
await post('/xrpc/com.atproto.memberspace.putMember', member(MEMBER));
expect(
await store.getSpaceRecord(
SPACE,
'com.atproto.memberspace.member',
MEMBER,
),
).not.toBeNull();
await post('/xrpc/com.atproto.memberspace.removeMember', {
space: SPACE,
did: MEMBER,
});
expect(
await store.getSpaceRecord(
SPACE,
'com.atproto.memberspace.member',
MEMBER,
),
).toBeNull();
});
it('refuses a space under another policy', async () => {
await store.putSpace(
makeSpaceRow(SPACE, {
isOwner: true,
policy: { $type: 'com.atproto.personalspace.policy' },
}),
);
for (const res of [
await post('/xrpc/com.atproto.memberspace.putMember', member(MEMBER)),
await post('/xrpc/com.atproto.memberspace.removeMember', {
space: SPACE,
did: MEMBER,
}),
await get('/xrpc/com.atproto.memberspace.listMembers', { space: SPACE }),
]) {
expect(res.status).toBe(400);
expect((await res.json()).error).toBe('PolicyMismatch');
}
});
it('denies on scope before revealing whether a foreign space exists', async () => {
// SCOPE is granted for OWNER's spaces only, so a space under another
// authority fails the scope check first — the caller learns nothing about
// whether it exists.
const other = `at://${MEMBER}/space/${TYPE}/x`;
const res = await post('/xrpc/com.atproto.memberspace.putMember', {
...member(MEMBER),
space: other,
});
expect(res.status).toBe(403);
expect((await res.json()).error).toBe('InvalidToken');
});
it('404s a space that does not exist but is covered by scope', async () => {
const res = await post('/xrpc/com.atproto.memberspace.putMember', {
...member(MEMBER),
space: `at://${OWNER}/space/${TYPE}/nonexistent`,
});
expect(res.status).toBe(404);
expect((await res.json()).error).toBe('SpaceNotFound');
});
it('403s a space present locally but owned elsewhere', async () => {
const foreign = `at://${OWNER}/space/${TYPE}/foreign`;
// isOwner false: we hold a repo in it, but we are not the authority.
await store.putSpace(makeSpaceRow(foreign, { isOwner: false }));
const res = await post('/xrpc/com.atproto.memberspace.putMember', {
...member(MEMBER),
space: foreign,
});
expect(res.status).toBe(403);
expect((await res.json()).error).toBe('NotSpaceOwner');
});
});
describe('removeWriter', () => {
it('drops an account from the writer set', async () => {
await store.putSpace(makeSpaceRow(SPACE, { isOwner: true }));
await store.putSpaceWriter(SPACE, MEMBER, REV, DATA_BYTES, REV);
const res = await post('/xrpc/com.atproto.space.removeWriter', {
space: SPACE,
did: MEMBER,
});
expect(res.status).toBe(200);
expect(await store.getSpaceWriter(SPACE, MEMBER)).toBeNull();
});
it('404s a deleted space', async () => {
await store.putSpace(
makeSpaceRow(SPACE, { isOwner: true, deletedAt: '2026-10-01' }),
);
const res = await post('/xrpc/com.atproto.space.removeWriter', {
space: SPACE,
did: MEMBER,
});
expect(res.status).toBe(404);
});
});
describe('updateSpace', () => {
beforeEach(async () => {
await store.putSpace(
makeSpaceRow(SPACE, {
isOwner: true,
policy: {
...MEMBERSPACE_STORED,
allowedApps: ['https://a.example/c.json'],
},
}),
);
await store.putMember(SPACE, MEMBER, { read: true, write: true });
});
it('replaces the policy wholesale', async () => {
await post('/xrpc/com.atproto.space.updateSpace', {
space: SPACE,
policy: { $type: 'com.atproto.memberspace.policy', publicRead: true },
});
expect((await store.getSpace(SPACE)).policy).toEqual({
...MEMBERSPACE_STORED,
publicRead: true,
});
// The same policy keeps its state.
expect(await store.getMember(SPACE, MEMBER)).not.toBeNull();
});
it('discards the member list when the policy changes type', async () => {
await post('/xrpc/com.atproto.space.updateSpace', {
space: SPACE,
policy: { $type: 'com.atproto.personalspace.policy' },
});
expect((await store.getSpace(SPACE)).policy).toEqual({
$type: 'com.atproto.personalspace.policy',
});
expect(await store.getMember(SPACE, MEMBER)).toBeNull();
});
it('publishes the member list when publishMembers turns on, and withdraws it', async () => {
await post('/xrpc/com.atproto.space.updateSpace', {
space: SPACE,
policy: { ...MEMBERSPACE, publishMembers: true, publishPolicy: true },
});
const listed = async () =>
(await store.listAllSpaceRecords(SPACE)).map(
(/** @type {any} */ r) => `${r.collection}/${r.rkey}`,
);
expect(await listed()).toEqual([
`com.atproto.memberspace.member/${MEMBER}`,
'com.atproto.space.policy/self',
]);
await post('/xrpc/com.atproto.space.updateSpace', {
space: SPACE,
policy: { $type: 'com.atproto.personalspace.policy' },
});
expect(await listed()).toEqual([]);
});
it('rejects a policy it does not implement', async () => {
const res = await post('/xrpc/com.atproto.space.updateSpace', {
space: SPACE,
policy: { $type: 'com.example.inviteOnly.policy' },
});
expect((await res.json()).error).toBe('UnsupportedPolicy');
});
});
describe('getSpace', () => {
it('describes a space it owns', async () => {
await store.putSpace(makeSpaceRow(SPACE, { isOwner: true }));
const res = await get('/xrpc/com.atproto.space.getSpace', {
space: SPACE,
});
expect(await res.json()).toEqual({
uri: SPACE,
policy: MEMBERSPACE_STORED,
});
});
it('404s a space it does not own', async () => {
await store.putSpace(makeSpaceRow(SPACE, { isOwner: false }));
const res = await get('/xrpc/com.atproto.space.getSpace', {
space: SPACE,
});
expect(res.status).toBe(404);
});
it('refuses a caller with no session', async () => {
await store.putSpace(makeSpaceRow(SPACE, { isOwner: true }));
const res = await get(
'/xrpc/com.atproto.space.getSpace',
{ space: SPACE },
null,
);
expect(res.status).toBe(401);
expect((await res.json()).error).toBe('AuthenticationRequired');
});
it('refuses another account', async () => {
await store.putSpace(makeSpaceRow(SPACE, { isOwner: true }));
const res = await get(
'/xrpc/com.atproto.space.getSpace',
{ space: SPACE },
{ did: MEMBER, scope: SCOPE },
);
expect(res.status).toBe(403);
});
it('tells a caller with no session that a space is not there', async () => {
// Existence comes before who may read: a space's address is its
// owner's DID and a key, so a miss is an answer anyone may have. The
// git helper asks this to tell a private repository from a public one.
const res = await get(
'/xrpc/com.atproto.space.getSpace',
{ space: `at://${OWNER}/space/com.example.forum/nowhere` },
null,
);
expect(res.status).toBe(404);
expect((await res.json()).error).toBe('SpaceNotFound');
});
it('refuses a space credential', async () => {
await store.putSpace(makeSpaceRow(SPACE, { isOwner: true }));
const res = await withCredential('/xrpc/com.atproto.space.getSpace', {
params: { space: SPACE },
});
expect(res.status).toBe(401);
});
});
describe('deleteSpace', () => {
it('soft-deletes and fans out to registered syncers', async () => {
await store.putSpace(makeSpaceRow(SPACE, { isOwner: true }));
await store.putCredentialRecipient(
SPACE,
MEMBER,
'https://syncer.example',
new Date(Date.now() + 3600_000).toISOString(),
);
const res = await post('/xrpc/com.atproto.space.deleteSpace', {
space: SPACE,
});
expect(res.status).toBe(200);
expect((await store.getSpace(SPACE)).deletedAt).toBeTruthy();
expect(sent).toHaveLength(1);
expect(sent[0].url).toBe(
'https://syncer.example/xrpc/com.atproto.space.notifySpaceDeleted',
);
expect(JSON.parse(sent[0].init.body).space).toBe(SPACE);
expect(sent[0].init.headers.authorization).toMatch(/^Bearer /);
expect(await store.nextSpaceNotifyAt()).toBeNull();
});
it('owes a syncer that could not be reached the notice, until it lands', async () => {
await store.putSpace(makeSpaceRow(SPACE, { isOwner: true }));
await store.putCredentialRecipient(
SPACE,
MEMBER,
'https://down.example',
new Date(Date.now() + 3600_000).toISOString(),
);
routes = createSpaceRoutes({
spaceStorage: store,
...createMemoryBlobContext(),
getDid: async () => OWNER,
getKemKey: async () => null,
getSigner: async () => owner.signer,
resolveDid: async (/** @type {string} */ d) => didDocs[d] ?? null,
verifier: createVerifier(),
fetch: async () => new Response('{}', { status: 503 }),
});
const res = await post('/xrpc/com.atproto.space.deleteSpace', {
space: SPACE,
});
expect(res.status).toBe(200);
expect(await store.nextSpaceNotifyAt()).not.toBeNull();
const [owed] = await store.listDueSpaceDeletedNotify(
new Date(Date.now() + 3600_000).toISOString(),
10,
);
expect(owed).toMatchObject({
space: SPACE,
service: MEMBER,
serviceEndpoint: 'https://down.example',
attempts: 1,
});
});
});
describe('notifyWrite', () => {
beforeEach(async () => {
await store.putSpace(makeSpaceRow(SPACE, { isOwner: true }));
await store.putMember(SPACE, MEMBER, { read: true, write: true });
});
/** @param {string} iss @param {any} identity @param {string} [lxm] */
async function serviceToken(
iss,
identity,
lxm = 'com.atproto.space.notifyWrite',
) {
return createServiceAuth({
iss,
aud: OWNER,
lxm,
signer: identity.signer,
});
}
it('records the writer', async () => {
const token = await serviceToken(MEMBER, member);
const res = await post(
'/xrpc/com.atproto.space.notifyWrite',
{
space: SPACE,
repo: MEMBER,
repoRev: REV,
data: DATA,
},
null,
{ authorization: `Bearer ${token}` },
);
expect(res.status).toBe(200);
const { writers } = await store.listSpaceWriters(SPACE, null, 50);
expect(writers[0].did).toBe(MEMBER);
expect(writers[0].repoRev).toBe(REV);
expect(writers[0].spaceRev).toMatch(/^[2-7a-z]{13}$/);
});
it('forwards the notice to every registered syncer', async () => {
await store.putCredentialRecipient(
SPACE,
`${MEMBER}${ATPROTO_PDS_ID}`,
'https://syncer.example',
new Date(Date.now() + 3600_000).toISOString(),
);
const token = await serviceToken(MEMBER, member);
await post(
'/xrpc/com.atproto.space.notifyWrite',
{
space: SPACE,
repo: MEMBER,
repoRev: REV,
data: DATA,
},
null,
{ authorization: `Bearer ${token}` },
);
const forwarded = sent.find((c) => c.url.includes('notifyWrite'));
expect(forwarded.url).toBe(
'https://syncer.example/xrpc/com.atproto.space.notifyWrite',
);
const body = JSON.parse(forwarded.init.body);
expect(body.space).toBe(SPACE);
expect(body.repo).toBe(MEMBER);
expect(body.repoRev).toBe(REV);
expect(body.spaceRev).toMatch(/^[2-7a-z]{13}$/);
// The space's first update has no predecessor to name.
expect(body).not.toHaveProperty('prevSpaceRev');
// The data arrives as it came in. `$bytes` carries no padding, so the
// strings differ where the bytes do not.
expect(
Uint8Array.from(atob(body.data.$bytes), (c) => c.charCodeAt(0)),
).toEqual(DATA_BYTES);
// Addressed to the syncer and signed by the authority, which is the only
// party that knows the registration exists.
expect(forwarded.init.headers.authorization).toMatch(/^Bearer /);
});
it('refuses a writer the space does not admit', async () => {
await store.putMember(SPACE, MEMBER, { read: true, write: false });
const token = await serviceToken(MEMBER, member);
const res = await post(
'/xrpc/com.atproto.space.notifyWrite',
{ space: SPACE, repo: MEMBER, repoRev: REV, data: DATA },
null,
{ authorization: `Bearer ${token}` },
);
expect(res.status).toBe(403);
const { writers } = await store.listSpaceWriters(SPACE, null, 50);
expect(writers).toEqual([]);
});
it('admits a writer under a public write policy', async () => {
await store.putSpace(
makeSpaceRow(SPACE, {
isOwner: true,
policy: { ...MEMBERSPACE_STORED, publicWrite: true },
}),
);
await store.removeMember(SPACE, MEMBER);
const token = await serviceToken(MEMBER, member);
const res = await post(
'/xrpc/com.atproto.space.notifyWrite',
{ space: SPACE, repo: MEMBER, repoRev: REV, data: DATA },
null,
{ authorization: `Bearer ${token}` },
);
expect(res.status).toBe(200);
});
it('does not forward to a lapsed registration', async () => {
await store.putCredentialRecipient(
SPACE,
`${MEMBER}${ATPROTO_PDS_ID}`,
'https://syncer.example',
new Date(Date.now() - 1000).toISOString(),
);
const token = await serviceToken(MEMBER, member);
await post(
'/xrpc/com.atproto.space.notifyWrite',
{ space: SPACE, repo: MEMBER, repoRev: REV, data: DATA },
null,
{ authorization: `Bearer ${token}` },
);
expect(sent.filter((c) => c.url.includes('notifyWrite'))).toHaveLength(0);
});
it('rejects an unsigned call', async () => {
const res = await post(
'/xrpc/com.atproto.space.notifyWrite',
{ space: SPACE, repo: MEMBER, repoRev: REV, data: DATA },
null,
);
expect(res.status).toBe(401);
});
it('rejects a token minted for a different method', async () => {
const token = await serviceToken(
MEMBER,
member,
'com.atproto.space.getRepo',
);
const res = await post(
'/xrpc/com.atproto.space.notifyWrite',
{ space: SPACE, repo: MEMBER, repoRev: REV, data: DATA },
null,
{ authorization: `Bearer ${token}` },
);
expect(res.status).toBe(401);
expect((await res.json()).error).toBe('BadJwtLexiconMethod');
});
it('refuses to let one account report another account writes', async () => {
const token = await serviceToken(MEMBER, member);
const res = await post(
'/xrpc/com.atproto.space.notifyWrite',
{ space: SPACE, repo: 'did:plc:someone-else', repoRev: REV, data: DATA },
null,
{ authorization: `Bearer ${token}` },
);
expect(res.status).toBe(403);
});
/**
* @param {Record<string, unknown>} body
* @param {string} [aud]
*/
async function notify(body, aud = OWNER) {
const token = await createServiceAuth({
iss: MEMBER,
aud,
lxm: 'com.atproto.space.notifyWrite',
signer: member.signer,
});
return post(
'/xrpc/com.atproto.space.notifyWrite',
{ space: SPACE, repo: MEMBER, data: DATA, ...body },
null,
{ authorization: `Bearer ${token}` },
);
}
it('accepts a token addressed to the space host service', async () => {
const res = await notify({ repoRev: REV }, `${OWNER}${ATPROTO_PDS_ID}`);
expect(res.status).toBe(200);
});
it('requires the commit data', async () => {
const token = await serviceToken(MEMBER, member);
for (const data of [undefined, { $bytes: '!!' }, 'AAAA']) {
const res = await post(
'/xrpc/com.atproto.space.notifyWrite',
{ space: SPACE, repo: MEMBER, repoRev: REV, data },
null,
{ authorization: `Bearer ${token}` },
);
expect(res.status).toBe(400);
expect((await res.json()).error).toBe('InvalidRequest');
}
});
it('refuses a repoRev that is not a TID', async () => {
const res = await notify({ repoRev: '3lkaaa' });
expect(res.status).toBe(400);
});
it('refuses a repoRev too far in the future', async () => {
const future = nextTid(null);
// Ten minutes ahead: the TID's timestamp is microseconds in its first
// eleven characters, so rebuild one from a later clock.
const later = tidAt(Date.now() + 10 * 60 * 1000);
expect(later > future).toBe(true);
const res = await notify({ repoRev: later });
expect(res.status).toBe(400);
expect((await res.json()).error).toBe('FutureRev');
});
it('answers SpaceNotFound for a deleted space', async () => {
await store.deleteSpace(SPACE, new Date().toISOString());
const res = await notify({ repoRev: REV });
expect(res.status).toBe(404);
expect((await res.json()).error).toBe('SpaceNotFound');
});
it('sequences updates across writers, each naming the one before', async () => {
await store.putSpaceWriter(SPACE, OWNER, REV, DATA_BYTES, '3lkaaaaaaaa23');
await store.putCredentialRecipient(
SPACE,
`${MEMBER}${ATPROTO_PDS_ID}`,
'https://syncer.example',
new Date(Date.now() + 3600_000).toISOString(),
);
await notify({ repoRev: REV });
await notify({ repoRev: LATER_REV });
const bodies = sent
.filter((c) => c.url.includes('notifyWrite'))
.map((c) => JSON.parse(c.init.body));
expect(bodies).toHaveLength(2);
expect(bodies[0].prevSpaceRev).toBe('3lkaaaaaaaa23');
expect(bodies[0].spaceRev > bodies[0].prevSpaceRev).toBe(true);
expect(bodies[1].prevSpaceRev).toBe(bodies[0].spaceRev);
expect(bodies[1].spaceRev > bodies[0].spaceRev).toBe(true);
});
it('neither resequences nor forwards a repeated or older rev', async () => {
await store.putCredentialRecipient(
SPACE,
`${MEMBER}${ATPROTO_PDS_ID}`,
'https://syncer.example',
new Date(Date.now() + 3600_000).toISOString(),
);
expect((await notify({ repoRev: LATER_REV })).status).toBe(200);
const before = await store.getSpaceWriter(SPACE, MEMBER);
expect((await notify({ repoRev: LATER_REV })).status).toBe(200);
expect((await notify({ repoRev: REV })).status).toBe(200);
const after = await store.getSpaceWriter(SPACE, MEMBER);
expect(after).toEqual(before);
expect(sent.filter((c) => c.url.includes('notifyWrite'))).toHaveLength(1);
});
});
describe('notifyCredentialRevoked', () => {
const PATH = '/xrpc/com.atproto.space.notifyCredentialRevoked';
/** A space MEMBER governs, which OWNER holds a repo in. */
const FOREIGN = `at://${MEMBER}/space/${TYPE}/general`;
/**
* @param {Record<string, unknown>} body
* @param {{iss?: string, identity?: any, aud?: string}} [opts]
*/
async function revoke(body, { iss = MEMBER, identity, aud = OWNER } = {}) {
const token = await createServiceAuth({
iss,
aud,
lxm: 'com.atproto.space.notifyCredentialRevoked',
signer: (identity ?? member).signer,
});
return post(PATH, body, null, { authorization: `Bearer ${token}` });
}
it('records revocations from the space authority', async () => {
const res = await revoke({ space: FOREIGN, credentials: ['j1', 'j2'] });
expect(res.status).toBe(200);
expect(await store.isCredentialRevoked(FOREIGN, 'j1')).toBe(true);
expect(await store.isCredentialRevoked(FOREIGN, 'j2')).toBe(true);
expect(await store.isCredentialRevoked(FOREIGN, 'j3')).toBe(false);
});
it('takes the same revocation twice', async () => {
const body = { space: FOREIGN, credentials: ['j1', 'j1'] };
expect((await revoke(body)).status).toBe(200);
expect((await revoke(body)).status).toBe(200);
expect(await store.isCredentialRevoked(FOREIGN, 'j1')).toBe(true);
});
it('refuses anybody but the space authority', async () => {
const res = await revoke(
{ space: SPACE, credentials: ['j1'] },
{ iss: MEMBER },
);
expect(res.status).toBe(403);
expect(await store.isCredentialRevoked(SPACE, 'j1')).toBe(false);
});
it('refuses a token addressed to another account', async () => {
const res = await revoke(
{ space: FOREIGN, credentials: ['j1'] },
{ aud: 'did:plc:someone-else' },
);
expect(res.status).toBe(401);
});
it('refuses a body without 1 to 100 credential ids', async () => {
for (const credentials of [
[],
[''],
Array.from({ length: 101 }, (_, i) => `j${i}`),
'j1',
]) {
const res = await revoke({ space: FOREIGN, credentials });
expect(res.status).toBe(400);
}
});
});
describe('registerNotify', () => {
// The test DID documents publish one space-host entry each, so a subscriber
// names itself with that fragment.
const SYNCER = `${MEMBER}${ATPROTO_PDS_ID}`;
beforeEach(async () => {
await store.putSpace(makeSpaceRow(SPACE, { isOwner: true }));
});
it('records a subscriber holding a valid credential', async () => {
const res = await withCredential('/xrpc/com.atproto.space.registerNotify', {
body: { space: SPACE, service: SYNCER },
});
expect(res.status).toBe(200);
expect((await res.json()).expiresAt).toMatch(/^\d{4}-/);
const list = await store.listCredentialRecipients(SPACE);
expect(list[0].serviceDid).toBe(SYNCER);
// Resolved from the subscriber's DID document, not taken from the request.
expect(list[0].serviceEndpoint).toBe('https://member.example');
});
it('refuses a service it cannot resolve to an endpoint', async () => {
const res = await withCredential('/xrpc/com.atproto.space.registerNotify', {
body: { space: SPACE, service: 'did:web:nowhere.example#syncer' },
});
expect((await res.json()).error).toBe('ServiceNotResolvable');
});
it('refuses a space this server does not govern', async () => {
const foreign = `at://${MEMBER}/space/${TYPE}/general`;
await store.putSpace(makeSpaceRow(foreign, { isOwner: false }));
const res = await withCredential('/xrpc/com.atproto.space.registerNotify', {
body: { space: foreign, service: SYNCER },
iss: MEMBER,
sub: foreign,
signer: member.signer,
audience: MEMBER,
});
expect(res.status).toBe(404);
expect((await res.json()).error).toBe('SpaceNotFound');
});
it('refuses without a credential', async () => {
const res = await post(
'/xrpc/com.atproto.space.registerNotify',
{ space: SPACE, service: SYNCER },
null,
);
expect(res.status).toBe(401);
});
});
describe('unregisterNotify', () => {
const SYNCER = `${MEMBER}${ATPROTO_PDS_ID}`;
beforeEach(async () => {
await store.putSpace(makeSpaceRow(SPACE, { isOwner: true }));
await store.putCredentialRecipient(
SPACE,
SYNCER,
'https://member.example',
new Date(Date.now() + 3600_000).toISOString(),
);
});
/** @param {any} body */
const unregister = (body) =>
withCredential('/xrpc/com.atproto.space.unregisterNotify', { body });
it('withdraws a registration', async () => {
const res = await unregister({ space: SPACE, service: SYNCER });
expect(res.status).toBe(200);
expect(await store.listCredentialRecipients(SPACE)).toEqual([]);
});
it('succeeds when nothing was registered', async () => {
const res = await unregister({
space: SPACE,
service: 'did:web:stranger.example#syncer',
});
expect(res.status).toBe(200);
expect(await store.listCredentialRecipients(SPACE)).toHaveLength(1);
});
it('refuses without a credential', async () => {
const res = await post(
'/xrpc/com.atproto.space.unregisterNotify',
{ space: SPACE, service: SYNCER },
null,
);
expect(res.status).toBe(401);
});
});
describe('notifySpaceDeleted', () => {
it('flags the local space when its authority says so', async () => {
const foreign = `at://${MEMBER}/space/${TYPE}/general`;
await store.putSpace(makeSpaceRow(foreign, { isOwner: false }));
const token = await createServiceAuth({
iss: MEMBER,
aud: OWNER,
lxm: 'com.atproto.space.notifySpaceDeleted',
signer: member.signer,
});
const res = await post(
'/xrpc/com.atproto.space.notifySpaceDeleted',
{ space: foreign },
null,
{ authorization: `Bearer ${token}` },
);
expect(res.status).toBe(200);
// Flagged, not erased — the records are the member's own.
const row = await store.getSpace(foreign);
expect(row).not.toBeNull();
expect(row.deletedAt).toBeTruthy();
});
it('refuses a caller that is not the space authority', async () => {
const foreign = `at://${MEMBER}/space/${TYPE}/general`;
await store.putSpace(makeSpaceRow(foreign, { isOwner: false }));
const impostor = await makeIdentity();
didDocs['did:plc:impostor'] = {
id: 'did:plc:impostor',
verificationMethod: [
{
id: '#atproto',
type: 'Multikey',
publicKeyMultibase: impostor.multibase,
},
],
};
const token = await createServiceAuth({
iss: 'did:plc:impostor',
aud: OWNER,
lxm: 'com.atproto.space.notifySpaceDeleted',
signer: impostor.signer,
});
const res = await post(
'/xrpc/com.atproto.space.notifySpaceDeleted',
{ space: foreign },
null,
{ authorization: `Bearer ${token}` },
);
expect(res.status).toBe(403);
});
});
describe('outbound notifyWrite', () => {
it('tells the space authority after a write', async () => {
// A space owned by MEMBER, written to by us (OWNER).
const foreign = `at://${MEMBER}/space/${TYPE}/general`;
const url = new URL(
'https://pds.example/xrpc/com.atproto.space.createRecord',
);
const res = await routes['/xrpc/com.atproto.space.createRecord'].handler(
new Request(url, {
method: 'POST',
body: JSON.stringify({
space: foreign,
repo: OWNER,
collection: 'com.example.post',
rkey: 'a',
record: { v: 1 },
}),
}),
url,
{
did: OWNER,
scope: `space:${TYPE}?authority=*&collection=com.example.post`,
},
);
expect(res.status).toBe(200);
// fireNotifyWrite swallows its own errors by design, so assert the call
// actually happened — otherwise a broken notifier looks identical to a
// working one.
const notify = sent.find((c) =>
c.url.endsWith('/xrpc/com.atproto.space.notifyWrite'),
);
expect(notify).toBeTruthy();
expect(notify.url).toBe(
'https://member.example/xrpc/com.atproto.space.notifyWrite',
);
const body = JSON.parse(notify.init.body);
expect(body.space).toBe(foreign);
expect(body.repo).toBe(OWNER);
expect(body.repoRev).toBeTruthy();
// The data is sha256 of the LtHash state, i.e. the commit data.
expect(atob(body.data.$bytes)).toHaveLength(32);
expect(notify.init.headers.authorization).toMatch(/^Bearer /);
const token = notify.init.headers.authorization.slice('Bearer '.length);
const claims = JSON.parse(atob(token.split('.')[1]));
expect(claims.aud).toBe(`${MEMBER}${ATPROTO_PDS_ID}`);
// Delivered, so nothing is left owed.
expect(await store.nextSpaceNotifyAt()).toBeNull();
});
/** Write one record into the hosted account's own space. */
async function writeToOwnSpace() {
const url = new URL(
'https://pds.example/xrpc/com.atproto.space.createRecord',
);
return routes['/xrpc/com.atproto.space.createRecord'].handler(
new Request(url, {
method: 'POST',
body: JSON.stringify({
space: SPACE,
repo: OWNER,
collection: 'com.example.post',
rkey: 'a',
record: { v: 1 },
}),
}),
url,
{
did: OWNER,
scope: `space:${TYPE}?authority=${OWNER}&collection=com.example.post`,
},
);
}
it('records itself as a writer in its own space without HTTP', async () => {
await store.putSpace(makeSpaceRow(SPACE, { isOwner: true }));
await writeToOwnSpace();
// No notification round-trip to ourselves...
expect(sent.filter((c) => c.url.includes('notifyWrite'))).toHaveLength(0);
// ...but the writer set must include us, or listRepos would omit the
// authority's own repo and syncers walking it would miss this repo.
const { writers } = await store.listSpaceWriters(SPACE, null, 50);
expect(writers.map((/** @type {any} */ w) => w.did)).toEqual([OWNER]);
});
// The case an application actually meets: its own space, on its own PDS, with
// one syncer following it. Nothing arrives over HTTP for this server to
// forward, so the write path is the only thing that can send the notice.
it('notifies a syncer of a write into the authority own space', async () => {
await store.putSpace(makeSpaceRow(SPACE, { isOwner: true }));
await store.putCredentialRecipient(
SPACE,
`${MEMBER}${ATPROTO_PDS_ID}`,
'https://syncer.example',
new Date(Date.now() + 3600_000).toISOString(),
);
await writeToOwnSpace();
const forwarded = sent.filter((c) => c.url.includes('notifyWrite'));
expect(forwarded).toHaveLength(1);
expect(forwarded[0].url).toBe(
'https://syncer.example/xrpc/com.atproto.space.notifyWrite',
);
const body = JSON.parse(forwarded[0].init.body);
expect(body.space).toBe(SPACE);
expect(body.repo).toBe(OWNER);
expect(body.repoRev).toBeTruthy();
// The same data the writer set records, which is what a syncer compares.
const { writers } = await store.listSpaceWriters(SPACE, null, 50);
expect(atob(body.data.$bytes)).toHaveLength(32);
expect(body.repoRev).toBe(writers[0].repoRev);
expect(body.spaceRev).toBe(writers[0].spaceRev);
});
/**
* Rebuild the routes with a background hook, and answer the work it holds.
* @returns {Promise<unknown>[]}
*/
function withBackground() {
/** @type {Promise<unknown>[]} */
const held = [];
routes = createSpaceRoutes({
spaceStorage: store,
...createMemoryBlobContext(),
getDid: async () => OWNER,
getKemKey: async () => null,
getSigner: async () => owner.signer,
resolveDid: async (/** @type {string} */ d) => didDocs[d] ?? null,
verifier: createVerifier(),
fetch: async (/** @type {any} */ url, /** @type {any} */ init) => {
sent.push({ url: String(url), init });
return new Response('{}', { status: 200 });
},
background: (/** @type {Promise<unknown>} */ work) => {
held.push(work);
},
});
return held;
}
it('answers a write into its own space before notifying the syncers', async () => {
const held = withBackground();
await store.putSpace(makeSpaceRow(SPACE, { isOwner: true }));
await store.putCredentialRecipient(
SPACE,
`${MEMBER}${ATPROTO_PDS_ID}`,
'https://syncer.example',
new Date(Date.now() + 3600_000).toISOString(),
);
const res = await writeToOwnSpace();
expect(res.status).toBe(200);
expect(held).toHaveLength(1);
await Promise.all(held);
const forwarded = sent.filter((c) => c.url.includes('notifyWrite'));
expect(forwarded).toHaveLength(1);
expect(forwarded[0].init.signal).toBeInstanceOf(AbortSignal);
});
it('owes the authority a write before trying it after the answer', async () => {
const held = withBackground();
const foreign = `at://${MEMBER}/space/${TYPE}/general`;
const url = new URL(
'https://pds.example/xrpc/com.atproto.space.createRecord',
);
const res = await routes['/xrpc/com.atproto.space.createRecord'].handler(
new Request(url, {
method: 'POST',
body: JSON.stringify({
space: foreign,
repo: OWNER,
collection: 'com.example.post',
rkey: 'a',
record: { v: 1 },
}),
}),
url,
{
did: OWNER,
scope: `space:${TYPE}?authority=*&collection=com.example.post`,
},
);
expect(res.status).toBe(200);
expect(held).toHaveLength(1);
expect(await store.nextSpaceNotifyAt()).not.toBeNull();
await Promise.all(held);
expect(
sent.filter((c) => c.url.endsWith('com.atproto.space.notifyWrite')),
).toHaveLength(1);
expect(await store.nextSpaceNotifyAt()).toBeNull();
});
});