chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
/**
* Read-only git smart HTTP (protocol v0) served from dev.pdsjs.git.repo
* records, so stock git can clone without the remote helper:
*
* git clone https://<pds-host>/git/<handle-or-did>/<repo>
*
* Refs are advertised from the record; fetch responses are built by merging
* the bundle chain's packfiles (see pack.js), so no negotiation state and no
* git binary is needed server-side. Pushes are refused with a pointer to
* git-remote-atproto.
*
* This module must stay free of node-only imports; it runs on Cloudflare
* Workers. All PDS access goes through an injected fetch, typically the
* server's own fetch function.
*/
import { createGitBrowser } from './browser.js';
import { GIT_REPO_COLLECTION } from './lexicon.js';
import { mergePackStream, parsePackHeader } from './pack.js';
import { concatBytes, FLUSH, parsePkts, pktLine, pktText } from './pkt.js';
import { concatChunks, parseRepoRecord, repoNameOf, rkeyOf } from './record.js';
const ZERO_SHA = '0'.repeat(40);
/** Repositories read from one account's listing to resolve a name. */
const MAX_REPOS = 100;
/**
* Raw files are served as text whatever their extension, the way every other
* raw code host does it. A repository can hold a page that would otherwise
* run as script on this origin, and an agent reading source wants the source
* rather than a rendering of it.
*/
const RAW_TEXT_TYPE = 'text/plain; charset=utf-8';
const RAW_BINARY_TYPE = 'application/octet-stream';
const UPLOAD_PACK = 'git-upload-pack';
const RECEIVE_PACK = 'git-receive-pack';
const PUSH_HINT =
'this repository is read-only over HTTP. Push with git-remote-atproto: git remote set-url --push origin atproto://<handle>/<repo>';
// Payload byte budget for one side-band pkt-line: 65519 frame minus the
// 4-byte length prefix and 1-byte band marker.
const SIDE_BAND_MAX = 65514;
// Band 1 carries pack data, band 3 an error the client prints and fails on.
const BAND_PACK = 1;
const BAND_ERROR = 3;
// A bundle's text header ends at a blank line. It is small next to the
// pack, so one read of this size finds the pack start on any ref list a
// repository is likely to carry.
const BUNDLE_HEAD_READ = 64 * 1024;
/**
* The offset of the packfile inside a bundle: the byte after the blank line
* that ends the bundle's text header.
* @param {Uint8Array} bytes - the start of a bundle
* @returns {number} -1 when the buffer stops short of the blank line
*/
function packOffset(bytes) {
for (let i = 1; i < bytes.length; i++) {
if (bytes[i] === 0x0a && bytes[i - 1] === 0x0a) return i + 1;
}
return -1;
}
/**
* @param {number} band
* @param {Uint8Array} payload - at most SIDE_BAND_MAX bytes
* @returns {Uint8Array} one side-band pkt-line
*/
function bandLine(band, payload) {
const framed = new Uint8Array(payload.length + 1);
framed[0] = band;
framed.set(payload, 1);
return pktLine(framed);
}
/**
* The upload-pack response body after the header, in order: the NAK, the
* merged pack, and the terminating flush when the client took side-band.
* An error after the first byte is out of reach of a status code, so it
* goes out as a band 3 line the client reports.
* @param {import('./pack.js').PackSource[]} sources
* @param {boolean} sideBand
* @returns {AsyncGenerator<Uint8Array>}
*/
async function* uploadPackBody(sources, sideBand) {
yield pktLine('NAK\n');
try {
for await (const chunk of mergePackStream(sources)) {
if (!sideBand) {
yield chunk;
continue;
}
for (let offset = 0; offset < chunk.length; offset += SIDE_BAND_MAX) {
yield bandLine(
BAND_PACK,
chunk.subarray(offset, offset + SIDE_BAND_MAX),
);
}
}
} catch (err) {
if (!sideBand) throw err;
const message = err instanceof Error ? err.message : String(err);
yield bandLine(BAND_ERROR, new TextEncoder().encode(`${message}\n`));
return;
}
if (sideBand) yield FLUSH;
}
/**
* @param {AsyncGenerator<Uint8Array>} chunks
* @returns {ReadableStream<Uint8Array>}
*/
function toStream(chunks) {
return new ReadableStream({
async pull(controller) {
const { value, done } = await chunks.next();
if (done) controller.close();
else controller.enqueue(value);
},
async cancel() {
await chunks.return(undefined);
},
});
}
/**
* @typedef {Object} GitHttpContext
* @property {(request: Request) => Promise<Response>} xrpc - the PDS fetch
* function; requests are self-addressed XRPC reads
* @property {string} [collection] - record collection, for lexicon forks
*/
/**
* @param {GitHttpContext} ctx
* @returns {{handle: (request: Request) => Promise<Response|null>}}
*/
export function createGitHttp(ctx) {
if (!ctx || typeof ctx.xrpc !== 'function') {
throw new Error('createGitHttp requires an xrpc fetch function');
}
const collection = ctx.collection ?? GIT_REPO_COLLECTION;
/**
* @param {string} path
* @returns {Promise<Response>}
*/
const xrpcGet = (path) =>
ctx.xrpc(new Request(`https://pds.internal${path}`));
/**
* @param {string} account - handle or DID from the URL
* @returns {Promise<string|null>}
*/
async function resolveDid(account) {
if (account.startsWith('did:')) return account;
const res = await xrpcGet(
`/xrpc/com.atproto.identity.resolveHandle?handle=${encodeURIComponent(account)}`,
);
if (!res.ok) return null;
const body = /** @type {{did: string}} */ (await res.json());
return body.did;
}
/**
* The repository one name addresses, from the account's own listing. The
* URL carries a name and the record carries a key, so the listing is what
* joins them.
* @param {string} did
* @param {string} repoName
* @returns {Promise<{rkey: string, cid: string, repo: import('./record.js').GitRepoRecord}|null>}
*/
async function loadRepo(did, repoName) {
const params = new URLSearchParams({
repo: did,
collection,
limit: String(MAX_REPOS),
});
const res = await xrpcGet(`/xrpc/com.atproto.repo.listRecords?${params}`);
if (!res.ok) return null;
const body =
/** @type {{records?: Array<{uri?: unknown, cid?: unknown, value?: unknown}>}} */ (
await res.json()
);
for (const row of body.records ?? []) {
if (typeof row.uri !== 'string' || typeof row.cid !== 'string') continue;
const rkey = rkeyOf(row.uri);
if (repoNameOf(row.value, rkey) !== repoName) continue;
return { rkey, cid: row.cid, repo: parseRepoRecord(row.value) };
}
return null;
}
/**
* @param {string} account
* @param {string} repoName
* @returns {Promise<{did: string, rkey: string, cid: string, repo: import('./record.js').GitRepoRecord}|null>}
*/
async function resolveRepo(account, repoName) {
const did = await resolveDid(account);
if (!did) return null;
const found = await loadRepo(did, repoName);
return found ? { did, ...found } : null;
}
/**
* True when If-None-Match names the record's cid. The cid is the strong
* validator for every read here: any push writes a new record cid.
* @param {Request} request
* @param {string} cid
* @returns {boolean}
*/
function etagMatches(request, cid) {
const raw = request.headers.get('if-none-match');
if (!raw) return false;
return raw
.split(',')
.map((tag) => tag.trim())
.some((tag) => tag === '*' || tag.replace(/^W\//, '') === `"${cid}"`);
}
/**
* @param {string} cid
* @param {Record<string, string>} [extra]
* @returns {Response}
*/
function notModified(cid, extra = {}) {
return new Response(null, {
status: 304,
headers: { ETag: `"${cid}"`, 'Cache-Control': 'no-cache', ...extra },
});
}
/**
* One byte range of a blob. A 206 answers the range itself; a server that
* ignores Range answers 200 with the whole blob, sliced here so both
* behave the same.
* @param {string} did
* @param {string} cid
* @param {number} start
* @param {number} end - exclusive
* @returns {Promise<Uint8Array>}
*/
async function fetchBlobRange(did, cid, start, end) {
const params = new URLSearchParams({ did, cid });
const res = await ctx.xrpc(
new Request(
`https://pds.internal/xrpc/com.atproto.sync.getBlob?${params}`,
{ headers: { range: `bytes=${start}-${end - 1}` } },
),
);
if (!res.ok) throw new Error(`bundle blob ${cid} unavailable`);
const bytes = new Uint8Array(await res.arrayBuffer());
if (res.status === 206) return bytes;
return bytes.subarray(start, end);
}
/**
* @param {import('./record.js').BundleEntry} bundle
* @returns {number} the bundle's length in bytes
*/
function bundleLength(bundle) {
return bundle.parts.reduce((sum, part) => sum + part.size, 0);
}
/**
* Bundle bytes in [start, end), one chunk blob at a time, so a reader
* holds one chunk however long the bundle is.
* @param {string} did
* @param {import('./record.js').BundleEntry} bundle
* @param {number} start
* @param {number} end - exclusive
* @returns {AsyncGenerator<Uint8Array>}
*/
async function* bundleBytes(did, bundle, start, end) {
let offset = 0;
for (const part of bundle.parts) {
if (offset >= end) return;
const partEnd = offset + part.size;
if (partEnd > start) {
yield await fetchBlobRange(
did,
part.ref.$link,
Math.max(start, offset) - offset,
Math.min(end, partEnd) - offset,
);
}
offset = partEnd;
}
}
/**
* Where the packfile starts inside the bundle, and how many objects it
* holds. Both come from the bundle's first bytes, so the pack itself
* stays unread until the response streams.
* @param {string} did
* @param {import('./record.js').BundleEntry} bundle
* @returns {Promise<{packStart: number, count: number}>}
*/
async function readPackHeader(did, bundle) {
const total = bundleLength(bundle);
/** @type {Uint8Array[]} */
const head = [];
let read = 0;
while (read < total) {
const end = Math.min(total, read + BUNDLE_HEAD_READ);
for await (const chunk of bundleBytes(did, bundle, read, end)) {
head.push(chunk);
}
read = end;
const bytes = concatChunks(head);
const packStart = packOffset(bytes);
if (packStart !== -1 && packStart + 12 <= bytes.length) {
return {
packStart,
count: parsePackHeader(bytes.subarray(packStart, packStart + 12)),
};
}
}
throw new Error('bundle carries no packfile');
}
/**
* @param {string} did
* @param {import('./record.js').BundleEntry} bundle
* @returns {Promise<import('./pack.js').PackSource>}
*/
async function bundlePackSource(did, bundle) {
const total = bundleLength(bundle);
const { packStart, count } = await readPackHeader(did, bundle);
return {
count,
length: total - packStart,
read: (start, end) =>
bundleBytes(did, bundle, packStart + start, packStart + end),
};
}
/**
* Advertised ref list: HEAD first (when resolvable) so clients pick the
* default branch, then refs in name order.
* @param {import('./record.js').GitRepoRecord} repo
* @returns {{lines: Array<{sha: string, name: string}>, symref: string|null}}
*/
function advertisedRefs(repo) {
const sorted = [...repo.refs].sort((a, b) =>
a.name < b.name ? -1 : a.name > b.name ? 1 : 0,
);
const headRef = repo.defaultBranch
? sorted.find((r) => r.name === `refs/heads/${repo.defaultBranch}`)
: sorted.find((r) => r.name.startsWith('refs/heads/'));
/** @type {Array<{sha: string, name: string}>} */
const lines = [];
let symref = null;
if (headRef) {
lines.push({ sha: headRef.sha, name: 'HEAD' });
symref = headRef.name;
}
lines.push(...sorted);
return { lines, symref };
}
/**
* @param {import('./record.js').GitRepoRecord} repo
* @param {string} cid
* @returns {Response}
*/
function refAdvertisement(repo, cid) {
const { lines, symref } = advertisedRefs(repo);
const caps = ['side-band-64k', 'ofs-delta', 'agent=pdsjs-git'];
if (symref) caps.unshift(`symref=HEAD:${symref}`);
const capString = caps.join(' ');
/** @type {Uint8Array[]} */
const parts = [pktLine(`# service=${UPLOAD_PACK}\n`), FLUSH];
if (lines.length === 0) {
parts.push(pktLine(`${ZERO_SHA} capabilities^{}\0${capString}\n`));
} else {
parts.push(pktLine(`${lines[0].sha} ${lines[0].name}\0${capString}\n`));
for (const ref of lines.slice(1)) {
parts.push(pktLine(`${ref.sha} ${ref.name}\n`));
}
}
parts.push(FLUSH);
return new Response(/** @type {BodyInit} */ (concatBytes(parts)), {
headers: {
'Content-Type': `application/x-${UPLOAD_PACK}-advertisement`,
'Cache-Control': 'no-cache',
ETag: `"${cid}"`,
// The advertisement is a public read, like the raw file paths. The
// open origin lets a page on the account's site ask whether this
// server clones over HTTP before it offers an https clone URL.
'Access-Control-Allow-Origin': '*',
},
});
}
/**
* Pick the shortest chain suffix that covers what the client is missing:
* walk backward collecting bundles until one's heads are all in the
* client's haves. No haves in common means the whole chain (a clone).
* @param {import('./record.js').GitRepoRecord} repo
* @param {Set<string>} haves
* @returns {import('./record.js').BundleEntry[]}
*/
function neededBundles(repo, haves) {
const chain = repo.bundles;
let start = 0;
for (let i = chain.length - 1; i >= 0; i--) {
const heads = chain[i].heads;
if (heads.length > 0 && heads.every((sha) => haves.has(sha))) {
start = i + 1;
break;
}
}
return chain.slice(start);
}
/**
* @param {Request} request
* @returns {Promise<Uint8Array>} request body, gunzipped when necessary
*/
async function readBody(request) {
const encoding = request.headers.get('content-encoding') ?? '';
if (/gzip/i.test(encoding) && request.body) {
const stream = request.body.pipeThrough(new DecompressionStream('gzip'));
return new Uint8Array(await new Response(stream).arrayBuffer());
}
return new Uint8Array(await request.arrayBuffer());
}
/**
* @param {string} message
* @param {number} [status]
* @returns {Response}
*/
function errPkt(message, status = 200) {
return new Response(
/** @type {BodyInit} */ (concatBytes([pktLine(`ERR ${message}\n`)])),
{
status,
headers: { 'Content-Type': `application/x-${UPLOAD_PACK}-result` },
},
);
}
/**
* @param {string} did
* @param {import('./record.js').GitRepoRecord} repo
* @param {Request} request
* @returns {Promise<Response>}
*/
async function uploadPack(did, repo, request) {
const body = await readBody(request);
/** @type {Set<string>} */
const wants = new Set();
/** @type {Set<string>} */
const haves = new Set();
let sideBand = false;
for (const pkt of parsePkts(body)) {
if (pkt.type === 'flush') continue;
const line = pktText(pkt.data);
if (line.startsWith('want ')) {
wants.add(line.slice(5, 45));
if (/\bside-band-64k\b/.test(line)) sideBand = true;
} else if (line.startsWith('have ')) {
haves.add(line.slice(5, 45));
} else if (line.startsWith('deepen') || line.startsWith('shallow')) {
return errPkt('shallow fetches are not supported');
} else if (line !== 'done' && line !== '') {
return errPkt(`unsupported request line: ${line.split(' ')[0]}`);
}
}
if (wants.size === 0) return errPkt('no want lines');
const known = new Set(repo.refs.map((r) => r.sha));
for (const want of wants) {
if (!known.has(want)) {
return errPkt(`want ${want} is not an advertised tip`);
}
}
// An empty suffix is still a valid fetch: a refs-only update (branch or
// tag created on objects the client already has) needs no objects sent.
const bundles = neededBundles(repo, haves);
// The pack headers are read before the response starts, so a bundle
// this server cannot read is still a status code rather than a body
// that stops halfway.
/** @type {import('./pack.js').PackSource[]} */
const sources = [];
for (const bundle of bundles) {
sources.push(await bundlePackSource(did, bundle));
}
return new Response(toStream(uploadPackBody(sources, sideBand)), {
headers: {
'Content-Type': `application/x-${UPLOAD_PACK}-result`,
'Cache-Control': 'no-cache',
},
});
}
const browser = createGitBrowser({ xrpc: ctx.xrpc, collection });
/**
* Split a raw URL tail into its ref and the path inside that ref. Both may
* hold slashes, so the split runs against the repository's own ref names,
* longest first. A tail matching no ref reads as a ref with no path, which
* is what reports the bad ref rather than a missing file.
* @param {string[]} refs - full ref names, longest first
* @param {string} tail
* @returns {{ref: string, path: string}}
*/
function splitRefPath(refs, tail) {
for (const full of refs) {
const short = full.replace(/^refs\/(heads|tags)\//, '');
for (const name of new Set([full, short])) {
if (tail === name) return { ref: name, path: '' };
if (tail.startsWith(`${name}/`)) {
return { ref: name, path: tail.slice(name.length + 1) };
}
}
}
return { ref: tail, path: '' };
}
/** @param {string} body @param {number} status */
const rawError = (body, status) =>
new Response(body, {
status,
headers: {
'Content-Type': RAW_TEXT_TYPE,
'Access-Control-Allow-Origin': '*',
},
});
/**
* GET /git/<account>/<repo>/raw/<ref>/<path>
*
* A file answers with its bytes. A directory answers with its entries, one
* per line and a trailing slash on each directory, so a crawler can walk
* the tree with the same route it reads files through.
* @param {Request} request
* @param {string} account
* @param {string} repoName
* @param {string} tail - everything after `/raw/`
*/
async function serveRaw(request, account, repoName, tail) {
const found = await resolveRepo(account, repoName);
if (!found) return rawError('repository not found\n', 404);
const { did, rkey, cid, repo } = found;
if (etagMatches(request, cid)) {
return notModified(cid, { 'Access-Control-Allow-Origin': '*' });
}
const etag = `"${cid}"`;
// Longest first, so splitRefPath matches refs/heads/x before x.
const refs = repo.refs
.map((r) => r.name)
.sort((a, b) => b.length - a.length);
const { ref, path } = splitRefPath(refs, tail.replace(/\/+$/, ''));
// A trailing slash asks for a listing, which is how a caller reads a
// directory whose name a file could also have. A ref's own root is a
// directory whether or not the caller wrote the slash.
const wantsListing = tail === '' || tail.endsWith('/') || path === '';
if (wantsListing) {
/** @type {Awaited<ReturnType<typeof browser.listTree>>} */
let listing;
try {
listing = await browser.listTree(did, rkey, ref, path);
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
return rawError(`${message}\n`, 404);
}
if (!listing) return rawError('repository not found\n', 404);
const body = listing.entries
.map((entry) => (entry.type === 'dir' ? `${entry.name}/` : entry.name))
.join('\n');
return new Response(body ? `${body}\n` : '', {
headers: {
'Content-Type': RAW_TEXT_TYPE,
'Cache-Control': 'no-cache',
ETag: etag,
'X-Content-Type-Options': 'nosniff',
'Access-Control-Allow-Origin': '*',
},
});
}
/** @type {Awaited<ReturnType<typeof browser.readBytes>>} */
let file;
try {
file = await browser.readBytes(did, rkey, ref, path);
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
// A directory asked for without its trailing slash still answers, so a
// crawler following a bare path is not sent away.
if (message.startsWith('not a file')) {
const listing = await browser.listTree(did, rkey, ref, path);
const body = (listing?.entries ?? [])
.map((entry) =>
entry.type === 'dir' ? `${entry.name}/` : entry.name,
)
.join('\n');
return new Response(body ? `${body}\n` : '', {
headers: {
'Content-Type': RAW_TEXT_TYPE,
'Cache-Control': 'no-cache',
ETag: etag,
'X-Content-Type-Options': 'nosniff',
'Access-Control-Allow-Origin': '*',
},
});
}
// An unknown ref or path is a missing file, not a broken server.
return rawError(`${message}\n`, 404);
}
if (!file) return rawError('repository not found\n', 404);
return new Response(/** @type {BodyInit} */ (file.bytes), {
headers: {
'Content-Type': file.binary ? RAW_BINARY_TYPE : RAW_TEXT_TYPE,
'Content-Length': String(file.size),
'Cache-Control': 'no-cache',
ETag: etag,
'X-Content-Type-Options': 'nosniff',
'Content-Security-Policy': "default-src 'none'; sandbox",
'Access-Control-Allow-Origin': '*',
},
});
}
return {
/**
* @param {Request} request
* @returns {Promise<Response|null>} null when the path is not a git route
*/
async handle(request) {
const url = new URL(request.url);
const match = /^\/git\/([^/]+)\/([^/]+?)(?:\.git)?(\/.*)?$/.exec(
decodeURIComponent(url.pathname),
);
if (!match) return null;
const [, account, repoName, subPath = ''] = match;
try {
if (subPath === '/info/refs') {
const service = url.searchParams.get('service');
if (service === RECEIVE_PACK) {
// A push probe. Answer with a well-formed advertisement whose
// only pkt is an ERR, so git surfaces the hint as a remote error
// instead of falling back to the dumb protocol.
return new Response(
/** @type {BodyInit} */ (
concatBytes([
pktLine(`# service=${RECEIVE_PACK}\n`),
FLUSH,
pktLine(`ERR ${PUSH_HINT}\n`),
FLUSH,
])
),
{
headers: {
'Content-Type': `application/x-${RECEIVE_PACK}-advertisement`,
'Cache-Control': 'no-cache',
},
},
);
}
if (service !== UPLOAD_PACK) {
return new Response(
'smart HTTP only; use a git client that supports it\n',
{ status: 400, headers: { 'Content-Type': 'text/plain' } },
);
}
const found = await resolveRepo(account, repoName);
if (!found) {
return new Response('repository not found\n', {
status: 404,
headers: { 'Content-Type': 'text/plain' },
});
}
if (etagMatches(request, found.cid)) {
return notModified(found.cid);
}
return refAdvertisement(found.repo, found.cid);
}
if (subPath === `/${UPLOAD_PACK}` && request.method === 'POST') {
const found = await resolveRepo(account, repoName);
if (!found) {
return new Response('repository not found\n', { status: 404 });
}
return await uploadPack(found.did, found.repo, request);
}
if (subPath === `/${RECEIVE_PACK}`) {
return errPkt(PUSH_HINT, 403);
}
if (subPath === '/raw' || subPath.startsWith('/raw/')) {
if (request.method !== 'GET' && request.method !== 'HEAD') {
return rawError('raw is read-only\n', 405);
}
return await serveRaw(
request,
account,
repoName,
subPath.slice('/raw/'.length),
);
}
return new Response('not found\n', { status: 404 });
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
return new Response(`git http error: ${message}\n`, { status: 500 });
}
},
};
}