chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
/**
* npm registry HTTP API served from dev.pdsjs.npm.package records, so the
* stock npm client can publish and install:
*
* npm config set registry https://<pds-host>/npm/
* npm publish
* npm install <package>
*
* The surface lives under /npm/ because the npm protocol wants the URL
* root, which the PDS owns. Tarball and manifest bytes are stored as blobs
* exactly as published; the record holds the dist-tags, the integrity
* strings and a blob reference per version, which is what holds the blobs
* against orphan cleanup. The packument is assembled from the manifest
* blobs, with each tarball URL rewritten to the requesting origin.
*
* Reads go through an injected fetch, typically the server's own fetch
* function. Writes go straight to the ports, the way the drive writer
* writes. Without the write ports the registry serves installs only.
*
* Authentication is the account's own: `npm login` verifies through
* createSession and answers the password back as the token, so the stored
* `_authToken` is the app password itself and nothing server-side has to
* survive a restart. Installs are anonymous.
*
* This module must stay free of node-only imports; it runs on Cloudflare
* Workers.
*/
import {
cborEncodeDagCbor,
cidToString,
createBlobCid,
createCid,
} from '@pdsjs/core/repo';
import { createSha256 } from '@pdsjs/core/sha256';
import { fromBase64, tarballHashes } from './integrity.js';
import { isValidName, NPM_PACKAGE_COLLECTION, nameToRkey } from './lexicon.js';
import {
blobRef,
parsePackageRecord,
recordBlobCids,
serializePackageRecord,
} from './record.js';
const TARBALL_MIME_TYPE = 'application/octet-stream';
const MANIFEST_MIME_TYPE = 'application/json';
/** Largest accepted tarball, when the caller sets no other bound. */
const DEFAULT_MAX_TARBALL_SIZE = 64 * 1024 * 1024;
/** How long one verified credential pair stays good without re-checking. */
const AUTH_CACHE_TTL = 5 * 60 * 1000;
const VERSION_PATTERN = /^[0-9A-Za-z.+-]{1,256}$/;
const TAG_PATTERN = /^[a-zA-Z0-9_][a-zA-Z0-9._-]{0,127}$/;
/**
* @typedef {Object} NpmRegistryContext
* @property {(request: Request) => Promise<Response>} xrpc - the PDS fetch
* function; requests are self-addressed XRPC calls
* @property {() => Promise<string|null>} getDid
* @property {import('@pdsjs/core/ports').ActorStoragePort} [actorStorage] - publishes need it
* @property {import('@pdsjs/core/ports').BlobPort} [blobs] - publishes need it
* @property {(did: string, ops: Array<{action: string, uri: string, cid?: string}>) => Promise<Object>} [createCommit] - publishes need it
* @property {import('@pdsjs/core/ports').HasherFactory} [hasher] - SHA-256 factory; absent, the plain JavaScript one
* @property {number} [maxTarballSize] - largest accepted tarball, in bytes
* @property {string} [collection] - record collection, for lexicon forks
*/
/**
* @param {string} message
* @param {number} status
* @returns {Response}
*/
function npmError(message, status) {
return Response.json({ error: message }, { status });
}
/**
* The name without its scope, which names the tarball file.
* @param {string} name
* @returns {string}
*/
function basename(name) {
return name.includes('/') ? name.split('/')[1] : name;
}
/**
* @param {URL} url
* @param {string} name
* @param {string} version
* @returns {string}
*/
function tarballUrl(url, name, version) {
return `${url.origin}/npm/${name}/-/${basename(name)}-${version}.tgz`;
}
/**
* @param {NpmRegistryContext} ctx
* @returns {{handle: (request: Request) => Promise<Response|null>}}
*/
export function createNpmRegistry(ctx) {
if (!ctx || typeof ctx.xrpc !== 'function') {
throw new Error('createNpmRegistry requires an xrpc fetch function');
}
if (typeof ctx.getDid !== 'function') {
throw new Error('createNpmRegistry requires getDid');
}
const {
xrpc,
getDid,
actorStorage,
blobs,
createCommit,
hasher = createSha256,
maxTarballSize = DEFAULT_MAX_TARBALL_SIZE,
collection = NPM_PACKAGE_COLLECTION,
} = ctx;
const publishEnabled = Boolean(actorStorage && blobs && createCommit);
/**
* @param {string} path
* @param {RequestInit} [init]
* @returns {Promise<Response>}
*/
const xrpcFetch = (path, init) =>
xrpc(new Request(`https://pds.internal${path}`, init));
// ── Authentication ─────────────────────────────────────────────────────
/** Verified credential digests, value = epoch ms the entry expires. */
const authCache = new Map();
/**
* @param {string} identifier
* @param {string} password
* @returns {Promise<boolean>}
*/
async function credentialsValid(identifier, password) {
const hash = hasher();
await hash.update(new TextEncoder().encode(`${identifier}\0${password}`));
const digest = await hash.digest();
let key = '';
for (const byte of digest) key += byte.toString(16).padStart(2, '0');
const expires = authCache.get(key);
if (expires && expires > Date.now()) return true;
authCache.delete(key);
const res = await xrpcFetch('/xrpc/com.atproto.server.createSession', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ identifier, password }),
});
if (!res.ok) return false;
authCache.set(key, Date.now() + AUTH_CACHE_TTL);
return true;
}
/**
* True when the request carries credentials createSession accepts. A
* Bearer token is the password alone; the account is the identifier,
* since this server holds exactly one.
* @param {Request} request
* @returns {Promise<boolean>}
*/
async function authorized(request) {
const header = request.headers.get('authorization') ?? '';
if (/^bearer\s/i.test(header)) {
const did = await getDid();
if (!did) return false;
return credentialsValid(did, header.slice(6).trim());
}
if (/^basic\s/i.test(header)) {
let decoded;
try {
decoded = atob(header.slice(5).trim());
} catch {
return false;
}
const didMatch = /^(did:[a-z0-9]+:[^:]+):(.*)$/s.exec(decoded);
const at = decoded.indexOf(':');
if (!didMatch && at < 0) return false;
return credentialsValid(
didMatch ? didMatch[1] : decoded.slice(0, at),
didMatch ? didMatch[2] : decoded.slice(at + 1),
);
}
return false;
}
// ── Reads ──────────────────────────────────────────────────────────────
/**
* @param {string} did
* @param {string} name
* @returns {Promise<{cid: string, record: import('./record.js').NpmPackageRecord}|null>}
*/
async function loadPackage(did, name) {
const params = new URLSearchParams({
repo: did,
collection,
rkey: nameToRkey(name),
});
const res = await xrpcFetch(`/xrpc/com.atproto.repo.getRecord?${params}`);
if (!res.ok) return null;
const body = /** @type {{cid: string, value: unknown}} */ (
await res.json()
);
const record = parsePackageRecord(body.value);
return record ? { cid: body.cid, record } : null;
}
/**
* @param {string} did
* @param {string} cid
* @returns {Promise<Uint8Array|null>}
*/
async function readBlob(did, cid) {
const params = new URLSearchParams({ did, cid });
const res = await xrpcFetch(`/xrpc/com.atproto.sync.getBlob?${params}`);
if (!res.ok) return null;
return new Uint8Array(await res.arrayBuffer());
}
/**
* GET /npm/<name>: the packument, assembled from the manifest blobs with
* tarball URLs rewritten to the requesting origin.
* @param {Request} request
* @param {URL} url
* @param {string} did
* @param {string} name
* @returns {Promise<Response>}
*/
async function servePackument(request, url, did, name) {
const found = await loadPackage(did, name);
if (!found) return npmError(`package not found: ${name}`, 404);
const { cid, record } = found;
const ifNoneMatch = request.headers.get('if-none-match');
if (ifNoneMatch?.includes(cid)) {
return new Response(null, { status: 304, headers: { ETag: `"${cid}"` } });
}
/** @type {Record<string, unknown>} */
const versions = {};
/** @type {Record<string, string>} */
const time = {};
for (const entry of record.versions) {
const bytes = await readBlob(did, entry.manifest.ref.$link);
if (!bytes) continue;
/** @type {Record<string, unknown>} */
let manifest;
try {
manifest = JSON.parse(new TextDecoder().decode(bytes));
} catch {
continue;
}
manifest.dist = {
.../** @type {Record<string, unknown>} */ (manifest.dist ?? {}),
tarball: tarballUrl(url, name, entry.version),
integrity: entry.integrity,
...(entry.shasum ? { shasum: entry.shasum } : {}),
};
versions[entry.version] = manifest;
if (entry.publishedAt) time[entry.version] = entry.publishedAt;
}
/** @type {Record<string, string>} */
const distTags = {};
for (const tag of record.distTags) distTags[tag.tag] = tag.version;
return Response.json(
{
name,
'dist-tags': distTags,
versions,
time: {
created: record.createdAt,
modified: record.updatedAt ?? record.createdAt,
...time,
},
},
{ headers: { ETag: `"${cid}"` } },
);
}
/**
* GET /npm/<name>/-/<file>.tgz: the version's tarball, streamed from
* blob storage.
* @param {URL} _url
* @param {string} did
* @param {string} name
* @param {string} file
* @returns {Promise<Response>}
*/
async function serveTarball(_url, did, name, file) {
const prefix = `${basename(name)}-`;
if (!file.startsWith(prefix) || !file.endsWith('.tgz')) {
return npmError(`tarball not found: ${file}`, 404);
}
const version = file.slice(prefix.length, -4);
const found = await loadPackage(did, name);
const entry = found?.record.versions.find((v) => v.version === version);
if (!entry) return npmError(`tarball not found: ${file}`, 404);
const params = new URLSearchParams({ did, cid: entry.tarball.ref.$link });
const res = await xrpcFetch(`/xrpc/com.atproto.sync.getBlob?${params}`);
if (!res.ok) return npmError(`tarball not found: ${file}`, 404);
return new Response(res.body, {
headers: {
'Content-Type': TARBALL_MIME_TYPE,
'Content-Length':
res.headers.get('content-length') ?? String(entry.tarball.size),
},
});
}
// ── Writes ─────────────────────────────────────────────────────────────
/**
* One write to a package record at a time, so concurrent publishes
* read-modify-write in sequence rather than losing updates.
* @type {Map<string, Promise<unknown>>}
*/
const recordLocks = new Map();
/**
* @template T
* @param {string} name
* @param {() => Promise<T>} fn
* @returns {Promise<T>}
*/
function withRecordLock(name, fn) {
const previous = recordLocks.get(name) ?? Promise.resolve();
const next = previous.then(fn, fn);
recordLocks.set(
name,
next.catch(() => {}),
);
return next;
}
/**
* Store the record, link its blobs and commit, the way the drive writer
* does. The link runs before the commit, so a reap between the two
* cannot take a blob the record already holds.
* @param {string} did
* @param {string} name
* @param {import('./record.js').NpmPackageRecord} record
* @param {boolean} existed
*/
async function writePackageRecord(did, name, record, existed) {
const storage = /** @type {NonNullable<typeof actorStorage>} */ (
actorStorage
);
const rkey = nameToRkey(name);
const uri = `at://${did}/${collection}/${rkey}`;
const encoded = cborEncodeDagCbor(serializePackageRecord(record));
const cid = cidToString(await createCid(encoded));
await storage.putBlock(cid, encoded);
await storage.putRecord(uri, cid, collection, rkey, encoded);
await storage.unlinkBlobsFromRecord(uri);
for (const blobCid of recordBlobCids(record)) {
await storage.linkBlobToRecord(blobCid, uri, Date.now());
}
await /** @type {NonNullable<typeof createCommit>} */ (createCommit)(did, [
{ action: existed ? 'update' : 'create', uri, cid },
]);
}
/**
* @param {string} did
* @param {Uint8Array} bytes
* @param {string} mimeType
* @returns {Promise<string>} the blob's CID
*/
async function storeBlob(did, bytes, mimeType) {
const cid = cidToString(await createBlobCid(bytes));
await /** @type {NonNullable<typeof blobs>} */ (blobs).put(
did,
cid,
bytes,
mimeType,
);
await /** @type {NonNullable<typeof actorStorage>} */ (
actorStorage
).putBlob(cid, mimeType, bytes.length);
return cid;
}
/**
* Read the whole request body, refusing one past the cap. The publish
* body carries the tarball base64-encoded, so the cap allows for the
* encoding overhead.
* @param {Request} request
* @param {number} cap
* @returns {Promise<Uint8Array|null>} null when the body exceeds the cap
*/
async function readBody(request, cap) {
if (!request.body) return new Uint8Array(0);
const chunks = [];
let size = 0;
const reader = request.body.getReader();
while (true) {
const { done, value } = await reader.read();
if (done) break;
const chunk = value instanceof Uint8Array ? value : new Uint8Array(value);
size += chunk.length;
if (size > cap) {
await reader.cancel();
return null;
}
chunks.push(chunk);
}
const joined = new Uint8Array(size);
let at = 0;
for (const chunk of chunks) {
joined.set(chunk, at);
at += chunk.length;
}
return joined;
}
/**
* PUT /npm/<name>: publish. The body is the package document npm sends:
* version manifests plus the tarballs base64-encoded in `_attachments`.
* @param {Request} request
* @param {URL} url
* @param {string} did
* @param {string} name
* @returns {Promise<Response>}
*/
async function publish(request, url, did, name) {
// base64 inflates the tarball by a third; the slack covers manifests.
const body = await readBody(
request,
Math.ceil(maxTarballSize * 1.4) + 1024 * 1024,
);
if (!body) {
return npmError(
`package exceeds the maximum size of ${maxTarballSize} bytes`,
413,
);
}
/** @type {Record<string, unknown>} */
let doc;
try {
doc = JSON.parse(new TextDecoder().decode(body));
} catch {
return npmError('the publish body is not valid JSON', 400);
}
if (doc.name !== name) {
return npmError('the document name does not match the URL', 400);
}
const versions = /** @type {Record<string, Record<string, unknown>>} */ (
doc.versions && typeof doc.versions === 'object' ? doc.versions : {}
);
const attachments =
/** @type {Record<string, {data?: unknown, length?: number}>} */ (
doc._attachments && typeof doc._attachments === 'object'
? doc._attachments
: {}
);
const docTags = /** @type {Record<string, unknown>} */ (
doc['dist-tags'] && typeof doc['dist-tags'] === 'object'
? doc['dist-tags']
: {}
);
if (Object.keys(versions).length === 0) {
return npmError('the publish body names no versions', 400);
}
return withRecordLock(name, async () => {
const existing = await loadPackage(did, name);
const now = new Date().toISOString();
/** @type {import('./record.js').NpmPackageRecord} */
const record = existing?.record ?? {
name,
distTags: [],
versions: [],
createdAt: now,
};
for (const [version, manifest] of Object.entries(versions)) {
if (!VERSION_PATTERN.test(version)) {
return npmError(`invalid version: ${version}`, 400);
}
if (record.versions.some((v) => v.version === version)) {
// npm expects 403 here and reports the version conflict itself.
return npmError(
`cannot publish over the previously published version ${version}`,
403,
);
}
const attachment =
attachments[`${basename(name)}-${version}.tgz`] ??
Object.values(attachments)[0];
if (typeof attachment?.data !== 'string') {
return npmError(`no tarball attached for version ${version}`, 400);
}
/** @type {Uint8Array} */
let tarball;
try {
tarball = fromBase64(attachment.data);
} catch {
return npmError('the tarball attachment is not valid base64', 400);
}
if (tarball.length > maxTarballSize) {
return npmError(
`package exceeds the maximum size of ${maxTarballSize} bytes`,
413,
);
}
const hashes = await tarballHashes(tarball);
const declared = /** @type {{integrity?: unknown}} */ (
manifest.dist ?? {}
).integrity;
if (typeof declared === 'string' && declared !== hashes.integrity) {
return npmError(
`the tarball hashes to ${hashes.integrity}, not the declared integrity`,
400,
);
}
const tarballCid = await storeBlob(did, tarball, TARBALL_MIME_TYPE);
const manifestBytes = new TextEncoder().encode(
JSON.stringify(manifest),
);
const manifestCid = await storeBlob(
did,
manifestBytes,
MANIFEST_MIME_TYPE,
);
record.versions.push({
version,
tarball: blobRef(tarballCid, TARBALL_MIME_TYPE, tarball.length),
manifest: blobRef(
manifestCid,
MANIFEST_MIME_TYPE,
manifestBytes.length,
),
integrity: hashes.integrity,
shasum: hashes.shasum,
publishedAt: now,
});
}
for (const [tag, version] of Object.entries(docTags)) {
if (typeof version !== 'string' || !TAG_PATTERN.test(tag)) continue;
const entry = record.distTags.find((t) => t.tag === tag);
if (entry) entry.version = version;
else record.distTags.push({ tag, version });
}
record.updatedAt = now;
await writePackageRecord(did, name, record, Boolean(existing));
return Response.json({ ok: true, id: name }, { status: 201 });
});
}
/**
* Change one dist-tag: PUT with the version as a JSON string, or DELETE.
* @param {Request} request
* @param {string} did
* @param {string} name
* @param {string} tag
* @returns {Promise<Response>}
*/
async function changeDistTag(request, did, name, tag) {
return withRecordLock(name, async () => {
const found = await loadPackage(did, name);
if (!found) return npmError(`package not found: ${name}`, 404);
const { record } = found;
if (request.method === 'DELETE') {
if (!record.distTags.some((t) => t.tag === tag)) {
return npmError(`tag not found: ${tag}`, 404);
}
record.distTags = record.distTags.filter((t) => t.tag !== tag);
} else {
let version;
try {
version = JSON.parse(await request.text());
} catch {
return npmError('the body is not a JSON version string', 400);
}
if (
typeof version !== 'string' ||
!record.versions.some((v) => v.version === version)
) {
return npmError(`version not found: ${version}`, 404);
}
const entry = record.distTags.find((t) => t.tag === tag);
if (entry) entry.version = version;
else record.distTags.push({ tag, version });
}
record.updatedAt = new Date().toISOString();
await writePackageRecord(did, name, record, true);
return Response.json({ ok: true });
});
}
// ── Dispatch ───────────────────────────────────────────────────────────
/**
* @param {string[]} segments - path segments after /npm/
* @returns {{name: string, rest: string[]}|null}
*/
function parseName(segments) {
if (segments.length === 0) return null;
if (segments[0].startsWith('@')) {
if (segments.length < 2) return null;
return {
name: `${segments[0]}/${segments[1]}`,
rest: segments.slice(2),
};
}
return { name: segments[0], rest: segments.slice(1) };
}
return {
/**
* @param {Request} request
* @returns {Promise<Response|null>} null when the path is not a
* registry route
*/
async handle(request) {
const url = new URL(request.url);
const pathname = decodeURIComponent(url.pathname);
if (pathname !== '/npm' && !pathname.startsWith('/npm/')) return null;
try {
const did = await getDid();
if (!did) return npmError('server not initialized', 404);
const tail = pathname.slice('/npm'.length);
const segments = tail.split('/').filter((part) => part !== '');
if (segments.length === 0) {
return Response.json({});
}
// Meta routes, the ones npm serves under /-/.
if (segments[0] === '-') {
if (segments[1] === 'ping') return Response.json({});
// PUT /-/user/org.couchdb.user:<name>, the npm login shape.
if (
request.method === 'PUT' &&
segments[1] === 'user' &&
segments.length === 3
) {
/** @type {{name?: unknown, password?: unknown}} */
let body;
try {
body = JSON.parse(await request.text());
} catch {
return npmError('the login body is not valid JSON', 400);
}
if (
typeof body.name !== 'string' ||
typeof body.password !== 'string' ||
!(await credentialsValid(body.name, body.password))
) {
return npmError('invalid credentials', 401);
}
// The token is the password itself: createSession verifies it on
// every use, so nothing server-side has to survive a restart.
return Response.json(
{
ok: true,
id: `org.couchdb.user:${body.name}`,
token: body.password,
},
{ status: 201 },
);
}
if (segments[1] === 'whoami') {
if (!(await authorized(request))) {
return npmError('authentication required', 401);
}
const res = await xrpcFetch(
`/xrpc/com.atproto.repo.describeRepo?repo=${encodeURIComponent(did)}`,
);
const handle = res.ok
? /** @type {{handle?: string}} */ (await res.json()).handle
: null;
return Response.json({ username: handle || did });
}
if (segments[1] === 'package') {
const parsed = parseName(segments.slice(2));
if (!parsed || !isValidName(parsed.name)) {
return npmError('invalid package name', 400);
}
if (parsed.rest[0] !== 'dist-tags') {
return npmError('unsupported route', 404);
}
if (request.method === 'GET') {
const found = await loadPackage(did, parsed.name);
if (!found) {
return npmError(`package not found: ${parsed.name}`, 404);
}
/** @type {Record<string, string>} */
const tags = {};
for (const tag of found.record.distTags) {
tags[tag.tag] = tag.version;
}
return Response.json(tags);
}
const tag = parsed.rest[1];
if (!tag || !TAG_PATTERN.test(tag)) {
return npmError('invalid tag', 400);
}
if (!publishEnabled) {
return npmError('this registry is read-only', 405);
}
if (!(await authorized(request))) {
return npmError('authentication required', 401);
}
return await changeDistTag(request, did, parsed.name, tag);
}
// npm probes audit and search endpoints during installs; a plain
// 404 makes the client skip the feature.
return npmError('unsupported route', 404);
}
const parsed = parseName(segments);
if (!parsed || !isValidName(parsed.name)) {
return npmError('invalid package name', 400);
}
const { name, rest } = parsed;
if (rest.length === 2 && rest[0] === '-') {
if (request.method !== 'GET') {
return npmError('unsupported method', 405);
}
return await serveTarball(url, did, name, rest[1]);
}
if (rest.length === 0) {
if (request.method === 'GET') {
return await servePackument(request, url, did, name);
}
if (request.method === 'PUT') {
if (!publishEnabled) {
return npmError('this registry is read-only', 405);
}
if (!(await authorized(request))) {
return npmError('authentication required', 401);
}
return await publish(request, url, did, name);
}
if (request.method === 'DELETE') {
return npmError(
'unpublish is not supported; delete the dev.pdsjs.npm.package record instead',
403,
);
}
}
return npmError('unsupported route', 404);
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
return npmError(message, 500);
}
},
};
}