diff --git a/packages/bitcore-cli/src/tss.ts b/packages/bitcore-cli/src/tss.ts index a69be034cd1..93df3d3b2da 100644 --- a/packages/bitcore-cli/src/tss.ts +++ b/packages/bitcore-cli/src/tss.ts @@ -49,6 +49,15 @@ export async function sign(args: { tssKey: walletData.key as TssKeyType }); + // Reduces the noise of transient connection errors + const connResilience = (count: number, e: Error): number => { + if (count > 10) { + prompt.log.warn(e.message); + return 0; + } + return ++count; + }; + // Restore a previously-interrupted TSS session if it exists if (fs.existsSync(storedSessionFile)) { const storedSession = Encryption.decryptWithPassword(fs.readFileSync(storedSessionFile, 'utf8'), password); @@ -57,27 +66,36 @@ export async function sign(args: { // ...otherwise, start a new TSS session } else { - try { - await tssSign.start({ - id, - messageHash, - derivationPath, - password - }); - storeSession(tssSign.exportSession()); - } catch (err) { - if (err.message?.startsWith('TSS_ROUND_ALREADY_DONE')) { - const sig = await tssSign.getSignatureFromServer(); - if (!sig) { - throw new Error('It looks like the TSS signature session was interrupted. Try deleting this proposal and creating a new one.'); + let isTransientError = false; + let connErrs = 0; + do { + try { + await tssSign.start({ + id, + messageHash, + derivationPath, + password + }); + storeSession(tssSign.exportSession()); + } catch (err) { + isTransientError = false; // reset + if (err.message?.startsWith('TSS_ROUND_ALREADY_DONE')) { + const sig = await tssSign.getSignatureFromServer(); + if (!sig) { + throw new Error('It looks like the TSS signature session was interrupted. Try deleting this proposal and creating a new one.'); + } + return { + signature: transformISignature(sig), + publicKey: sig.pubKey + }; + } else if (err instanceof Errors.CONNECTION_ERROR) { + isTransientError = true; + connErrs = connResilience(connErrs, err); + } else { + throw err; } - return { - signature: transformISignature(sig), - publicKey: sig.pubKey - }; } - throw err; - } + } while (isTransientError); } const spinner = prompt.spinner({ indicator: 'timer', onCancel: () => { tssSign.unsubscribe(); } }); @@ -87,8 +105,10 @@ export async function sign(args: { let rejected = false; const reject = (err) => { if (!rejected) { rejected = true; _reject(err); } }; + let connErrs = 0; tssSign.subscribe(); tssSign.on('roundsubmitted', (round) => { + connErrs = 0; storeSession(tssSign.exportSession()); spinner.message(`Round ${round} submitted`); }); @@ -103,8 +123,13 @@ export async function sign(args: { spinner.cancel(e.message); rmSessionState(); return reject(new ProcessCancelled()); + } else if (e instanceof Errors.CONNECTION_ERROR) { + // Reduce the noise of transient errors + connErrs = connResilience(connErrs, e); + return; + } else { + prompt.log.error('Unexpected error during TSS signing: ' + (e.stack || e)); } - prompt.log.error('Unexpected error during TSS signing: ' + (e.stack || e)); }); tssSign.on('complete', async () => { try { diff --git a/packages/bitcore-wallet-client/src/lib/errors/spec.ts b/packages/bitcore-wallet-client/src/lib/errors/spec.ts index 59022dc90ec..6291b1264ba 100644 --- a/packages/bitcore-wallet-client/src/lib/errors/spec.ts +++ b/packages/bitcore-wallet-client/src/lib/errors/spec.ts @@ -147,7 +147,7 @@ export const errorSpec: IErrorSpec[] = [ 'We were unable to parse your payment. Please try again or contact your wallet provider.' }, { - name: 'NO_TRASACTION', + name: 'NO_TRANSACTION', message: 'Your request did not include a transaction. Please try again or contact your wallet provider.' }, diff --git a/packages/bitcore-wallet-client/src/lib/payproV2.ts b/packages/bitcore-wallet-client/src/lib/payproV2.ts index 7898f83f7f6..1ca69f89cf1 100644 --- a/packages/bitcore-wallet-client/src/lib/payproV2.ts +++ b/packages/bitcore-wallet-client/src/lib/payproV2.ts @@ -118,7 +118,7 @@ export class PayProV2 { case errMsg.includes('We were unable to parse your payment.'): return new Errors.UNABLE_TO_PARSE_PAYMENT(); case errMsg.includes('Request must include exactly one'): - return new Errors.NO_TRASACTION(); + return new Errors.NO_TRANSACTION(); case errMsg.includes('Your transaction was an in an invalid format'): return new Errors.INVALID_TX_FORMAT(); case errMsg.includes('We were unable to parse the transaction you sent'): diff --git a/packages/bitcore-wallet-client/test/bulkclient.test.ts b/packages/bitcore-wallet-client/test/bulkclient.test.ts index 761471823bb..2e19986a53b 100644 --- a/packages/bitcore-wallet-client/test/bulkclient.test.ts +++ b/packages/bitcore-wallet-client/test/bulkclient.test.ts @@ -21,18 +21,21 @@ describe('Bulk Client', function() { let clients, app, sandbox, storage, keys, i; let db; let connection; - this.timeout(8000); + this.timeout(Math.max(this['_timeout'], 8000)); before(done => { i = 0; clients = []; keys = []; helpers.newDb('', (err, database, conn) => { + if (err) return done(err); db = database; connection = conn; storage = new Storage({ db }); Storage.createIndexes(db); - return done(err); + helpers.finishIndexCreation(db) + .then(() => done()) + .catch(done); }); }); diff --git a/packages/bitcore-wallet-client/test/helpers.ts b/packages/bitcore-wallet-client/test/helpers.ts index f22ab2ab0ae..550853c665f 100644 --- a/packages/bitcore-wallet-client/test/helpers.ts +++ b/packages/bitcore-wallet-client/test/helpers.ts @@ -222,6 +222,17 @@ export const helpers = { return cb(err, db, connection); }); }); + }, + finishIndexCreation: async (db) => { + let inprog; + do { + ({ inprog } = await db.admin().command({ + currentOp: 1, + 'command.createIndexes': { $exists: true }, + 'command.$db': db.databaseName + })); + if (inprog.length > 0) await new Promise(r2 => setTimeout(r2, 10)); + } while (inprog.length > 0); } }; diff --git a/packages/bitcore-wallet-client/test/tss.test.ts b/packages/bitcore-wallet-client/test/tss.test.ts index e97a8fb91c5..707177b60f0 100644 --- a/packages/bitcore-wallet-client/test/tss.test.ts +++ b/packages/bitcore-wallet-client/test/tss.test.ts @@ -4,6 +4,10 @@ import sinon from 'sinon'; import * as chai from 'chai'; import BWS from '@bitpay-labs/bitcore-wallet-service'; import { Defaults as BwsDefaults } from '@bitpay-labs/bitcore-wallet-service/ts_build/src/lib/common/defaults'; +import { + TssKeyGen as BwsTssKeyGen, + TssSign as BwsTssSign +} from '@bitpay-labs/bitcore-wallet-service/ts_build/src/lib/tss'; import request from 'supertest'; import crypto from 'crypto'; import fs from 'fs'; @@ -71,6 +75,17 @@ describe('TSS', function() { }); }); + beforeEach(function() { + sandbox.stub(BwsTssKeyGen, 'getMessagesForParty').callsFake(async function(params) { + params.maxWaitTimeSec = 1; + return (BwsTssKeyGen.getMessagesForParty as any).wrappedMethod.call(this, params); + }); + sandbox.stub(BwsTssSign, 'getMessagesForParty').callsFake(async function(params) { + params.maxWaitTimeSec = 1; + return (BwsTssSign.getMessagesForParty as any).wrappedMethod.call(this, params); + }); + }); + after(function(done) { dbConnection.close(done); }); diff --git a/packages/bitcore-wallet-service/README.md b/packages/bitcore-wallet-service/README.md index 7365d7b68f4..a1d079c5ffd 100644 --- a/packages/bitcore-wallet-service/README.md +++ b/packages/bitcore-wallet-service/README.md @@ -23,7 +23,7 @@ More about BWS at https://blog.bitpay.com/announcing-the-bitcore-wallet-suite/ ```sh git clone https://github.com/bitpay/bitcore.git cd bitcore -npm install +npm ci npm run bws ``` @@ -86,9 +86,6 @@ There are plenty examples of creating and sending proposals in the `/test/integr - copay running on port: 8100 - bitcoin-core running on regtest mode (blue icon logo) -> mongo topology crashes sometimes due to notifications being incompatible in a web browser -> **bitcore-wallet-service/lib/notificationbroadcaster.js** -> Note: If testing on a PC browser, comment out notificationbroadcaster.js to disable notifications. ### Steps: diff --git a/packages/bitcore-wallet-service/src/lib/messagebroker.ts b/packages/bitcore-wallet-service/src/lib/messagebroker.ts index 82060ac3605..573b44b1757 100644 --- a/packages/bitcore-wallet-service/src/lib/messagebroker.ts +++ b/packages/bitcore-wallet-service/src/lib/messagebroker.ts @@ -2,10 +2,12 @@ import { EventEmitter } from 'events'; import * as io from 'socket.io-client'; import 'source-map-support/register'; import logger from './logger'; +import type { Notification } from './model/notification'; export class MessageBroker extends EventEmitter { - remote: boolean; + remote: boolean = false; mq: io.Socket; + constructor(opts) { super(); @@ -28,7 +30,7 @@ export class MessageBroker extends EventEmitter { } } - send(data) { + send(data: Notification) { if (this.remote) { this.mq.emit('msg', data); } else { @@ -36,7 +38,11 @@ export class MessageBroker extends EventEmitter { } } - onMessage(handler) { + onMessage(handler: (data: Notification) => void) { this.on('msg', handler); } + + offMessage(handler: (data: Notification) => void) { + this.off('msg', handler); + } } diff --git a/packages/bitcore-wallet-service/src/lib/model/notification.ts b/packages/bitcore-wallet-service/src/lib/model/notification.ts index bb02fec5904..cb6c10c5b81 100644 --- a/packages/bitcore-wallet-service/src/lib/model/notification.ts +++ b/packages/bitcore-wallet-service/src/lib/model/notification.ts @@ -24,7 +24,7 @@ export interface INotification { version: string; createdOn: number; - id: number; + id: string | number; type: string; data: any; walletId: string; @@ -32,7 +32,7 @@ export interface INotification { isCreator: boolean; } -export class Notification { +export class Notification implements INotification { version: string; createdOn: number; id: string | number; diff --git a/packages/bitcore-wallet-service/src/lib/routes/helpers/error.ts b/packages/bitcore-wallet-service/src/lib/routes/helpers/error.ts index 2115d0dcf8f..6177a459b49 100644 --- a/packages/bitcore-wallet-service/src/lib/routes/helpers/error.ts +++ b/packages/bitcore-wallet-service/src/lib/routes/helpers/error.ts @@ -13,6 +13,7 @@ export class ApiErrorHelper { returnError(err: any, res: express.Response, req: express.Request): void { // make sure headers have not been sent as this leads to an uncaught error if (res.headersSent) { + res.end(); return; } if (err instanceof ClientError) { diff --git a/packages/bitcore-wallet-service/src/lib/routes/tss.ts b/packages/bitcore-wallet-service/src/lib/routes/tss.ts index b246a62fdf3..c4e8f965736 100644 --- a/packages/bitcore-wallet-service/src/lib/routes/tss.ts +++ b/packages/bitcore-wallet-service/src/lib/routes/tss.ts @@ -37,17 +37,34 @@ export class TssRouter { }); router.get('/v1/tss/keygen/:id/:round', authTssRequest(), async function(req, res) { + let interval: NodeJS.Timeout; try { const { id, round } = req.params as { [key: string]: string }; + const { maxWaitTime } = req.query as { [key: string]: string }; const copayerId = req.headers['x-identity']; if (round === 'secret') { const secret = await TssKeyGen.getBwsJoinSecret({ id, copayerId }); return res.json({ secret }); } - const { messages, publicKey } = await TssKeyGen.getMessagesForParty({ id, round: parseInt(round), copayerId }); - return res.json({ messages, publicKey }); + + // Validate access and fetch session before committing to a streaming response, so that errors like + // "session not found" or "not a participant" can still be returned with a proper + // HTTP status instead of silently becoming an empty 200 (see below). + const session = await TssKeyGen.getSessionForCopayer({ id, copayerId }); + + // Keep the connection alive while waiting for the change stream to return a result. + // Headers must be finalized before writing the first heartbeat byte. + // Flush ensures the heartbeat is sent immediately to keep the connection alive. + res.writeHead(200, { 'Content-Type': 'application/json' }); + interval = setInterval(() => { res.write('\n'); res.flush(); }, 1000); + req.on('close', () => clearInterval(interval)); + + const { messages, publicKey } = await TssKeyGen.getMessagesForParty({ session, round: parseInt(round), copayerId, maxWaitTimeSec: parseInt(maxWaitTime) }); + return res.end(JSON.stringify({ messages, publicKey })); } catch (err) { return returnError(err ?? 'unknown', res, req); + } finally { + clearInterval(interval); } }); @@ -104,13 +121,31 @@ export class TssRouter { }); router.get('/v1/tss/sign/:id/:round', authTssRequest(), async function(req, res) { + let interval: NodeJS.Timeout; try { const { id, round } = req.params as { [key: string]: string }; + const { maxWaitTime } = req.query as { [key: string]: string }; const copayerId = req.headers['x-identity']; - const { messages, signature, participants } = await TssSign.getMessagesForParty({ id, round: parseInt(round), copayerId }); - return res.json({ messages, signature, participants }); + + // Validate access and fetch session before committing to a streaming response, so that errors like + // "session not found" or "not a participant" can still be returned with a proper + // HTTP status instead of silently becoming an empty 200 (see below). + const session = await TssSign.getSessionForCopayer({ id, copayerId }); + + // Keep the connection alive while waiting for the change stream to return a result. + // Headers must be finalized before writing the first heartbeat byte. + // Flush ensures the heartbeat is sent immediately to keep the connection alive. + res.writeHead(200, { 'Content-Type': 'application/json' }); + interval = setInterval(() => { res.write('\n'); res.flush(); }, 1000); + req.on('close', () => clearInterval(interval)); + + const { messages, signature, participants } = await TssSign.getMessagesForParty({ session, round: parseInt(round), copayerId, maxWaitTimeSec: parseInt(maxWaitTime) }); + clearInterval(interval); + return res.end(JSON.stringify({ messages, signature, participants })); } catch (err) { return returnError(err ?? 'unknown', res, req); + } finally { + clearInterval(interval); } }); diff --git a/packages/bitcore-wallet-service/src/lib/server.ts b/packages/bitcore-wallet-service/src/lib/server.ts index 8edd5eb1896..a5b71d2f004 100644 --- a/packages/bitcore-wallet-service/src/lib/server.ts +++ b/packages/bitcore-wallet-service/src/lib/server.ts @@ -326,12 +326,9 @@ export class WalletService implements IWalletService { ); } - static handleIncomingNotifications(notification, cb) { - cb = cb || function() { }; - + static handleIncomingNotifications(_notification: INotification) { // do nothing here.... // bc height cache is cleared on bcmonitor - return cb(); } static shutDown(cb) { @@ -474,6 +471,13 @@ export class WalletService implements IWalletService { return storage; } + static getMessageBroker() { + if (!initialized) { + throw new Error('Message broker requested before server was initialized'); + } + return messageBroker; + } + _runLocked(cb, task, waitTime?: number) { $.checkState(this.walletId, 'Failed state: this.walletId undefined at <_runLocked()>'); this.lock.runLocked(this.walletId, { waitTime }, cb, task); @@ -1152,7 +1156,7 @@ export class WalletService implements IWalletService { try { const isValid = this._verifyRequestPubKey(opts.requestPubKey, opts.signature, target.xPubKey); if (!isValid) return cb(Errors.NOT_AUTHORIZED); - } catch (e) { + } catch { return cb(Errors.NOT_AUTHORIZED); } @@ -3857,7 +3861,7 @@ export class WalletService implements IWalletService { const notifications = res .flat() .map((n: INotification) => ({ ...n, walletId: this.walletId })) - .sort((a, b) => a.id - b.id); + .sort((a, b) => a.id?.toString()?.localeCompare(b.id?.toString())); return cb(null, notifications); } diff --git a/packages/bitcore-wallet-service/src/lib/storage.ts b/packages/bitcore-wallet-service/src/lib/storage.ts index 0b03a8b2bfc..950c659ea18 100644 --- a/packages/bitcore-wallet-service/src/lib/storage.ts +++ b/packages/bitcore-wallet-service/src/lib/storage.ts @@ -1,6 +1,5 @@ import * as async from 'async'; import _ from 'lodash'; -import { Db } from 'mongodb'; import * as mongodb from 'mongodb'; import preconditions from 'preconditions'; import { BCHAddressTranslator } from './bchaddresstranslator'; // only for migration @@ -19,8 +18,11 @@ import { TxProposal, Wallet } from './model'; -import { ITssKeyMessageObject, TssKeyGenModel } from './model/tsskeygen'; -import { ITssSigMessageObject, TssSigGenModel } from './model/tsssign'; +import { TssKeyGenModel } from './model/tsskeygen'; +import { TssSigGenModel } from './model/tsssign'; +import type { ITssKeyMessageObject } from './model/tsskeygen'; +import type { ITssSigMessageObject } from './model/tsssign'; +import type { Db } from 'mongodb'; const $ = preconditions.singleton(); diff --git a/packages/bitcore-wallet-service/src/lib/tss.ts b/packages/bitcore-wallet-service/src/lib/tss.ts index 5cde1216ed4..f4521c2a843 100644 --- a/packages/bitcore-wallet-service/src/lib/tss.ts +++ b/packages/bitcore-wallet-service/src/lib/tss.ts @@ -1,36 +1,178 @@ +import { EventEmitter } from 'events'; import { BitcoreLib } from '@bitpay-labs/crypto-wallet-core'; import { Constants } from './common/constants'; import { Errors } from './errors/errordefinitions'; import logger from './logger'; -import { ITssKeyMessageObject, TssKeyGenModel } from './model/tsskeygen'; -import { ITssSigMessageObject, TssSigGenModel } from './model/tsssign'; +import { TssKeyGenModel } from './model/tsskeygen'; +import { TssSigGenModel } from './model/tsssign'; import { WalletService, checkRequired } from './server'; import { Storage } from './storage'; +import type { INotification } from './model/notification'; +import type { ITssKeyMessageObject } from './model/tsskeygen'; +import type { ITssSigMessageObject } from './model/tsssign'; + +type SessionHandler = (message: INotification) => Promise; + +const sessionRegistry = new Map>(); +let dispatcherRegistered = false; + +function sessionKey(type: string, id: string | number): string { + return JSON.stringify([type, id]); +} + +function dispatchSessionMessage(message: INotification): void { + const handlers = sessionRegistry.get(sessionKey(message.type, message.id)); + for (const notify of [...(handlers ?? [])]) { + void notify(message).catch(err => { + logger.error('Error handling TSS session update: %o', err); + }); + } +} + +function subscribeToSession(type: string, id: string, handler: SessionHandler): () => void { + if (!dispatcherRegistered) { + WalletService.getMessageBroker().onMessage(dispatchSessionMessage); + dispatcherRegistered = true; + } + + const key = sessionKey(type, id); + let handlers = sessionRegistry.get(key); + if (!handlers) { + handlers = new Set(); + sessionRegistry.set(key, handlers); + } + handlers.add(handler); + + return () => { + if (!handlers.delete(handler)) { + return; + } + if (handlers.size === 0) { + sessionRegistry.delete(key); + } + }; +} + +/** + * Get a bounded wait time in milliseconds for TSS message retrieval. The wait time is bounded between 0 and 20 seconds. + * If no maxWaitTimeSec is provided, the default is maxSec seconds (default: 20). + * @param {number} maxWaitTimeSec The value to be bounded + * @param {number} maxSec The maximum wait time in seconds (default: 20) + * @param {number} minSec The minimum wait time in seconds (default: 0) + */ +function getBoundedWaitTime(maxWaitTimeSec?: number, maxSec = 20, minSec = 0): number { + maxWaitTimeSec = Math.max(isNaN(maxWaitTimeSec) ? maxSec : maxWaitTimeSec, minSec); + const maxWaitTime = Math.min(maxWaitTimeSec, maxSec) * 1000; + return maxWaitTime; +} + +/** + * Common function for listening to the completion of a TSS session's round. + */ +async function listenForSessionComplete(params: { + /** Message type to listen for */ + messageType: typeof TssKeyGenClass.TSS_KEYGEN_MESSAGE_TYPE | typeof TssSignClass.TSS_SIGGEN_MESSAGE_TYPE; + /** Session to listen for updates */ + session: T; + /** Function to determine if the round is complete */ + isComplete: (session: T) => boolean; + /** Function to fetch the latest session state */ + fetchSession: (params: { id: string }) => Promise; + /** Maximum time (in milliseconds) to wait for the round to complete */ + maxWaitTime: number; +}): Promise { + const { messageType, isComplete, fetchSession, maxWaitTime } = params; + let { session } = params; + + const events = new EventEmitter(); + const sessionUpdateHandler = async () => { + try { + const _session = await fetchSession({ id: session.id }); + if (isComplete(_session)) { + unsubscribe(); + events.emit('session', _session); + } + } catch (err) { + // Do not throw on possibly transient db connection errors. At worst, this runs until the maxWaitTime expires + logger.error('Error fetching updated TSS session: %o - %o', session.id, err); + } + }; + const unsubscribe = subscribeToSession(messageType, session.id, sessionUpdateHandler); + + // Listen for session update events over the message broker service + let timer: NodeJS.Timeout; + const sessionUpdate = Promise.race([ + new Promise(r => events.once('session', r)), + new Promise(r => timer = setTimeout(() => { unsubscribe(); r(session); }, maxWaitTime)) + ]); + + try { + // Check for an updated session one last time before awaiting the subscription. + // This is to prevent a race condition where the update arrives before we start listening for it. + const _session = await fetchSession({ id: session.id }); + if (isComplete(_session)) { + session = _session; + } else { + session = await sessionUpdate; + } + return session; + } finally { + unsubscribe(); + clearTimeout(timer); + events.removeAllListeners(); + } +} class TssKeyGenClass { + static TSS_KEYGEN_MESSAGE_TYPE = 'TssKeyGenMessage' as const; + + /** + * Check if a copayer is a participant in a TSS keygen session and return the session. + * Throws an error if the copayer is not a participant or if the session does not exist. + */ + async getSessionForCopayer(params: { + /** Session ID */ + id: string; + /** Copayer ID of the requesting party */ + copayerId: string; + }): Promise { + const { id, copayerId } = params; + const storage = WalletService.getStorage(); + const session = await storage.fetchTssKeyGenSession({ id }); + if (!session) { + throw Errors.TSS_SESSION_NOT_FOUND; + } + + const partyId = session.participants.indexOf(copayerId); + if (partyId === -1) { + throw Errors.TSS_NON_PARTICIPANT; + } + + return session; + } + /** * Get messages for a given party in a TSS keygen session. * Only returns messages if all other parties have sent their messages for the round. */ async getMessagesForParty(params: { - /** Session ID */ - id: string; + /** Session */ + session: TssKeyGenModel; /** Round number */ round: number; /** Copayer ID of the requesting party */ copayerId: string; + /** Maximum time (in seconds) to wait for a complete round */ + maxWaitTimeSec?: number; }): Promise<{ messages?: ITssKeyMessageObject[]; publicKey?: string; hasKeyBackup?: boolean; }> { - const { id, round, copayerId } = params; + const { round, copayerId } = params; + let { session } = params; + const maxWaitTime = getBoundedWaitTime(params.maxWaitTimeSec); - const storage = WalletService.getStorage(); - const session = await storage.fetchTssKeyGenSession({ id }); - if (!session) { - throw Errors.TSS_SESSION_NOT_FOUND; - } if (!session.rounds[round]) { return {}; } @@ -40,15 +182,32 @@ class TssKeyGenClass { throw Errors.TSS_NON_PARTICIPANT; } - const otherPartyMsgs = session.rounds[round].filter(m => m.fromPartyId != partyId); + const isRoundComplete = (session) => { + const otherPartyMsgs = session.rounds[round].filter(m => m.fromPartyId != partyId); + return otherPartyMsgs.length === session.n - 1; + }; + + if (!isRoundComplete(session)) { + const storage = WalletService.getStorage(); + session = await listenForSessionComplete({ + messageType: TssKeyGenClass.TSS_KEYGEN_MESSAGE_TYPE, + session, + isComplete: isRoundComplete, + fetchSession: storage.fetchTssKeyGenSession.bind(storage), + maxWaitTime + }); + } + + // Only return message if all other parties have sent their messages. // This is to prevent complexity in TSS session management when processing rounds. There's // no value in partially processing rounds with missing messages and messages can't be // re-processed, so it makes sense to only return messages when the round is complete. - if (otherPartyMsgs.length !== session.n - 1) { + if (!isRoundComplete(session)) { return {}; } + const otherPartyMsgs = session.rounds[round].filter(m => m.fromPartyId != partyId); const messages = otherPartyMsgs.map(m => m.messages); for (const m of messages) { m.p2pMessages = m.p2pMessages.filter(m => m.to == partyId); @@ -121,7 +280,7 @@ class TssKeyGenClass { let result = false; while (!result) { - result = await this._pushMessage({ id, session, message, storage }); + result = await this._pushMessage({ session, message, storage }); if (!result) { session = await storage.fetchTssKeyGenSession({ id }); } @@ -225,8 +384,6 @@ class TssKeyGenClass { * This will fail if the round is already complete or if the message is from a party that has already sent a message for the round. */ private async _pushMessage(params: { - /** Session ID */ - id: string; /** TSS keygen session fetched from BWS storage */ session: TssKeyGenModel; /** Message to push to the session */ @@ -234,7 +391,8 @@ class TssKeyGenClass { /** BWS storage instance */ storage: Storage; }) { - const { id, session, message, storage } = params; + const { session, message, storage } = params; + const { id } = session; const { round } = message; const currentRound = session.getCurrentRound(); @@ -248,6 +406,8 @@ class TssKeyGenClass { if (existing) { throw Errors.TSS_ROUND_MESSAGE_EXISTS; } + + const messageBroker = WalletService.getMessageBroker(); try { const result = await storage.storeTssKeyGenMessage({ id, message, __v: session.__v }); @@ -255,6 +415,7 @@ class TssKeyGenClass { logger.error('Failed to store TSS key generation message %o %o %o', id, result, message); throw Errors.TSS_GENERIC_ERROR.withMessage('Failed to store TSS key generation message'); } + messageBroker.send({ type: TssKeyGenClass.TSS_KEYGEN_MESSAGE_TYPE, id } as INotification); return true; } catch (e) { if (e?.message?.startsWith('MONGO_DOC_OUTDATED')) { @@ -373,25 +534,52 @@ class TssKeyGenClass { export const TssKeyGen = new TssKeyGenClass(); class TssSignClass { + static TSS_SIGGEN_MESSAGE_TYPE = 'TssSigMessage' as const; + + /** + * Check if a copayer is a participant in a TSS signature session and return the session. + * Throws an error if the copayer is not a participant or if the session does not exist. + */ + async getSessionForCopayer(params: { + /** Session ID */ + id: string; + /** Copayer ID of the requesting party */ + copayerId: string; + }): Promise { + const { id, copayerId } = params; + const storage = WalletService.getStorage(); + const session = await storage.fetchTssSigSession({ id }); + if (!session) { + throw Errors.TSS_SESSION_NOT_FOUND; + } + + const party = session.participants.find(p => p.copayerId === copayerId); + if (!party) { + throw Errors.TSS_NON_PARTICIPANT; + } + + return session; + } + + /** * Get messages for a given party in a TSS signature session. * Only returns messages if all other parties have sent their messages for the round. */ async getMessagesForParty(params: { /** Session ID */ - id: string; + session: TssSigGenModel; /** Round number */ round: number; /** Copayer ID of the requesting party */ copayerId: string; + /** Maximum time (in seconds) to wait for a complete round */ + maxWaitTimeSec?: number; }): Promise<{ messages?: ITssSigMessageObject[]; signature?: ITssSigMessageObject['signature']; participants?: string[] }> { - const { id, round, copayerId } = params; + const { round, copayerId } = params; + let { session } = params; + const maxWaitTime = getBoundedWaitTime(params.maxWaitTimeSec); - const storage = WalletService.getStorage(); - const session = await storage.fetchTssSigSession({ id }); - if (!session) { - throw Errors.TSS_SESSION_NOT_FOUND; - } if (!session.rounds[round]) { return {}; } @@ -401,20 +589,37 @@ class TssSignClass { throw Errors.TSS_NON_PARTICIPANT; } + const isRoundComplete = (session) => { + const otherPartyMsgs = session.rounds[round].filter(m => m.fromPartyId != party.partyId); + return otherPartyMsgs.length === session.m - 1; + }; + + if (!isRoundComplete(session)) { + const storage = WalletService.getStorage(); + session = await listenForSessionComplete({ + messageType: TssSignClass.TSS_SIGGEN_MESSAGE_TYPE, + session, + isComplete: isRoundComplete, + fetchSession: storage.fetchTssSigSession.bind(storage), + maxWaitTime + }); + } + const otherPartyMsgs = session.rounds[round].filter(m => m.fromPartyId != party.partyId); const participants = otherPartyMsgs.map(m => { const p = session.participants.find(p => p.partyId === m.fromPartyId); return p?.copayerId; }).filter(Boolean) as string[]; - if (otherPartyMsgs.length === session.m - 1) { - const messages = otherPartyMsgs.map(m => m.messages); - for (const m of messages) { - m.p2pMessages = m.p2pMessages.filter(m => m.to == party.partyId); - } - return { messages, signature: session.signature, participants }; + if (!isRoundComplete(session)) { + return { participants }; + } + + const messages = otherPartyMsgs.map(m => m.messages); + for (const m of messages) { + m.p2pMessages = m.p2pMessages.filter(m => m.to == party.partyId); } - return { participants }; + return { messages, signature: session.signature, participants }; } /** @@ -486,7 +691,7 @@ class TssSignClass { let result = false; while (!result) { - result = await this._pushMessage({ id, session, message, storage }); + result = await this._pushMessage({ session, message, storage }); // `result` will be false if the session was stale (version conflict) and we need to retry // Any other failure of the message state will result in a throw (e.g. same-message race condition, round already done, etc.) if (!result) { @@ -563,8 +768,6 @@ class TssSignClass { * Push a TSS signature message to the session. */ private async _pushMessage(params: { - /** Session ID */ - id: string; /** TSS sig generation session fetched from BWS storage */ session: TssSigGenModel; /** TSS signature message to be pushed */ @@ -572,7 +775,8 @@ class TssSignClass { /** BWS storage instance */ storage: Storage; }): Promise { - const { id, session, message, storage } = params; + const { session, message, storage } = params; + const { id } = session; const { round } = message; const currentRound = session.getCurrentRound(); @@ -587,12 +791,15 @@ class TssSignClass { throw Errors.TSS_ROUND_MESSAGE_EXISTS; } + const messageBroker = WalletService.getMessageBroker(); + try { const result = await storage.storeTssSigMessage({ id, message, __v: session.__v }); if (!result.result.ok) { logger.error('Failed to store TSS key generation message %o %o %o', id, result, message); throw Errors.TSS_GENERIC_ERROR.withMessage('Failed to store TSS key generation message'); } + messageBroker.send({ type: TssSignClass.TSS_SIGGEN_MESSAGE_TYPE, id } as INotification); return true; } catch (e) { if (e?.message?.startsWith('MONGO_DOC_OUTDATED')) { diff --git a/packages/bitcore-wallet-service/test/integration/server.test.ts b/packages/bitcore-wallet-service/test/integration/server.test.ts index 37a878ae677..922606a1f78 100644 --- a/packages/bitcore-wallet-service/test/integration/server.test.ts +++ b/packages/bitcore-wallet-service/test/integration/server.test.ts @@ -1665,7 +1665,7 @@ describe('Wallet service', function() { message: { partyId: 0, broadcastMessages: [], p2pMessages: [], publicKey: 'dummy', round: 0 }, n: 1, copayerId: legitCopayerId, - version: 1.1, + version: Defaults.TSS_KEYGEN_SCHEME_VERSION }); session.sharedPublicKey = 'dummy-shared-public-key'; await server.storage.db.collection('tss_keygen').deleteMany({ id: session.id }); @@ -1732,7 +1732,7 @@ describe('Wallet service', function() { message: { partyId: 0, broadcastMessages: [], p2pMessages: [], publicKey: 'dummy', round: 0 }, n: 1, copayerId: ancillaryDerivedCopayerId, - version: 1.1, + version: Defaults.TSS_KEYGEN_SCHEME_VERSION }); session.sharedPublicKey = 'dummy-shared-public-key'; await server.storage.db.collection('tss_keygen').deleteMany({ id: session.id }); diff --git a/packages/bitcore-wallet-service/test/messagebroker.test.ts b/packages/bitcore-wallet-service/test/messagebroker.test.ts new file mode 100644 index 00000000000..ee226ecfcce --- /dev/null +++ b/packages/bitcore-wallet-service/test/messagebroker.test.ts @@ -0,0 +1,137 @@ +'use strict'; + +import 'chai/register-should'; +import * as sinon from 'sinon'; +import * as chai from 'chai'; +import io from 'socket.io-client'; +import logger from '../src/lib/logger'; +import { MessageBroker } from '../src/lib/messagebroker'; +import type { INotification } from '../src/lib/model/notification'; + +const should = chai.should(); + +describe('MessageBroker', function() { + const sandbox = sinon.createSandbox(); + const opts = { + messageBrokerServer: { + url: 'http://dummy:3380' + } + }; + + beforeEach(function() { + sandbox.stub(io, 'connect').returns({ + on: sandbox.stub(), + emit: sandbox.stub() + }); + sandbox.stub(logger, 'info'); + sandbox.stub(logger, 'warn'); + sandbox.stub(logger, 'error'); + }); + + afterEach(function() { + sandbox.restore(); + }); + + describe('#constructor', function() { + it('should create a local MessageBroker instance without opts', function() { + const mb = new MessageBroker(null); + mb.should.be.instanceof(MessageBroker); + should.not.exist(mb.mq); + mb.remote.should.equal(false); + }); + + it('should create a fleshed out MessageBroker instance', function() { + const mb = new MessageBroker(opts); + mb.should.be.instanceof(MessageBroker); + should.exist(mb.mq); + mb.remote.should.equal(true); + const onStub = mb.mq.on as sinon.SinonStub; + onStub.callCount.should.equal(3); + onStub.getCall(0).args[0].should.equal('connect'); + onStub.getCall(1).args[0].should.equal('connect_error'); + onStub.getCall(2).args[0].should.equal('msg'); + (logger.info as sinon.SinonStub).callCount.should.equal(1); + }); + }); + + describe('#send', function() { + it('should emit a message when remote is false', function() { + const mb = new MessageBroker(null); + const emitSpy = sandbox.spy(mb, 'emit'); + const data = { type: 'test' }; + mb.send(data as INotification); + emitSpy.calledOnce.should.equal(true); + emitSpy.calledWith('msg', data).should.equal(true); + }); + + it('should emit a message when remote is true', function() { + const mb = new MessageBroker(opts); + const emitSpy = mb.mq.emit as sinon.SinonSpy; + const data = { type: 'test' }; + mb.send(data as INotification); + emitSpy.calledOnce.should.equal(true); + emitSpy.calledWith('msg', data).should.equal(true); + }); + }); + + describe('#onMessage', function() { + it('should register a message handler', function() { + const mb = new MessageBroker(null); + const handler = sandbox.stub(); + mb.onMessage(handler); + mb.emit('msg', { type: 'test' }); + handler.calledOnce.should.equal(true); + }); + + it('should register multiple message handlers', function() { + const mb = new MessageBroker(null); + const handler1 = sandbox.stub(); + const handler2 = sandbox.stub(); + mb.onMessage(handler1); + mb.onMessage(handler2); + mb.emit('msg', { type: 'test' }); + handler1.calledOnce.should.equal(true); + handler2.calledOnce.should.equal(true); + }); + + it('should register the same handler twice', function() { + const mb = new MessageBroker(null); + const handler = sandbox.stub(); + mb.onMessage(handler); + mb.onMessage(handler); + mb.emit('msg', { type: 'test' }); + handler.calledTwice.should.equal(true); + }); + }); + + describe('#offMessage', function() { + it('should unregister a message handler', function() { + const mb = new MessageBroker(null); + const handler = sandbox.stub(); + mb.onMessage(handler); + mb.offMessage(handler); + mb.emit('msg', { type: 'test' }); + handler.called.should.equal(false); + }); + + it('should unregister a message handler closure', function() { + const mb = new MessageBroker(null); + const closure = sandbox.stub(); + const handler = (message: INotification) => { + closure(message); + }; + + // Register the same closure more than once. + mb.onMessage(handler); + mb.onMessage(handler); + mb.listeners('msg').length.should.equal(2); + + // Unsubscribe should remove only one instance of the same closure. + mb.offMessage(handler); + mb.listeners('msg').length.should.equal(1); + + mb.emit('msg', { type: 'test' }); + closure.calledOnce.should.equal(true); + }); + }); +}); diff --git a/packages/bitcore-wallet-service/test/tss.test.ts b/packages/bitcore-wallet-service/test/tss.test.ts new file mode 100644 index 00000000000..b33a58e2ce4 --- /dev/null +++ b/packages/bitcore-wallet-service/test/tss.test.ts @@ -0,0 +1,165 @@ +import { expect } from 'chai'; +import sinon from 'sinon'; +import { MessageBroker } from '../src/lib/messagebroker'; +import { TssKeyGenModel } from '../src/lib/model/tsskeygen'; +import { TssSigGenModel } from '../src/lib/model/tsssign'; +import { WalletService } from '../src/lib/server'; +import { Storage } from '../src/lib/storage'; + +describe('TSS session subscriptions', function() { + const sandbox = sinon.createSandbox(); + const broker = new MessageBroker(null); + let TssKeyGen: typeof import('../src/lib/tss').TssKeyGen; + let TssSign: typeof import('../src/lib/tss').TssSign; + let storage: Storage; + let clock: sinon.SinonFakeTimers; + let session: TssKeyGenModel; + let fetchSession: sinon.SinonStub; + + function poll(maxWaitTimeSec = 20) { + return TssKeyGen.getMessagesForParty({ session, round: 0, copayerId: 'alice', maxWaitTimeSec }); + } + + function completeSession() { + return Object.assign(new TssKeyGenModel(), session, { + rounds: [[{ + fromPartyId: 1, + messages: { partyId: 1, round: 0, publicKey: '', broadcastMessages: [], p2pMessages: [] } + }]] + }); + } + + before(async function() { + const modulePath = require.resolve('../src/lib/tss'); + const cachedModule = require.cache[modulePath]; + delete require.cache[modulePath]; + try { + ({ TssKeyGen, TssSign } = await import(modulePath)); + } finally { + if (cachedModule) { + require.cache[modulePath] = cachedModule; + } else { + delete require.cache[modulePath]; + } + } + }); + + beforeEach(function() { + clock = sandbox.useFakeTimers(); + storage = new Storage(); + sandbox.stub(WalletService, 'getMessageBroker').returns(broker); + sandbox.stub(WalletService, 'getStorage').returns(storage); + session = Object.assign(new TssKeyGenModel(), { + id: 'session', n: 2, participants: ['alice', 'bob'], rounds: [[]] + }); + fetchSession = sandbox.stub(storage, 'fetchTssKeyGenSession').resolves(session); + }); + + afterEach(function() { + sandbox.restore(); + }); + + it('uses one broker listener for 20 polls and removes completed subscriptions', async function() { + const existingHandler = sandbox.stub(); + broker.onMessage(existingHandler); + try { + const polls = Array.from({ length: 20 }, () => poll()); + expect(broker.listenerCount('msg')).to.equal(2); + + fetchSession.resolves(completeSession()); + broker.emit('msg', { type: 'TssKeyGenMessage', id: session.id }); + const results = await Promise.all(polls); + expect(results.every(result => result.messages.length === 1)).to.equal(true); + expect(existingHandler.calledOnce).to.equal(true); + expect(clock.countTimers()).to.equal(0); + + fetchSession.resetHistory(); + broker.emit('msg', { type: 'TssKeyGenMessage', id: session.id }); + expect(fetchSession.called).to.equal(false); + expect(broker.listenerCount('msg')).to.equal(2); + } finally { + broker.offMessage(existingHandler); + } + }); + + it('routes by session ID and message type for keygen and signing', async function() { + const signingSession = Object.assign(new TssSigGenModel(), { + id: session.id, m: 2, participants: [{ partyId: 0, copayerId: 'alice' }], rounds: [[]] + }); + const fetchSigning = sandbox.stub(storage, 'fetchTssSigSession').resolves(signingSession); + const keygenPoll = poll(1); + const signingPoll = TssSign.getMessagesForParty({ + session: signingSession, round: 0, copayerId: 'alice', maxWaitTimeSec: 1 + }); + fetchSession.resetHistory(); + fetchSigning.resetHistory(); + + broker.emit('msg', { type: 'TssKeyGenMessage', id: 'other-session' }); + broker.emit('msg', { type: 'unrelated', id: session.id }); + expect(fetchSession.called).to.equal(false); + expect(fetchSigning.called).to.equal(false); + broker.emit('msg', { type: 'TssKeyGenMessage', id: session.id }); + expect(fetchSession.calledOnce).to.equal(true); + expect(fetchSigning.called).to.equal(false); + broker.emit('msg', { type: 'TssSigMessage', id: session.id }); + expect(fetchSigning.calledOnce).to.equal(true); + expect(fetchSession.calledOnce).to.equal(true); + expect(broker.listenerCount('msg')).to.equal(1); + + clock.tick(1000); + await Promise.all([keygenPoll, signingPoll]); + fetchSession.resetHistory(); + fetchSigning.resetHistory(); + broker.emit('msg', { type: 'TssKeyGenMessage', id: session.id }); + broker.emit('msg', { type: 'TssSigMessage', id: session.id }); + expect(fetchSession.called).to.equal(false); + expect(fetchSigning.called).to.equal(false); + expect(broker.listenerCount('msg')).to.equal(1); + }); + + it('removes only the timed-out poll and allows later subscriptions', async function() { + const shortPoll = poll(1); + const longPoll = poll(2); + clock.tick(1000); + expect(await shortPoll).to.deep.equal({}); + expect(broker.listenerCount('msg')).to.equal(1); + fetchSession.resetHistory(); + fetchSession.resolves(completeSession()); + broker.emit('msg', { type: 'TssKeyGenMessage', id: session.id }); + expect((await longPoll).messages).to.have.length(1); + expect(fetchSession.calledOnce).to.equal(true); + expect(broker.listenerCount('msg')).to.equal(1); + + fetchSession.resolves(session); + const laterPoll = poll(1); + expect(broker.listenerCount('msg')).to.equal(1); + fetchSession.resolves(completeSession()); + broker.emit('msg', { type: 'TssKeyGenMessage', id: session.id }); + expect((await laterPoll).messages).to.have.length(1); + expect(broker.listenerCount('msg')).to.equal(1); + expect(clock.countTimers()).to.equal(0); + }); + + it('cleans up when the final database recheck finds a complete round', async function() { + fetchSession.resolves(completeSession()); + expect((await poll()).messages).to.have.length(1); + expect(broker.listenerCount('msg')).to.equal(1); + expect(clock.countTimers()).to.equal(0); + fetchSession.resetHistory(); + broker.emit('msg', { type: 'TssKeyGenMessage', id: session.id }); + expect(fetchSession.called).to.equal(false); + }); + + it('cleans up when the final database recheck rejects', async function() { + const failure = new Error('database unavailable'); + fetchSession.rejects(failure); + const result = await poll().then(() => undefined, err => err); + expect(result).to.equal(failure); + expect(broker.listenerCount('msg')).to.equal(1); + expect(clock.countTimers()).to.equal(0); + fetchSession.resetHistory(); + broker.emit('msg', { type: 'TssKeyGenMessage', id: session.id }); + expect(fetchSession.called).to.equal(false); + }); + +}); \ No newline at end of file