chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
// @pdsjs/spaces/handlers/manage - space management, the memberspace member
// list, and notifications.
//
// A space is anchored on its owner's own DID and governed by one of the
// baseline policies in ../policy.js. The member list belongs to the
// memberspace policy, and com.atproto.memberspace manages it.
//
// Notifications are best-effort and are not required for eventual consistency.
// A dropped one is recovered by a later write's notification or by a syncer's
// periodic sweep, because sync correctness rests on comparing set hashes rather
// than on receiving every event.
import { createTid } from '@pdsjs/core/repo';
import { ScopePermissions } from '@pdsjs/core/scope';
import { namedServiceEndpoint, spaceHostAudience } from '../authority.js';
import { createWriteNotifier, forwardToSyncers } from '../notify.js';
import { createNotifyOutbox } from '../outbox.js';
import {
areMembersPublished,
assertMemberspace,
memberListWrites,
memberRecordWrites,
normalizePolicy,
policyRecordWrites,
SpacePolicyError,
} from '../policy.js';
import { isFutureRev, isTid, recordWriterUpdate } from '../sequence.js';
import { verifyServiceAuth } from '../service-auth.js';
import { makeSpaceRow } from '../space-row.js';
import {
CLOCK_SKEW_SEC,
MAX_CREDENTIAL_LIFETIME_SEC,
SpaceTokenError,
} from '../token.js';
import { formatSpaceUri, parseSpaceUri } from '../uri.js';
import { applyWrites, MAX_WRITES_PER_COMMIT } from '../writer.js';
import { authorizeSpaceUser, createCredentialAuthorizer } from './auth.js';
// How long a write-notification registration lasts before a syncer has to
// renew it.
const REGISTRATION_TTL_MS = 24 * 60 * 60 * 1000;
// A revocation must outlive every credential it can match, allowing for clock
// skew at both ends.
const REVOCATION_RETENTION_MS =
(MAX_CREDENTIAL_LIFETIME_SEC + 2 * CLOCK_SKEW_SEC) * 1000;
/**
* @param {string} error
* @param {string} message
* @param {number} [status]
* @returns {Response}
*/
function errorResponse(error, message, status = 400) {
return Response.json({ error, message }, { status });
}
/**
* @param {unknown} err
* @returns {Response}
*/
function policyErrorResponse(err) {
if (err instanceof SpacePolicyError) {
return errorResponse(err.code, err.message);
}
throw err;
}
/**
* @param {Request} request
* @returns {Promise<any>}
*/
async function readJson(request) {
try {
return JSON.parse(await request.text());
} catch {
return null;
}
}
/**
* @param {Object} ctx
* @param {import('@pdsjs/core/ports').SpaceStoragePort} ctx.spaceStorage
* @param {() => Promise<string|null>} ctx.getDid
* @param {() => Promise<import('../token.js').SpaceSigner>} ctx.getSigner
* @param {import('@pdsjs/core/ports').DidResolverPort} ctx.resolveDid
* @param {import('@pdsjs/core/ports').SignatureVerifierPort} ctx.verifier
* @param {typeof fetch} [ctx.fetch]
* @param {ReturnType<typeof createNotifyOutbox>} [ctx.outbox] - built here
* from the rest of ctx when absent
* @returns {import('@pdsjs/core/pds').Routes}
*/
export function createManageRoutes(ctx) {
const { spaceStorage, getDid, resolveDid, verifier } = ctx;
const outbox = ctx.outbox ?? createNotifyOutbox(ctx);
const fireNotifyWrite = createWriteNotifier(ctx, outbox);
const requireSpaceCredential = createCredentialAuthorizer({
spaceStorage,
resolveDid,
verifier,
});
/**
* Management verbs are governed by the `manage=` scope param and are not
* implied by any record action.
* @param {{did: string, scope?: string}|null} auth
* @param {string} space
* @param {string} op
* @returns {Response|null}
*/
function checkManageScope(auth, space, op) {
const { spaceDid, spaceType, skey } = parseSpaceUri(space);
const permissions = new ScopePermissions(auth?.scope);
const allowed = permissions.allowsSpaceManage({
spaceType,
authority: spaceDid,
skey,
op,
userDid: auth?.did,
});
if (!allowed) {
return errorResponse(
'InvalidToken',
`Scope does not permit ${op} on ${space}`,
403,
);
}
return null;
}
/**
* Registrations are the authority's to hold, so this server must govern the
* space.
* @param {string} space
* @returns {Promise<import('@pdsjs/core/ports').SpaceRow|Response>}
*/
async function hostedSpace(space) {
const row = await spaceStorage.getSpace(space);
if (!row?.isOwner || row.deletedAt) {
return errorResponse('SpaceNotFound', 'Space not found', 404);
}
return row;
}
/**
* Load a space this server owns, or the Response explaining why not.
* @param {string} space
* @param {{did: string}|null} auth
* @returns {Promise<import('@pdsjs/core/ports').SpaceRow|Response>}
*/
async function ownedSpace(space, auth) {
const row = await spaceStorage.getSpace(space);
if (!row) return errorResponse('SpaceNotFound', 'Space not found', 404);
if (!row.isOwner) {
return errorResponse(
'NotSpaceOwner',
'This server is not the authority for that space',
403,
);
}
const hosted = await getDid();
if (auth && auth.did !== hosted) {
return errorResponse('NotSpaceOwner', 'Not the space owner', 403);
}
return row;
}
/**
* Write the PDS's own records in the authority's repo in a space: the
* published policy and member list. Applied in commits of the largest size
* a commit allows, then notified once.
* @param {string} space
* @param {import('../writer.js').SpaceWriteInput[]} writes
* @returns {Promise<void>}
*/
async function publish(space, writes) {
/** @type {{rev: string, setHash: Uint8Array}|null} */
let commit = null;
for (let i = 0; i < writes.length; i += MAX_WRITES_PER_COMMIT) {
commit = await applyWrites(spaceStorage, {
space,
writes: writes.slice(i, i + MAX_WRITES_PER_COMMIT),
owner: await getDid(),
});
}
if (commit) await fireNotifyWrite(space, commit);
}
/**
* The policy a request states, or the refusal.
* @param {unknown} value
* @returns {import('@pdsjs/core/ports').SpacePolicy|Response}
*/
function requestedPolicy(value) {
try {
return normalizePolicy(value);
} catch (err) {
return policyErrorResponse(err);
}
}
/**
* Load a live space this server owns, governed by the memberspace policy.
* @param {string} space
* @param {{did: string}|null} auth
* @returns {Promise<import('@pdsjs/core/ports').SpaceRow|Response>}
*/
async function memberspace(space, auth) {
const row = await ownedSpace(space, auth);
if (row instanceof Response) return row;
if (row.deletedAt) {
return errorResponse('SpaceNotFound', 'Space not found', 404);
}
try {
assertMemberspace(row.policy);
} catch (err) {
return policyErrorResponse(err);
}
return row;
}
return {
'/xrpc/com.atproto.space.createSpace': {
method: 'POST',
auth: 'required',
handler: async (request, _url, auth) => {
const body = await readJson(request);
if (!body) return errorResponse('InvalidRequest', 'Invalid JSON body');
const { skey, spaceType: type } = body;
if (typeof type !== 'string') {
return errorResponse('InvalidRequest', 'spaceType is required');
}
if (!type.includes('.')) {
return errorResponse('InvalidType', `Not an NSID: ${type}`);
}
// The space is anchored on the authenticated user, who becomes its
// owner. This server hosts one account, so any other caller is a miss.
const hosted = await getDid();
const caller = /** @type {{did: string}} */ (auth).did;
if (caller !== hosted) {
return errorResponse(
'NotSpaceOwner',
'Can only create spaces anchored on the hosted account',
403,
);
}
if (typeof skey === 'string' && skey.length > 512) {
return errorResponse('InvalidRequest', 'skey is too long');
}
const uri = formatSpaceUri({
spaceDid: caller,
spaceType: type,
skey: skey || createTid(),
});
const scopeError = checkManageScope(auth, uri, 'create');
if (scopeError) return scopeError;
const policy = requestedPolicy(body.policy);
if (policy instanceof Response) return policy;
if (await spaceStorage.getSpace(uri)) {
return errorResponse('SpaceAlreadyExists', `Exists: ${uri}`);
}
await spaceStorage.putSpace(
makeSpaceRow(uri, { isOwner: true, policy }),
);
await publish(uri, await policyRecordWrites(spaceStorage, uri, policy));
return Response.json({ uri });
},
},
// Replaces the policy wholesale. A policy of another `$type` migrates the
// space and discards the state the previous policy held, such as a
// member list.
'/xrpc/com.atproto.space.updateSpace': {
method: 'POST',
auth: 'required',
handler: async (request, _url, auth) => {
const body = await readJson(request);
if (!body?.space) {
return errorResponse('InvalidRequest', 'space is required');
}
const scopeError = checkManageScope(auth, body.space, 'update');
if (scopeError) return scopeError;
const row = await ownedSpace(body.space, auth);
if (row instanceof Response) return row;
if (row.deletedAt) {
return errorResponse('SpaceNotFound', 'Space not found', 404);
}
const next = requestedPolicy(body.policy);
if (next instanceof Response) return next;
const writes = await policyRecordWrites(spaceStorage, body.space, next);
// Only memberspace has a member list, so this also withdraws a
// published list when a migration discards it.
if (areMembersPublished(row.policy) !== areMembersPublished(next)) {
writes.push(
...(await memberListWrites(spaceStorage, body.space, next)),
);
}
await spaceStorage.putSpace({ ...row, policy: next });
if (row.policy?.$type !== next.$type) {
await spaceStorage.removeAllMembers(body.space);
}
await publish(body.space, writes);
return Response.json({});
},
},
'/xrpc/com.atproto.space.deleteSpace': {
method: 'POST',
auth: 'required',
handler: async (request, _url, auth) => {
const body = await readJson(request);
if (!body?.space) {
return errorResponse('InvalidRequest', 'space is required');
}
const scopeError = checkManageScope(auth, body.space, 'delete');
if (scopeError) return scopeError;
const row = await ownedSpace(body.space, auth);
if (row instanceof Response) return row;
const recipients = await spaceStorage.listCredentialRecipients(
body.space,
);
await spaceStorage.deleteSpace(body.space, new Date().toISOString());
// Retried until each registered syncer has it or its registration
// expires. A syncer that still misses it learns from SpaceDeleted on
// its next credential request.
await outbox.deliverSpaceDeleted(body.space, recipients);
return Response.json({});
},
},
// The policy is the owner's own state, so only the owner's session reads
// it. A policy can publish itself to readers instead.
'/xrpc/com.atproto.space.getSpace': {
auth: 'optional',
handler: async (_request, url, auth) => {
const space = url.searchParams.get('space');
if (!space) return errorResponse('InvalidRequest', 'space is required');
// Whether the space exists is answered before who may read it: a
// space's address is its owner's DID and a key, so existence is not a
// secret, and the git helper tells a private repository from a public
// one by asking this with no session at all.
const row = await spaceStorage.getSpace(space);
if (!row?.isOwner || row.deletedAt) {
return errorResponse('SpaceNotFound', 'Space not found', 404);
}
if (!auth) {
return errorResponse(
'AuthenticationRequired',
"The space owner's session is required",
401,
);
}
const owned = await ownedSpace(space, auth);
if (owned instanceof Response) return owned;
return Response.json({ uri: row.uri, policy: row.policy });
},
},
'/xrpc/com.atproto.space.removeWriter': {
method: 'POST',
auth: 'required',
handler: async (request, _url, auth) => {
const body = await readJson(request);
if (!body?.space || !body?.did) {
return errorResponse('InvalidRequest', 'space and did are required');
}
const scopeError = checkManageScope(auth, body.space, 'update');
if (scopeError) return scopeError;
const row = await ownedSpace(body.space, auth);
if (row instanceof Response) return row;
if (row.deletedAt) {
return errorResponse('SpaceNotFound', 'Space not found', 404);
}
// The account is tracked again if it writes while still authorized.
await spaceStorage.removeSpaceWriter(body.space, body.did);
return Response.json({});
},
},
'/xrpc/com.atproto.memberspace.putMember': {
method: 'POST',
auth: 'required',
handler: async (request, _url, auth) => {
const body = await readJson(request);
if (!body?.space || !body?.did) {
return errorResponse('InvalidRequest', 'space and did are required');
}
if (typeof body.read !== 'boolean' || typeof body.write !== 'boolean') {
return errorResponse('InvalidRequest', 'read and write are booleans');
}
const scopeError = checkManageScope(auth, body.space, 'update');
if (scopeError) return scopeError;
const row = await memberspace(body.space, auth);
if (row instanceof Response) return row;
const access = { read: body.read, write: body.write };
await spaceStorage.putMember(body.space, body.did, access);
if (!access.write) {
await spaceStorage.removeSpaceWriter(body.space, body.did);
}
await publish(
body.space,
await memberRecordWrites(
spaceStorage,
body.space,
row.policy,
body.did,
access,
),
);
return Response.json({});
},
},
'/xrpc/com.atproto.memberspace.removeMember': {
method: 'POST',
auth: 'required',
handler: async (request, _url, auth) => {
const body = await readJson(request);
if (!body?.space || !body?.did) {
return errorResponse('InvalidRequest', 'space and did are required');
}
const scopeError = checkManageScope(auth, body.space, 'update');
if (scopeError) return scopeError;
const row = await memberspace(body.space, auth);
if (row instanceof Response) return row;
// Credentials already issued stay valid until they expire.
await spaceStorage.removeMember(body.space, body.did);
await spaceStorage.removeSpaceWriter(body.space, body.did);
await publish(
body.space,
await memberRecordWrites(
spaceStorage,
body.space,
row.policy,
body.did,
),
);
return Response.json({});
},
},
'/xrpc/com.atproto.memberspace.listMembers': {
auth: 'required',
handler: async (_request, url, auth) => {
const space = url.searchParams.get('space');
if (!space) return errorResponse('InvalidRequest', 'space is required');
const raw = url.searchParams.get('limit');
const limit = raw === null ? 100 : Number(raw);
if (!Number.isInteger(limit) || limit < 1 || limit > 1000) {
return errorResponse(
'InvalidRequest',
'limit must be an integer between 1 and 1000',
);
}
const row = await memberspace(space, auth);
if (row instanceof Response) return row;
const { members, cursor } = await spaceStorage.listMembers(
space,
url.searchParams.get('cursor'),
limit,
);
return Response.json({
...(cursor ? { cursor } : {}),
members,
});
},
},
// A repo host telling this authority that one of its users wrote. Maintains
// the writer set, which is what listRepos enumerates as the sync boundary.
'/xrpc/com.atproto.space.notifyWrite': {
method: 'POST',
handler: async (request) => {
const body = await readJson(request);
const repoRev = body?.repoRev;
if (
!body?.space ||
!body?.repo ||
!repoRev ||
typeof body.data?.$bytes !== 'string'
) {
return errorResponse(
'InvalidRequest',
'space, repo, repoRev and data are required',
);
}
if (!isTid(repoRev)) {
return errorResponse('InvalidRequest', 'repoRev must be a TID');
}
const hosted = await getDid();
if (!hosted) {
return errorResponse('SpaceNotFound', 'Server not initialised', 404);
}
/** @type {{iss: string}} */
let caller;
try {
caller = await verifyServiceAuth({
jwt: (request.headers.get('authorization') ?? '').replace(
/^Bearer\s+/i,
'',
),
aud: [spaceHostAudience(hosted), hosted],
lxm: 'com.atproto.space.notifyWrite',
resolveDid,
verifier,
});
} catch (err) {
if (err instanceof SpaceTokenError) {
return errorResponse(err.code, err.message, 401);
}
throw err;
}
// A repo host may only report its own account's writes.
if (caller.iss !== body.repo) {
return errorResponse(
'Forbidden',
'Can only report writes for the calling account',
403,
);
}
const row = await spaceStorage.getSpace(body.space);
if (!row?.isOwner || row.deletedAt) {
return errorResponse('SpaceNotFound', 'Space not found', 404);
}
if (isFutureRev(repoRev)) {
return errorResponse('FutureRev', 'repoRev is too far in the future');
}
// The write perimeter, which the authority holds alone. A writer the
// space does not admit stays out of the writer set, so listRepos never
// names them and no syncer pulls their repo.
const authorized = await authorizeSpaceUser({
spaceStorage,
spaceRow: row,
access: 'write',
userDid: body.repo,
resolveDid,
getSigner: ctx.getSigner,
fetch: ctx.fetch,
});
if (!authorized) {
return errorResponse(
'Forbidden',
'Writer is not authorized for this space',
403,
);
}
/** @type {Uint8Array} */
let data;
try {
data = Uint8Array.from(atob(body.data.$bytes), (c) =>
c.charCodeAt(0),
);
} catch {
return errorResponse('InvalidRequest', 'data is not base64');
}
const sequenced = await recordWriterUpdate(spaceStorage, {
space: body.space,
did: body.repo,
repoRev,
data,
});
// A repeated or out-of-order notice: the writer set already holds this
// rev or a later one.
if (!sequenced) return Response.json({});
// The authority is the only party that knows who is following this
// space, so forwarding is its job. A writer's own host cannot do it: the
// registrations are held here.
await forwardToSyncers(ctx, {
authorityDid: hosted,
space: body.space,
repo: body.repo,
repoRev,
data,
...sequenced,
});
return Response.json({});
},
},
// A space's authority telling this repo host that credentials it issued
// are revoked before they expire. Kept long enough to outlive any of them.
'/xrpc/com.atproto.space.notifyCredentialRevoked': {
method: 'POST',
handler: async (request) => {
const body = await readJson(request);
const credentials = body?.credentials;
if (
typeof body?.space !== 'string' ||
!Array.isArray(credentials) ||
credentials.length < 1 ||
credentials.length > 100 ||
!credentials.every((jti) => typeof jti === 'string' && jti)
) {
return errorResponse(
'InvalidRequest',
'space and 1 to 100 credential ids are required',
);
}
/** @type {string} */
let authority;
try {
authority = parseSpaceUri(body.space).spaceDid;
} catch {
return errorResponse(
'InvalidSpaceUri',
`Not a space uri: ${body.space}`,
);
}
const hosted = await getDid();
if (!hosted) {
return errorResponse('Forbidden', 'Server not initialised', 403);
}
/** @type {{iss: string}} */
let caller;
try {
caller = await verifyServiceAuth({
jwt: (request.headers.get('authorization') ?? '').replace(
/^Bearer\s+/i,
'',
),
aud: hosted,
lxm: 'com.atproto.space.notifyCredentialRevoked',
resolveDid,
verifier,
});
} catch (err) {
if (err instanceof SpaceTokenError) {
return errorResponse(err.code, err.message, 401);
}
throw err;
}
if (caller.iss !== authority) {
return errorResponse(
'Forbidden',
'Only the space authority may revoke its credentials',
403,
);
}
await spaceStorage.putCredentialRevocations(
body.space,
[...new Set(/** @type {string[]} */ (credentials))],
new Date(Date.now() + REVOCATION_RETENTION_MS).toISOString(),
);
return Response.json({});
},
},
// A syncer subscribing to write notifications. Authenticated with a space
// credential, so only someone the authority already admitted can subscribe.
'/xrpc/com.atproto.space.registerNotify': {
method: 'POST',
handler: async (request) => {
const body = await readJson(request);
if (!body?.space || !body?.service) {
return errorResponse(
'InvalidRequest',
'space and service are required',
);
}
const denied = await requireSpaceCredential(request, body.space);
if (denied) return denied;
const owned = await hostedSpace(body.space);
if (owned instanceof Response) return owned;
// The delivery URL comes from the subscriber's own DID document, so a
// registration can only ever point at an endpoint that service
// published for itself.
const didDoc = await resolveDid(body.service.split('#')[0]);
const endpoint = didDoc && namedServiceEndpoint(didDoc, body.service);
if (!endpoint) {
return errorResponse(
'ServiceNotResolvable',
`Could not resolve a service endpoint for ${body.service}`,
);
}
const expiresAt = new Date(Date.now() + REGISTRATION_TTL_MS)
.toISOString()
.replace(/\.\d{3}Z$/, '.000Z');
await spaceStorage.putCredentialRecipient(
body.space,
body.service,
endpoint,
expiresAt,
);
return Response.json({ expiresAt });
},
},
// Withdrawing a registration. The service identifier is not resolved here:
// a subscriber whose DID document has changed must still be able to leave.
'/xrpc/com.atproto.space.unregisterNotify': {
method: 'POST',
handler: async (request) => {
const body = await readJson(request);
if (!body?.space || !body?.service) {
return errorResponse(
'InvalidRequest',
'space and service are required',
);
}
const denied = await requireSpaceCredential(request, body.space);
if (denied) return denied;
const owned = await hostedSpace(body.space);
if (owned instanceof Response) return owned;
// Idempotent: succeeds whether or not a registration existed.
await spaceStorage.deleteCredentialRecipient(body.space, body.service);
return Response.json({});
},
},
// An authority telling us one of its spaces is gone.
'/xrpc/com.atproto.space.notifySpaceDeleted': {
method: 'POST',
handler: async (request) => {
const body = await readJson(request);
if (!body?.space) {
return errorResponse('InvalidRequest', 'space is required');
}
const hosted = await getDid();
if (!hosted) {
return errorResponse('SpaceNotFound', 'Server not initialised', 404);
}
/** @type {{iss: string}} */
let caller;
try {
caller = await verifyServiceAuth({
jwt: (request.headers.get('authorization') ?? '').replace(
/^Bearer\s+/i,
'',
),
aud: hosted,
lxm: 'com.atproto.space.notifySpaceDeleted',
resolveDid,
verifier,
});
} catch (err) {
if (err instanceof SpaceTokenError) {
return errorResponse(err.code, err.message, 401);
}
throw err;
}
// Only the space's own authority may declare it deleted.
const { spaceDid } = parseSpaceUri(body.space);
if (caller.iss !== spaceDid) {
return errorResponse(
'Forbidden',
'Only the space authority may delete a space',
403,
);
}
const row = await spaceStorage.getSpace(body.space);
if (row) {
// Flag the member's repo rather than erasing it: the records are the
// member's own, and the proposal has repo hosts keep them.
await spaceStorage.deleteSpace(body.space, new Date().toISOString());
}
return Response.json({});
},
},
};
}