feat(sdk): [NET-1663] Signature validation in worker thread (draft)

packages/sdk/package.json

@@ -13,7 +13,9 @@
   "script": "./dist/streamr-sdk.web.min.js",
   "browser": {
     "./src/utils/persistence/ServerPersistence.ts": "./src/utils/persistence/BrowserPersistence.mts",
-    "./dist/src/utils/persistence/ServerPersistence.js": "./dist/src/utils/persistence/BrowserPersistence.mjs"
+    "./dist/src/utils/persistence/ServerPersistence.js": "./dist/src/utils/persistence/BrowserPersistence.mjs",
+    "./src/signature/ServerSignatureValidation.ts": "./src/signature/BrowserSignatureValidation.mts",
+    "./dist/src/signature/ServerSignatureValidation.js": "./dist/src/signature/BrowserSignatureValidation.mjs"
   },
   "exports": {
     "default": {
@@ -103,6 +105,7 @@
     "@streamr/proto-rpc": "103.2.0",
     "@streamr/trackerless-network": "103.2.0",
     "@streamr/utils": "103.2.0",
+    "comlink": "^4.4.2",
     "core-js": "^3.47.0",
     "env-paths": "^2.2.1",
     "ethers": "^6.13.0",

packages/sdk/src/signature/BrowserSignatureValidation.mts

@@ -0,0 +1,35 @@
+/**
+ * Browser implementation of signature validation using Web Worker.
+ * This offloads CPU-intensive cryptographic operations to a separate thread.
+ */
+import * as Comlink from 'comlink'
+import { SignatureValidationContext } from './SignatureValidationContext.js'
+import { SignatureValidationResult, toSignatureValidationData } from './signatureValidation.js'
+import type { SignatureValidationWorkerApi } from './SignatureValidationWorker.js'
+import { StreamMessage } from '../protocol/StreamMessage.js'
+
+export default class BrowserSignatureValidation implements SignatureValidationContext {
+    private worker: Worker
+    private workerApi: Comlink.Remote<SignatureValidationWorkerApi> 
+
+    constructor() {
+        // Webpack 5 handles this pattern automatically, creating a separate chunk for the worker
+        this.worker = new Worker(
+            /* webpackChunkName: "signature-worker" */
+            new URL('./SignatureValidationWorker.js', import.meta.url)
+        )
+        this.workerApi = Comlink.wrap<SignatureValidationWorkerApi>(this.worker)
+    }
+
+    async validateSignature(message: StreamMessage): Promise<SignatureValidationResult> {
+        // Convert class instance to plain serializable data before sending to worker
+        const data = toSignatureValidationData(message)
+        return this.workerApi.validateSignature(data)
+    }
+
+    destroy(): void {
+        if (this.worker) {
+            this.worker.terminate()
+        }
+    }
+}

packages/sdk/src/signature/ServerSignatureValidation.ts

@@ -0,0 +1,35 @@
+import * as Comlink from 'comlink'
+// eslint-disable-next-line no-restricted-imports
+import nodeEndpoint from 'comlink/dist/umd/node-adapter'
+import { Worker } from 'worker_threads'
+import { StreamMessage } from '../protocol/StreamMessage'
+import { SignatureValidationContext } from './SignatureValidationContext'
+import { SignatureValidationWorkerApi } from './SignatureValidationWorker'
+import { SignatureValidationResult, toSignatureValidationData } from './signatureValidation'
+import { join } from 'path'
+
+export default class ServerSignatureValidation implements SignatureValidationContext {
+
+    private readonly worker: Worker
+    private readonly workerApi: Comlink.Remote<SignatureValidationWorkerApi>
+    constructor() {
+        const isRunningFromDist = __dirname.includes('/dist/')
+        const workerPath = isRunningFromDist
+            ? join(__dirname, 'SignatureValidationWorker.js')
+            : join(__dirname, '../../dist/src/signature/SignatureValidationWorker.js')
+        this.worker = new Worker(workerPath)
+        this.workerApi = Comlink.wrap<SignatureValidationWorkerApi>(nodeEndpoint(this.worker))
+    }
+
+    async validateSignature(message: StreamMessage): Promise<SignatureValidationResult> {
+        // Convert class instance to plain serializable data before sending to worker
+        const data = toSignatureValidationData(message)
+        return this.workerApi.validateSignature(data)
+    }
+
+    // eslint-disable-next-line class-methods-use-this
+    destroy(): void {
+        // No-op for server implementation
+    }
+}
+

packages/sdk/src/signature/SignatureValidationContext.ts

@@ -0,0 +1,12 @@
+/**
+ * Interface for signature validation backend.
+ * Browser implementation uses a Web Worker, Node.js runs on main thread.
+ */
+import { StreamMessage } from '../protocol/StreamMessage'
+import { SignatureValidationResult } from './signatureValidation'
+
+export interface SignatureValidationContext {
+    validateSignature(message: StreamMessage): Promise<SignatureValidationResult>
+    destroy(): void
+}
+

packages/sdk/src/signature/SignatureValidationWorker.ts

@@ -0,0 +1,31 @@
+import * as Comlink from 'comlink'
+import { validateSignatureData, SignatureValidationResult, SignatureValidationData } from './signatureValidation'
+
+const workerApi = {
+    validateSignature: async (data: SignatureValidationData): Promise<SignatureValidationResult> => {
+        return validateSignatureData(data)
+    }
+}
+
+export type SignatureValidationWorkerApi = typeof workerApi
+
+// Detect environment and expose accordingly
+// Check for Node.js worker_threads first, since `self` is defined in both environments
+// but only browser Web Workers have WorkerGlobalScope with addEventListener
+let parentPort: import('worker_threads').MessagePort | null = null
+try {
+    // eslint-disable-next-line @typescript-eslint/no-require-imports
+    parentPort = require('worker_threads').parentPort
+} catch {
+    // Not in Node.js environment
+}
+
+if (parentPort) {
+    // Node.js Worker Thread
+    // eslint-disable-next-line @typescript-eslint/no-require-imports
+    const nodeEndpoint = require('comlink/dist/umd/node-adapter')
+    Comlink.expose(workerApi, nodeEndpoint(parentPort))
+} else {
+    // Browser Web Worker
+    Comlink.expose(workerApi)
+}

packages/sdk/src/signature/SignatureValidator.ts

@@ -1,26 +1,30 @@
-import { toEthereumAddress, toUserIdRaw, SigningUtil } from '@streamr/utils'
+import { toEthereumAddress } from '@streamr/utils'
 import { Lifecycle, scoped } from 'tsyringe'
 import { ERC1271ContractFacade } from '../contracts/ERC1271ContractFacade'
+import { DestroySignal } from '../DestroySignal'
 import { StreamMessage } from '../protocol/StreamMessage'
 import { StreamrClientError } from '../StreamrClientError'
-import { createLegacySignaturePayload } from './createLegacySignaturePayload'
 import { createSignaturePayload } from './createSignaturePayload'
+import { SignatureValidationContext } from './SignatureValidationContext'
+// This import will be swapped to BrowserSignatureValidation.mts in browser builds
+import SignatureValidation from './ServerSignatureValidation'
 import { SignatureType } from '@streamr/trackerless-network'
-import { IDENTITY_MAPPING } from '../identity/IdentityMapping'
-
-// Lookup structure SignatureType -> SigningUtil
-const signingUtilBySignatureType: Record<number, SigningUtil> = Object.fromEntries(
-    IDENTITY_MAPPING.map((idMapping) => [idMapping.signatureType, SigningUtil.getInstance(idMapping.keyType)])
-)
-
-const evmSigner = SigningUtil.getInstance('ECDSA_SECP256K1_EVM')
 
 @scoped(Lifecycle.ContainerScoped)
 export class SignatureValidator {
     private readonly erc1271ContractFacade: ERC1271ContractFacade
+    private validationContext: SignatureValidationContext | undefined
 
-    constructor(erc1271ContractFacade: ERC1271ContractFacade) {
+    constructor(
+        erc1271ContractFacade: ERC1271ContractFacade,
+        destroySignal: DestroySignal
+    ) {
         this.erc1271ContractFacade = erc1271ContractFacade
+        destroySignal.onDestroy.listen(() => this.destroy())
+    }
+
+    private getValidationContext(): SignatureValidationContext {
+        return this.validationContext ??= new SignatureValidation()
     }
 
     /**
@@ -41,36 +45,33 @@ export class SignatureValidator {
     }
 
     private async validate(streamMessage: StreamMessage): Promise<boolean> {
-        const signingUtil = signingUtilBySignatureType[streamMessage.signatureType]
-
-        // Common case
-        if (signingUtil) {
-            return signingUtil.verifySignature(
-                toUserIdRaw(streamMessage.getPublisherId()),
-                createSignaturePayload(streamMessage),
-                streamMessage.signature
-            )
-        }
-
-        // Special handling: different payload computation, same SigningUtil
-        if (streamMessage.signatureType === SignatureType.ECDSA_SECP256K1_LEGACY) {
-            return evmSigner.verifySignature(
-                // publisherId is hex encoded Ethereum address string
-                toUserIdRaw(streamMessage.getPublisherId()),
-                createLegacySignaturePayload(streamMessage),
-                streamMessage.signature
-            )
-        }
-
-        // Special handling: check signature with ERC-1271 contract facade
         if (streamMessage.signatureType === SignatureType.ERC_1271) {
             return this.erc1271ContractFacade.isValidSignature(
-                toEthereumAddress(streamMessage.getPublisherId()),
+                toEthereumAddress(streamMessage.messageId.publisherId),
                 createSignaturePayload(streamMessage),
                 streamMessage.signature
             )
         }
+        const result = await this.getValidationContext().validateSignature(streamMessage)
+        switch (result.type) {
+            case 'valid':
+                return true
+            case 'invalid':
+                return false
+            case 'error':
+                throw new Error(result.message)
+            default:
+                throw new Error(`Unknown signature validation result type '${result}'`)
+        }
+    }
 
-        throw new Error(`Cannot validate message signature, unsupported signatureType: "${streamMessage.signatureType}"`)
+    /**
+     * Cleanup worker resources when the validator is no longer needed.
+     */
+    destroy(): void {
+        if (this.validationContext) {
+            this.validationContext.destroy()
+            this.validationContext = undefined
+        }
     }
 }

packages/sdk/src/signature/createLegacySignaturePayload.ts

@@ -1,7 +1,6 @@
 import { binaryToHex, binaryToUtf8 } from '@streamr/utils'
 import { EncryptedGroupKey, EncryptionType } from '@streamr/trackerless-network'
-import { MessageID } from '../protocol/MessageID'
-import { MessageRef } from '../protocol/MessageRef'
+import { MessageIdLike, MessageRefLike } from './createSignaturePayload'
 
 const serializeGroupKey = ({ id, data }: EncryptedGroupKey): string => {
     return JSON.stringify([id, binaryToHex(data)])
@@ -11,10 +10,10 @@ const serializeGroupKey = ({ id, data }: EncryptedGroupKey): string => {
  * Only to be used for LEGACY_SECP256K1 signature type.
  */
 export const createLegacySignaturePayload = (opts: {
-    messageId: MessageID
+    messageId: MessageIdLike
     content: Uint8Array
     encryptionType: EncryptionType
-    prevMsgRef?: MessageRef
+    prevMsgRef?: MessageRefLike
     newGroupKey?: EncryptedGroupKey
 }): Uint8Array => {
     const prev = ((opts.prevMsgRef !== undefined) ? `${opts.prevMsgRef.timestamp}${opts.prevMsgRef.sequenceNumber}` : '')

packages/sdk/src/signature/createSignaturePayload.ts

@@ -3,16 +3,34 @@ import {
     GroupKeyRequest as NewGroupKeyRequest,
     GroupKeyResponse as NewGroupKeyResponse
 } from '@streamr/trackerless-network'
-import { utf8ToBinary } from '@streamr/utils'
-import { MessageID } from '../protocol/MessageID'
-import { MessageRef } from '../protocol/MessageRef'
+import { StreamID, UserID, utf8ToBinary } from '@streamr/utils'
 import { StreamMessageType } from '../protocol/StreamMessage'
 
+/**
+ * Plain data for message ID - accepts class instances or plain objects with the same properties.
+ */
+export interface MessageIdLike {
+    streamId: StreamID
+    streamPartition: number
+    timestamp: number
+    sequenceNumber: number
+    publisherId: UserID
+    msgChainId: string
+}
+
+/**
+ * Plain data for message reference - accepts class instances or plain objects with the same properties.
+ */
+export interface MessageRefLike {
+    timestamp: number
+    sequenceNumber: number
+}
+
 export const createSignaturePayload = (opts: {
-    messageId: MessageID
+    messageId: MessageIdLike
     content: Uint8Array
     messageType: StreamMessageType
-    prevMsgRef?: MessageRef
+    prevMsgRef?: MessageRefLike
     newGroupKey?: EncryptedGroupKey
 }): Uint8Array | never => {
     const header = Buffer.concat([