diff --git a/.changeset/bright-remotes-verify.md b/.changeset/bright-remotes-verify.md new file mode 100644 index 00000000..5eaf8aff --- /dev/null +++ b/.changeset/bright-remotes-verify.md @@ -0,0 +1,11 @@ +--- +"@enbox/dwn-sdk-js": patch +"@enbox/agent": patch +"@enbox/api": patch +--- + +Authenticate protocol configurations used for remote encryption-policy resolution and record artifacts returned through app-facing remote query, read, and initial subscription snapshot calls, bind record results to the original request filter, and verify inline or streamed record bytes against their signed CID and size. Remote protocol definitions used for encryption policy must now be signed directly by the target DID. Anonymous subscriptions now use the current transport request shape, and lazy read-only records reject data from a different record version. + +These checks authenticate returned artifacts; they do not prove result completeness or freshness because DWN query replies do not yet carry a tenant-authenticated state commitment. `RecordsCount` replies carry no signed artifacts, so their aggregate values remain assertions by the remote DWN. Initial `RecordsSubscribe` snapshots are verified, but subsequent live events remain outside this response-verification boundary. + +Streamed reads are authenticated at successful end-of-stream, so callers can observe chunks before the final CID check completes. Integrity-sensitive consumers that cannot tolerate an unauthenticated prefix must buffer the stream through successful completion before using its bytes. diff --git a/.changeset/remove-social-protocols.md b/.changeset/remove-social-protocols.md index b98ff6b0..4a3afb65 100644 --- a/.changeset/remove-social-protocols.md +++ b/.changeset/remove-social-protocols.md @@ -2,4 +2,6 @@ "@enbox/protocols": patch --- -Remove `SocialGraphProtocol`, `StatusProtocol`, and `ListsProtocol` from the shipped catalog together with their definitions, data types, codecs, and JSON Schemas. Profile no longer composes with Social Graph and no longer exposes `PrivateNoteData` or the friend-gated `privateNote` path. No compatibility aliases are retained. +Remove `SocialGraphProtocol`, `StatusProtocol`, and `ListsProtocol` from the shipped catalog together with their definitions, data types, codecs, and JSON Schemas. The removed Social Graph definition allowed anyone to create a `friend` role record, while current role invocation accepts a matching role recipient without independently binding the role to a trusted issuer. A caller could therefore self-assign friend authority and reach friend-gated data in protocols composed with that role. Profile no longer composes with Social Graph and no longer exposes `PrivateNoteData` or the friend-gated `privateNote` path. + +This removal does not change engine role verification, uninstall definitions, revoke existing role records, or migrate stored data. Applications that installed these protocols should stop treating their role-gated records as confidential, install a corrected protocol under a new URI, migrate and re-encrypt sensitive records, and retire the old records. No replacement or compatibility aliases are provided. diff --git a/packages/agent/src/anonymous-dwn-api.ts b/packages/agent/src/anonymous-dwn-api.ts index bb8bcec6..9b702581 100644 --- a/packages/agent/src/anonymous-dwn-api.ts +++ b/packages/agent/src/anonymous-dwn-api.ts @@ -1,7 +1,9 @@ -import type { DidUrlDereferencer } from '@enbox/dids'; +import type { DidResolver, DidUrlDereferencer } from '@enbox/dids'; +import type { DwnRpcRequest, EnboxRpc } from '@enbox/dwn-clients'; import type { DateSort, + GenericMessage, Pagination, ProtocolsQueryFilter, ProtocolsQueryReply, @@ -13,18 +15,18 @@ import type { RecordsSubscribeReply, SubscriptionListener, } from '@enbox/dwn-sdk-js'; -import type { DwnRpcRequest, EnboxRpc } from '@enbox/dwn-clients'; import { ProtocolsQuery, RecordsCount, RecordsQuery, RecordsRead, RecordsSubscribe } from '@enbox/dwn-sdk-js'; -import { getDwnServiceEndpointUrls } from './utils.js'; +import { verifyRemoteDwnResponse } from './remote-dwn-response.js'; +import { getDwnServiceEndpointUrls, resolveDwnSubscriptionUrl } from './utils.js'; /** * Parameters for constructing an {@link AnonymousDwnApi}. */ export type AnonymousDwnApiParams = { - /** A DID URL dereferencer for resolving target DID service endpoints. */ - didResolver: DidUrlDereferencer; + /** Resolver used for target DWN discovery and returned-message authentication. */ + didResolver: DidResolver & DidUrlDereferencer; /** An RPC client for sending messages to remote DWNs. */ rpcClient: EnboxRpc; }; @@ -98,7 +100,7 @@ export type AnonymousProtocolsQueryParams = { * ``` */ export class AnonymousDwnApi { - private readonly _didResolver: DidUrlDereferencer; + private readonly _didResolver: DidResolver & DidUrlDereferencer; private readonly _rpcClient: EnboxRpc; constructor({ didResolver, rpcClient }: AnonymousDwnApiParams) { @@ -219,40 +221,40 @@ export class AnonymousDwnApi { */ private async sendRequest( target: string, - message: unknown, + message: GenericMessage, data?: Blob, subscriptionHandler?: SubscriptionListener, ): Promise { const dwnEndpointUrls = await getDwnServiceEndpointUrls(target, this._didResolver); const errorMessages: { url: string; message: string }[] = []; - for (let dwnUrl of dwnEndpointUrls) { + for (const endpointUrl of dwnEndpointUrls) { try { - // For subscriptions, upgrade to WebSocket transport if available. - if (subscriptionHandler !== undefined) { - const serverInfo = await this._rpcClient.getServerInfo(dwnUrl); - if (!serverInfo.webSocketSupport) { - errorMessages.push({ url: dwnUrl, message: 'WebSocket support is not enabled on the server.' }); - continue; - } - - const parsedUrl = new URL(dwnUrl); - parsedUrl.protocol = parsedUrl.protocol === 'http:' ? 'ws:' : 'wss:'; - dwnUrl = parsedUrl.toString(); - } + const dwnUrl = subscriptionHandler === undefined + ? endpointUrl + : await resolveDwnSubscriptionUrl(endpointUrl, this._rpcClient); const reply = await this._rpcClient.sendDwnRequest({ dwnUrl, - targetDid : target, + targetDid : target, message, data, - subscriptionHandler : subscriptionHandler, + subscription : subscriptionHandler === undefined ? undefined : { + handler: subscriptionHandler, + }, } as DwnRpcRequest); + await verifyRemoteDwnResponse({ + didResolver : this._didResolver, + message, + reply, + targetDid : target, + }); + return reply as TReply; } catch (error: unknown) { errorMessages.push({ - url : dwnUrl, + url : endpointUrl, message : (error instanceof Error) ? error.message : 'Unknown error', }); } diff --git a/packages/agent/src/dwn-api.ts b/packages/agent/src/dwn-api.ts index 103aaff3..ec5e126d 100644 --- a/packages/agent/src/dwn-api.ts +++ b/packages/agent/src/dwn-api.ts @@ -77,9 +77,10 @@ import type { import { DwnDiscoveryFile } from './dwn-discovery-file.js'; import { PermissionGrantNotFoundError } from './permissions-api.js'; +import { verifyRemoteDwnResponse } from './remote-dwn-response.js'; import { DEFAULT_LOCAL_DWN_STRATEGY, LocalDwnDiscovery } from './local-dwn.js'; import { DwnInterface, dwnMessageConstructors } from './types/dwn.js'; -import { getDwnServiceEndpointUrls, isRecordsWrite } from './utils.js'; +import { getDwnServiceEndpointUrls, isRecordsWrite, resolveDwnSubscriptionUrl } from './utils.js'; // Re-export DWN type guards from the agent API surface. export { isDwnMessage, isDwnRequest, isMessagesPermissionScope, isRecordPermissionScope, isRecordsType } from './dwn-type-guards.js'; @@ -689,6 +690,8 @@ export class AgentDwnApi { message, data : dataStream, subscriptionHandler, + // The configured local DWN already verifies artifacts on admission; this check is for untrusted remote replies. + verifyResponse : false, resubscribeFactory : subscriptionHandler === undefined ? undefined : this.createResubscribeFactory(request, 'local'), @@ -917,6 +920,8 @@ export class AgentDwnApi { dwnEndpointUrls : [this._localDwnEndpoint!], message : message as DwnMessage[DwnInterface], data : options?.dataStream, + // The configured local DWN already verifies artifacts on admission; this check is for untrusted remote replies. + verifyResponse : false, }); } @@ -1006,12 +1011,13 @@ export class AgentDwnApi { // Send the RPC request to the target DID's DWN service endpoint using the Agent's RPC client. const reply = await this.sendDwnRpcRequest({ - targetDid: request.target, + targetDid : request.target, dwnEndpointUrls, message, data, subscriptionHandler, resubscribeFactory, + verifyResponse : true, }); this.invalidateAcceptedProtocolDefinition(request.target, message, reply.status.code); @@ -1038,36 +1044,8 @@ export class AgentDwnApi { }; } - /** - * Resolves the URL to send a subscription request to: upgrades `dwnUrl` to a - * WebSocket transport when the server supports it (unsecured `ws` for `http`, - * secured `wss` for `https`), or records an error and returns `undefined` to - * signal that this endpoint should be skipped. - */ - private async resolveSubscriptionDwnUrl( - dwnUrl: string, - errorMessages: { url: string, message: string }[], - ): Promise { - // we get the server info to check if the server supports WebSocket for subscription requests - const serverInfo = await this.agent.rpc.getServerInfo(dwnUrl); - if (!serverInfo.webSocketSupport) { - // If the server does not support WebSocket, add an error message and continue to the next URL. - errorMessages.push({ - url : dwnUrl, - message : 'WebSocket support is not enabled on the server.' - }); - return undefined; - } - - // If the server supports WebSocket, replace the subscription URL with a socket transport. - // For `http` we use the unsecured `ws` protocol, and for `https` we use the secured `wss` protocol. - const parsedUrl = new URL(dwnUrl); - parsedUrl.protocol = parsedUrl.protocol === 'http:' ? 'ws:' : 'wss:'; - return parsedUrl.toString(); - } - private async sendDwnRpcRequest({ - targetDid, dwnEndpointUrls, message, data, subscriptionHandler, resubscribeFactory + targetDid, dwnEndpointUrls, message, data, subscriptionHandler, resubscribeFactory, verifyResponse }: { targetDid: string; dwnEndpointUrls: string[]; @@ -1075,6 +1053,7 @@ export class AgentDwnApi { data?: DwnRpcData; subscriptionHandler?: MessageHandler[T]; resubscribeFactory?: ResubscribeFactory; + verifyResponse: boolean; } ): Promise { const errorMessages: { url: string, message: string }[] = []; @@ -1084,15 +1063,11 @@ export class AgentDwnApi { } // Try sending to author's publicly addressable DWNs until the first request succeeds. - for (let dwnUrl of dwnEndpointUrls) { + for (const endpointUrl of dwnEndpointUrls) { try { - if (subscriptionHandler !== undefined) { - const subscriptionDwnUrl = await this.resolveSubscriptionDwnUrl(dwnUrl, errorMessages); - if (subscriptionDwnUrl === undefined) { - continue; - } - dwnUrl = subscriptionDwnUrl; - } + const dwnUrl = subscriptionHandler === undefined + ? endpointUrl + : await resolveDwnSubscriptionUrl(endpointUrl, this.agent.rpc); const dwnReply = await this.agent.rpc.sendDwnRequest({ dwnUrl, @@ -1105,10 +1080,19 @@ export class AgentDwnApi { } : undefined, }); + if (verifyResponse) { + await verifyRemoteDwnResponse({ + didResolver : this.agent.did, + message, + reply : dwnReply, + targetDid, + }); + } + return dwnReply; } catch (error: any) { errorMessages.push({ - url : dwnUrl, + url : endpointUrl, message : (error instanceof Error) ? error.message : 'Unknown error', }); } @@ -2562,6 +2546,8 @@ export class AgentDwnApi { targetDid : author, dwnEndpointUrls : [this._localDwnEndpoint!], message : messagesRead.message, + // The configured local DWN already verifies artifacts on admission; this check is for untrusted remote replies. + verifyResponse : false, }); } diff --git a/packages/agent/src/dwn-protocol-cache.ts b/packages/agent/src/dwn-protocol-cache.ts index 506f54e8..a385a703 100644 --- a/packages/agent/src/dwn-protocol-cache.ts +++ b/packages/agent/src/dwn-protocol-cache.ts @@ -40,6 +40,7 @@ type SendDwnRpcRequestFn = (params: { message: DwnMessage[T]; data?: Blob; subscriptionHandler?: MessageHandler[T]; + verifyResponse: boolean; }) => Promise; /** Minimal DWN interface needed for local `processMessage` calls. */ @@ -107,7 +108,8 @@ export async function getProtocolDefinition( * Fetches a protocol definition from a **remote** DWN. * * Uses an unsigned `ProtocolsQuery` since public protocols can be queried - * anonymously. + * anonymously. The returned configuration must carry a valid signature made + * directly by the target DID before its definition can enter the cache. * * @param targetDid - The remote DWN owner * @param protocolUri - The protocol URI to look up @@ -138,6 +140,7 @@ export async function fetchRemoteProtocolDefinition( targetDid, dwnEndpointUrls : await getDwnEndpointUrls(targetDid), message : protocolsQuery.message, + verifyResponse : true, }) as ProtocolsQueryReply; if (reply.status.code !== 200 || !reply.entries?.length) { diff --git a/packages/agent/src/remote-dwn-response.ts b/packages/agent/src/remote-dwn-response.ts new file mode 100644 index 00000000..e80c8397 --- /dev/null +++ b/packages/agent/src/remote-dwn-response.ts @@ -0,0 +1,468 @@ +import type { DidResolver } from '@enbox/dids'; +import type { + GenericMessage, + GenericMessageReply, + ProtocolsQueryMessage, + ProtocolsQueryReply, + RecordsDeleteMessage, + RecordsFilter, + RecordsQueryMessage, + RecordsQueryReply, + RecordsQueryReplyEntry, + RecordsReadMessage, + RecordsReadReply, + RecordsSubscribeMessage, + RecordsSubscribeReply, + RecordsWriteMessage, +} from '@enbox/dwn-sdk-js'; + +import { + authenticate, + Cid, + DataStream, + DateSort, + DwnInterfaceName, + DwnMethodName, + Encoder, + FilterUtility, + Message, + ProtocolsConfigure, + Records, + RecordsDelete, + RecordsWrite, +} from '@enbox/dwn-sdk-js'; + +/** + * Verifies signed artifacts carried by a remote DWN response. + * + * This establishes artifact authorship, signed-field integrity, request-filter + * membership, and record-data integrity. It cannot prove that a remote retained + * every matching message or that a participant-authored record was admitted by + * the target tenant; those guarantees require a tenant-authenticated feed head + * or admission receipt. + */ +export async function verifyRemoteDwnResponse({ + didResolver, + message, + reply, + targetDid, +}: { + didResolver: DidResolver; + message: GenericMessage; + reply: GenericMessageReply; + targetDid: string; +}): Promise { + const { interface: messageInterface, method } = message.descriptor; + + if (messageInterface === DwnInterfaceName.Protocols && method === DwnMethodName.Query) { + await verifyProtocolsQueryReply({ + didResolver, + message : message as ProtocolsQueryMessage, + reply : reply as ProtocolsQueryReply, + targetDid, + }); + } else if (messageInterface === DwnInterfaceName.Records && method === DwnMethodName.Query) { + await verifyRecordsCollectionReply({ + didResolver, + message : message as RecordsQueryMessage, + reply : reply as RecordsQueryReply, + }); + } else if (messageInterface === DwnInterfaceName.Records && method === DwnMethodName.Subscribe) { + await verifyRecordsSubscribeReply({ + didResolver, + message : message as RecordsSubscribeMessage, + reply : reply as RecordsSubscribeReply, + }); + } else if (messageInterface === DwnInterfaceName.Records && method === DwnMethodName.Read) { + await verifyRecordsReadReply({ + didResolver, + message : message as RecordsReadMessage, + reply : reply as RecordsReadReply, + }); + } + +} + +async function verifyProtocolsQueryReply({ + didResolver, + message, + reply, + targetDid, +}: { + didResolver: DidResolver; + message: ProtocolsQueryMessage; + reply: ProtocolsQueryReply; + targetDid: string; +}): Promise { + const entries = reply.entries ?? []; + if (reply.status.code !== 200 && entries.length > 0) { + throw verificationError(`ProtocolsQuery response used status ${reply.status.code} with entries`); + } + + const requestedProtocol = message.descriptor.filter?.protocol; + if (requestedProtocol !== undefined && entries.length > 1) { + throw verificationError(`remote returned multiple configurations for protocol '${requestedProtocol}'`); + } + + for (const entry of entries) { + const protocol = await ProtocolsConfigure.parse(entry); + await authenticate(entry.authorization, didResolver); + + if (protocol.signer !== targetDid) { + throw verificationError( + `protocol '${entry.descriptor.definition.protocol}' was signed by '${protocol.signer ?? 'an unknown DID'}' instead of '${targetDid}'` + ); + } + if (requestedProtocol !== undefined && entry.descriptor.definition.protocol !== requestedProtocol) { + throw verificationError( + `remote returned protocol '${entry.descriptor.definition.protocol}' while '${requestedProtocol}' was requested` + ); + } + if (message.authorization === undefined && entry.descriptor.definition.published !== true) { + throw verificationError(`anonymous query returned unpublished protocol '${entry.descriptor.definition.protocol}'`); + } + } +} + +async function verifyRecordsCollectionReply({ + didResolver, + message, + reply, +}: { + didResolver: DidResolver; + message: RecordsQueryMessage | RecordsSubscribeMessage; + reply: RecordsQueryReply | RecordsSubscribeReply; +}): Promise { + const entries = reply.entries ?? []; + if (reply.status.code !== 200 && entries.length > 0) { + throw verificationError(`Records${message.descriptor.method} response used status ${reply.status.code} with entries`); + } + + for (const entry of entries) { + await verifyRecordsWriteEntry({ + didResolver, + entry, + filter : message.descriptor.filter, + dateSort : message.descriptor.dateSort, + publishedOnly : message.authorization === undefined, + }); + } +} + +async function verifyRecordsSubscribeReply({ + didResolver, + message, + reply, +}: { + didResolver: DidResolver; + message: RecordsSubscribeMessage; + reply: RecordsSubscribeReply; +}): Promise { + try { + await verifyRecordsCollectionReply({ didResolver, message, reply }); + } catch (error) { + if (reply.subscription !== undefined) { + await Promise.allSettled([reply.subscription.close()]); + } + throw error; + } +} + +async function verifyRecordsReadReply({ + didResolver, + message, + reply, +}: { + didResolver: DidResolver; + message: RecordsReadMessage; + reply: RecordsReadReply; +}): Promise { + const entry = reply.entry; + if (entry === undefined) { + if (reply.status.code === 200) { + throw verificationError('successful RecordsRead response did not contain an entry'); + } + return; + } + + if (entry.recordsWrite !== undefined && entry.recordsDelete === undefined) { + await verifyRecordsWriteReadEntry({ + didResolver, + entry, + message, + recordsWriteMessage: entry.recordsWrite, + reply, + }); + return; + } + + if (entry.recordsDelete !== undefined && entry.recordsWrite === undefined) { + await verifyRecordsDeleteReadEntry({ + didResolver, + entry, + message, + recordsDeleteMessage: entry.recordsDelete, + reply, + }); + return; + } + + throw verificationError('RecordsRead entry must contain exactly one RecordsWrite or RecordsDelete'); +} + +async function verifyRecordsWriteReadEntry({ + didResolver, + entry, + message, + recordsWriteMessage, + reply, +}: { + didResolver: DidResolver; + entry: NonNullable; + message: RecordsReadMessage; + recordsWriteMessage: RecordsWriteMessage; + reply: RecordsReadReply; +}): Promise { + const recordsWriteEntry: RecordsQueryReplyEntry = { + ...recordsWriteMessage, + ...(entry.initialWrite === undefined ? {} : { initialWrite: entry.initialWrite }), + }; + const recordsWrite = await verifyRecordsWriteEntry({ + allowMissingInitialWrite : reply.status.code === 410, + dateSort : message.descriptor.dateSort, + didResolver, + entry : recordsWriteEntry, + filter : message.descriptor.filter, + }); + + if (reply.status.code === 410) { + if (entry.data !== undefined) { + throw verificationError(`unavailable RecordsRead response for '${recordsWrite.message.recordId}' contained record data`); + } + return; + } + if (reply.status.code !== 200) { + throw verificationError( + `RecordsRead response for '${recordsWrite.message.recordId}' used status ${reply.status.code} with a RecordsWrite entry` + ); + } + if (entry.data === undefined) { + throw verificationError(`RecordsRead response for '${recordsWrite.message.recordId}' did not contain record data`); + } + + entry.data = DataStream.fromAsyncIterable(verifyDataStream(entry.data, recordsWrite.message)); +} + +async function verifyRecordsDeleteReadEntry({ + didResolver, + entry, + message, + recordsDeleteMessage, + reply, +}: { + didResolver: DidResolver; + entry: NonNullable; + message: RecordsReadMessage; + recordsDeleteMessage: RecordsDeleteMessage; + reply: RecordsReadReply; +}): Promise { + if (entry.initialWrite === undefined) { + throw verificationError('RecordsRead response containing a RecordsDelete did not include its initial RecordsWrite'); + } + if (entry.data !== undefined) { + throw verificationError('RecordsRead response containing a RecordsDelete also contained record data'); + } + if (reply.status.code !== 404) { + throw verificationError(`RecordsRead response used status ${reply.status.code} with a RecordsDelete entry`); + } + + const recordsDelete = await verifyRecordsDelete(recordsDeleteMessage, didResolver); + const initialWrite = await verifyInitialWrite(entry.initialWrite, didResolver); + if (recordsDelete.message.descriptor.recordId !== initialWrite.message.recordId) { + throw verificationError( + `RecordsDelete for '${recordsDelete.message.descriptor.recordId}' was paired with initial write '${initialWrite.message.recordId}'` + ); + } + + assertDeleteFilterVerifiable(message); + const indexes = recordsDelete.constructIndexes(initialWrite.message, initialWrite.message); + if (!FilterUtility.matchFilter(indexes, Records.convertFilter(message.descriptor.filter))) { + throw verificationError(`RecordsDelete '${recordsDelete.message.descriptor.recordId}' does not match the RecordsRead filter`); + } +} + +async function verifyRecordsWriteEntry({ + didResolver, + entry, + filter, + dateSort, + publishedOnly = false, + allowMissingInitialWrite = false, +}: { + didResolver: DidResolver; + entry: RecordsQueryReplyEntry; + filter: RecordsFilter; + dateSort?: RecordsQueryMessage['descriptor']['dateSort']; + publishedOnly?: boolean; + allowMissingInitialWrite?: boolean; +}): Promise { + const recordsWrite = await verifyRecordsWrite(entry, didResolver); + await verifyInitialWriteAttachment({ allowMissingInitialWrite, didResolver, entry, recordsWrite }); + + const indexes = await recordsWrite.constructIndexes(true); + if (!FilterUtility.matchFilter(indexes, Records.convertFilter(filter, dateSort))) { + throw verificationError(`RecordsWrite '${entry.recordId}' does not match the request filter`); + } + if (publishedOnly && recordsWrite.message.descriptor.published !== true) { + throw verificationError(`anonymous query returned unpublished RecordsWrite '${entry.recordId}'`); + } + + if (entry.encodedData !== undefined) { + await verifyDataBytes(Encoder.base64UrlToBytes(entry.encodedData), recordsWrite.message); + } + + return recordsWrite; +} + +async function verifyInitialWriteAttachment({ + allowMissingInitialWrite, + didResolver, + entry, + recordsWrite, +}: { + allowMissingInitialWrite: boolean; + didResolver: DidResolver; + entry: RecordsQueryReplyEntry; + recordsWrite: RecordsWrite; +}): Promise { + if (await recordsWrite.isInitialWrite()) { + if (entry.initialWrite !== undefined) { + throw verificationError(`initial RecordsWrite '${entry.recordId}' contained a redundant initialWrite attachment`); + } + return; + } + + if (entry.initialWrite === undefined) { + if (allowMissingInitialWrite) { + return; + } + throw verificationError(`updated RecordsWrite '${entry.recordId}' did not include its initial write`); + } + + const initialWrite = await verifyInitialWrite(entry.initialWrite, didResolver); + if (initialWrite.message.recordId !== recordsWrite.message.recordId || + initialWrite.message.contextId !== recordsWrite.message.contextId) { + throw verificationError(`updated RecordsWrite '${entry.recordId}' does not match its attached initial write`); + } + RecordsWrite.verifyEqualityOfImmutableProperties(initialWrite.message, recordsWrite.message); +} + +async function verifyRecordsWrite(message: RecordsWriteMessage, didResolver: DidResolver): Promise { + const recordsWrite = await RecordsWrite.parse(message); + await authenticate(recordsWrite.message.authorization, didResolver, recordsWrite.message.attestation); + return recordsWrite; +} + +async function verifyInitialWrite(message: RecordsWriteMessage, didResolver: DidResolver): Promise { + const initialWrite = await verifyRecordsWrite(message, didResolver); + if (!await initialWrite.isInitialWrite()) { + throw verificationError(`attached initial write for '${message.recordId}' is not an initial RecordsWrite`); + } + return initialWrite; +} + +async function verifyRecordsDelete(message: RecordsDeleteMessage, didResolver: DidResolver): Promise { + Message.validateJsonSchema(message); + const recordsDelete = await RecordsDelete.parse(message); + await authenticate(recordsDelete.message.authorization, didResolver); + return recordsDelete; +} + +async function verifyDataBytes(data: Uint8Array, recordsWrite: RecordsWriteMessage): Promise { + const dataCid = await Cid.computeDagPbCidFromBytes(data); + RecordsWrite.validateDataIntegrity( + recordsWrite.descriptor.dataCid, + recordsWrite.descriptor.dataSize, + dataCid, + data.byteLength, + ); +} + +/** Verifies streamed bytes as they are consumed, before allowing a successful end-of-stream. */ +async function* verifyDataStream( + source: ReadableStream, + recordsWrite: RecordsWriteMessage, +): AsyncGenerator { + const reader = source.getReader(); + const hashingStream = new TransformStream(); + const hashWriter = hashingStream.writable.getWriter(); + const dataCidPromise = Cid.computeDagPbCidFromStream(hashingStream.readable); + let dataSize = 0; + let completed = false; + + try { + while (true) { + const { done, value } = await reader.read(); + if (done) { + break; + } + + dataSize += value.byteLength; + if (dataSize > recordsWrite.descriptor.dataSize) { + RecordsWrite.validateDataIntegrity( + recordsWrite.descriptor.dataCid, + recordsWrite.descriptor.dataSize, + recordsWrite.descriptor.dataCid, + dataSize, + ); + } + + await hashWriter.write(value); + yield value; + } + + await hashWriter.close(); + const dataCid = await dataCidPromise; + RecordsWrite.validateDataIntegrity( + recordsWrite.descriptor.dataCid, + recordsWrite.descriptor.dataSize, + dataCid, + dataSize, + ); + completed = true; + } finally { + if (!completed) { + await Promise.allSettled([reader.cancel(), hashWriter.abort()]); + void dataCidPromise.catch((): void => {}); + } + reader.releaseLock(); + hashWriter.releaseLock(); + } +} + +/** Rejects tombstones that cannot be bound to the request without the deleted record's last write. */ +function assertDeleteFilterVerifiable(message: RecordsReadMessage): void { + if (message.descriptor.dateSort === DateSort.PublishedAscending || + message.descriptor.dateSort === DateSort.PublishedDescending) { + throw verificationError('RecordsDelete cannot be authenticated against a published date sort without its last RecordsWrite'); + } + + const unverifiableProperties: (keyof RecordsFilter)[] = [ + 'attester', + 'dataCid', + 'dataFormat', + 'dataSize', + 'datePublished', + 'published', + 'tags', + ]; + const property = unverifiableProperties.find((name) => message.descriptor.filter[name] !== undefined); + + if (property !== undefined) { + throw verificationError('RecordsDelete cannot be authenticated against a filter that depends on its last RecordsWrite'); + } +} + +function verificationError(detail: string): Error { + return new Error(`Remote DWN response verification failed: ${detail}.`); +} diff --git a/packages/agent/src/utils.ts b/packages/agent/src/utils.ts index 389a38fc..22376d25 100644 --- a/packages/agent/src/utils.ts +++ b/packages/agent/src/utils.ts @@ -1,4 +1,5 @@ import type { DidUrlDereferencer } from '@enbox/dids'; +import type { EnboxRpc } from '@enbox/dwn-clients'; import type { PaginationCursor, RecordsDeleteMessage, RecordsWrite, RecordsWriteMessage } from '@enbox/dwn-sdk-js'; import { utils as didUtils } from '@enbox/dids'; @@ -35,6 +36,18 @@ export async function getDwnServiceEndpointUrls(didUri: string, dereferencer: Di return []; } +/** Resolves the WebSocket transport advertised for a DWN HTTP endpoint. */ +export async function resolveDwnSubscriptionUrl(dwnUrl: string, rpcClient: EnboxRpc): Promise { + const serverInfo = await rpcClient.getServerInfo(dwnUrl); + if (!serverInfo.webSocketSupport) { + throw new Error('WebSocket support is not enabled on the server.'); + } + + const parsedUrl = new URL(dwnUrl); + parsedUrl.protocol = parsedUrl.protocol === 'http:' ? 'ws:' : 'wss:'; + return parsedUrl.toString(); +} + function normalizeDwnServiceEndpointUrl(endpoint: string): string { try { return normalizeBaseUrl(new URL(endpoint).toString()); diff --git a/packages/agent/tests/anonymous-dwn-api.spec.ts b/packages/agent/tests/anonymous-dwn-api.spec.ts index 604b82d4..c7f37fa8 100644 --- a/packages/agent/tests/anonymous-dwn-api.spec.ts +++ b/packages/agent/tests/anonymous-dwn-api.spec.ts @@ -1,27 +1,48 @@ import type { EnboxRpc } from '@enbox/dwn-clients'; -import type { DidDereferencingResult, DidUrlDereferencer } from '@enbox/dids'; +import type { BearerDid, DidDereferencingResult, DidResolver, DidUrlDereferencer } from '@enbox/dids'; +import type { MessageSigner, ProtocolDefinition } from '@enbox/dwn-sdk-js'; import sinon from 'sinon'; -import { afterEach, beforeEach, describe, expect, it } from 'bun:test'; +import { afterEach, beforeAll, beforeEach, describe, expect, it } from 'bun:test'; +import { DidJwk, UniversalResolver } from '@enbox/dids'; +import { Encoder, ProtocolsConfigure, RecordsWrite } from '@enbox/dwn-sdk-js'; import { AnonymousDwnApi } from '../src/anonymous-dwn-api.js'; describe('AnonymousDwnApi', () => { let anonymousDwn: AnonymousDwnApi; + let dereferenceStub: sinon.SinonStub; let rpcStub: sinon.SinonStubbedInstance; - let resolverStub: sinon.SinonStubbedInstance; + let resolver: DidResolver & DidUrlDereferencer; + let targetDid: string; + let targetSigner: MessageSigner; - const targetDid = 'did:example:alice'; const dwnEndpoint = 'https://dwn.example.com'; + const protocolUri = 'https://blog.example/posts'; + const textEncoder = new TextEncoder(); + + const protocolDefinition: ProtocolDefinition = { + protocol : protocolUri, + published : true, + types : { + post: { + schema : 'https://blog.example/schemas/post', + dataFormats : ['text/plain'], + }, + }, + structure: { + post: {}, + }, + }; /** * Helper: create a stubbed DidUrlDereferencer that returns a DWN service * endpoint for the given DID. */ - function createResolverStub(): sinon.SinonStubbedInstance { - const stub = { dereference: sinon.stub() } as sinon.SinonStubbedInstance; - stub.dereference.resolves({ + function createResolver(): DidResolver & DidUrlDereferencer { + const universalResolver = new UniversalResolver({ didResolvers: [DidJwk] }); + dereferenceStub = sinon.stub().resolves({ dereferencingMetadata : {}, contentMetadata : {}, contentStream : { @@ -30,7 +51,11 @@ describe('AnonymousDwnApi', () => { serviceEndpoint : [dwnEndpoint], }, } as DidDereferencingResult); - return stub; + + return { + dereference : dereferenceStub, + resolve : universalResolver.resolve.bind(universalResolver), + }; } /** @@ -49,11 +74,17 @@ describe('AnonymousDwnApi', () => { } as unknown as sinon.SinonStubbedInstance; } + beforeAll(async () => { + const target = await DidJwk.create(); + targetDid = target.uri; + targetSigner = await signerForDid(target); + }); + beforeEach(() => { - resolverStub = createResolverStub(); + resolver = createResolver(); rpcStub = createRpcStub(); anonymousDwn = new AnonymousDwnApi({ - didResolver : resolverStub, + didResolver : resolver, rpcClient : rpcStub, }); }); @@ -64,23 +95,17 @@ describe('AnonymousDwnApi', () => { describe('recordsQuery()', () => { it('should create an unsigned RecordsQuery and send it via RPC', async () => { - const mockEntries = [ - { - recordId : 'record-1', - descriptor : { interface: 'Records', method: 'Write', dataFormat: 'text/plain', dataCid: 'cid1', dataSize: 5, dateCreated: '2024-01-01', messageTimestamp: '2024-01-01' }, - contextId : undefined, - encodedData : undefined, - }, - ]; + const data = textEncoder.encode('published post'); + const recordsWrite = await createPublishedRecord(data); rpcStub.sendDwnRequest.resolves({ status : { code: 200, detail: 'OK' }, - entries : mockEntries, + entries : [{ ...recordsWrite.message, encodedData: Encoder.bytesToBase64Url(data) }], cursor : undefined, }); const reply = await anonymousDwn.recordsQuery(targetDid, { - filter: { protocol: 'https://blog.example/posts' }, + filter: { protocol: protocolUri }, }); expect(reply.status.code).toBe(200); @@ -126,15 +151,14 @@ describe('AnonymousDwnApi', () => { describe('recordsRead()', () => { it('should create an unsigned RecordsRead and send it via RPC', async () => { - const recordId = 'bafyrei-test-record'; + const data = textEncoder.encode('public value'); + const recordsWrite = await createPublishedRecord(data); + const recordId = recordsWrite.message.recordId; rpcStub.sendDwnRequest.resolves({ status : { code: 200, detail: 'OK' }, entry : { - recordsWrite: { - recordId : recordId, - descriptor : { interface: 'Records', method: 'Write', dataFormat: 'text/plain', dataCid: 'cid1', dataSize: 12, dateCreated: '2024-01-01', messageTimestamp: '2024-01-01' }, - }, - data: new ReadableStream(), + recordsWrite : recordsWrite.message, + data : new Blob([data]).stream(), }, }); @@ -185,20 +209,18 @@ describe('AnonymousDwnApi', () => { describe('protocolsQuery()', () => { it('should create an unsigned ProtocolsQuery and send it via RPC', async () => { - const mockDefinition = { - protocol : 'https://blog.example/posts', - published : true, - types : { post: { schema: 'https://blog.example/schemas/post' } }, - structure : { post: {} }, - }; + const protocolsConfigure = await ProtocolsConfigure.create({ + definition : protocolDefinition, + signer : targetSigner, + }); rpcStub.sendDwnRequest.resolves({ status : { code: 200, detail: 'OK' }, - entries : [{ descriptor: { definition: mockDefinition } }], + entries : [protocolsConfigure.message], }); const reply = await anonymousDwn.protocolsQuery(targetDid, { - filter: { protocol: 'https://blog.example/posts' }, + filter: { protocol: protocolUri }, }); expect(reply.status.code).toBe(200); @@ -223,6 +245,30 @@ describe('AnonymousDwnApi', () => { }); describe('recordsSubscribe()', () => { + it('should pass the listener in the current subscription request shape', async () => { + const handler = sinon.stub(); + rpcStub.getServerInfo.resolves({ + registrationRequirements : [], + maxFileSize : 10_000_000, + webSocketSupport : true, + }); + rpcStub.sendDwnRequest.resolves({ + status : { code: 200, detail: 'OK' }, + entries : [], + subscription : { close: sinon.stub() }, + }); + + await anonymousDwn.recordsSubscribe( + targetDid, + { filter: { protocol: protocolUri } }, + handler, + ); + + const rpcArgs = rpcStub.sendDwnRequest.args[0][0]; + expect(rpcArgs.dwnUrl).toBe('wss://dwn.example.com/'); + expect(rpcArgs.subscription).toEqual({ handler }); + }); + it('should throw when WebSocket is not supported and no other endpoints available', async () => { rpcStub.getServerInfo.resolves({ registrationRequirements : [], @@ -239,6 +285,8 @@ describe('AnonymousDwnApi', () => { () => {}, ) ).rejects.toThrow('AnonymousDwnApi: Failed to send request'); + + expect(rpcStub.sendDwnRequest.called).toBe(false); }); }); @@ -254,13 +302,13 @@ describe('AnonymousDwnApi', () => { }); // Verify the DID resolver was called with the target DID's #dwn fragment. - expect(resolverStub.dereference.calledOnce).toBe(true); - const dereferenceArg = resolverStub.dereference.args[0][0]; + expect(dereferenceStub.calledOnce).toBe(true); + const dereferenceArg = dereferenceStub.args[0][0]; expect(dereferenceArg).toContain(targetDid); }); it('should throw when DID has no DWN service endpoints', async () => { - resolverStub.dereference.resolves({ + dereferenceStub.resolves({ dereferencingMetadata : {}, contentMetadata : {}, contentStream : undefined, @@ -273,7 +321,7 @@ describe('AnonymousDwnApi', () => { it('should try multiple DWN endpoints on failure', async () => { // DID resolves to two endpoints. - resolverStub.dereference.resolves({ + dereferenceStub.resolves({ dereferencingMetadata : {}, contentMetadata : {}, contentStream : { @@ -298,5 +346,56 @@ describe('AnonymousDwnApi', () => { expect(reply.status.code).toBe(200); expect(rpcStub.sendDwnRequest.calledTwice).toBe(true); }); + + it('should try the next DWN endpoint when response verification fails', async () => { + dereferenceStub.resolves({ + dereferencingMetadata : {}, + contentMetadata : {}, + contentStream : { + id : '#dwn', + type : 'DecentralizedWebNode', + serviceEndpoint : ['https://dwn1.example.com', 'https://dwn2.example.com'], + }, + } as DidDereferencingResult); + + const attacker = await DidJwk.create(); + const [attackerSigner, validConfigure] = await Promise.all([ + signerForDid(attacker), + ProtocolsConfigure.create({ definition: protocolDefinition, signer: targetSigner }), + ]); + const invalidConfigure = await ProtocolsConfigure.create({ + definition : protocolDefinition, + signer : attackerSigner, + }); + rpcStub.sendDwnRequest + .onFirstCall().resolves({ status: { code: 200, detail: 'OK' }, entries: [invalidConfigure.message] }) + .onSecondCall().resolves({ status: { code: 200, detail: 'OK' }, entries: [validConfigure.message] }); + + const reply = await anonymousDwn.protocolsQuery(targetDid, { filter: { protocol: protocolUri } }); + + expect(reply.entries).toEqual([validConfigure.message]); + expect(rpcStub.sendDwnRequest.calledTwice).toBe(true); + }); }); + + async function createPublishedRecord(data: Uint8Array): Promise { + return RecordsWrite.create({ + data, + dataFormat : 'text/plain', + protocol : protocolDefinition.protocol, + protocolPath : 'post', + published : true, + schema : protocolDefinition.types.post.schema, + signer : targetSigner, + }); + } }); + +async function signerForDid(did: BearerDid): Promise { + const signer = await did.getSigner(); + return { + algorithm : signer.algorithm, + keyId : signer.keyId, + sign : async (content: Uint8Array): Promise => signer.sign({ data: content }), + }; +} diff --git a/packages/agent/tests/dwn-api.spec.ts b/packages/agent/tests/dwn-api.spec.ts index f211c267..137c3191 100644 --- a/packages/agent/tests/dwn-api.spec.ts +++ b/packages/agent/tests/dwn-api.spec.ts @@ -13,6 +13,7 @@ import { ENCRYPTION_CONTROL_DELIVERY_PATH, KeyDerivationScheme, Message, + RecordsWrite, TestDataGenerator, Time } from '@enbox/dwn-sdk-js'; @@ -2373,6 +2374,45 @@ describe('AgentDwnApi', () => { expect(queryReply.entries?.[0]).toHaveProperty('recordId', writeMessage.recordId); }); + it('rejects an invalid remote response and tries the next DWN endpoint', async () => { + const dataBytes = Convert.string('authentic remote record').toUint8Array(); + const recordsWrite = await RecordsWrite.create({ + data : dataBytes, + dataFormat : 'text/plain', + protocol : 'http://free-for-all.xyz', + protocolPath : 'post', + signer : await testHarness.agent.dwn['getSigner'](alice.did.uri), + }); + const tamperedWrite = structuredClone(recordsWrite.message); + tamperedWrite.descriptor.dataFormat = 'application/json'; + + sinon.stub(testHarness.agent.dwn, 'getDwnEndpointUrlsForTarget').resolves([ + 'https://invalid.example', + 'https://valid.example', + ]); + const sendDwnRequest = sinon.stub(testHarness.agent.rpc, 'sendDwnRequest'); + sendDwnRequest.onFirstCall().resolves({ + entries : [tamperedWrite], + status : { code: 200, detail: 'OK' }, + }); + sendDwnRequest.onSecondCall().resolves({ + entries : [], + status : { code: 200, detail: 'OK' }, + }); + + const { reply } = await testHarness.agent.dwn.sendRequest({ + author : alice.did.uri, + target : alice.did.uri, + messageType : DwnInterface.RecordsQuery, + messageParams : { filter: { recordId: recordsWrite.message.recordId } }, + }); + + expect(reply.entries).toEqual([]); + expect(sendDwnRequest.callCount).toBe(2); + expect(sendDwnRequest.firstCall.args[0].dwnUrl).toBe('https://invalid.example'); + expect(sendDwnRequest.secondCall.args[0].dwnUrl).toBe('https://valid.example'); + }); + it('handles RecordsRead messages', async () => { // Create test data to write. const dataBytes = Convert.string('Hello, world!').toUint8Array(); diff --git a/packages/agent/tests/dwn-protocol-cache.spec.ts b/packages/agent/tests/dwn-protocol-cache.spec.ts index 5a2021eb..c6c9084c 100644 --- a/packages/agent/tests/dwn-protocol-cache.spec.ts +++ b/packages/agent/tests/dwn-protocol-cache.spec.ts @@ -131,6 +131,7 @@ describe('dwn-protocol-cache', () => { ); expect(result).toEqual(mockDefinition); expect(cache.get(`remote~${targetDid}~${protocolUri}`)).toEqual(mockDefinition); + expect(sendDwnRpcRequest.firstCall.args[0].verifyResponse).toBe(true); }); it('should throw when remote DWN returns non-200 status', async () => { diff --git a/packages/agent/tests/remote-dwn-response.spec.ts b/packages/agent/tests/remote-dwn-response.spec.ts new file mode 100644 index 00000000..6603438b --- /dev/null +++ b/packages/agent/tests/remote-dwn-response.spec.ts @@ -0,0 +1,553 @@ +import type { BearerDid, DidResolver } from '@enbox/dids'; +import type { + GenericMessage, + GenericMessageReply, + MessageSigner, + ProtocolDefinition, + ProtocolsConfigureMessage, + ProtocolsQueryReply, + RecordsQueryReply, + RecordsReadReply, + RecordsSubscribeReply, + RecordsWriteMessage, +} from '@enbox/dwn-sdk-js'; + +import { beforeAll, describe, expect, it } from 'bun:test'; + +import { + DataStream, + DateSort, + Encoder, + ProtocolsConfigure, + ProtocolsQuery, + RecordsDelete, + RecordsQuery, + RecordsRead, + RecordsSubscribe, + RecordsWrite, + Time, +} from '@enbox/dwn-sdk-js'; +import { DidJwk, UniversalResolver } from '@enbox/dids'; + +import { verifyRemoteDwnResponse } from '../src/remote-dwn-response.js'; + +const protocolUri = 'https://example.com/remote-response'; +const otherProtocolUri = 'https://example.com/other-protocol'; +const recordSchema = 'https://example.com/schemas/note'; +const textEncoder = new TextEncoder(); + +describe('verifyRemoteDwnResponse', () => { + let attacker: BearerDid; + let didResolver: DidResolver; + let target: BearerDid; + let attackerSigner: MessageSigner; + let targetSigner: MessageSigner; + + beforeAll(async () => { + [target, attacker] = await Promise.all([DidJwk.create(), DidJwk.create()]); + [targetSigner, attackerSigner] = await Promise.all([signerForDid(target), signerForDid(attacker)]); + didResolver = new UniversalResolver({ didResolvers: [DidJwk] }); + }); + + function verifyResponse(message: GenericMessage, reply: GenericMessageReply): Promise { + return verifyRemoteDwnResponse({ didResolver, message, reply, targetDid: target.uri }); + } + + it('preserves unsuccessful query responses that contain no artifacts', async () => { + const [protocolsQuery, recordsQuery] = await Promise.all([ + ProtocolsQuery.create({ filter: { protocol: protocolUri } }), + RecordsQuery.create({ filter: { protocol: protocolUri }, signer: targetSigner }), + ]); + const reply = { status: { code: 500, detail: 'Internal Server Error' } }; + + await verifyResponse(protocolsQuery.message, reply); + await verifyResponse(recordsQuery.message, reply); + }); + + describe('ProtocolsQuery', () => { + it('accepts a configuration signed directly by the target DID', async () => { + const query = await ProtocolsQuery.create({ filter: { protocol: protocolUri } }); + const configure = await ProtocolsConfigure.create({ + definition : protocolDefinition(protocolUri), + signer : targetSigner, + }); + const reply: ProtocolsQueryReply = { + entries : [configure.message], + status : { code: 200, detail: 'OK' }, + }; + + await verifyResponse(query.message, reply); + }); + + it('rejects a configuration signed by an attacker', async () => { + const query = await ProtocolsQuery.create({ filter: { protocol: protocolUri } }); + const configure = await ProtocolsConfigure.create({ + definition : protocolDefinition(protocolUri), + signer : attackerSigner, + }); + + await expect(verifyResponse(query.message, protocolReply(configure.message))) + .rejects.toThrow(`instead of '${target.uri}'`); + }); + + it('rejects a signed configuration whose definition was changed in transit', async () => { + const query = await ProtocolsQuery.create({ filter: { protocol: protocolUri } }); + const configure = await ProtocolsConfigure.create({ + definition : protocolDefinition(protocolUri), + signer : targetSigner, + }); + const tampered = structuredClone(configure.message); + tampered.descriptor.definition.published = false; + + await expect(verifyResponse(query.message, protocolReply(tampered))).rejects.toThrow(); + }); + + it('rejects a valid configuration for a different requested protocol', async () => { + const query = await ProtocolsQuery.create({ filter: { protocol: protocolUri } }); + const configure = await ProtocolsConfigure.create({ + definition : protocolDefinition(otherProtocolUri), + signer : targetSigner, + }); + + await expect(verifyResponse(query.message, protocolReply(configure.message))) + .rejects.toThrow(`while '${protocolUri}' was requested`); + }); + + it('rejects multiple configurations for one exact protocol query', async () => { + const query = await ProtocolsQuery.create({ filter: { protocol: protocolUri } }); + const configure = await ProtocolsConfigure.create({ + definition : protocolDefinition(protocolUri), + signer : targetSigner, + }); + const reply = protocolReply(configure.message); + reply.entries!.push(configure.message); + + await expect(verifyResponse(query.message, reply)).rejects.toThrow('multiple configurations'); + }); + + it('rejects entries attached to an unsuccessful response', async () => { + const query = await ProtocolsQuery.create({ filter: { protocol: protocolUri } }); + const configure = await ProtocolsConfigure.create({ + definition : protocolDefinition(protocolUri), + signer : targetSigner, + }); + const reply = protocolReply(configure.message); + reply.status = { code: 500, detail: 'Internal Server Error' }; + + await expect(verifyResponse(query.message, reply)).rejects.toThrow('status 500 with entries'); + }); + + it('rejects an unpublished configuration returned to an anonymous query', async () => { + const query = await ProtocolsQuery.create({ filter: { protocol: protocolUri } }); + const configure = await ProtocolsConfigure.create({ + definition : { ...protocolDefinition(protocolUri), published: false }, + signer : targetSigner, + }); + + await expect(verifyResponse(query.message, protocolReply(configure.message))) + .rejects.toThrow('anonymous query returned unpublished protocol'); + }); + }); + + describe('RecordsQuery', () => { + it('accepts an authenticated matching record whose inline bytes match its descriptor', async () => { + const data = textEncoder.encode('authentic record'); + const recordsWrite = await createRecordsWrite(data, targetSigner); + const query = await RecordsQuery.create({ + filter : { recordId: recordsWrite.message.recordId }, + signer : targetSigner, + }); + const reply = recordsQueryReply(recordsWrite.message, data); + + await verifyResponse(query.message, reply); + }); + + it('rejects signed records whose descriptor or signature was changed in transit', async () => { + const data = textEncoder.encode('signed record'); + const recordsWrite = await createRecordsWrite(data, targetSigner); + const query = await RecordsQuery.create({ filter: { recordId: recordsWrite.message.recordId }, signer: targetSigner }); + const changedDescriptor = structuredClone(recordsWrite.message); + changedDescriptor.descriptor.dataFormat = 'application/json'; + const changedSignature = structuredClone(recordsWrite.message); + const signature = changedSignature.authorization!.signature.signatures[0].signature; + changedSignature.authorization!.signature.signatures[0].signature = replaceLastCharacter(signature); + + for (const message of [changedDescriptor, changedSignature]) { + await expect(verifyResponse(query.message, recordsQueryReply(message, data))).rejects.toThrow(); + } + }); + + it('rejects an authenticated record outside the requested filter', async () => { + const data = textEncoder.encode('wrong record'); + const recordsWrite = await createRecordsWrite(data, targetSigner); + const query = await RecordsQuery.create({ filter: { recordId: 'not-the-returned-record' }, signer: targetSigner }); + + await expect(verifyResponse(query.message, recordsQueryReply(recordsWrite.message, data))) + .rejects.toThrow('does not match the request filter'); + }); + + it('rejects inline bytes that do not match the signed data CID', async () => { + const data = textEncoder.encode('expected bytes'); + const recordsWrite = await createRecordsWrite(data, targetSigner); + const query = await RecordsQuery.create({ filter: { recordId: recordsWrite.message.recordId }, signer: targetSigner }); + + await expect(verifyResponse( + query.message, + recordsQueryReply(recordsWrite.message, textEncoder.encode('tampered bytes')), + )).rejects.toThrow(); + }); + + it('rejects entries attached to an unsuccessful response', async () => { + const data = textEncoder.encode('injected error entry'); + const recordsWrite = await createRecordsWrite(data, targetSigner); + const query = await RecordsQuery.create({ filter: { recordId: recordsWrite.message.recordId }, signer: targetSigner }); + const reply = recordsQueryReply(recordsWrite.message, data); + reply.status = { code: 500, detail: 'Internal Server Error' }; + + await expect(verifyResponse(query.message, reply)).rejects.toThrow('status 500 with entries'); + }); + + it('requires anonymous results to be both published and matched by the original filter', async () => { + const data = textEncoder.encode('visibility checks'); + const unpublishedWrite = await createRecordsWrite(data, targetSigner); + const publishedWrite = await createRecordsWrite(data, targetSigner, true); + const anonymousQuery = await RecordsQuery.create({ filter: { protocol: protocolUri } }); + const unpublishedQuery = await RecordsQuery.create({ filter: { protocol: protocolUri, published: false } }); + + await expect(verifyResponse(anonymousQuery.message, recordsQueryReply(unpublishedWrite.message, data))) + .rejects.toThrow('anonymous query returned unpublished'); + + await expect(verifyResponse(unpublishedQuery.message, recordsQueryReply(publishedWrite.message, data))) + .rejects.toThrow('does not match the request filter'); + }); + + it('accepts an update only when it carries its matching authenticated initial write', async () => { + const initialData = textEncoder.encode('initial data'); + const updatedData = textEncoder.encode('updated data'); + const initialWrite = await createRecordsWrite(initialData, targetSigner); + await Time.minimalSleep(); + const update = await RecordsWrite.createFrom({ + data : updatedData, + recordsWriteMessage : initialWrite.message, + signer : targetSigner, + }); + const query = await RecordsQuery.create({ filter: { recordId: update.message.recordId }, signer: targetSigner }); + const reply: RecordsQueryReply = { + entries : [{ ...update.message, encodedData: Encoder.bytesToBase64Url(updatedData), initialWrite: initialWrite.message }], + status : { code: 200, detail: 'OK' }, + }; + + await verifyResponse(query.message, reply); + }); + + it('rejects an update with a missing or mismatched initial write', async () => { + const initialData = textEncoder.encode('initial data'); + const updatedData = textEncoder.encode('updated data'); + const initialWrite = await createRecordsWrite(initialData, targetSigner); + const unrelatedInitialWrite = await createRecordsWrite(textEncoder.encode('unrelated data'), targetSigner); + await Time.minimalSleep(); + const update = await RecordsWrite.createFrom({ + data : updatedData, + recordsWriteMessage : initialWrite.message, + signer : targetSigner, + }); + const query = await RecordsQuery.create({ filter: { recordId: update.message.recordId }, signer: targetSigner }); + const entries = [ + { ...update.message, encodedData: Encoder.bytesToBase64Url(updatedData) }, + { ...update.message, encodedData: Encoder.bytesToBase64Url(updatedData), initialWrite: unrelatedInitialWrite.message }, + ]; + + for (const entry of entries) { + const reply: RecordsQueryReply = { entries: [entry], status: { code: 200, detail: 'OK' } }; + await expect(verifyResponse(query.message, reply)).rejects.toThrow(); + } + }); + }); + + describe('RecordsSubscribe snapshot', () => { + it('accepts an authenticated matching initial snapshot', async () => { + const data = textEncoder.encode('snapshot record'); + const recordsWrite = await createRecordsWrite(data, targetSigner, true); + const subscribe = await RecordsSubscribe.create({ filter: { protocol: protocolUri } }); + const reply: RecordsSubscribeReply = { + entries : [{ ...recordsWrite.message, encodedData: Encoder.bytesToBase64Url(data) }], + status : { code: 200, detail: 'OK' }, + subscription : { id: 'valid-snapshot', close: async (): Promise => {} }, + }; + + await verifyResponse(subscribe.message, reply); + }); + + it('closes a subscription whose initial snapshot fails verification', async () => { + const data = textEncoder.encode('tampered snapshot record'); + const recordsWrite = await createRecordsWrite(data, targetSigner, true); + const tamperedWrite = structuredClone(recordsWrite.message); + const signature = tamperedWrite.authorization!.signature.signatures[0].signature; + tamperedWrite.authorization!.signature.signatures[0].signature = replaceLastCharacter(signature); + const subscribe = await RecordsSubscribe.create({ filter: { protocol: protocolUri } }); + let closed = false; + const reply: RecordsSubscribeReply = { + entries : [{ ...tamperedWrite, encodedData: Encoder.bytesToBase64Url(data) }], + status : { code: 200, detail: 'OK' }, + subscription : { id: 'invalid-snapshot', close: async (): Promise => { closed = true; } }, + }; + + await expect(verifyResponse(subscribe.message, reply)).rejects.toThrow(); + expect(closed).toBe(true); + }); + }); + + describe('RecordsRead', () => { + it('streams authentic bytes and verifies them before successful end-of-stream', async () => { + const data = textEncoder.encode('streamed record'); + const recordsWrite = await createRecordsWrite(data, targetSigner); + const read = await RecordsRead.create({ filter: { recordId: recordsWrite.message.recordId }, signer: targetSigner }); + const reply = recordsReadReply(recordsWrite.message, data); + + await verifyResponse(read.message, reply); + + expect(await DataStream.toBytes(reply.entry!.data!)).toEqual(data); + }); + + it('fails the returned stream when bytes are tampered with or differ from the signed size', async () => { + const data = textEncoder.encode('expected stream'); + const recordsWrite = await createRecordsWrite(data, targetSigner); + const read = await RecordsRead.create({ filter: { recordId: recordsWrite.message.recordId }, signer: targetSigner }); + const invalidPayloads = [ + textEncoder.encode('tampered stream'), + textEncoder.encode('short'), + textEncoder.encode('expected stream plus extra bytes'), + ]; + + for (const invalidData of invalidPayloads) { + const reply = recordsReadReply(recordsWrite.message, invalidData); + await verifyResponse(read.message, reply); + + await expect(DataStream.toBytes(reply.entry!.data!)).rejects.toThrow(); + } + }); + + it('accepts an authenticated unpublished record returned by an anonymous read', async () => { + const data = textEncoder.encode('anyone-readable record'); + const recordsWrite = await createRecordsWrite(data, targetSigner); + const read = await RecordsRead.create({ filter: { recordId: recordsWrite.message.recordId } }); + const reply = recordsReadReply(recordsWrite.message, data); + + await verifyResponse(read.message, reply); + + expect(await DataStream.toBytes(reply.entry!.data!)).toEqual(data); + }); + + it('rejects an unpublished record from a published-sorted read', async () => { + const data = textEncoder.encode('unpublished record'); + const recordsWrite = await createRecordsWrite(data, targetSigner); + const read = await RecordsRead.create({ + dateSort : DateSort.PublishedDescending, + filter : { recordId: recordsWrite.message.recordId }, + signer : targetSigner, + }); + + await expect(verifyResponse(read.message, recordsReadReply(recordsWrite.message, data))) + .rejects.toThrow('does not match the request filter'); + }); + + it('authenticates and preserves 410 responses whose record data is unavailable', async () => { + const data = textEncoder.encode('unavailable record'); + const initialWrite = await createRecordsWrite(data, targetSigner); + await Time.minimalSleep(); + const update = await RecordsWrite.createFrom({ + data, + recordsWriteMessage : initialWrite.message, + signer : targetSigner, + }); + const read = await RecordsRead.create({ filter: { recordId: initialWrite.message.recordId }, signer: targetSigner }); + + for (const recordsWrite of [initialWrite, update]) { + const reply: RecordsReadReply = { + entry : { recordsWrite: recordsWrite.message }, + status : { code: 410, detail: 'Record data not available' }, + }; + + await verifyResponse(read.message, reply); + } + }); + + it('authenticates a deleted record and its initial write in a 404 response', async () => { + const data = textEncoder.encode('deleted record'); + const initialWrite = await createRecordsWrite(data, targetSigner); + const recordsDelete = await RecordsDelete.create({ + recordId : initialWrite.message.recordId, + signer : targetSigner, + }); + const read = await RecordsRead.create({ filter: { recordId: initialWrite.message.recordId }, signer: targetSigner }); + const reply: RecordsReadReply = { + entry: { + initialWrite : initialWrite.message, + recordsDelete : recordsDelete.message, + }, + status: { code: 404, detail: 'Not Found' }, + }; + + await verifyResponse(read.message, reply); + }); + + it('rejects a tombstone when the read filter depends on the missing last write', async () => { + const data = textEncoder.encode('deleted record'); + const initialWrite = await createRecordsWrite(data, targetSigner); + const recordsDelete = await RecordsDelete.create({ recordId: initialWrite.message.recordId, signer: targetSigner }); + const read = await RecordsRead.create({ + filter : { recordId: initialWrite.message.recordId, dataCid: initialWrite.message.descriptor.dataCid }, + signer : targetSigner, + }); + const reply: RecordsReadReply = { + entry: { + initialWrite : initialWrite.message, + recordsDelete : recordsDelete.message, + }, + status: { code: 404, detail: 'Not Found' }, + }; + + await expect(verifyResponse(read.message, reply)) + .rejects.toThrow('cannot be authenticated against a filter'); + }); + + it('rejects a tombstone from a published-sorted read', async () => { + const initialWrite = await createRecordsWrite(textEncoder.encode('deleted record'), targetSigner, true); + const recordsDelete = await RecordsDelete.create({ recordId: initialWrite.message.recordId, signer: targetSigner }); + const read = await RecordsRead.create({ + dateSort : DateSort.PublishedAscending, + filter : { recordId: initialWrite.message.recordId }, + signer : targetSigner, + }); + const reply: RecordsReadReply = { + entry : { initialWrite: initialWrite.message, recordsDelete: recordsDelete.message }, + status : { code: 404, detail: 'Not Found' }, + }; + + await expect(verifyResponse(read.message, reply)).rejects.toThrow('published date sort'); + }); + + it('rejects inconsistent RecordsRead entry and status combinations', async () => { + const data = textEncoder.encode('record state'); + const initialWrite = await createRecordsWrite(data, targetSigner); + const recordsDelete = await RecordsDelete.create({ + recordId : initialWrite.message.recordId, + signer : targetSigner, + }); + const read = await RecordsRead.create({ filter: { recordId: initialWrite.message.recordId }, signer: targetSigner }); + const invalidReplies: RecordsReadReply[] = [ + { status: { code: 200, detail: 'OK' } }, + { + entry : { recordsWrite: initialWrite.message }, + status : { code: 200, detail: 'OK' }, + }, + { + entry : { data: DataStream.fromBytes(data), recordsWrite: initialWrite.message }, + status : { code: 410, detail: 'Record data not available' }, + }, + { + entry : { recordsDelete: recordsDelete.message, recordsWrite: initialWrite.message }, + status : { code: 404, detail: 'Not Found' }, + }, + { + entry : {}, + status : { code: 200, detail: 'OK' }, + }, + { + entry : { initialWrite: initialWrite.message, recordsDelete: recordsDelete.message }, + status : { code: 200, detail: 'OK' }, + }, + ]; + + for (const reply of invalidReplies) { + await expect(verifyResponse(read.message, reply)).rejects.toThrow(); + } + }); + + it('rejects a tombstone with an unauthenticated or unrelated initial write', async () => { + const initialWrite = await createRecordsWrite(textEncoder.encode('deleted record'), targetSigner); + const unrelatedWrite = await createRecordsWrite(textEncoder.encode('unrelated record'), targetSigner); + const recordsDelete = await RecordsDelete.create({ + recordId : initialWrite.message.recordId, + signer : targetSigner, + }); + const tamperedDelete = structuredClone(recordsDelete.message); + const deleteSignature = tamperedDelete.authorization.signature.signatures[0].signature; + tamperedDelete.authorization.signature.signatures[0].signature = replaceLastCharacter(deleteSignature); + const read = await RecordsRead.create({ filter: { recordId: initialWrite.message.recordId }, signer: targetSigner }); + const invalidReplies: RecordsReadReply[] = [ + { + entry : { initialWrite: unrelatedWrite.message, recordsDelete: recordsDelete.message }, + status : { code: 404, detail: 'Not Found' }, + }, + { + entry : { initialWrite: initialWrite.message, recordsDelete: tamperedDelete }, + status : { code: 404, detail: 'Not Found' }, + }, + ]; + + for (const reply of invalidReplies) { + await expect(verifyResponse(read.message, reply)).rejects.toThrow(); + } + }); + }); +}); + +async function signerForDid(did: BearerDid): Promise { + const signer = await did.getSigner(); + return { + algorithm : signer.algorithm, + keyId : signer.keyId, + sign : async (content: Uint8Array): Promise => signer.sign({ data: content }), + }; +} + +function protocolDefinition(protocol: string): ProtocolDefinition { + return { + protocol, + published : true, + structure : {}, + types : {}, + }; +} + +function protocolReply(message: ProtocolsConfigureMessage): ProtocolsQueryReply { + return { + entries : [message], + status : { code: 200, detail: 'OK' }, + }; +} + +async function createRecordsWrite(data: Uint8Array, signer: MessageSigner, published?: boolean): Promise { + return RecordsWrite.create({ + data, + dataFormat : 'text/plain', + protocol : protocolUri, + protocolPath : 'note', + schema : recordSchema, + signer, + published, + }); +} + +function recordsQueryReply(message: RecordsWriteMessage, data: Uint8Array): RecordsQueryReply { + return { + entries : [{ ...message, encodedData: Encoder.bytesToBase64Url(data) }], + status : { code: 200, detail: 'OK' }, + }; +} + +function recordsReadReply(message: RecordsWriteMessage, data: Uint8Array): RecordsReadReply { + return { + entry: { + data : DataStream.fromBytes(data), + recordsWrite : message, + }, + status: { code: 200, detail: 'OK' }, + }; +} + +function replaceLastCharacter(value: string): string { + const replacement = value.endsWith('A') ? 'B' : 'A'; + return `${value.slice(0, -1)}${replacement}`; +} diff --git a/packages/api/src/read-only-record.ts b/packages/api/src/read-only-record.ts index b3c98d79..17c1033c 100644 --- a/packages/api/src/read-only-record.ts +++ b/packages/api/src/read-only-record.ts @@ -272,10 +272,10 @@ export class ReadOnlyRecord { private async readRecordData(): Promise { try { const reply = await this._anonymousDwn.recordsRead(this._remoteOrigin, { - filter: { recordId: this._recordId }, + filter: { recordId: this._recordId, dataCid: this.dataCid }, }); - if (reply.status.code !== 200 || !reply.entry?.data) { + if (reply.status.code !== 200 || !reply.entry?.recordsWrite || !reply.entry.data) { throw new Error(`${reply.status.code}: ${reply.status.detail}`); } diff --git a/packages/api/tests/dwn-reader-api.spec.ts b/packages/api/tests/dwn-reader-api.spec.ts index 67fbf39d..5bb1bdab 100644 --- a/packages/api/tests/dwn-reader-api.spec.ts +++ b/packages/api/tests/dwn-reader-api.spec.ts @@ -372,10 +372,10 @@ describe('ReadOnlyRecord', () => { expect(text).toBe('re-fetched data'); expect(anonStub.recordsRead.calledOnce).toBe(true); - // Verify the re-fetch used the correct recordId and target. + // Pin the re-fetch to the exact record version represented by this object. const readArgs = anonStub.recordsRead.args[0]; expect(readArgs[0]).toBe(targetDid); - expect(readArgs[1].filter.recordId).toBe('refetch-test'); + expect(readArgs[1].filter).toEqual({ recordId: 'refetch-test', dataCid: msg.descriptor.dataCid }); }); it('should coalesce concurrent convenience reads from a captured data accessor', async () => { diff --git a/packages/dwn-sdk-js/src/handlers/records-subscribe.ts b/packages/dwn-sdk-js/src/handlers/records-subscribe.ts index 2bdd5136..1d375962 100644 --- a/packages/dwn-sdk-js/src/handlers/records-subscribe.ts +++ b/packages/dwn-sdk-js/src/handlers/records-subscribe.ts @@ -351,6 +351,9 @@ export class RecordsSubscribeHandler implements MethodHandler { ); } } catch (error) { + // Open-time authorization uses the subscription's signed timestamp, while delivery uses + // the current time. A future-dated grant can therefore become valid without authority changing, + // so a not-yet-active delivery check is retryable while other authorization failures are terminal. const authorizationFailure = error instanceof DwnError && error.code !== DwnErrorCode.GrantAuthorizationGrantNotYetActive; emitTerminalDeliveryError( diff --git a/packages/dwn-sdk-js/src/index.ts b/packages/dwn-sdk-js/src/index.ts index 272b2b8a..4e0e6cc5 100644 --- a/packages/dwn-sdk-js/src/index.ts +++ b/packages/dwn-sdk-js/src/index.ts @@ -52,7 +52,7 @@ export { DwnError, DwnErrorCode } from './core/dwn-error.js'; export type { DwnErrorInfo } from './core/dwn-error.js'; export { DwnInterfaceName, DwnMethodName } from './enums/dwn-interface-method.js'; export { Encoder } from './utils/encoder.js'; -export { assertValidSubtreeFilters, isSubtreeFilter } from './utils/filter.js'; +export { assertValidSubtreeFilters, FilterUtility, isSubtreeFilter } from './utils/filter.js'; export { MessagesSubscribe } from './interfaces/messages-subscribe.js'; export type { MessagesSubscribeOptions } from './interfaces/messages-subscribe.js'; export { Encryption, ContentEncryptionAlgorithm, KeyAgreementAlgorithm, ROLE_AUDIENCE_DERIVATION_SCHEME, SEAL_DERIVATION_SCHEME } from './utils/encryption.js';