From a02a3f4ce3675b08254b72e4e8aa5ab532722d3b Mon Sep 17 00:00:00 2001 From: Florian <45694132+flo-bit@users.noreply.github.com> Date: Sat, 8 Aug 2026 00:14:43 +0200 Subject: [PATCH] Generate connected service clients --- .changeset/atproto-service-auth.md | 2 +- packages/contrail/src/cli/commands/connect.ts | 181 ++++++++++- packages/contrail/src/public-client.ts | 291 +++++++++++++++++- packages/contrail/tests/built-client.mjs | 7 + packages/contrail/tests/connect.test.ts | 109 ++++++- packages/contrail/tests/public-client.test.ts | 232 +++++++++++++- .../contrail/tests/public-service-e2e.test.ts | 21 +- 7 files changed, 799 insertions(+), 44 deletions(-) diff --git a/.changeset/atproto-service-auth.md b/.changeset/atproto-service-auth.md index d677108..51ea2de 100644 --- a/.changeset/atproto-service-auth.md +++ b/.changeset/atproto-service-auth.md @@ -2,4 +2,4 @@ "@atmo-dev/contrail": minor --- -Add discoverable AT Protocol service authentication for personalized feeds and authoritative update notifications. +Add discoverable AT Protocol service authentication, unified authenticated clients for PDS and provider methods, and automatic authoritative update notifications after tracked record writes. diff --git a/packages/contrail/src/cli/commands/connect.ts b/packages/contrail/src/cli/commands/connect.ts index 7a6e0da..1397534 100644 --- a/packages/contrail/src/cli/commands/connect.ts +++ b/packages/contrail/src/cli/commands/connect.ts @@ -45,7 +45,10 @@ interface ConnectOptions { root: string; out: string; lock: string; + client: string; + clientTypes: string; generate?: boolean; + skipClient?: boolean; update?: boolean; } @@ -57,6 +60,7 @@ export interface ProviderLock { contractDigest: string; lexiconDigest: string; methods: string[]; + collections: string[]; serviceAuth: PublicServiceAuthContract | null; lexiconRoot: string; } @@ -158,28 +162,154 @@ async function exists(path: string): Promise { } } -/** Create a dependency-free Atcute config for ordinary consumer projects. - * Existing JavaScript or TypeScript configs remain entirely consumer-owned. */ +function formatStringArray( + values: readonly string[], + indentation: number, +): string { + if (values.length === 0) return "[]"; + const items = values + .map((value) => `${" ".repeat(indentation + 2)}${JSON.stringify(value)},`) + .join("\n"); + return `[\n${items}\n${" ".repeat(indentation)}]`; +} + +const GENERATED_LEXICON_CONFIG_HEADER = + "// Generated by `contrail connect`. Re-run the command to update; do not edit.\n"; + +/** Create a dependency-free Atcute config containing the connected service + * metadata. Unmarked JavaScript or TypeScript configs remain consumer-owned. */ export async function ensureConsumerLexiconConfig(options: { root: string; out: string; -}): Promise<{ path: string; created: boolean }> { + types?: string; + lock: ProviderLock; +}): Promise<{ path: string; created: boolean; updated: boolean }> { const root = resolve(options.root); + let path = join(root, "lex.config.js"); for (const name of LEXICON_CONFIG_NAMES) { - const path = join(root, name); - if (await exists(path)) return { path, created: false }; + const candidate = join(root, name); + if (await exists(candidate)) { + path = candidate; + break; + } } const lexiconRoot = resolveInsideRoot(root, options.out); const patternRoot = relative(root, lexiconRoot).replaceAll("\\", "/"); - const path = join(root, "lex.config.js"); - const source = `// Generated by \`contrail connect\`. Customize as needed.\nexport default {\n generate: {\n files: [${JSON.stringify(`${patternRoot}/**/*.json`)}],\n outdir: "src/lexicons/",\n },\n};\n`; + const typesIndex = resolveInsideRoot( + root, + options.types ?? "src/contrail/types/index.ts", + ); + const typesRoot = relative(root, dirname(typesIndex)).replaceAll("\\", "/"); + const serviceDid = options.lock.serviceAuth?.audience ?? null; + const scope = serviceDid ? `rpc?lxm=*&aud=${serviceDid}` : null; + const source = `${GENERATED_LEXICON_CONFIG_HEADER}export default {\n contrail: {\n endpoint: ${JSON.stringify(options.lock.endpoint)},\n serviceDid: ${JSON.stringify(serviceDid)},\n scope: ${JSON.stringify(scope)},\n collections: ${formatStringArray(options.lock.collections, 4)},\n },\n generate: {\n files: [${JSON.stringify(`${patternRoot}/**/*.json`)}],\n outdir: ${JSON.stringify(`${typesRoot}/`)},\n },\n};\n`; + + if (await exists(path)) { + const current = await readFile(path, "utf8"); + if ( + !current.startsWith(GENERATED_LEXICON_CONFIG_HEADER) || + current === source + ) { + return { path, created: false, updated: false }; + } + const stagedDirectory = await mkdtemp(join(dirname(path), ".contrail-lex-")); + const staged = join(stagedDirectory, basename(path)); + try { + await writeFile(staged, source); + await rename(staged, path); + } finally { + await rm(stagedDirectory, { recursive: true, force: true }); + } + return { path, created: false, updated: true }; + } + try { await writeFile(path, source, { flag: "wx" }); - return { path, created: true }; + return { path, created: true, updated: false }; } catch (error) { if ((error as NodeJS.ErrnoException).code === "EEXIST") { - return { path, created: false }; + return { path, created: false, updated: false }; + } + throw error; + } +} + +const GENERATED_CLIENT_HEADER = + "// Generated by `contrail connect`. Re-run the command to update; do not edit.\n"; + +/** Create the small provider-specific module used by application and OAuth + * setup code. Unmarked TypeScript or JavaScript modules remain consumer-owned. */ +export async function ensureConsumerClientModule(options: { + root: string; + file?: string; + types?: string; + lock: ProviderLock; +}): Promise<{ path: string; created: boolean; updated: boolean }> { + const root = resolve(options.root); + const requested = options.file ?? "src/contrail/index.ts"; + const defaultNames = ["src/contrail/index.ts", "src/contrail/index.js"]; + let selected = requested; + if (defaultNames.includes(requested)) { + for (const name of defaultNames) { + if (await exists(join(root, name))) { + selected = name; + break; + } + } + } + if (!/\.(?:ts|js)$/.test(selected)) { + throw new Error("Contrail client module must end in .ts or .js"); + } + + const path = resolveInsideRoot(root, selected); + const isTypeScript = selected.endsWith(".ts"); + let generatedImport = ""; + if (isTypeScript) { + const types = resolveInsideRoot( + root, + options.types ?? "src/contrail/types/index.ts", + ); + let specifier = relative(dirname(path), types).replaceAll("\\", "/"); + specifier = specifier.replace(/\.(?:ts|js)$/, ".js"); + if (!specifier.startsWith(".")) specifier = `./${specifier}`; + generatedImport = `import type {} from ${JSON.stringify(specifier)};\n`; + } + const serviceDid = options.lock.serviceAuth?.audience; + const scope = serviceDid ? `rpc?lxm=*&aud=${serviceDid}` : null; + const protectedMethods = + options.lock.serviceAuth?.methods.map(({ id }) => id) ?? []; + const serviceMethods = [ + ...new Set([...options.lock.methods, ...protectedMethods]), + ].sort(); + const notifyMethod = protectedMethods.find( + (method) => method === `${options.lock.namespace}.notifyOfUpdate`, + ); + const source = `${GENERATED_CLIENT_HEADER}import { createPublicServiceClient } from "@atmo-dev/contrail/client";\n${generatedImport}\nexport const contrail = createPublicServiceClient({\n endpoint: ${JSON.stringify(options.lock.endpoint)},${serviceDid ? `\n serviceDid: ${JSON.stringify(serviceDid)},\n scope: ${JSON.stringify(scope)},` : ""}\n serviceMethods: ${formatStringArray(serviceMethods, 2)},\n collections: ${formatStringArray(options.lock.collections, 2)},${notifyMethod ? `\n notifyMethod: ${JSON.stringify(notifyMethod)},` : ""}\n});\n`; + await mkdir(dirname(path), { recursive: true }); + + if (await exists(path)) { + const current = await readFile(path, "utf8"); + if (!current.startsWith(GENERATED_CLIENT_HEADER) || current === source) { + return { path, created: false, updated: false }; + } + const stagedDirectory = await mkdtemp(join(dirname(path), ".contrail-client-")); + const staged = join(stagedDirectory, basename(path)); + try { + await writeFile(staged, source); + await rename(staged, path); + } finally { + await rm(stagedDirectory, { recursive: true, force: true }); + } + return { path, created: false, updated: true }; + } + + try { + await writeFile(path, source, { flag: "wx" }); + return { path, created: true, updated: false }; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "EEXIST") { + return { path, created: false, updated: false }; } throw error; } @@ -295,6 +425,9 @@ export async function connectPublicService(options: { contractDigest: manifest.contract.digest, lexiconDigest: manifest.lexicons.digest, methods: [...manifest.methods].sort(), + collections: [ + ...new Set(manifest.collections.map(({ nsid }) => nsid)), + ].sort(), serviceAuth: manifest.serviceAuth ?? null, lexiconRoot: relative(projectRoot, providerRoot), }; @@ -346,19 +479,26 @@ export function registerConnect(cli: CAC): void { cli .command( "connect ", - "Discover a public Contrail, lock its API, pull Lexicons, and generate types", + "Discover a public Contrail, lock its API, and generate a typed client", ) .option("--root ", "Consumer project root", { default: process.cwd(), }) .option("--out ", "Provider-owned Lexicon storage relative to root", { - default: "lexicons/pulled", + default: "src/contrail/lexicons", }) .option("--lock ", "Provider lock file relative to root", { default: "contrail.lock.json", }) + .option("--client ", "Generated client module (.ts or .js)", { + default: "src/contrail/index.ts", + }) + .option("--client-types ", "Generated Lexicon index imported by a TypeScript client", { + default: "src/contrail/types/index.ts", + }) + .option("--skip-client", "Do not create a provider client module") .option("--update", "Replace an existing provider lock and owned Lexicons") - .option("--no-generate", "Pull and lock without running Atcute lex-cli") + .option("--no-generate", "Pull and lock without generating types or a client module") .action(async (endpoint: string, options: ConnectOptions) => { const result = await connectPublicService({ endpoint, @@ -374,11 +514,28 @@ export function registerConnect(cli: CAC): void { const config = await ensureConsumerLexiconConfig({ root: options.root, out: options.out, + types: options.clientTypes, + lock: result.lock, }); if (config.created) { console.log(`created ${relative(resolve(options.root), config.path)}`); + } else if (config.updated) { + console.log(`updated ${relative(resolve(options.root), config.path)}`); } generateLexiconTypesWithAtcute(resolve(options.root)); + if (!options.skipClient) { + const client = await ensureConsumerClientModule({ + root: options.root, + file: options.client, + types: options.clientTypes, + lock: result.lock, + }); + if (client.created) { + console.log(`created ${relative(resolve(options.root), client.path)}`); + } else if (client.updated) { + console.log(`updated ${relative(resolve(options.root), client.path)}`); + } + } } }); } diff --git a/packages/contrail/src/public-client.ts b/packages/contrail/src/public-client.ts index 8979cc3..af41414 100644 --- a/packages/contrail/src/public-client.ts +++ b/packages/contrail/src/public-client.ts @@ -22,15 +22,61 @@ interface CachedToken { export interface PublicServiceClientOptions { /** Canonical public Contrail HTTPS origin. */ endpoint: string; - /** Existing authenticated PDS client used to mint service tokens. Omit when - * the consumer only needs anonymous methods. */ - authenticatedPds?: Client; + /** Existing authenticated AT Protocol client used to mint service tokens. + * Omit when the consumer only needs anonymous methods. */ + authenticatedClient?: Client; /** Optional contract pin from `contrail.lock.json`. */ contractDigest?: string; + /** Optional receiving-service DID from `lex.config.js`. Supplying it also + * makes the required OAuth permission available as `client.scope`. */ + serviceDid?: Did; + /** Optional precomputed OAuth permission. Must match `serviceDid`. */ + scope?: `rpc?lxm=*&aud=${string}`; + /** Exact XRPC methods served by this provider. Supplying the verified list + * lets authenticated clients route all other methods to the user's PDS. */ + serviceMethods?: readonly Nsid[]; + /** Record collections whose successful PDS writes should notify Contrail. */ + collections?: readonly Nsid[]; + /** Protected notification procedure advertised by the provider. */ + notifyMethod?: Nsid; /** Browser, test, or instrumented fetch implementation. */ fetch?: typeof globalThis.fetch; } +export interface PublicServiceNotificationErrorContext { + method: Nsid; + uris: readonly string[]; +} + +export interface PublicServiceAuthenticatedOptions { + onNotificationError?: ( + error: unknown, + context: PublicServiceNotificationErrorContext, + ) => void; +} + +export type PublicServiceClient = Client & { + /** Canonical public Contrail origin. */ + readonly endpoint: string; + /** OAuth permission required by protected methods, or null when unconfigured. */ + readonly scope: `rpc?lxm=*&aud=${string}` | null; + /** Record collections whose successful writes trigger notification. */ + readonly collections: readonly Nsid[]; + /** Combine this provider with an authenticated PDS client. Provider methods + * route to Contrail; other methods route to the PDS; successful tracked + * record writes notify Contrail before returning their original response. */ + authenticated( + authenticatedClient: Client, + options?: PublicServiceAuthenticatedOptions, + ): PublicServiceClient; +}; + +export function publicServiceOAuthScope( + audience: Did, +): `rpc?lxm=*&aud=${string}` { + return `rpc?lxm=*&aud=${audience}`; +} + function xrpcMethod(pathname: string): Nsid | null { const path = pathname.startsWith("http") ? new URL(pathname).pathname @@ -111,7 +157,16 @@ export function publicServiceFetchHandler( `Contrail contract digest mismatch: expected ${options.contractDigest}, received ${value.contract.digest}`, ); } - return value.serviceAuth ?? null; + const serviceAuth = value.serviceAuth ?? null; + if ( + options.serviceDid && + serviceAuth?.audience !== options.serviceDid + ) { + throw new Error( + `Contrail service DID mismatch: expected ${options.serviceDid}, received ${serviceAuth?.audience ?? "none"}`, + ); + } + return serviceAuth; })(); return serviceAuthPromise; }; @@ -128,9 +183,9 @@ export function publicServiceFetchHandler( auth: PublicServiceAuthContract, force = false, ): Promise => { - if (!options.authenticatedPds) { + if (!options.authenticatedClient) { throw new Error( - `Contrail method ${method} requires an authenticated PDS client`, + `Contrail method ${method} requires an authenticated AT Protocol client`, ); } const cached = tokens.get(method); @@ -143,7 +198,7 @@ export function publicServiceFetchHandler( } const pending = (async () => { - const response = await options.authenticatedPds!.get( + const response = await options.authenticatedClient!.get( "com.atproto.server.getServiceAuth", { params: { @@ -171,7 +226,7 @@ export function publicServiceFetchHandler( return async (pathname, init) => { const method = xrpcMethod(pathname); - if (!method || !options.authenticatedPds) return base(pathname, init); + if (!method || !options.authenticatedClient) return base(pathname, init); // Once discovery has been loaded, avoid the initial challenge on subsequent // protected calls. Anonymous calls never wait for discovery. @@ -198,10 +253,226 @@ export function publicServiceFetchHandler( }; } +interface UntypedRequestOptions { + input?: unknown; + [key: string]: unknown; +} + +interface UntypedClientResponse { + ok: boolean; + status: number; + headers: Headers; + data: unknown; +} + +type UntypedMethod = ( + name: string, + options?: UntypedRequestOptions, +) => Promise; +type UntypedCall = ( + schema: unknown, + options?: UntypedRequestOptions, +) => Promise; + +const NOTIFIED_WRITE_METHODS = new Set([ + "com.atproto.repo.createRecord", + "com.atproto.repo.putRecord", + "com.atproto.repo.deleteRecord", +]); + +function objectValue(value: unknown): Record | null { + return value && typeof value === "object" && !Array.isArray(value) + ? (value as Record) + : null; +} + +function writtenRecordUri( + method: string, + request: UntypedRequestOptions | undefined, + response: UntypedClientResponse, + collections: ReadonlySet, +): string | null { + if (!response.ok || !NOTIFIED_WRITE_METHODS.has(method)) return null; + const input = objectValue(request?.input); + const collection = input?.collection; + if (typeof collection !== "string" || !collections.has(collection)) { + return null; + } + + if ( + method === "com.atproto.repo.createRecord" || + method === "com.atproto.repo.putRecord" + ) { + const uri = objectValue(response.data)?.uri; + return typeof uri === "string" ? uri : null; + } + + const repo = input?.repo; + const rkey = input?.rkey; + return typeof repo === "string" && + repo.startsWith("did:") && + typeof rkey === "string" + ? `at://${repo}/${collection}/${rkey}` + : null; +} + +function schemaNsid(schema: unknown): string | null { + const namespace = objectValue(schema); + const value = objectValue(namespace?.mainSchema) ?? namespace; + return typeof value?.nsid === "string" ? value.nsid : null; +} + +function createClient( + options: PublicServiceClientOptions, + authenticatedOptions: PublicServiceAuthenticatedOptions = {}, +): PublicServiceClient { + const endpoint = normalizePublicServiceEndpoint(options.endpoint); + const authenticatedClients = new WeakMap(); + const client = new Client({ + handler: publicServiceFetchHandler({ ...options, endpoint }), + }) as PublicServiceClient; + const expectedScope = options.serviceDid + ? publicServiceOAuthScope(options.serviceDid) + : null; + if (options.scope && options.scope !== expectedScope) { + throw new Error( + `Contrail OAuth scope mismatch: expected ${expectedScope ?? "none"}, received ${options.scope}`, + ); + } + const scope = options.scope ?? expectedScope; + const collections = Object.freeze([...(options.collections ?? [])]); + + Object.defineProperties(client, { + endpoint: { value: endpoint, enumerable: true }, + scope: { value: scope, enumerable: true }, + collections: { value: collections, enumerable: true }, + authenticated: { + enumerable: true, + value( + authenticatedClient: Client, + childOptions: PublicServiceAuthenticatedOptions = {}, + ) { + if ( + options.authenticatedClient === authenticatedClient && + !childOptions.onNotificationError + ) { + return client; + } + if (!childOptions.onNotificationError) { + const existing = authenticatedClients.get(authenticatedClient); + if (existing) return existing; + } + const created = createClient( + { ...options, endpoint, authenticatedClient }, + childOptions, + ); + if (!childOptions.onNotificationError) { + authenticatedClients.set(authenticatedClient, created); + } + return created; + }, + }, + }); + + if (options.authenticatedClient && options.serviceMethods) { + const serviceMethods = new Set(options.serviceMethods); + const trackedCollections = new Set(collections); + const serviceGet = client.get.bind(client) as unknown as UntypedMethod; + const servicePost = client.post.bind(client) as unknown as UntypedMethod; + const serviceCall = client.call.bind(client) as unknown as UntypedCall; + const pdsGet = options.authenticatedClient.get.bind( + options.authenticatedClient, + ) as unknown as UntypedMethod; + const pdsPost = options.authenticatedClient.post.bind( + options.authenticatedClient, + ) as unknown as UntypedMethod; + const pdsCall = options.authenticatedClient.call.bind( + options.authenticatedClient, + ) as unknown as UntypedCall; + + const reportNotificationError = ( + error: unknown, + method: string, + uris: readonly string[], + ) => { + try { + authenticatedOptions.onNotificationError?.(error, { + method: method as Nsid, + uris, + }); + } catch { + // A reporting callback must never turn a committed PDS write into a + // failed write response. + } + }; + + const notifyWrite = async (method: string, uri: string) => { + if (!options.notifyMethod) return; + try { + const notified = await servicePost(options.notifyMethod, { + input: { uris: [uri] }, + }); + if (!notified.ok) { + throw new Error( + `Contrail notification failed with status ${notified.status}`, + ); + } + const errors = objectValue(notified.data)?.errors; + if (Array.isArray(errors) && errors.length > 0) { + throw new Error( + `Contrail notification reported errors: ${errors.join("; ")}`, + ); + } + } catch (error) { + reportNotificationError(error, method, [uri]); + } + }; + + Object.defineProperties(client, { + get: { + value: ((name: string, request?: UntypedRequestOptions) => + serviceMethods.has(name) + ? serviceGet(name, request) + : pdsGet(name, request)) as Client["get"], + }, + post: { + value: (async (name: string, request?: UntypedRequestOptions) => { + if (serviceMethods.has(name)) return servicePost(name, request); + const response = await pdsPost(name, request); + const uri = writtenRecordUri( + name, + request, + response, + trackedCollections, + ); + if (uri) await notifyWrite(name, uri); + return response; + }) as Client["post"], + }, + call: { + value: ((schema: unknown, request?: UntypedRequestOptions) => { + const method = schemaNsid(schema); + return method && serviceMethods.has(method) + ? serviceCall(schema, request) + : pdsCall(schema, request); + }) as Client["call"], + }, + }); + } + + return client; +} + /** Create a typed Atcute client for anonymous and service-auth Contrail methods. * Generated Lexicon imports still supply the method-specific TypeScript API. */ +export function createPublicServiceClient( + options: PublicServiceClientOptions & { serviceDid: Did }, +): PublicServiceClient & { readonly scope: `rpc?lxm=*&aud=${string}` }; +export function createPublicServiceClient( + options: PublicServiceClientOptions, +): PublicServiceClient; export function createPublicServiceClient( options: PublicServiceClientOptions, -): Client { - return new Client({ handler: publicServiceFetchHandler(options) }); +): PublicServiceClient { + return createClient(options); } diff --git a/packages/contrail/tests/built-client.mjs b/packages/contrail/tests/built-client.mjs index 1bc59d4..479d61d 100644 --- a/packages/contrail/tests/built-client.mjs +++ b/packages/contrail/tests/built-client.mjs @@ -3,8 +3,15 @@ import { createPublicServiceClient } from "../dist/public-client.js"; const client = createPublicServiceClient({ endpoint: "https://api.example.com", + serviceDid: "did:web:api.example.com", + scope: "rpc?lxm=*&aud=did:web:api.example.com", + serviceMethods: ["com.example.listRecords"], + collections: ["community.example.event"], fetch: async () => Response.json({ records: [] }), }); +assert.equal(client.endpoint, "https://api.example.com"); +assert.equal(client.scope, "rpc?lxm=*&aud=did:web:api.example.com"); +assert.deepEqual(client.collections, ["community.example.event"]); const response = await client.get("com.example.listRecords"); assert.equal(response.ok, true); assert.deepEqual(response.data, { records: [] }); diff --git a/packages/contrail/tests/connect.test.ts b/packages/contrail/tests/connect.test.ts index ee73699..a3a9781 100644 --- a/packages/contrail/tests/connect.test.ts +++ b/packages/contrail/tests/connect.test.ts @@ -4,6 +4,7 @@ import { join } from "node:path"; import { afterEach, describe, expect, it, vi } from "vitest"; import { connectPublicService, + ensureConsumerClientModule, ensureConsumerLexiconConfig, type ProviderLock, } from "../src/cli/commands/connect"; @@ -49,8 +50,13 @@ function providerLock(): ProviderLock { contractDigest: `sha256:${"a".repeat(64)}`, lexiconDigest: `sha256:${"b".repeat(64)}`, methods: [method], - serviceAuth: null, - lexiconRoot: "lexicons/pulled/api.atmo.rsvp", + collections: ["community.lexicon.calendar.event"], + serviceAuth: { + type: "atproto-service-auth", + audience: "did:web:api.atmo.rsvp", + methods: [{ id: notifyMethod, type: "procedure" }], + }, + lexiconRoot: "src/contrail/lexicons/api.atmo.rsvp", }; } @@ -103,21 +109,44 @@ describe("contrail connect", () => { const root = await temporaryRoot(); const generated = await ensureConsumerLexiconConfig({ root, - out: "lexicons/providers", + out: "src/contrail/lexicons", + lock: providerLock(), }); expect(generated.created).toBe(true); expect(await readFile(generated.path, "utf8")).toContain( - 'files: ["lexicons/providers/**/*.json"]', + 'files: ["src/contrail/lexicons/**/*.json"]', + ); + expect(await readFile(generated.path, "utf8")).toContain( + 'outdir: "src/contrail/types/"', + ); + expect(await readFile(generated.path, "utf8")).toContain( + 'endpoint: "https://api.atmo.rsvp"', + ); + expect(await readFile(generated.path, "utf8")).toContain( + 'serviceDid: "did:web:api.atmo.rsvp"', ); expect(await readFile(generated.path, "utf8")).toContain( - 'outdir: "src/lexicons/"', + 'scope: "rpc?lxm=*&aud=did:web:api.atmo.rsvp"', + ); + expect(await readFile(generated.path, "utf8")).toContain( + '"community.lexicon.calendar.event",', + ); + const updated = await ensureConsumerLexiconConfig({ + root, + out: "src/contrail/lexicons", + lock: { ...providerLock(), endpoint: "https://next.example.com" }, + }); + expect(updated.updated).toBe(true); + expect(await readFile(updated.path, "utf8")).toContain( + 'endpoint: "https://next.example.com"', ); await writeFile(join(root, "lex.config.ts"), "export default { mine: true }"); await rm(generated.path); const existing = await ensureConsumerLexiconConfig({ root, - out: "lexicons/other", + out: "src/other", + lock: providerLock(), }); expect(existing.created).toBe(false); expect(existing.path).toBe(join(root, "lex.config.ts")); @@ -126,6 +155,74 @@ describe("contrail connect", () => { ); }); + it("generates TypeScript or JavaScript client modules without replacing one", async () => { + const root = await temporaryRoot(); + const fixture = await serviceFixture([ + methodLexicon, + sourceLexicon, + notifyLexicon, + ]); + fixture.manifest.serviceAuth = { + type: "atproto-service-auth", + audience: "did:web:api.atmo.rsvp", + methods: [{ id: notifyMethod, type: "procedure" }], + }; + fixture.manifest.contract.digest = await digestPublicContract( + contractFromManifest(fixture.manifest), + ); + const { lock } = await connectPublicService({ + endpoint, + root, + out: "lexicons/pulled", + lock: "contrail.lock.json", + fetcher: fixture.fetcher, + }); + + const generated = await ensureConsumerClientModule({ root, lock }); + expect(generated.created).toBe(true); + const source = await readFile(generated.path, "utf8"); + expect(source).toContain( + 'import type {} from "./types/index.js"', + ); + expect(source).toContain(`endpoint: ${JSON.stringify(endpoint)}`); + expect(source).toContain( + 'serviceDid: "did:web:api.atmo.rsvp"', + ); + expect(source).toContain( + 'scope: "rpc?lxm=*&aud=did:web:api.atmo.rsvp"', + ); + expect(source).toContain('"community.lexicon.calendar.event",'); + expect(source).toContain( + `notifyMethod: ${JSON.stringify(notifyMethod)}`, + ); + + const updated = await ensureConsumerClientModule({ + root, + lock: { ...lock, endpoint: "https://next.example.com" }, + }); + expect(updated.updated).toBe(true); + expect(await readFile(updated.path, "utf8")).toContain( + 'endpoint: "https://next.example.com"', + ); + + await writeFile(generated.path, "export const mine = true;\n"); + const existing = await ensureConsumerClientModule({ root, lock }); + expect(existing.created).toBe(false); + expect(await readFile(existing.path, "utf8")).toBe( + "export const mine = true;\n", + ); + + const javascript = await ensureConsumerClientModule({ + root, + file: "client/contrail.js", + lock, + }); + expect(javascript.created).toBe(true); + expect(await readFile(javascript.path, "utf8")).not.toContain( + "lexicons/index", + ); + }); + it("verifies and atomically locks a discovered service", async () => { const root = await temporaryRoot(); const { fetcher, manifest, values } = await serviceFixture(); diff --git a/packages/contrail/tests/public-client.test.ts b/packages/contrail/tests/public-client.test.ts index 97187c5..7cae200 100644 --- a/packages/contrail/tests/public-client.test.ts +++ b/packages/contrail/tests/public-client.test.ts @@ -9,6 +9,8 @@ import type { PublicServiceManifest } from "../src/public-service"; const endpoint = "https://api.example.com"; const method = "com.example.getFeed"; +const notifyMethod = "com.example.notifyOfUpdate"; +const collection = "community.example.event"; const digest = `sha256:${"a".repeat(64)}`; function token() { @@ -39,7 +41,7 @@ function manifest(): PublicServiceManifest { }; } -function authenticatedPds(jwt: string) { +function authenticatedClient(jwt: string) { const handler = vi.fn(async (pathname: string) => { const url = new URL(pathname, "https://pds.example.com"); expect(url.pathname).toBe("/xrpc/com.atproto.server.getServiceAuth"); @@ -51,6 +53,16 @@ function authenticatedPds(jwt: string) { } describe("public service client", () => { + it("rejects a configured OAuth scope for a different service DID", () => { + expect(() => + createPublicServiceClient({ + endpoint, + serviceDid: "did:web:api.example.com", + scope: "rpc?lxm=*&aud=did:web:other.example.com", + }), + ).toThrow("OAuth scope mismatch"); + }); + it("keeps anonymous requests anonymous", async () => { const fetcher = vi.fn(async () => Response.json({ records: [] })); const handler = publicServiceFetchHandler({ endpoint, fetch: fetcher }); @@ -65,7 +77,7 @@ describe("public service client", () => { it("discovers, mints, caches, and attaches method-bound tokens", async () => { const jwt = token(); - const pds = authenticatedPds(jwt); + const pds = authenticatedClient(jwt); const requests: Array<{ url: string; authorization: string | null }> = []; const fetcher = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => { const url = String(input); @@ -81,12 +93,20 @@ describe("public service client", () => { { status: 401, headers: { "www-authenticate": "Bearer" } }, ); }); - const client = createPublicServiceClient({ + const publicClient = createPublicServiceClient({ endpoint, - authenticatedPds: pds.client, contractDigest: digest, + serviceDid: "did:web:api.example.com", + scope: "rpc?lxm=*&aud=did:web:api.example.com", + serviceMethods: [method], fetch: fetcher, }); + expect(publicClient.endpoint).toBe(endpoint); + expect(publicClient.scope).toBe( + "rpc?lxm=*&aud=did:web:api.example.com", + ); + const client = publicClient.authenticated(pds.client); + expect(publicClient.authenticated(pds.client)).toBe(client); const first = await (client as any).get(method, { params: { actor: "did:plc:test", feed: "network" }, @@ -105,8 +125,208 @@ describe("public service client", () => { ).toEqual([null, `Bearer ${jwt}`, `Bearer ${jwt}`]); }); + it("routes PDS writes and service methods through one authenticated client", async () => { + const jwt = token(); + const discovered = manifest(); + discovered.serviceAuth = { + type: "atproto-service-auth", + audience: "did:web:api.example.com", + methods: [ + { id: method, type: "query" }, + { id: notifyMethod, type: "procedure" }, + ], + }; + let pdsWriteShouldFail = false; + const pdsHandler = vi.fn(async (pathname: string) => { + const url = new URL(pathname, "https://pds.example.com"); + if (url.pathname === "/xrpc/com.atproto.server.getServiceAuth") { + return Response.json({ token: jwt }); + } + if (url.pathname === "/xrpc/com.atproto.repo.getRecord") { + return Response.json({ + uri: `at://did:plc:test/${collection}/3test`, + cid: "bafyreicid", + value: { $type: collection, name: "Test event" }, + }); + } + if (url.pathname === "/xrpc/com.atproto.repo.deleteRecord") { + return Response.json({}); + } + if ( + url.pathname === "/xrpc/com.atproto.repo.createRecord" || + url.pathname === "/xrpc/com.atproto.repo.putRecord" + ) { + if (pdsWriteShouldFail) { + return Response.json({ error: "InvalidRecord" }, { status: 400 }); + } + return Response.json({ + uri: `at://did:plc:test/${collection}/3test`, + cid: "bafyreicid", + }); + } + return Response.json({ error: "MethodNotFound" }, { status: 404 }); + }); + const pds = new Client({ handler: pdsHandler }); + const serviceRequests: Array<{ url: string; body: string | null }> = []; + let notificationShouldFail = false; + const fetcher = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => { + const url = String(input); + if (url.endsWith("/.well-known/contrail")) { + return Response.json(discovered); + } + serviceRequests.push({ + url, + body: typeof init?.body === "string" ? init.body : null, + }); + const authorization = new Headers(init?.headers).get("authorization"); + if (authorization !== `Bearer ${jwt}`) { + return Response.json( + { error: "AuthenticationRequired" }, + { status: 401 }, + ); + } + return url.endsWith(`/xrpc/${notifyMethod}`) + ? Response.json({ + indexed: notificationShouldFail ? 0 : 1, + deleted: 0, + ...(notificationShouldFail ? { errors: ["retry later"] } : {}), + }) + : Response.json({ records: [] }); + }); + const notificationError = vi.fn(); + const client = createPublicServiceClient({ + endpoint, + serviceDid: "did:web:api.example.com", + serviceMethods: [method, notifyMethod], + collections: [collection], + notifyMethod, + fetch: fetcher, + }).authenticated(pds, { onNotificationError: notificationError }); + + const write = await client.post("com.atproto.repo.createRecord", { + input: { + repo: "did:plc:test", + collection, + record: { $type: collection, name: "Test event" }, + }, + }); + const feed = await (client as any).get(method, { + params: { actor: "did:plc:test", feed: "network" }, + }); + const record = await client.get("com.atproto.repo.getRecord", { + params: { + repo: "did:plc:test", + collection, + rkey: "3test", + }, + }); + + expect(write.ok).toBe(true); + expect(feed.ok).toBe(true); + expect(record.ok).toBe(true); + expect(notificationError).not.toHaveBeenCalled(); + + notificationShouldFail = true; + const updated = await client.post("com.atproto.repo.putRecord", { + input: { + repo: "did:plc:test", + collection, + rkey: "3test", + record: { $type: collection, name: "Updated event" }, + }, + }); + expect(updated.ok).toBe(true); + expect(notificationError).toHaveBeenCalledOnce(); + + notificationShouldFail = false; + const notificationsBeforeDelete = serviceRequests.filter(({ url }) => + url.endsWith(`/xrpc/${notifyMethod}`), + ).length; + const deleted = await client.post("com.atproto.repo.deleteRecord", { + input: { + repo: "did:plc:test", + collection, + rkey: "3test", + }, + }); + expect(deleted.ok).toBe(true); + expect( + serviceRequests.filter(({ url }) => + url.endsWith(`/xrpc/${notifyMethod}`), + ).length, + ).toBe(notificationsBeforeDelete + 1); + + pdsWriteShouldFail = true; + const notificationsBeforeFailedWrite = serviceRequests.filter(({ url }) => + url.endsWith(`/xrpc/${notifyMethod}`), + ).length; + const failed = await client.post("com.atproto.repo.createRecord", { + input: { + repo: "did:plc:test", + collection, + record: { $type: collection, name: "Invalid" }, + }, + }); + expect(failed.ok).toBe(false); + expect( + serviceRequests.filter(({ url }) => + url.endsWith(`/xrpc/${notifyMethod}`), + ).length, + ).toBe(notificationsBeforeFailedWrite); + pdsWriteShouldFail = false; + + const notificationsBeforeUntrackedWrite = serviceRequests.filter( + ({ url }) => url.endsWith(`/xrpc/${notifyMethod}`), + ).length; + const untracked = await client.post("com.atproto.repo.createRecord", { + input: { + repo: "did:plc:test", + collection: "app.bsky.feed.post", + record: { $type: "app.bsky.feed.post", text: "Not tracked" }, + }, + }); + expect(untracked.ok).toBe(true); + expect( + serviceRequests.filter(({ url }) => + url.endsWith(`/xrpc/${notifyMethod}`), + ).length, + ).toBe(notificationsBeforeUntrackedWrite); + expect( + serviceRequests.some( + ({ url, body }) => + url.endsWith(`/xrpc/${notifyMethod}`) && + body?.includes(`at://did:plc:test/${collection}/3test`), + ), + ).toBe(true); + expect( + pdsHandler.mock.calls.some(([pathname]) => + String(pathname).includes("com.atproto.repo.createRecord"), + ), + ).toBe(true); + }); + + it("refuses a discovered service-auth audience that differs from its lock", async () => { + const pds = authenticatedClient(token()); + const fetcher = vi.fn(async (input: RequestInfo | URL) => + String(input).endsWith("/.well-known/contrail") + ? Response.json(manifest()) + : new Response(null, { status: 401 }), + ); + const client = createPublicServiceClient({ + endpoint, + authenticatedClient: pds.client, + serviceDid: "did:web:other.example.com", + fetch: fetcher, + }); + + await expect( + (client as any).get(method), + ).rejects.toThrow("service DID mismatch"); + expect(pds.handler).not.toHaveBeenCalled(); + }); + it("refuses runtime discovery that differs from an optional lock pin", async () => { - const pds = authenticatedPds(token()); + const pds = authenticatedClient(token()); const fetcher = vi.fn(async (input: RequestInfo | URL) => String(input).endsWith("/.well-known/contrail") ? Response.json(manifest()) @@ -114,7 +334,7 @@ describe("public service client", () => { ); const handler = publicServiceFetchHandler({ endpoint, - authenticatedPds: pds.client, + authenticatedClient: pds.client, contractDigest: `sha256:${"b".repeat(64)}`, fetch: fetcher, }); diff --git a/packages/contrail/tests/public-service-e2e.test.ts b/packages/contrail/tests/public-service-e2e.test.ts index 881be2d..6d3c14c 100644 --- a/packages/contrail/tests/public-service-e2e.test.ts +++ b/packages/contrail/tests/public-service-e2e.test.ts @@ -4,6 +4,7 @@ import { join } from "node:path"; import { afterEach, describe, expect, it } from "vitest"; import { connectPublicService, + ensureConsumerClientModule, ensureConsumerLexiconConfig, } from "../src/cli/commands/connect"; import { generateLexiconTypesWithAtcute } from "../src/cli/atcute"; @@ -104,29 +105,31 @@ describe("public service consumer integration", () => { const fetcher: typeof fetch = (input, init) => app.fetch(new Request(input, init)); - await connectPublicService({ + const connection = await connectPublicService({ endpoint: "https://api.example.com", root: consumerRoot, - out: "lexicons/pulled", + out: "src/contrail/lexicons", lock: "contrail.lock.json", fetcher, }); const generatedConfig = await ensureConsumerLexiconConfig({ root: consumerRoot, - out: "lexicons/pulled", + out: "src/contrail/lexicons", + lock: connection.lock, }); expect(generatedConfig.created).toBe(true); generateLexiconTypesWithAtcute(consumerRoot); + const generatedClient = await ensureConsumerClientModule({ + root: consumerRoot, + lock: connection.lock, + }); + expect(generatedClient.created).toBe(true); mkdirSync(join(consumerRoot, "src"), { recursive: true }); writeFileSync( join(consumerRoot, "src", "consumer.ts"), - `import { Client, simpleFetchHandler } from "@atcute/client"; -import "./lexicons/index.js"; -const client = new Client({ - handler: simpleFetchHandler({ service: "https://api.example.com" }), -}); -const response = await client.get("com.example.event.listRecords", { + `import { contrail } from "./contrail/index.js"; +const response = await contrail.get("com.example.event.listRecords", { params: { name: "Typed event", limit: 1 }, }); if (response.ok) { -- 2.51.2