feat: lit protocol key exchange

package.json

@@ -42,6 +42,7 @@
         "husky": "^6.0.0",
         "jest": "^29.4.3",
         "jest-extended": "^3.2.3",
+        "jest-mock-extended": "^3.0.1",
         "lerna": "^4.0.0",
         "node-gyp-build": "^4.3.0",
         "semver": "^7.3.5",

packages/client/CHANGELOG.md

@@ -8,6 +8,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Add support for experimental encryption key exchange via [Lit Protocol](https://litprotocol.com/). Enabled by
+  setting configuration option `encryption.litProtocolEnabled` to be true.
+
 ### Changed
 
 - All contract providers are used to query the tracker registry, storage node registry and stream storage registry

packages/client/package.json

@@ -97,6 +97,7 @@
     "@ethersproject/transactions": "^5.5.0",
     "@ethersproject/wallet": "^5.5.0",
     "@ethersproject/web": "^5.5.0",
+    "@lit-protocol/lit-node-client": "2.1.38",
     "@streamr/network-node": "7.3.0",
     "@streamr/protocol": "7.3.0",
     "@streamr/utils": "7.3.0",
@@ -107,6 +108,7 @@
     "eventemitter3": "^4.0.7",
     "heap": "^0.2.6",
     "idb-keyval": "^5.1.5",
+    "lit-siwe": "^1.1.8",
     "lodash": "^4.17.21",
     "mem": "^8.1.1",
     "node-fetch": "^2.6.6",

packages/client/src/Config.ts

@@ -67,6 +67,21 @@ export interface StreamrClientConfig {
     retryResendAfter?: number
     gapFillTimeout?: number
 
+    encryption?: {
+        /**
+         * Enable experimental Lit Protocol key exchange.
+         *
+         * When enabled encryption key storing and fetching will be primarily done through the Lit Protocol and
+         * secondarily through the standard Streamr key-exchange system.
+         */
+        litProtocolEnabled?: boolean
+
+        /**
+         * Enable log messages of the Lit Protocol library to be printed to stdout.
+         */
+        litProtocolLogging?: boolean
+    }
+
     network?: {
         id?: string
         acceptProxyConnections?: boolean
@@ -140,6 +155,7 @@ export interface StreamrClientConfig {
 export type StrictStreamrClientConfig = MarkOptional<Required<StreamrClientConfig>, 'auth' | 'metrics'> & {
     network: MarkOptional<Exclude<Required<StreamrClientConfig['network']>, undefined>, 'location'>
     contracts: Exclude<Required<StreamrClientConfig['contracts']>, undefined>
+    encryption: Exclude<Required<StreamrClientConfig['encryption']>, undefined>
     decryption: Exclude<Required<StreamrClientConfig['decryption']>, undefined>
     cache: Exclude<Required<StreamrClientConfig['cache']>, undefined>
     _timeouts: Exclude<DeepRequired<StreamrClientConfig['_timeouts']>, undefined>
@@ -158,6 +174,11 @@ export const STREAM_CLIENT_DEFAULTS:
     maxGapRequests: 5,
     retryResendAfter: 5000,
     gapFillTimeout: 5000,
+
+    encryption: {
+        litProtocolEnabled: false,
+        litProtocolLogging: false
+    },
 
     network: {
         acceptProxyConnections: false,

packages/client/src/StreamrClient.ts

@@ -35,6 +35,7 @@ import { LoggerFactory } from './utils/LoggerFactory'
 import { convertStreamMessageToMessage, Message } from './Message'
 import { ErrorCode } from './HttpUtil'
 import { omit } from 'lodash'
+import { StreamrClientError } from './StreamrClientError'
 
 /**
  * The main API used to interact with Streamr.
@@ -123,6 +124,9 @@ export class StreamrClient {
         if (opts.streamId === undefined) {
             throw new Error('streamId required')
         }
+        if (opts.key !== undefined && this.config.encryption.litProtocolEnabled) {
+            throw new StreamrClientError('cannot pass "key" when Lit Protocol is enabled', 'UNSUPPORTED_OPERATION')
+        }
         const streamId = await this.streamIdBuilder.toStreamID(opts.streamId)
         const queue = await this.publisher.getGroupKeyQueue(streamId)
         if (opts.distributionMethod === 'rotate') {

packages/client/src/StreamrClientError.ts

@@ -4,6 +4,7 @@ export type StreamrClientErrorCode =
     'INVALID_ARGUMENT' | 
     'CLIENT_DESTROYED' | 
     'PIPELINE_ERROR' |
+    'UNSUPPORTED_OPERATION' |
     'UNKNOWN_ERROR'
 
 export class StreamrClientError extends Error {

packages/client/src/config.schema.json

@@ -306,6 +306,21 @@
             },
             "default": {}
         },
+        "encryption": {
+            "type": "object",
+            "additionalProperties": false,
+            "properties": {
+                "litProtocolEnabled": {
+                    "type": "boolean",
+                    "default": false
+                },
+                "litProtocolLogging": {
+                    "type": "boolean",
+                    "default": false
+                }
+            },
+            "default": {}
+        },
         "decryption": {
             "type": "object",
             "additionalProperties": false,

packages/client/src/encryption/GroupKey.ts

@@ -23,9 +23,9 @@ export class GroupKey {
     /** @internal */
     readonly id: string
     /** @internal */
-    readonly data: Uint8Array
+    readonly data: Buffer
 
-    constructor(groupKeyId: string, data: Uint8Array) {
+    constructor(groupKeyId: string, data: Buffer) {
         this.id = groupKeyId
         if (!groupKeyId) {
             throw new GroupKeyError(`groupKeyId must not be falsey ${groupKeyId}`)

packages/client/src/encryption/GroupKeyManager.ts

@@ -0,0 +1,88 @@
+import { LitProtocolFacade } from './LitProtocolFacade'
+import { inject, Lifecycle, scoped } from 'tsyringe'
+import { GroupKeyStore } from './GroupKeyStore'
+import { GroupKey } from './GroupKey'
+import { StreamID, StreamPartID, StreamPartIDUtils } from '@streamr/protocol'
+import { EthereumAddress, waitForEvent } from '@streamr/utils'
+import { SubscriberKeyExchange } from './SubscriberKeyExchange'
+import { ConfigInjectionToken, StrictStreamrClientConfig } from '../Config'
+import { StreamrClientEventEmitter } from '../events'
+import { DestroySignal } from '../DestroySignal'
+import crypto from 'crypto'
+import { uuid } from '../utils/uuid'
+
+@scoped(Lifecycle.ContainerScoped)
+export class GroupKeyManager {
+    private readonly groupKeyStore: GroupKeyStore
+    private readonly litProtocolFacade: LitProtocolFacade
+    private readonly subscriberKeyExchange: SubscriberKeyExchange
+    private readonly eventEmitter: StreamrClientEventEmitter
+    private readonly destroySignal: DestroySignal
+    private readonly config: Pick<StrictStreamrClientConfig, 'decryption' | 'encryption'>
+
+    constructor(
+        groupKeyStore: GroupKeyStore,
+        litProtocolFacade: LitProtocolFacade,
+        subscriberKeyExchange: SubscriberKeyExchange,
+        eventEmitter: StreamrClientEventEmitter,
+        destroySignal: DestroySignal,
+        @inject(ConfigInjectionToken) config: Pick<StrictStreamrClientConfig, 'decryption' | 'encryption'>
+    ) {
+        this.groupKeyStore = groupKeyStore
+        this.litProtocolFacade = litProtocolFacade
+        this.subscriberKeyExchange = subscriberKeyExchange
+        this.eventEmitter = eventEmitter
+        this.destroySignal = destroySignal
+        this.config = config
+    }
+
+    async fetchKey(streamPartId: StreamPartID, groupKeyId: string, publisherId: EthereumAddress): Promise<GroupKey> {
+        const streamId = StreamPartIDUtils.getStreamID(streamPartId)
+
+        // 1st try: local storage
+        let groupKey = await this.groupKeyStore.get(groupKeyId, streamId)
+        if (groupKey !== undefined) {
+            return groupKey
+        }
+
+        // 2nd try: lit-protocol
+        if (this.config.encryption.litProtocolEnabled) {
+            groupKey = await this.litProtocolFacade.get(streamId, groupKeyId)
+            if (groupKey !== undefined) {
+                await this.groupKeyStore.add(groupKey, streamId)
+                return groupKey
+            }
+        }
+
+        // 3rd try: Streamr key-exchange
+        await this.subscriberKeyExchange.requestGroupKey(groupKeyId, publisherId, streamPartId)
+        const groupKeys = await waitForEvent(
+            // TODO remove "as any" type casing in NET-889
+            this.eventEmitter as any,
+            'addGroupKey',
+            this.config.decryption.keyRequestTimeout,
+            (storedGroupKey: GroupKey) => storedGroupKey.id === groupKeyId,
+            this.destroySignal.abortSignal
+        )
+        return groupKeys[0] as GroupKey
+    }
+
+    async storeKey(groupKey: GroupKey | undefined, streamId: StreamID): Promise<GroupKey> { // TODO: name
+        if (groupKey === undefined) {
+            const keyData = crypto.randomBytes(32)
+            // 1st try lit-protocol, if a key cannot be generated and stored, then generate group key locally
+            if (this.config.encryption.litProtocolEnabled) {
+                groupKey = await this.litProtocolFacade.store(streamId, keyData)
+            }
+            if (groupKey === undefined) {
+                groupKey = new GroupKey(uuid('GroupKey'), keyData)
+            }
+        }
+        await this.groupKeyStore.add(groupKey, streamId)
+        return groupKey
+    }
+
+    addKeyToLocalStore(groupKey: GroupKey, streamId: StreamID): Promise<void> {
+        return this.groupKeyStore.add(groupKey, streamId)
+    }
+}

packages/client/src/encryption/GroupKeyStore.ts

@@ -43,6 +43,9 @@ export interface UpdateEncryptionKeyOptions {
     key?: GroupKey
 }
 
+/**
+ * TODO: rename to e.g. `LocalGroupKeyStore` for clarity
+ */
 @scoped(Lifecycle.ContainerScoped)
 export class GroupKeyStore {
 

packages/client/src/encryption/LitProtocolFacade.ts

@@ -0,0 +1,159 @@
+import * as LitJsSdk from '@lit-protocol/lit-node-client'
+import { inject, Lifecycle, scoped } from 'tsyringe'
+import * as siwe from 'lit-siwe'
+import { Authentication, AuthenticationInjectionToken } from '../Authentication'
+import { ethers } from 'ethers'
+import { StreamID } from '@streamr/protocol'
+import { StreamPermission, streamPermissionToSolidityType } from '../permission'
+import { ConfigInjectionToken, StrictStreamrClientConfig } from '../Config'
+import { GroupKey } from './GroupKey'
+import { Logger, withRateLimit } from '@streamr/utils'
+import { LoggerFactory } from '../utils/LoggerFactory'
+
+const logger = new Logger(module)
+
+const chain = 'polygon'
+
+const LIT_PROTOCOL_CONNECT_INTERVAL = 60 * 60 * 1000 // 1h
+
+const formEvmContractConditions = (streamRegistryChainAddress: string, streamId: StreamID) => ([
+    {
+        contractAddress: streamRegistryChainAddress,
+        chain,
+        functionName: 'hasPermission',
+        functionParams: [streamId, ':userAddress', `${streamPermissionToSolidityType(StreamPermission.SUBSCRIBE)}`],
+        functionAbi: {
+            inputs: [
+                {
+                    name: "streamId",
+                    type: "string"
+                },
+                {
+                    name: "user",
+                    type: "address"
+                },
+                {
+                    name: "permissionType",
+                    type: "uint8"
+                }
+            ],
+            name: "hasPermission",
+            outputs: [
+                {
+                    name: "userHasPermission",
+                    type: "bool"
+                }
+            ],
+            stateMutability: "view",
+            type: "function"
+        },
+        returnValueTest: {
+            key: "userHasPermission",
+            comparator: '=',
+            value: "true",
+        },
+    }
+])
+
+const signAuthMessage = async (authentication: Authentication) => {
+    const domain = "dummy.com"
+    const uri = "https://dummy.com"
+    const statement = "dummy"
+    const addressInChecksumCase = ethers.utils.getAddress(await authentication.getAddress())
+    const siweMessage = new siwe.SiweMessage({
+        domain,
+        uri,
+        statement,
+        address: addressInChecksumCase,
+        version: "1",
+        chainId: 1
+    })
+    const messageToSign = siweMessage.prepareMessage()
+    const signature = await authentication.createMessageSignature(messageToSign)
+    return {
+        sig: signature,
+        derivedVia: "web3.eth.personal.sign",
+        signedMessage: messageToSign,
+        address: addressInChecksumCase
+    }
+}
+
+/**
+ * This class only operates with Polygon production network and therefore ignores contracts config.
+ */
+@scoped(Lifecycle.ContainerScoped)
+export class LitProtocolFacade {
+    private readonly authentication: Authentication
+    private readonly config: Pick<StrictStreamrClientConfig, 'contracts' | 'encryption'>
+    private readonly logger: Logger
+    private litNodeClient?: LitJsSdk.LitNodeClient
+    private connectLitNodeClient?: () => Promise<void>
+
+    constructor(
+        loggerFactory: LoggerFactory,
+        @inject(ConfigInjectionToken) config: Pick<StrictStreamrClientConfig, 'contracts' | 'encryption'>,
+        @inject(AuthenticationInjectionToken) authentication: Authentication,
+    ) {
+        this.authentication = authentication
+        this.config = config
+        this.logger = loggerFactory.createLogger(module)
+    }
+
+    async getLitNodeClient(): Promise<LitJsSdk.LitNodeClient> {
+        if (this.litNodeClient === undefined) {
+            this.litNodeClient = new LitJsSdk.LitNodeClient({
+                alertWhenUnauthorized: false,
+                debug: this.config.encryption.litProtocolLogging
+            })
+            // Add a rate limiter to avoid calling `connect` each time we want to use lit protocol as this would cause an unnecessary handshake.
+            this.connectLitNodeClient = withRateLimit(() => this.litNodeClient!.connect(), LIT_PROTOCOL_CONNECT_INTERVAL)
+        }
+        await this.connectLitNodeClient!()
+        return this.litNodeClient
+    }
+
+    async store(streamId: StreamID, symmetricKey: Uint8Array): Promise<GroupKey | undefined> {
+        this.logger.debug('storing key: %j', { streamId })
+        try {
+            const authSig = await signAuthMessage(this.authentication)
+            const client = await this.getLitNodeClient()
+            const encryptedSymmetricKey = await client.saveEncryptionKey({
+                evmContractConditions: formEvmContractConditions(this.config.contracts.streamRegistryChainAddress, streamId),
+                symmetricKey,
+                authSig,
+                chain
+            })
+            if (encryptedSymmetricKey === undefined) {
+                return undefined
+            }
+            const groupKeyId = LitJsSdk.uint8arrayToString(encryptedSymmetricKey, 'base16')
+            this.logger.debug('stored key: %j', { groupKeyId, streamId })
+            return new GroupKey(groupKeyId, Buffer.from(symmetricKey))
+        } catch (e) {
+            logger.warn('encountered error when trying to store key on lit-protocol: %s', e?.message)
+            return undefined
+        }
+    }
+
+    async get(streamId: StreamID, groupKeyId: string): Promise<GroupKey | undefined> {
+        this.logger.debug('get key: %j', { groupKeyId, streamId })
+        try {
+            const authSig = await signAuthMessage(this.authentication)
+            const client = await this.getLitNodeClient()
+            const symmetricKey = await client.getEncryptionKey({
+                evmContractConditions: formEvmContractConditions(this.config.contracts.streamRegistryChainAddress, streamId),
+                toDecrypt: groupKeyId,
+                chain,
+                authSig
+            })
+            if (symmetricKey === undefined) {
+                return undefined
+            }
+            this.logger.debug('got key: %j', { groupKeyId, streamId })
+            return new GroupKey(groupKeyId, Buffer.from(symmetricKey))
+        } catch (e) {
+            logger.warn('encountered error when trying to get key from lit-protocol: %s', e?.message)
+            return undefined
+        }
+    }
+}