Event and Records Subscriptions

build/compile-validators.js

@@ -24,6 +24,7 @@ import Definitions from '../json-schemas/definitions.json' assert { type: 'json'
 import EventsFilter from '../json-schemas/interface-methods/events-filter.json' assert { type: 'json' };
 import EventsGet from '../json-schemas/interface-methods/events-get.json' assert { type: 'json' };
 import EventsQuery from '../json-schemas/interface-methods/events-query.json' assert { type: 'json' };
+import EventsSubscribe from '../json-schemas/interface-methods/events-subscribe.json' assert { type: 'json' };
 import GeneralJwk from '../json-schemas/jwk/general-jwk.json' assert { type: 'json' };
 import GeneralJws from '../json-schemas/general-jws.json' assert { type: 'json' };
 import GenericSignaturePayload from '../json-schemas/signature-payloads/generic-signature-payload.json' assert { type: 'json' };
@@ -44,6 +45,7 @@ import RecordsDelete from '../json-schemas/interface-methods/records-delete.json
 import RecordsFilter from '../json-schemas/interface-methods/records-filter.json' assert { type: 'json' };
 import RecordsQuery from '../json-schemas/interface-methods/records-query.json' assert { type: 'json' };
 import RecordsRead from '../json-schemas/interface-methods/records-read.json' assert { type: 'json' };
+import RecordsSubscribe from '../json-schemas/interface-methods/records-subscribe.json' assert { type: 'json' };
 import RecordsWrite from '../json-schemas/interface-methods/records-write.json' assert { type: 'json' };
 import RecordsWriteSignaturePayload from '../json-schemas/signature-payloads/records-write-signature-payload.json' assert { type: 'json' };
 import RecordsWriteUnidentified from '../json-schemas/interface-methods/records-write-unidentified.json' assert { type: 'json' };
@@ -54,10 +56,12 @@ const schemas = {
   AuthorizationOwner,
   RecordsDelete,
   RecordsQuery,
+  RecordsSubscribe,
   RecordsWrite,
   RecordsWriteUnidentified,
   EventsFilter,
   EventsGet,
+  EventsSubscribe,
   EventsQuery,
   Definitions,
   GeneralJwk,
@@ -92,4 +96,4 @@ const moduleCode = standaloneCode(ajv);
 const __dirname = url.fileURLToPath(new URL('.', import.meta.url));
 
 await mkdirp(path.join(__dirname, '../generated'));
-fs.writeFileSync(path.join(__dirname, '../generated/precompiled-validators.js'), moduleCode);
+fs.writeFileSync(path.join(__dirname, '../generated/precompiled-validators.js'), moduleCode);

json-schemas/interface-methods/events-subscribe.json

@@ -0,0 +1,47 @@
+{
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "$id": "https://identity.foundation/dwn/json-schemas/events-subscribe.json",
+    "type": "object",
+    "additionalProperties": false,
+    "required": [
+      "descriptor"
+    ],
+    "properties": {
+      "authorization": {
+        "$ref": "https://identity.foundation/dwn/json-schemas/authorization.json"
+      },
+      "descriptor": {
+        "type": "object",
+        "additionalProperties": false,
+        "required": [
+          "interface",
+          "method",
+          "messageTimestamp",
+          "filters"
+        ],
+        "properties": {
+          "interface": {
+            "enum": [
+                "Events"
+            ],
+             "type": "string"
+          },
+          "method": {
+            "enum": [
+              "Subscribe"
+            ],
+            "type": "string"
+          },
+          "messageTimestamp": {
+            "type": "string"
+          },
+          "filters": {
+            "type": "array",
+            "items": {
+              "$ref": "https://identity.foundation/dwn/json-schemas/events-filter.json"
+            }
+          }
+        }
+      }
+    }
+  }

json-schemas/interface-methods/protocol-rule-set.json

@@ -66,6 +66,7 @@
                 "enum": [
                   "delete",
                   "query",
+                  "subscribe",
                   "read",
                   "update",
                   "write"

json-schemas/interface-methods/records-subscribe.json

@@ -0,0 +1,44 @@
+{
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "$id": "https://identity.foundation/dwn/json-schemas/records-subscribe.json",
+  "type": "object",
+  "additionalProperties": false,
+  "required": [
+    "descriptor"
+  ],
+  "properties": {
+    "authorization": {
+      "$ref": "https://identity.foundation/dwn/json-schemas/authorization-delegated-grant.json"
+    },
+    "descriptor": {
+      "type": "object",
+      "additionalProperties": false,
+      "required": [
+        "interface",
+        "method",
+        "messageTimestamp",
+        "filter"
+      ],
+      "properties": {
+        "interface": {
+          "enum": [
+            "Records"
+          ],
+          "type": "string"
+        },
+        "method": {
+          "enum": [
+            "Subscribe"
+          ],
+          "type": "string"
+        },
+        "messageTimestamp": {
+          "$ref": "https://identity.foundation/dwn/json-schemas/defs.json#/definitions/date-time"
+        },
+        "filter": {
+          "$ref": "https://identity.foundation/dwn/json-schemas/records-filter.json"
+        }
+      }
+    }
+  }
+}

json-schemas/permissions/permissions-definitions.json

@@ -28,6 +28,9 @@
         },
         {
           "$ref": "https://identity.foundation/dwn/json-schemas/permissions/scopes.json#/definitions/records-query-scope"
+        },
+        {
+          "$ref": "https://identity.foundation/dwn/json-schemas/permissions/scopes.json#/definitions/records-subscribe-scope"
         }
       ]
     },

json-schemas/permissions/scopes.json

@@ -106,6 +106,24 @@
           "type": "string"
         }
       }
+    },
+    "records-subscribe-scope": {
+      "type": "object",
+      "required": [
+        "interface",
+        "method"
+      ],
+      "properties": {
+        "interface": {
+          "const": "Records"
+        },
+        "method": {
+          "const": "Subscribe"
+        },
+        "protocol": {
+          "type": "string"
+        }
+      }
     }
   }
 }

src/core/dwn-error.ts

@@ -25,6 +25,7 @@ export enum DwnErrorCode {
   DidNotValid = 'DidNotValid',
   DidResolutionFailed = 'DidResolutionFailed',
   Ed25519InvalidJwk = 'Ed25519InvalidJwk',
+  EventStreamSubscriptionNotSupported = 'EventStreamSubscriptionNotSupported',
   GeneralJwsVerifierGetPublicKeyNotFound = 'GeneralJwsVerifierGetPublicKeyNotFound',
   GeneralJwsVerifierInvalidSignature = 'GeneralJwsVerifierInvalidSignature',
   GrantAuthorizationGrantExpired = 'GrantAuthorizationGrantExpired',
@@ -88,6 +89,7 @@ export enum DwnErrorCode {
   RecordsGrantAuthorizationDeleteProtocolScopeMismatch = 'RecordsGrantAuthorizationDeleteProtocolScopeMismatch',
   RecordsGrantAuthorizationQueryProtocolScopeMismatch = 'RecordsGrantAuthorizationQueryProtocolScopeMismatch',
   RecordsGrantAuthorizationScopeContextIdMismatch = 'RecordsGrantAuthorizationScopeContextIdMismatch',
+  RecordsGrantAuthorizationSubscribeProtocolScopeMismatch = 'RecordsGrantAuthorizationSubscribeProtocolScopeMismatch',
   RecordsGrantAuthorizationScopeNotProtocol = 'RecordsGrantAuthorizationScopeNotProtocol',
   RecordsGrantAuthorizationScopeProtocolMismatch = 'RecordsGrantAuthorizationScopeProtocolMismatch',
   RecordsGrantAuthorizationScopeProtocolPathMismatch = 'RecordsGrantAuthorizationScopeProtocolPathMismatch',
@@ -99,6 +101,9 @@ export enum DwnErrorCode {
   RecordsQueryFilterMissingRequiredProperties = 'RecordsQueryFilterMissingRequiredProperties',
   RecordsReadReturnedMultiple = 'RecordsReadReturnedMultiple',
   RecordsReadAuthorizationFailed = 'RecordsReadAuthorizationFailed',
+  RecordsSubscribeFilterMissingRequiredProperties = 'RecordsSubscribeFilterMissingRequiredProperties',
+  RecordsSubscribeUnauthorized = 'RecordsSubscribeUnauthorized',
+  RecordsSubscribeUnknownError = 'RecordsSubscribeUnknownError',
   RecordsSchemasDerivationSchemeMissingSchema = 'RecordsSchemasDerivationSchemeMissingSchema',
   RecordsValidateIntegrityDelegatedGrantAndIdExistenceMismatch = 'RecordsValidateIntegrityDelegatedGrantAndIdExistenceMismatch',
   RecordsValidateIntegrityGrantedToAndSignerMismatch = 'RecordsValidateIntegrityGrantedToAndSignerMismatch',

src/core/message-reply.ts

@@ -2,7 +2,7 @@ import type { MessagesGetReplyEntry } from '../types/messages-types.js';
 import type { ProtocolsConfigureMessage } from '../types/protocols-types.js';
 import type { Readable } from 'readable-stream';
 import type { RecordsWriteMessage } from '../types/records-types.js';
-import type { GenericMessageReply, QueryResultEntry } from '../types/message-types.js';
+import type { GenericMessageReply, GenericMessageSubscription, QueryResultEntry } from '../types/message-types.js';
 
 export function messageReplyFromError(e: unknown, code: number): GenericMessageReply {
 
@@ -39,4 +39,9 @@ export type UnionMessageReply = GenericMessageReply & {
    * Mutually exclusive with `record`.
    */
   cursor?: string;
+
+  /**
+   * A subscription object if a subscription was requested.
+   */
+  subscription?: GenericMessageSubscription;
 };

src/core/protocol-authorization.ts

@@ -3,6 +3,7 @@ import type { MessageStore } from '../types/message-store.js';
 import type { RecordsDelete } from '../interfaces/records-delete.js';
 import type { RecordsQuery } from '../interfaces/records-query.js';
 import type { RecordsRead } from '../interfaces/records-read.js';
+import type { RecordsSubscribe } from '../interfaces/records-subscribe.js';
 import type { RecordsWriteMessage } from '../types/records-types.js';
 import type { ProtocolActionRule, ProtocolDefinition, ProtocolRuleSet, ProtocolsConfigureMessage, ProtocolType, ProtocolTypes } from '../types/protocols-types.js';
 
@@ -152,6 +153,47 @@ export class ProtocolAuthorization {
     );
   }
 
+  public static async authorizeSubscription(
+    tenant: string,
+    incomingMessage: RecordsSubscribe,
+    messageStore: MessageStore,
+  ): Promise<void> {
+    // validate that required properties exist in subscription filter
+    const { protocol, protocolPath, contextId } = incomingMessage.message.descriptor.filter;
+
+    // fetch the protocol definition
+    const protocolDefinition = await ProtocolAuthorization.fetchProtocolDefinition(
+      tenant,
+      protocol!, // `authorizeSubscription` is only called if `protocol` is present
+      messageStore,
+    );
+
+    // get the rule set for the inbound message
+    const inboundMessageRuleSet = ProtocolAuthorization.getRuleSet(
+      protocolPath!, // presence of `protocolPath` is verified in `parse()`
+      protocolDefinition,
+    );
+
+    // If the incoming message has `protocolRole` in the descriptor, validate the invoked role
+    await ProtocolAuthorization.verifyInvokedRole(
+      tenant,
+      incomingMessage,
+      protocol!,
+      contextId,
+      protocolDefinition,
+      messageStore,
+    );
+
+    // verify method invoked against the allowed actions
+    await ProtocolAuthorization.verifyAllowedActions(
+      tenant,
+      incomingMessage,
+      inboundMessageRuleSet,
+      [], // ancestor chain is not relevant to subscriptions
+      messageStore,
+    );
+  }
+
   /**
    * Performs protocol-based authorization against the incoming RecordsQuery message.
    * @throws {Error} if authorization fails.
@@ -423,7 +465,7 @@ export class ProtocolAuthorization {
    */
   private static async verifyInvokedRole(
     tenant: string,
-    incomingMessage: RecordsDelete | RecordsQuery | RecordsRead | RecordsWrite,
+    incomingMessage: RecordsDelete | RecordsQuery | RecordsRead | RecordsSubscribe | RecordsWrite,
     protocolUri: string,
     contextId: string | undefined,
     protocolDefinition: ProtocolDefinition,
@@ -481,7 +523,7 @@ export class ProtocolAuthorization {
    */
   private static async getActionsSeekingARuleMatch(
     tenant: string,
-    incomingMessage: RecordsDelete | RecordsQuery | RecordsRead | RecordsWrite,
+    incomingMessage: RecordsDelete | RecordsQuery | RecordsRead | RecordsSubscribe | RecordsWrite,
     messageStore: MessageStore,
   ): Promise<ProtocolAction[]> {
 
@@ -495,6 +537,9 @@ export class ProtocolAuthorization {
     case DwnMethodName.Read:
       return [ProtocolAction.Read];
 
+    case DwnMethodName.Subscribe:
+      return [ProtocolAction.Subscribe];
+
     case DwnMethodName.Write:
       const incomingRecordsWrite = incomingMessage as RecordsWrite;
       if (await incomingRecordsWrite.isInitialWrite()) {
@@ -519,7 +564,7 @@ export class ProtocolAuthorization {
    */
   private static async verifyAllowedActions(
     tenant: string,
-    incomingMessage: RecordsDelete | RecordsQuery | RecordsRead | RecordsWrite,
+    incomingMessage: RecordsDelete | RecordsQuery | RecordsRead | RecordsSubscribe | RecordsWrite,
     inboundMessageRuleSet: ProtocolRuleSet,
     ancestorMessageChain: RecordsWriteMessage[],
     messageStore: MessageStore,

src/core/records-grant-authorization.ts

@@ -1,7 +1,7 @@
 import type { MessageStore } from '../types/message-store.js';
 import type { RecordsPermissionScope } from '../types/permissions-grant-descriptor.js';
 import type { PermissionsGrantMessage, RecordsPermissionsGrantMessage } from '../types/permissions-types.js';
-import type { RecordsDeleteMessage, RecordsQueryMessage, RecordsReadMessage, RecordsWriteMessage } from '../types/records-types.js';
+import type { RecordsDeleteMessage, RecordsQueryMessage, RecordsReadMessage, RecordsSubscribeMessage, RecordsWriteMessage } from '../types/records-types.js';
 
 import { GrantAuthorization } from './grant-authorization.js';
 import { PermissionsConditionPublication } from '../types/permissions-grant-descriptor.js';
@@ -96,6 +96,40 @@ export class RecordsGrantAuthorization {
     }
   }
 
+  /**
+   * Authorizes the scope of a PermissionsGrant for RecordsSubscribe.
+   * @param messageStore Used to check if the grant has been revoked.
+   */
+  public static async authorizeSubscribe(input: {
+    recordsSubscribeMessage: RecordsSubscribeMessage,
+    expectedGrantedToInGrant: string,
+    expectedGrantedForInGrant: string,
+    permissionsGrantMessage: PermissionsGrantMessage,
+    messageStore: MessageStore,
+  }): Promise<void> {
+    const {
+      recordsSubscribeMessage, expectedGrantedToInGrant, expectedGrantedForInGrant, permissionsGrantMessage, messageStore
+    } = input;
+
+    await GrantAuthorization.performBaseValidation({
+      incomingMessage: recordsSubscribeMessage,
+      expectedGrantedToInGrant,
+      expectedGrantedForInGrant,
+      permissionsGrantMessage,
+      messageStore
+    });
+
+    // If the grant specifies a protocol, the query must specify the same protocol.
+    const protocolInGrant = (permissionsGrantMessage.descriptor.scope as RecordsPermissionScope).protocol;
+    const protocolInSubscribe = recordsSubscribeMessage.descriptor.filter.protocol;
+    if (protocolInGrant !== undefined && protocolInSubscribe !== protocolInGrant) {
+      throw new DwnError(
+        DwnErrorCode.RecordsGrantAuthorizationSubscribeProtocolScopeMismatch,
+        `Grant protocol scope ${protocolInGrant} does not match protocol in subscribe ${protocolInSubscribe}`
+      );
+    }
+  }
+
   /**
    * Authorizes the scope of a PermissionsGrant for RecordsDelete.
    * @param messageStore Used to check if the grant has been revoked.