records subscribe initial

build/compile-validators.js

@@ -46,6 +46,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' };
@@ -56,6 +57,7 @@ const schemas = {
   AuthorizationOwner,
   RecordsDelete,
   RecordsQuery,
+  RecordsSubscribe,
   RecordsWrite,
   RecordsWriteUnidentified,
   EventsFilter,

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

@@ -91,6 +91,7 @@ export enum DwnErrorCode {
   RecordsGrantAuthorizationDeleteProtocolScopeMismatch = 'RecordsGrantAuthorizationDeleteProtocolScopeMismatch',
   RecordsGrantAuthorizationQueryProtocolScopeMismatch = 'RecordsGrantAuthorizationQueryProtocolScopeMismatch',
   RecordsGrantAuthorizationScopeContextIdMismatch = 'RecordsGrantAuthorizationScopeContextIdMismatch',
+  RecordsGrantAuthorizationSubscribeProtocolScopeMismatch = 'RecordsGrantAuthorizationSubscribeProtocolScopeMismatch',
   RecordsGrantAuthorizationScopeNotProtocol = 'RecordsGrantAuthorizationScopeNotProtocol',
   RecordsGrantAuthorizationScopeProtocolMismatch = 'RecordsGrantAuthorizationScopeProtocolMismatch',
   RecordsGrantAuthorizationScopeProtocolPathMismatch = 'RecordsGrantAuthorizationScopeProtocolPathMismatch',
@@ -102,6 +103,10 @@ export enum DwnErrorCode {
   RecordsQueryFilterMissingRequiredProperties = 'RecordsQueryFilterMissingRequiredProperties',
   RecordsReadReturnedMultiple = 'RecordsReadReturnedMultiple',
   RecordsReadAuthorizationFailed = 'RecordsReadAuthorizationFailed',
+  RecordsSubscribeEventStreamUnimplemented = 'RecordsSubscribeEventStreamUnimplemented',
+  RecordsSubscribeFilterMissingRequiredProperties = 'RecordsSubscribeFilterMissingRequiredProperties',
+  RecordsSubscribeUnauthorized = 'RecordsSubscribeUnauthorized',
+  RecordsSubscribeUnknownError = 'RecordsSubscribeUnknownError',
   RecordsSchemasDerivationSchemeMissingSchema = 'RecordsSchemasDerivationSchemeMissingSchema',
   RecordsValidateIntegrityDelegatedGrantAndIdExistenceMismatch = 'RecordsValidateIntegrityDelegatedGrantAndIdExistenceMismatch',
   RecordsValidateIntegrityGrantedToAndSignerMismatch = 'RecordsValidateIntegrityGrantedToAndSignerMismatch',

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.

src/dwn.ts

@@ -10,7 +10,7 @@ import type { GenericMessage, GenericMessageReply, MessageOptions } from './type
 import type { MessagesGetMessage, MessagesGetReply } from './types/messages-types.js';
 import type { PermissionsGrantMessage, PermissionsRequestMessage, PermissionsRevokeMessage } from './types/permissions-types.js';
 import type { ProtocolsConfigureMessage, ProtocolsQueryMessage, ProtocolsQueryReply } from './types/protocols-types.js';
-import type { RecordsDeleteMessage, RecordsQueryMessage, RecordsQueryReply, RecordsReadMessage, RecordsReadReply, RecordsWriteMessage, RecordsWriteMessageOptions } from './types/records-types.js';
+import type { RecordsDeleteMessage, RecordsQueryMessage, RecordsQueryReply, RecordsReadMessage, RecordsReadReply, RecordsSubscribeMessage, RecordsSubscribeMessageOptions, RecordsSubscribeReply, RecordsWriteMessage, RecordsWriteMessageOptions } from './types/records-types.js';
 
 import { AllowAllTenantGate } from './core/tenant-gate.js';
 import { DidResolver } from './did/did-resolver.js';
@@ -28,6 +28,7 @@ import { ProtocolsQueryHandler } from './handlers/protocols-query.js';
 import { RecordsDeleteHandler } from './handlers/records-delete.js';
 import { RecordsQueryHandler } from './handlers/records-query.js';
 import { RecordsReadHandler } from './handlers/records-read.js';
+import { RecordsSubscribeHandler } from './handlers/records-subscribe.js';
 import { RecordsWriteHandler } from './handlers/records-write.js';
 import { DwnInterfaceName, DwnMethodName } from './enums/dwn-interface-method.js';
 
@@ -43,6 +44,7 @@ export class Dwn {
   private constructor(config: DwnConfig) {
     this.didResolver = config.didResolver!;
     this.tenantGate = config.tenantGate!;
+    this.eventStream = config.eventStream!;
     this.messageStore = config.messageStore;
     this.dataStore = config.dataStore;
     this.eventLog = config.eventLog;
@@ -112,6 +114,11 @@ export class Dwn {
         this.messageStore,
         this.dataStore
       ),
+      [DwnInterfaceName.Records + DwnMethodName.Subscribe]: new RecordsSubscribeHandler(
+        this.didResolver,
+        this.messageStore,
+        this.eventStream
+      ),
       [DwnInterfaceName.Records + DwnMethodName.Write]: new RecordsWriteHandler(
         this.didResolver,
         this.messageStore,
@@ -165,6 +172,8 @@ export class Dwn {
   public async processMessage(tenant: string, rawMessage: PermissionsRevokeMessage): Promise<GenericMessageReply>;
   public async processMessage(tenant: string, rawMessage: RecordsDeleteMessage): Promise<GenericMessageReply>;
   public async processMessage(tenant: string, rawMessage: RecordsQueryMessage): Promise<RecordsQueryReply>;
+  public async processMessage(
+    tenant: string, rawMessage: RecordsSubscribeMessage, options: RecordsSubscribeMessageOptions): Promise<RecordsSubscribeReply>;
   public async processMessage(tenant: string, rawMessage: RecordsReadMessage): Promise<RecordsReadReply>;
   public async processMessage(tenant: string, rawMessage: RecordsWriteMessage, options?: RecordsWriteMessageOptions): Promise<GenericMessageReply>;
   public async processMessage(tenant: string, rawMessage: unknown, options?: MessageOptions): Promise<UnionMessageReply>;

src/handlers/events-subscribe.ts

@@ -65,4 +65,4 @@ export class EventsSubscribeHandler implements MethodHandler {
       subscription,
     };
   }
-}
+}