johnandersen777.bsky.social/atlibs

Switch repository

Loading…

// The run-module host call for every JS host: the Node server
// (javascript/src/server) and the pages (examples/web). No node: imports, so browsers load it as is.
import { ATTEST_NSID } from "../attest/index.js";
import { pdsOf, subscribeXrpc, transportName, withServiceAuth, wsUrl } from "./index.js";

const PSEUDO = "com.fedfork.exec.";

/**
 * The call-time refusal of a transport this host has none of (com.fedfork.exec.<name>.connect: iroh, webrtc),
 * the same text as rust/host's transport::unavailable. A host's transports are its `calls`: their NSIDs are
 * what it provides, known at load time (Object.keys(calls)).
 */
export const unavailable = (nsid) => {
	const name = transportName(nsid);
	return name && `${nsid}: not available on this host (built without ${name})`;
};

/** The did:web aud of a service-auth token for an https service URL: did:web:<host>[%3A<port>]. The only audience builder. */
export const serviceAud = (url) => {
	const { hostname, port } = new URL(url);
	return `did:web:${hostname}${port ? `%3A${port}` : ""}`;
};

/** text/event-stream frames as [{ event, data }] ("message" when a frame names none), data parsed when it is JSON. */
export function sseEvents(text) {
	return text.split(/\r?\n\r?\n/).flatMap((frame) => {
		let event = "message";
		const data = [];
		for (const line of frame.split(/\r?\n/)) {
			if (line.startsWith("event:")) event = line.slice(6).trim();
			else if (line.startsWith("data:")) data.push(line.slice(5).replace(/^ /, ""));
		}
		if (!data.length) return [];
		const raw = data.join("\n");
		try {
			return [{ event, data: JSON.parse(raw) }];
		} catch {
			return [{ event, data: raw }];
		}
	});
}

/**
 * An XRPC response's body; not ok throws "<nsid> failed with <error>: <message>", with error and status on it.
 * A text/event-stream answer (a procedure that streams, e.g. dev.cocore.inference.dispatch) is read to its
 * end and comes back as { events: sseEvents(body) }; a binary one (a content-type neither JSON nor text:
 * getBlob, a CAR) as { $bytes } (base64, no padding: the atproto JSON form).
 */
export async function json(nsid, response) {
	const type = response.headers.get("content-type")?.toLowerCase() ?? "";
	if (response.ok && type.startsWith("text/event-stream")) return { events: sseEvents(await response.text()) };
	if (response.ok && type && !type.includes("json") && !type.startsWith("text/")) return { $bytes: new Uint8Array(await response.arrayBuffer()).toBase64({ omitPadding: true }) };
	const body = await response.json().catch(() => ({}));
	// A 429 says when it lifts (ratelimit-reset: unix seconds), so nobody has to probe for it.
	const reset = response.status === 429 && response.headers.get("ratelimit-reset");
	const until = reset ? ` (${response.headers.get("ratelimit-policy") ?? "limit"}; resets ${new Date(Number(reset) * 1000).toISOString()})` : "";
	if (!response.ok) throw Object.assign(new Error(`${nsid} failed with ${body.error ?? response.status}${body.message ? `: ${body.message}` : ""}${until}`), { error: body.error, status: response.status });
	return body;
}

/**
 * An XRPC request's body and content-type: a { $bytes } input goes as those bytes with `encoding`
 * (default application/octet-stream: com.atproto.repo.uploadBlob), any other input as JSON.
 */
export function xrpcBody(input, encoding) {
	if (input === undefined) return { headers: {} };
	const bytes = typeof input?.$bytes === "string" && Object.keys(input).length === 1;
	return bytes
		? { body: Uint8Array.fromBase64(input.$bytes), headers: { "content-type": encoding ?? "application/octet-stream" } }
		: { body: JSON.stringify(input), headers: { "content-type": "application/json" } };
}

export const xrpcUrl = (base, nsid, params) => `${new URL(`/xrpc/${nsid}`, base).href}${params ? `?${new URLSearchParams(params)}` : ""}`;

// RFC 6750 / 9449: the resource server refused the access token (a DPoP nonce retry is not that)
const refused = (response) => response.status === 401 && /^(Bearer|DPoP) .*error="invalid_token"/.test(response.headers.get("www-authenticate") ?? "");

/**
 * An @atproto/oauth-client OAuthSession (BrowserOAuthClient's, NodeOAuthClient's) whose fetchHandler renews a refused
 * token once for every run, tab and process sharing its session store. The library's own refreshes after any refused
 * token without re-reading the saved session, so a second tab or process refused the same token refreshed again (refresh
 * tokens rotate: that can end the session). Here, under the client's per-DID lock (Web Locks in a page, the server's
 * requestLock), the saved session is marked expired only while it still holds the refused token; the next read
 * refreshes it once, whoever reads first, and the rest use that. Anything else (an Agent, a test's) is left as it is.
 */
export function renewOnce(session) {
	const { sub, sessionGetter, server, dpopFetch } = session ?? {};
	if (!sessionGetter?.getStored || !server?.runtime?.usingLock || !dpopFetch) return session;
	session.fetchHandler = async (pathname, init = {}) => {
		const send = (tokenSet) => {
			const headers = new Headers(init.headers);
			headers.set("authorization", `${tokenSet.token_type} ${tokenSet.access_token}`);
			return dpopFetch(new URL(pathname, tokenSet.aud), { ...init, headers });
		};
		const used = await session.getTokenSet("auto");
		const first = await send(used);
		if (!refused(first) || init.body instanceof ReadableStream) return first; // a stream cannot be sent twice
		let fresh;
		try {
			await server.runtime.usingLock(`@atproto-oauth-client-${sub}`, async () => {
				const saved = await sessionGetter.getStored(sub);
				if (saved?.tokenSet.access_token === used.access_token) await sessionGetter.setStored(sub, { ...saved, tokenSet: { ...saved.tokenSet, expires_at: new Date(0).toISOString() } });
			});
			fresh = await session.getTokenSet("auto");
		} catch {
			return first;
		}
		return send(fresh);
	};
	return session;
}

/** XRPC as a signed-in OAuth session, through its fetchHandler (DPoP, refresh: renewOnce): agent.call() knows no custom lexicons. */
export const sessionXrpc = (agent) => (renewOnce(agent), async (nsid, { params, input, encoding, headers } = {}) => {
	const body = xrpcBody(input, encoding);
	return json(nsid, await agent.fetchHandler(`/xrpc/${nsid}${params ? `?${new URLSearchParams(params)}` : ""}`, {
		method: input !== undefined ? "POST" : "GET",
		headers: { ...headers, ...body.headers },
		body: body.body,
	}));
});

/** One did#id call to its endpoints: fetchers[<kind>:] at <kind>://<address>, else an https endpoint with ctx.http.fetch. */
async function peerFetch({ ctx, fetchers, mint }, nsid, { params, input, encoding, service, endpoints, signal }) {
	const viaFetcher = endpoints.find((e) => e.kind !== "https" && typeof fetchers[`${e.kind}:`] === "function");
	const https = endpoints.find((e) => e.kind === "https" && e.address.startsWith("https://"));
	if (!viaFetcher && !https) throw new Error(`${nsid}: this host reaches none of ${service}'s endpoints`);
	const { token } = await mint();
	if (typeof token !== "string" || !token) throw new Error("getServiceAuth answered no service-auth token");
	const body = xrpcBody(input, encoding);
	const init = { method: input !== undefined ? "POST" : "GET", headers: { authorization: `Bearer ${token}`, ...body.headers }, body: body.body };
	if (!viaFetcher) return ctx.http.fetch(xrpcUrl(https.address, nsid, params), { ...init, signal });
	const answer = await fetchers[`${viaFetcher.kind}:`]({ ...init, url: xrpcUrl(`${viaFetcher.kind}://${viaFetcher.address}`, nsid, params), signal });
	return new Response([204, 205, 304].includes(answer.status) ? null : (answer.body ?? ""), { status: answer.status, headers: answer.headers });
}

/**
 * runModule's `call` for a host with (or without) an atproto session.
 * @param {{ ctx: { http: { fetch: typeof fetch } }, xrpc?: (nsid: string, init?: object) => Promise<any>, attest?: (input: unknown) => Promise<unknown>,
 *   pds?: string, did?: string, WebSocket?: unknown, canHeader?: boolean, calls?: Record<string, (request: object) => unknown>, signIn?: string }} host
 *   xrpc: the session's (login().xrpc, sessionXrpc(agent)), absent to run signed out (DID-service and proxied queries only);
 *   attest: attestHost(ctx, secp256k1, did), absent to refuse com.fedfork.exec.attest;
 *   pds, or did (its PDS is looked up): where a subscribe with no service connects;
 *   canHeader: false in browsers (a WebSocket takes no headers: serviceAuth only as { param });
 *   calls: the host's own pseudo-calls by NSID (request and subscribe: its transports, e.g. the atlibs-iroh and
 *     atlibs-webrtc repos' js/index.js); any other com.fedfork.exec.* is refused (a transport's connect with unavailable());
 *   signIn: what to do about a missing session, for error messages
 * A request: proxy goes as atproto-proxy through the session (signed out, a query goes to that did#id's
 * endpoint directly, unauthenticated); an https service is called directly with a getServiceAuth token
 * (aud serviceAud(service), lxm nsid); a DID service is a public GET at its PDS (did#id: that service).
 * A subscribe opens wss://<service, the DID's PDS, or pds>/xrpc/<nsid>?<params> with subscribeXrpc.
 */
export function hostCall({ ctx, xrpc, attest, pds, did, WebSocket, canHeader = true, calls = {}, fetchers = {}, signIn = "sign in: an OAuth session, or an explicit app-password config" }) {
	const mint = async (service, nsid) => {
		if (!xrpc) throw new Error(`${nsid} serviceAuth needs a session (${signIn})`);
		return (await xrpc("com.atproto.server.getServiceAuth", { params: { aud: serviceAud(service), lxm: nsid } })).token;
	};
	const xrpcOrFail = async (nsid) => {
		if (!xrpc) throw new Error(`${nsid} needs a session (${signIn})`);
		return xrpc;
	};
	return async (nsid, request = {}) => {
		const { params, input, encoding, proxy, service, endpoints, subscribe, signal, framing, keepalive } = request;
		if (Object.hasOwn(calls, nsid)) return calls[nsid](request);
		if (unavailable(nsid)) throw new Error(unavailable(nsid));
		if (subscribe) {
			if (nsid.startsWith(PSEUDO)) throw new Error(`${nsid} cannot be subscribed to on this host`);
			if (proxy) throw new Error("subscribe cannot go through atproto-proxy; call the service directly");
			const authed = await withServiceAuth(nsid, request, mint, { canHeader });
			const base = service?.startsWith("did:") ? await pdsOf(ctx, service) : (service ?? pds ?? (did && (await pdsOf(ctx, did))));
			if (!base) throw new Error(`${nsid} needs a service or a session`);
			return subscribeXrpc(wsUrl(base, nsid, authed.params), { nsid, signal, framing, keepalive, headers: authed.headers, WebSocket });
		}
		if (nsid === ATTEST_NSID) {
			if (!attest) throw new Error("attest needs a session and a module record that lists its calls");
			return attest(input);
		}
		// Other host pseudo-calls must not reach a PDS.
		if (nsid.startsWith(PSEUDO)) throw new Error(`${nsid} is not available on this host`);
		// did#id with its endpoints: the first of a scheme a fetcher serves, else the first https one, with service auth aud did#id.
		if (endpoints !== undefined && service) return json(nsid, await peerFetch({ ctx, fetchers, mint: async () => (await xrpcOrFail(nsid))("com.atproto.server.getServiceAuth", { params: { aud: service, lxm: nsid } }) }, nsid, request));
		// A DID service (did: its PDS; did#id: that service) is a public GET, never with credentials; so is a
		// proxied query signed out, sent to the proxy's service directly.
		const open = service?.startsWith("did:") ? service : !xrpc && proxy && input === undefined ? proxy : undefined;
		if (open) return json(nsid, await ctx.http.fetch(xrpcUrl(await pdsOf(ctx, open), nsid, params), { signal }));
		if (!xrpc) throw new Error(`${nsid} needs a session (${signIn})`);
		if (service) {
			const token = await mint(service, nsid);
			const body = xrpcBody(input, encoding);
			return json(nsid, await ctx.http.fetch(xrpcUrl(service, nsid, params), {
				method: input !== undefined ? "POST" : "GET",
				signal,
				headers: { authorization: `Bearer ${token}`, ...body.headers },
				body: body.body,
			}));
		}
		return xrpc(nsid, { params, input, ...(encoding && { encoding }), headers: proxy && { "atproto-proxy": proxy } });
	};
}