feat(sdk): Explicit encryption keys (#3284)

juslesan · web-flow · commit b90adea4121b · 2025-12-12T15:40:57.000+02:00
## Summary

Make it possible to provide explicit encryption keys to the
StreamrClient. These keys will be used in a given stream for all
publishers as the encryption key. Setting the explicit keys disables the
group key exchange. Therefore, all participants in the stream need to
agree on what key is being used for encryption and decryption to work
properly.
 
## Changes

- Added new configuration field `keys` to the `encryption` section
- The `GroupKeyManager#fetchKey` returns explicitly defined keys if they
are defined
- `Publisher` initializes the `GroupKeyQueue` with the explicit keys if
they are defined
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -15,6 +15,7 @@ Changes before Tatum release are not documented in this file.
 - Proxy connections now support bidirectionality and it is the default behavior (https://github.com/streamr-dev/network/pull/3260)
 - Add `StreamrClient#findProxyNodes()` function for discovering proxy nodes via Operator nodes (https://github.com/streamr-dev/network/pull/3257)
 - Add `StreamrClient#publishRaw()` for publishing raw messages (https://github.com/streamr-dev/network/pull/3280)
+- Add new `keys` configuration to the `encryption` section (https://github.com/streamr-dev/network/pull/3284)
 
 #### Changed
 
diff --git a/packages/sdk/src/Config.ts b/packages/sdk/src/Config.ts
@@ -390,6 +390,11 @@ export interface StreamrClientConfig {
          * Note that subscribers will still accept unencrypted (public) data despite this setting.
          */
         requireQuantumResistantEncryption?: boolean
+
+        /**
+         * If this is defined only these encryption keys will be used. This will disable the Streamr key-exchange.
+         */
+        keys?: Record<string, { id: string, data: HexString }>
     }
 
     contracts?: {
diff --git a/packages/sdk/src/StreamrClientError.ts b/packages/sdk/src/StreamrClientError.ts
@@ -19,7 +19,8 @@ export type StreamrClientErrorCode =
     'UNKNOWN_ERROR' |
     'ASSERTION_FAILED' |
     'SIGNATURE_POLICY_VIOLATION' |
-    'ENCRYPTION_POLICY_VIOLATION'
+    'ENCRYPTION_POLICY_VIOLATION' |
+    'UNEXPECTED_INPUT'
 
 export class StreamrClientError extends Error {
 
diff --git a/packages/sdk/src/config.schema.json b/packages/sdk/src/config.schema.json
@@ -375,6 +375,15 @@
                 "requireQuantumResistantEncryption": {
                     "type": "boolean",
                     "default": false
+                },
+                "keys": {
+                    "type": "object",
+                    "propertyNames": {
+                        "$ref": "#/definitions/streamIdOrPath"
+                    },
+                    "additionalProperties": {
+                        "$ref": "#/definitions/encryptionKey"
+                    }
                 }
             },
             "default": {}
@@ -563,6 +572,25 @@
                     "type": "number"
                 }
             }
+        },
+        "streamIdOrPath": {
+            "type": "string"
+        },
+        "encryptionKey": {
+            "type": "object",
+            "properties": {
+                "id": {
+                    "type": "string"
+                },
+                "data": {
+                    "type": "string",
+                    "format": "hex-string"
+                }
+            },
+            "required": [
+                "id",
+                "data"
+            ]
         }
 
     }
diff --git a/packages/sdk/src/encryption/GroupKeyManager.ts b/packages/sdk/src/encryption/GroupKeyManager.ts
@@ -1,4 +1,4 @@
-import { StreamID, StreamPartID, UserID, waitForEvent } from '@streamr/utils'
+import { hexToBinary, StreamID, StreamPartID, StreamPartIDUtils, UserID, waitForEvent } from '@streamr/utils'
 import crypto from 'crypto'
 import { Lifecycle, inject, scoped } from 'tsyringe'
 import { Identity, IdentityInjectionToken } from '../identity/Identity'
@@ -9,12 +9,35 @@ import { uuid } from '../utils/uuid'
 import { GroupKey } from './GroupKey'
 import { LocalGroupKeyStore } from './LocalGroupKeyStore'
 import { SubscriberKeyExchange } from './SubscriberKeyExchange'
+import { StreamIDBuilder } from '../StreamIDBuilder'
+import { createLazyMap, Mapping } from '../utils/Mapping'
+import { StreamrClientError } from '../StreamrClientError'
+
+/**
+ * Gets an explicit encryption key from config for a given stream.
+ * Returns undefined if no key is configured for the stream.
+ */
+export const getExplicitKey = async (
+    streamId: StreamID,
+    streamIdBuilder: StreamIDBuilder,
+    config: StrictStreamrClientConfig['encryption']
+): Promise<GroupKey | undefined> => {
+    if (config.keys !== undefined) {
+        for (const entry of Object.entries(config.keys)) {
+            if (await streamIdBuilder.toStreamID(entry[0]) === streamId) {
+                return new GroupKey(entry[1].id, Buffer.from(hexToBinary(entry[1].data)))
+            }
+        }
+    }
+    return undefined
+}
 
 @scoped(Lifecycle.ContainerScoped)
 export class GroupKeyManager {
 
     private readonly subscriberKeyExchange: SubscriberKeyExchange
     private readonly localGroupKeyStore: LocalGroupKeyStore
+    private readonly explicitKeys?: Mapping<StreamID, GroupKey | undefined>
     private readonly config: Pick<StrictStreamrClientConfig, 'encryption'>
     private readonly identity: Identity
     private readonly eventEmitter: StreamrClientEventEmitter
@@ -23,6 +46,7 @@ export class GroupKeyManager {
     constructor(
         subscriberKeyExchange: SubscriberKeyExchange,
         localGroupKeyStore: LocalGroupKeyStore,
+        streamIdBuilder: StreamIDBuilder,
         @inject(ConfigInjectionToken) config: Pick<StrictStreamrClientConfig, 'encryption'>,
         @inject(IdentityInjectionToken) identity: Identity,
         eventEmitter: StreamrClientEventEmitter,
@@ -34,16 +58,35 @@ export class GroupKeyManager {
         this.identity = identity
         this.eventEmitter = eventEmitter
         this.destroySignal = destroySignal
+        if (config.encryption.keys !== undefined) {
+            this.explicitKeys = createLazyMap({
+                valueFactory: async (streamId: StreamID) => {
+                    return getExplicitKey(streamId, streamIdBuilder, config.encryption)
+                }
+            })
+        }
     }
 
     async fetchKey(streamPartId: StreamPartID, groupKeyId: string, publisherId: UserID): Promise<GroupKey> {
-        // 1st try: local storage
+        // If explicit keys are defined only those keys are used.
+        if (this.explicitKeys !== undefined) {
+            const explicitKey = await this.explicitKeys.get(StreamPartIDUtils.getStreamID(streamPartId))
+            if (explicitKey !== undefined) {
+                return explicitKey
+            }
+            throw new StreamrClientError(
+                `No encryption key available for stream part ID: groupKeyId=${groupKeyId}, streamPartId=${streamPartId}`,
+                'UNEXPECTED_INPUT'
+            )
+        }
+
+        // 2nd try: local storage
         let groupKey = await this.localGroupKeyStore.get(groupKeyId, publisherId)
         if (groupKey !== undefined) {
             return groupKey
         }
 
-        // 2nd try: Streamr key-exchange
+        // 3rd try: Streamr key-exchange
         await this.subscriberKeyExchange.requestGroupKey(groupKeyId, publisherId, streamPartId)
         const groupKeyIds = await waitForEvent(
             // TODO remove "as any" type casing in NET-889
diff --git a/packages/sdk/src/encryption/PublisherKeyExchange.ts b/packages/sdk/src/encryption/PublisherKeyExchange.ts
@@ -72,19 +72,22 @@ export class PublisherKeyExchange {
         this.identity = identity
         this.logger = loggerFactory.createLogger(module)
         this.config = config
-        networkNodeFacade.once('start', async () => {
-            networkNodeFacade.addMessageListener((msg: StreamMessage) => this.onMessage(msg))
-            this.logger.debug('Started')
-        })
-        eventEmitter.on('messagePublished', (msg) => {
-            if (msg.signatureType === SignatureType.ERC_1271) {
-                const publisherId = msg.getPublisherId()
-                if (!this.erc1271Publishers.has(publisherId)) {
-                    logger.debug('Add ERC-1271 publisher', { publisherId })
-                    this.erc1271Publishers.add(publisherId)
+        // Setting explicit keys disables the key-exchange
+        if (config.encryption.keys === undefined) {
+            networkNodeFacade.once('start', async () => {
+                networkNodeFacade.addMessageListener((msg: StreamMessage) => this.onMessage(msg))
+                this.logger.debug('Started')
+            })
+            eventEmitter.on('messagePublished', (msg) => {
+                if (msg.signatureType === SignatureType.ERC_1271) {
+                    const publisherId = msg.getPublisherId()
+                    if (!this.erc1271Publishers.has(publisherId)) {
+                        logger.debug('Add ERC-1271 publisher', { publisherId })
+                        this.erc1271Publishers.add(publisherId)
+                    }
                 }
-            }
-        })
+            })
+        }
     }
 
     private async onMessage(request: StreamMessage): Promise<void> {
diff --git a/packages/sdk/src/encryption/SubscriberKeyExchange.ts b/packages/sdk/src/encryption/SubscriberKeyExchange.ts
@@ -20,6 +20,7 @@ import { EncryptionUtil } from './EncryptionUtil'
 import { AsymmetricEncryptionType, ContentType, EncryptionType, GroupKeyRequest, GroupKeyResponse, SignatureType } from '@streamr/trackerless-network'
 import { KeyExchangeKeyPair } from './KeyExchangeKeyPair'
 import { createCompliantExchangeKeys } from '../utils/encryptionCompliance'
+import { StreamrClientError } from '../StreamrClientError'
 
 const MAX_PENDING_REQUEST_COUNT = 50000 // just some limit, we can tweak the number if needed
 
@@ -62,14 +63,24 @@ export class SubscriberKeyExchange {
         this.subscriber = subscriber
         this.identity = identity
         this.logger = loggerFactory.createLogger(module)
-        this.ensureStarted = pOnce(async () => {
-            this.keyPair = await createCompliantExchangeKeys(identity, config)
-            networkNodeFacade.addMessageListener((msg: StreamMessage) => this.onMessage(msg))
-            this.logger.debug('Started')
-        })
-        this.requestGroupKey = withThrottling((groupKeyId: string, publisherId: UserID, streamPartId: StreamPartID) => {
-            return this.doRequestGroupKey(groupKeyId, publisherId, streamPartId)
-        }, config.encryption.maxKeyRequestsPerSecond)
+        // Setting explicit keys disables the key-exchange
+        if (config.encryption.keys === undefined) {
+            this.ensureStarted = pOnce(async () => {
+                this.keyPair = await createCompliantExchangeKeys(identity, config)
+                networkNodeFacade.addMessageListener((msg: StreamMessage) => this.onMessage(msg))
+                this.logger.debug('Started')
+            })
+            this.requestGroupKey = withThrottling((groupKeyId: string, publisherId: UserID, streamPartId: StreamPartID) => {
+                return this.doRequestGroupKey(groupKeyId, publisherId, streamPartId)
+            }, config.encryption.maxKeyRequestsPerSecond)
+        } else {
+            this.ensureStarted = async () => {
+                throw new StreamrClientError('Assertion failed', 'ASSERTION_FAILED')
+            }
+            this.requestGroupKey = async () => {
+                throw new StreamrClientError('Assertion failed', 'ASSERTION_FAILED')
+            }
+        }
     }
 
     private async doRequestGroupKey(groupKeyId: string, publisherId: UserID, streamPartId: StreamPartID): Promise<void> {
diff --git a/packages/sdk/src/generated/validateConfig.js b/packages/sdk/src/generated/validateConfig.js
diff --git a/packages/sdk/src/publish/GroupKeyQueue.ts b/packages/sdk/src/publish/GroupKeyQueue.ts
@@ -19,13 +19,14 @@ export class GroupKeyQueue {
     static async createInstance(
         streamId: StreamID,
         identity: Identity,
-        groupKeyManager: GroupKeyManager
+        groupKeyManager: GroupKeyManager,
+        currentGroupKey?: GroupKey
     ): Promise<GroupKeyQueue> {
         const instance = new GroupKeyQueue(streamId, identity, groupKeyManager)
-        instance.currentGroupKey = await instance.groupKeyManager.fetchLatestEncryptionKey(
+        instance.currentGroupKey = currentGroupKey ?? await instance.groupKeyManager.fetchLatestEncryptionKey(
             await identity.getUserId(),
             streamId,
-        )
+        ) ?? undefined
         return instance
     }
 
diff --git a/packages/sdk/src/publish/Publisher.ts b/packages/sdk/src/publish/Publisher.ts
@@ -7,7 +7,7 @@ import { NetworkNodeFacade } from '../NetworkNodeFacade'
 import { StreamIDBuilder } from '../StreamIDBuilder'
 import { StreamrClientError } from '../StreamrClientError'
 import { StreamRegistry } from '../contracts/StreamRegistry'
-import { GroupKeyManager } from '../encryption/GroupKeyManager'
+import { getExplicitKey, GroupKeyManager } from '../encryption/GroupKeyManager'
 import { StreamMessage } from '../protocol/StreamMessage'
 import { MessageSigner } from '../signature/MessageSigner'
 import { SignatureValidator } from '../signature/SignatureValidator'
@@ -80,7 +80,8 @@ export class Publisher {
         })
         this.groupKeyQueues = createLazyMap({
             valueFactory: async (streamId) => {
-                return GroupKeyQueue.createInstance(streamId, this.identity, groupKeyManager)
+                const explicitKey = await getExplicitKey(streamId, this.streamIdBuilder, this.config.encryption)
+                return await GroupKeyQueue.createInstance(streamId, this.identity, groupKeyManager, explicitKey)
             }
         })
     }
diff --git a/packages/sdk/test/integration/explicit-encryption-keys.test.ts b/packages/sdk/test/integration/explicit-encryption-keys.test.ts
@@ -0,0 +1,51 @@
+import 'reflect-metadata'
+
+import { GroupKey } from '../../src/encryption/GroupKey'
+import { StreamPermission } from '../../src/permission'
+import { FakeEnvironment } from '../test-utils/fake/FakeEnvironment'
+import { createTestStream, getLocalGroupKeyStore } from '../test-utils/utils'
+import { nextValue } from '../../src/utils/iterators'
+import { binaryToHex, toUserId } from '@streamr/utils'
+
+describe('explicit encryption keys', () => {
+
+    it('happy path', async () => {
+        const environment = new FakeEnvironment()
+        const stream = await createTestStream(environment.createClient(), module)
+        const explicitKey = GroupKey.generate()
+        const config = {
+            encryption: {
+                keys: {
+                    [stream.id]: {
+                        id: explicitKey.id,
+                        data: binaryToHex(explicitKey.data)
+                    }
+                }
+            }
+        }
+        const publisher = environment.createClient(config)
+        const subscriber = environment.createClient(config)
+        await stream.grantPermissions({
+            userId: await publisher.getUserId(),
+            permissions: [StreamPermission.PUBLISH]
+        }, {
+            userId: await subscriber.getUserId(),
+            permissions: [StreamPermission.SUBSCRIBE]
+        })
+        const subscription = await subscriber.subscribe(stream.id)
+        await publisher.publish(stream.id, {
+            foo: 'bar'
+        })
+        const message = await nextValue(subscription[Symbol.asyncIterator]())
+        expect(message!.content).toEqual({
+            foo: 'bar'
+        })
+        const store = getLocalGroupKeyStore(toUserId(await publisher.getUserId()))
+        for (const ownerId of [await publisher.getUserId(), await subscriber.getUserId()]) {
+            const key = await store.get(message!.streamMessage.groupKeyId!, toUserId(ownerId))
+            expect(key).toBeUndefined()
+        }
+        await publisher.destroy()
+        await subscriber.destroy()
+    })
+})
diff --git a/packages/sdk/test/test-utils/utils.ts b/packages/sdk/test/test-utils/utils.ts
@@ -56,6 +56,7 @@ import { counterId } from '../../src/utils/utils'
 import { FakeEnvironment } from './../test-utils/fake/FakeEnvironment'
 import { FakeStorageNode } from './../test-utils/fake/FakeStorageNode'
 import { addAfterFn } from './jest-utils'
+import { StreamIDBuilder } from '../../src/StreamIDBuilder'
 
 const logger = new Logger(module)
 
@@ -216,12 +217,13 @@ export const createStreamRegistry = (opts?: {
 }
 
 export const createGroupKeyManager = async (
-    groupKeyStore: LocalGroupKeyStore = mock<LocalGroupKeyStore>(),
-    identity?: Identity 
+    identity: Identity,
+    groupKeyStore: LocalGroupKeyStore = mock<LocalGroupKeyStore>()
 ): Promise<GroupKeyManager> => {
     return new GroupKeyManager(
         mock<SubscriberKeyExchange>(),
         groupKeyStore,
+        new StreamIDBuilder(identity),
         {
             encryption: {
                 maxKeyRequestsPerSecond: 10,
@@ -230,19 +232,20 @@ export const createGroupKeyManager = async (
                 requireQuantumResistantKeyExchange: false,
                 requireQuantumResistantSignatures: false,
                 requireQuantumResistantEncryption: false,
+                keys: undefined as any
             }
         },
-        identity ?? await createRandomIdentity(),
+        identity,
         new StreamrClientEventEmitter(),
-        new DestroySignal()
+        new DestroySignal(),
     )
 }
 
 export const createGroupKeyQueue = async (identity: Identity, current?: GroupKey, next?: GroupKey): Promise<GroupKeyQueue> => {
     const queue = await GroupKeyQueue.createInstance(
         undefined as any,
         identity,
-        await createGroupKeyManager(undefined, identity)
+        await createGroupKeyManager(identity)
     )
     if (current !== undefined) {
         await queue.rekey(current)
diff --git a/packages/sdk/test/unit/Decrypt.test.ts b/packages/sdk/test/unit/Decrypt.test.ts
@@ -43,7 +43,7 @@ describe('Decrypt', () => {
 
     it('group key not available: timeout while waiting', async () => {
         const wallet = await createTestWallet()
-        const groupKeyManager = await createGroupKeyManager(undefined, EthereumKeyPairIdentity.fromPrivateKey(wallet.privateKey))
+        const groupKeyManager = await createGroupKeyManager(EthereumKeyPairIdentity.fromPrivateKey(wallet.privateKey))
         const destroySignal = new DestroySignal()
         const groupKey = GroupKey.generate()
         const msg = await createMockMessage({
diff --git a/packages/sdk/test/unit/GroupKeyManager.test.ts b/packages/sdk/test/unit/GroupKeyManager.test.ts
diff --git a/packages/sdk/test/unit/GroupKeyQueue.test.ts b/packages/sdk/test/unit/GroupKeyQueue.test.ts
diff --git a/packages/sdk/test/unit/Publisher.test.ts b/packages/sdk/test/unit/Publisher.test.ts
diff --git a/packages/sdk/test/unit/messagePipeline.test.ts b/packages/sdk/test/unit/messagePipeline.test.ts
