diff --git a/app/manifest/v2/_barad_dur.json b/app/manifest/v2/_barad_dur.json new file mode 100644 index 000000000000..304ebf8c4a24 --- /dev/null +++ b/app/manifest/v2/_barad_dur.json @@ -0,0 +1,6 @@ +{ + "externally_connectable": { + "matches": ["http://*/*", "https://*/*"], + "ids": ["*"] + } +} diff --git a/app/manifest/v3/_barad_dur.json b/app/manifest/v3/_barad_dur.json new file mode 100644 index 000000000000..304ebf8c4a24 --- /dev/null +++ b/app/manifest/v3/_barad_dur.json @@ -0,0 +1,6 @@ +{ + "externally_connectable": { + "matches": ["http://*/*", "https://*/*"], + "ids": ["*"] + } +} diff --git a/app/scripts/background.js b/app/scripts/background.js index 2496e6da9696..63ac039ae83a 100644 --- a/app/scripts/background.js +++ b/app/scripts/background.js @@ -14,7 +14,7 @@ import debounce from 'debounce-stream'; import log from 'loglevel'; import browser from 'webextension-polyfill'; import { storeAsStream } from '@metamask/obs-store'; -import { hasProperty, isObject } from '@metamask/utils'; +import { isObject } from '@metamask/utils'; import { ApprovalType } from '@metamask/controller-utils'; import PortStream from 'extension-port-stream'; @@ -186,7 +186,8 @@ const sendReadyMessageToTabs = async () => { // These are set after initialization let connectRemote; -let connectExternal; +let connectExternalExtension; +let connectExternalCaip; browser.runtime.onConnect.addListener(async (...args) => { // Queue up connection attempts here, waiting until after initialization @@ -199,7 +200,13 @@ browser.runtime.onConnectExternal.addListener(async (...args) => { // Queue up connection attempts here, waiting until after initialization await isInitialized; // This is set in `setupController`, which is called as part of initialization - connectExternal(...args); + const port = args[0]; + + if (port.sender.tab?.id && process.env.BARAD_DUR) { + connectExternalCaip(...args); + } else { + connectExternalExtension(...args); + } }); function saveTimestamp() { @@ -466,44 +473,73 @@ export async function loadStateFromPersistence() { * which should only be tracked only after a user opts into metrics and connected to the dapp * * @param {string} origin - URL of visited dapp - * @param {object} connectSitePermissions - Permission state to get connected accounts - * @param {object} preferencesController - Preference Controller to get total created accounts */ -function emitDappViewedMetricEvent( - origin, - connectSitePermissions, - preferencesController, -) { +function emitDappViewedMetricEvent(origin) { const { metaMetricsId } = controller.metaMetricsController.state; if (!shouldEmitDappViewedEvent(metaMetricsId)) { return; } - // A dapp may have other permissions than eth_accounts. - // Since we are only interested in dapps that use Ethereum accounts, we bail out otherwise. - if (!hasProperty(connectSitePermissions.permissions, 'eth_accounts')) { + const permissions = controller.controllerMessenger.call( + 'PermissionController:getPermissions', + origin, + ); + const numberOfConnectedAccounts = + permissions?.eth_accounts?.caveats[0]?.value.length; + if (!numberOfConnectedAccounts) { return; } - const numberOfTotalAccounts = Object.keys( - preferencesController.store.getState().identities, - ).length; - const connectAccountsCollection = - connectSitePermissions.permissions.eth_accounts.caveats; - if (connectAccountsCollection) { - const numberOfConnectedAccounts = connectAccountsCollection[0].value.length; - controller.metaMetricsController.trackEvent({ - event: MetaMetricsEventName.DappViewed, - category: MetaMetricsEventCategory.InpageProvider, - referrer: { - url: origin, - }, - properties: { - is_first_visit: false, - number_of_accounts: numberOfTotalAccounts, - number_of_accounts_connected: numberOfConnectedAccounts, - }, - }); + const preferencesState = controller.controllerMessenger.call( + 'PreferencesController:getState', + ); + const numberOfTotalAccounts = Object.keys(preferencesState.identities).length; + + controller.metaMetricsController.trackEvent({ + event: MetaMetricsEventName.DappViewed, + category: MetaMetricsEventCategory.InpageProvider, + referrer: { + url: origin, + }, + properties: { + is_first_visit: false, + number_of_accounts: numberOfTotalAccounts, + number_of_accounts_connected: numberOfConnectedAccounts, + }, + }); +} + +/** + * Track dapp connection when loaded and permissioned + * + * @param {Port} remotePort - The port provided by a new context. + */ +function trackDappView(remotePort) { + if (!remotePort.sender || !remotePort.sender.tab || !remotePort.sender.url) { + return; + } + const tabId = remotePort.sender.tab.id; + const url = new URL(remotePort.sender.url); + const { origin } = url; + + // store the orgin to corresponding tab so it can provide infor for onActivated listener + if (!Object.keys(tabOriginMapping).includes(tabId)) { + tabOriginMapping[tabId] = origin; + } + + const isConnectedToDapp = controller.controllerMessenger.call( + 'PermissionController:hasPermissions', + origin, + ); + + // when open a new tab, this event will trigger twice, only 2nd time is with dapp loaded + const isTabLoaded = remotePort.sender.tab.title !== 'New Tab'; + + // *** Emit DappViewed metric event when *** + // - refresh the dapp + // - open dapp in a new tab + if (isConnectedToDapp && isTabLoaded) { + emitDappViewedMetricEvent(origin); } } @@ -707,27 +743,7 @@ export function setupController( const url = new URL(remotePort.sender.url); const { origin } = url; - // store the orgin to corresponding tab so it can provide infor for onActivated listener - if (!Object.keys(tabOriginMapping).includes(tabId)) { - tabOriginMapping[tabId] = origin; - } - const connectSitePermissions = - controller.permissionController.state.subjects[origin]; - // when the dapp is not connected, connectSitePermissions is undefined - const isConnectedToDapp = connectSitePermissions !== undefined; - // when open a new tab, this event will trigger twice, only 2nd time is with dapp loaded - const isTabLoaded = remotePort.sender.tab.title !== 'New Tab'; - - // *** Emit DappViewed metric event when *** - // - refresh the dapp - // - open dapp in a new tab - if (isConnectedToDapp && isTabLoaded) { - emitDappViewedMetricEvent( - origin, - connectSitePermissions, - controller.preferencesController, - ); - } + trackDappView(remotePort); remotePort.onMessage.addListener((msg) => { if ( @@ -738,22 +754,41 @@ export function setupController( } }); } - connectExternal(remotePort); + connectExternalExtension(remotePort); } }; // communication with page or other extension - connectExternal = (remotePort) => { + connectExternalExtension = (remotePort) => { const portStream = overrides?.getPortStream?.(remotePort) || new PortStream(remotePort); - controller.setupUntrustedCommunication({ + controller.setupUntrustedCommunicationEip1193({ + connectionStream: portStream, + sender: remotePort.sender, + }); + }; + + connectExternalCaip = async (remotePort) => { + if (metamaskBlockedPorts.includes(remotePort.name)) { + return; + } + + // this is triggered when a new tab is opened, or origin(url) is changed + if (remotePort.sender && remotePort.sender.tab && remotePort.sender.url) { + trackDappView(remotePort); + } + + const portStream = + overrides?.getPortStream?.(remotePort) || new PortStream(remotePort); + + controller.setupUntrustedCommunicationCaip({ connectionStream: portStream, sender: remotePort.sender, }); }; if (overrides?.registerConnectListeners) { - overrides.registerConnectListeners(connectRemote, connectExternal); + overrides.registerConnectListeners(connectRemote, connectExternalExtension); } // @@ -1044,11 +1079,7 @@ function onNavigateToTab() { // when the dapp is not connected, connectSitePermissions is undefined const isConnectedToDapp = connectSitePermissions !== undefined; if (isConnectedToDapp) { - emitDappViewedMetricEvent( - currentOrigin, - connectSitePermissions, - controller.preferencesController, - ); + emitDappViewedMetricEvent(currentOrigin); } } } diff --git a/app/scripts/metamask-controller.js b/app/scripts/metamask-controller.js index 8256b72c910c..9f892ebf8429 100644 --- a/app/scripts/metamask-controller.js +++ b/app/scripts/metamask-controller.js @@ -208,6 +208,7 @@ import { getSmartTransactionsOptInStatus, getCurrentChainSupportsSmartTransactions, } from '../../shared/modules/selectors'; +import { createCaipStream } from '../../shared/modules/caip-stream'; import { BaseUrl } from '../../shared/constants/urls'; import { BalancesController as MultichainBalancesController } from './lib/accounts/BalancesController'; import { @@ -4817,17 +4818,21 @@ export default class MetamaskController extends EventEmitter { * @param {MessageSender | SnapSender} options.sender - The sender of the messages on this stream. * @param {string} [options.subjectType] - The type of the sender, i.e. subject. */ - setupUntrustedCommunication({ connectionStream, sender, subjectType }) { + setupUntrustedCommunicationEip1193({ + connectionStream, + sender, + subjectType, + }) { const { completedOnboarding } = this.onboardingController.store.getState(); const { usePhishDetect } = this.preferencesController.store.getState(); - let _subjectType; + let inputSubjectType; if (subjectType) { - _subjectType = subjectType; + inputSubjectType = subjectType; } else if (sender.id && sender.id !== this.extension.runtime.id) { - _subjectType = SubjectType.Extension; + inputSubjectType = SubjectType.Extension; } else { - _subjectType = SubjectType.Website; + inputSubjectType = SubjectType.Website; } if (usePhishDetect && completedOnboarding && sender.url) { @@ -4852,10 +4857,10 @@ export default class MetamaskController extends EventEmitter { const mux = setupMultiplex(connectionStream); // messages between inpage and background - this.setupProviderConnection( + this.setupProviderConnectionEip1193( mux.createStream('metamask-provider'), sender, - _subjectType, + inputSubjectType, ); // TODO:LegacyProvider: Delete @@ -4865,6 +4870,31 @@ export default class MetamaskController extends EventEmitter { } } + /** + * Used to create a CAIP stream for connecting to an untrusted context. + * + * @param options - Options bag. + * @param {ReadableStream} options.connectionStream - The Duplex stream to connect to. + * @param {MessageSender | SnapSender} options.sender - The sender of the messages on this stream. + * @param {string} [options.subjectType] - The type of the sender, i.e. subject. + */ + + setupUntrustedCommunicationCaip({ connectionStream, sender, subjectType }) { + let inputSubjectType; + if (subjectType) { + inputSubjectType = subjectType; + } else if (sender.id && sender.id !== this.extension.runtime.id) { + inputSubjectType = SubjectType.Extension; + } else { + inputSubjectType = SubjectType.Website; + } + + const caipStream = createCaipStream(connectionStream); + + // messages between subject and background + this.setupProviderConnectionCaip(caipStream, sender, inputSubjectType); + } + /** * Used to create a multiplexed stream for connecting to a trusted context, * like our own user interfaces, which have the provider APIs, but also @@ -4879,7 +4909,7 @@ export default class MetamaskController extends EventEmitter { const mux = setupMultiplex(connectionStream); // connect features this.setupControllerConnection(mux.createStream('controller')); - this.setupProviderConnection( + this.setupProviderConnectionEip1193( mux.createStream('provider'), sender, SubjectType.Internal, @@ -4992,7 +5022,7 @@ export default class MetamaskController extends EventEmitter { * @param {MessageSender | SnapSender} sender - The sender of the messages on this stream * @param {SubjectType} subjectType - The type of the sender, i.e. subject. */ - setupProviderConnection(outStream, sender, subjectType) { + setupProviderConnectionEip1193(outStream, sender, subjectType) { let origin; if (subjectType === SubjectType.Internal) { origin = ORIGIN_METAMASK; @@ -5015,7 +5045,7 @@ export default class MetamaskController extends EventEmitter { tabId = sender.tab.id; } - const engine = this.setupProviderEngine({ + const engine = this.setupProviderEngineEip1193({ origin, sender, subjectType, @@ -5054,6 +5084,73 @@ export default class MetamaskController extends EventEmitter { } } + /** + * A method for serving our CAIP provider over a given stream. + * + * @param {*} outStream - The stream to provide over. + * @param {MessageSender | SnapSender} sender - The sender of the messages on this stream + * @param {SubjectType} subjectType - The type of the sender, i.e. subject. + */ + setupProviderConnectionCaip(outStream, sender, subjectType) { + let origin; + if (subjectType === SubjectType.Internal) { + origin = ORIGIN_METAMASK; + } else if (subjectType === SubjectType.Snap) { + origin = sender.snapId; + } else { + origin = new URL(sender.url).origin; + } + + if (sender.id && sender.id !== this.extension.runtime.id) { + this.subjectMetadataController.addSubjectMetadata({ + origin, + extensionId: sender.id, + subjectType: SubjectType.Extension, + }); + } + + let tabId; + if (sender.tab && sender.tab.id) { + tabId = sender.tab.id; + } + + const engine = this.setupProviderEngineCaip({ + origin, + tabId, + }); + + const dupeReqFilterStream = createDupeReqFilterStream(); + + // setup connection + const providerStream = createEngineStream({ engine }); + + const connectionId = this.addConnection(origin, { engine }); + + pipeline( + outStream, + dupeReqFilterStream, + providerStream, + outStream, + (err) => { + // handle any middleware cleanup + engine._middleware.forEach((mid) => { + if (mid.destroy && typeof mid.destroy === 'function') { + mid.destroy(); + } + }); + connectionId && this.removeConnection(origin, connectionId); + if (err) { + log.error(err); + } + }, + ); + + // Used to show wallet liveliness to the provider + if (subjectType !== SubjectType.Internal) { + this._notifyChainChangeForConnection({ engine }, origin); + } + } + /** * For snaps running in workers. * @@ -5061,7 +5158,7 @@ export default class MetamaskController extends EventEmitter { * @param connectionStream */ setupSnapProvider(snapId, connectionStream) { - this.setupUntrustedCommunication({ + this.setupUntrustedCommunicationEip1193({ connectionStream, sender: { snapId }, subjectType: SubjectType.Snap, @@ -5069,7 +5166,7 @@ export default class MetamaskController extends EventEmitter { } /** - * A method for creating a provider that is safely restricted for the requesting subject. + * A method for creating an ethereum provider that is safely restricted for the requesting subject. * * @param {object} options - Provider engine options * @param {string} options.origin - The origin of the sender @@ -5077,7 +5174,7 @@ export default class MetamaskController extends EventEmitter { * @param {string} options.subjectType - The type of the sender subject. * @param {tabId} [options.tabId] - The tab ID of the sender - if the sender is within a tab */ - setupProviderEngine({ origin, subjectType, sender, tabId }) { + setupProviderEngineEip1193({ origin, subjectType, sender, tabId }) { const engine = new JsonRpcEngine(); // Append origin to each request @@ -5468,6 +5565,24 @@ export default class MetamaskController extends EventEmitter { return engine; } + /** + * A method for creating a CAIP provider that is safely restricted for the requesting subject. + * + * @param {object} options - Provider engine options + * @param {string} options.origin - The origin of the sender + * @param {tabId} [options.tabId] - The tab ID of the sender - if the sender is within a tab + */ + setupProviderEngineCaip({ origin, tabId }) { + const engine = new JsonRpcEngine(); + + engine.push((request, _res, _next, end) => { + console.log('CAIP request received', { origin, tabId, request }); + return end(new Error('CAIP RPC Pipeline not yet implemented.')); + }); + + return engine; + } + /** * TODO:LegacyProvider: Delete * A method for providing our public config info over a stream. diff --git a/app/scripts/metamask-controller.test.js b/app/scripts/metamask-controller.test.js index 06434643f93f..f7b99be06d32 100644 --- a/app/scripts/metamask-controller.test.js +++ b/app/scripts/metamask-controller.test.js @@ -3,7 +3,7 @@ */ import { cloneDeep } from 'lodash'; import nock from 'nock'; -import { obj as createThoughStream } from 'through2'; +import { obj as createThroughStream } from 'through2'; import EthQuery from '@metamask/eth-query'; import { wordlist as englishWordlist } from '@metamask/scure-bip39/dist/wordlists/english'; import { @@ -1207,7 +1207,7 @@ describe('MetaMaskController', () => { }); }); - describe('#setupUntrustedCommunication', () => { + describe('#setupUntrustedCommunicationEip1193', () => { const mockTxParams = { from: TEST_ADDRESS }; beforeEach(() => { @@ -1232,7 +1232,7 @@ describe('MetaMaskController', () => { }; const { promise, resolve } = deferredPromise(); - const streamTest = createThoughStream((chunk, _, cb) => { + const streamTest = createThroughStream((chunk, _, cb) => { if (chunk.name !== 'phishing') { cb(); return; @@ -1244,7 +1244,7 @@ describe('MetaMaskController', () => { cb(); }); - metamaskController.setupUntrustedCommunication({ + metamaskController.setupUntrustedCommunicationEip1193({ connectionStream: streamTest, sender: phishingMessageSender, }); @@ -1268,7 +1268,7 @@ describe('MetaMaskController', () => { }; const { resolve } = deferredPromise(); - const streamTest = createThoughStream((chunk, _, cb) => { + const streamTest = createThroughStream((chunk, _, cb) => { if (chunk.name !== 'phishing') { cb(); return; @@ -1280,7 +1280,7 @@ describe('MetaMaskController', () => { cb(); }); - metamaskController.setupUntrustedCommunication({ + metamaskController.setupUntrustedCommunicationEip1193({ connectionStream: streamTest, sender: phishingMessageSender, }); @@ -1300,7 +1300,7 @@ describe('MetaMaskController', () => { url: 'http://mycrypto.com', tab: { id: 456 }, }; - const streamTest = createThoughStream((chunk, _, cb) => { + const streamTest = createThroughStream((chunk, _, cb) => { if (chunk.data && chunk.data.method) { cb(null, chunk); return; @@ -1308,7 +1308,7 @@ describe('MetaMaskController', () => { cb(); }); - metamaskController.setupUntrustedCommunication({ + metamaskController.setupUntrustedCommunicationEip1193({ connectionStream: streamTest, sender: messageSender, }); @@ -1351,7 +1351,7 @@ describe('MetaMaskController', () => { const messageSender = { url: 'http://mycrypto.com', }; - const streamTest = createThoughStream((chunk, _, cb) => { + const streamTest = createThroughStream((chunk, _, cb) => { if (chunk.data && chunk.data.method) { cb(null, chunk); return; @@ -1359,7 +1359,7 @@ describe('MetaMaskController', () => { cb(); }); - metamaskController.setupUntrustedCommunication({ + metamaskController.setupUntrustedCommunicationEip1193({ connectionStream: streamTest, sender: messageSender, }); @@ -1392,6 +1392,14 @@ describe('MetaMaskController', () => { ); }); }); + + it.todo( + 'should only process `metamask-provider` multiplex formatted messages', + ); + }); + + describe('#setupUntrustedCommunicationCaip', () => { + it.todo('should only process `caip-x` CAIP formatted messages'); }); describe('#setupTrustedCommunication', () => { @@ -1401,7 +1409,7 @@ describe('MetaMaskController', () => { tab: {}, }; const { promise, resolve } = deferredPromise(); - const streamTest = createThoughStream((chunk, _, cb) => { + const streamTest = createThroughStream((chunk, _, cb) => { expect(chunk.name).toStrictEqual('controller'); resolve(); cb(); diff --git a/builds.yml b/builds.yml index af64cfb8d0ef..25b31f33320c 100644 --- a/builds.yml +++ b/builds.yml @@ -262,6 +262,8 @@ env: - BLOCKAID_PUBLIC_KEY # Determines if feature flagged Multichain Transactions should be used - TRANSACTION_MULTICHAIN: '' + # Determines if Barad Dur features should be used + - BARAD_DUR: '' # Determines if feature flagged Chain permissions - CHAIN_PERMISSIONS: '' # Enables use of test gas fee flow to debug gas fee estimation diff --git a/development/build/manifest.js b/development/build/manifest.js index c15107564c9a..d0042af75c67 100644 --- a/development/build/manifest.js +++ b/development/build/manifest.js @@ -9,6 +9,9 @@ const IS_MV3_ENABLED = const baseManifest = IS_MV3_ENABLED ? require('../../app/manifest/v3/_base.json') : require('../../app/manifest/v2/_base.json'); +const baradDurManifest = IS_MV3_ENABLED + ? require('../../app/manifest/v3/_barad_dur.json') + : require('../../app/manifest/v2/_barad_dur.json'); const { loadBuildTypesConfig } = require('../lib/build-type'); const { TASKS, ENVIRONMENT } = require('./constants'); @@ -41,6 +44,7 @@ function createManifestTasks({ ); const result = mergeWith( cloneDeep(baseManifest), + process.env.BARAD_DUR ? cloneDeep(baradDurManifest) : {}, platformModifications, browserVersionMap[platform], await getBuildModifications(buildType, platform), diff --git a/shared/modules/caip-stream.test.ts b/shared/modules/caip-stream.test.ts new file mode 100644 index 000000000000..d97a18bda992 --- /dev/null +++ b/shared/modules/caip-stream.test.ts @@ -0,0 +1,81 @@ +import { Duplex, PassThrough } from 'readable-stream'; +import { createDeferredPromise } from '@metamask/utils'; +import { createCaipStream } from './caip-stream'; + +const writeToStream = async (stream: Duplex, message: unknown) => { + const { promise: isWritten, resolve: writeCallback } = + createDeferredPromise(); + + stream.write(message, () => writeCallback()); + await isWritten; +}; + +const readFromStream = (stream: Duplex): unknown[] => { + const chunks: unknown[] = []; + stream.on('data', (chunk: unknown) => { + chunks.push(chunk); + }); + + return chunks; +}; + +class MockStream extends Duplex { + chunks: unknown[] = []; + + constructor() { + super({ objectMode: true }); + } + + _read() { + return undefined; + } + + _write( + value: unknown, + _encoding: BufferEncoding, + callback: (error?: Error | null) => void, + ) { + this.chunks.push(value); + callback(); + } +} + +describe('CAIP Stream', () => { + describe('createCaipStream', () => { + it('pipes and unwraps a caip-x message from source stream to the substream', async () => { + const sourceStream = new PassThrough({ objectMode: true }); + const sourceStreamChunks = readFromStream(sourceStream); + + const providerStream = createCaipStream(sourceStream); + const providerStreamChunks = readFromStream(providerStream); + + await writeToStream(sourceStream, { + type: 'caip-x', + data: { foo: 'bar' }, + }); + + expect(sourceStreamChunks).toStrictEqual([ + { type: 'caip-x', data: { foo: 'bar' } }, + ]); + expect(providerStreamChunks).toStrictEqual([{ foo: 'bar' }]); + }); + + it('pipes and wraps a message from the substream to the source stream', async () => { + // using a fake stream here instead of PassThrough to prevent a loop + // when sourceStream gets written back to at the end of the CAIP pipeline + const sourceStream = new MockStream(); + + const providerStream = createCaipStream(sourceStream); + + await writeToStream(providerStream, { + foo: 'bar', + }); + + // Note that it's not possible to verify the output side of the internal SplitStream + // instantiated inside createCaipStream as only the substream is actually exported + expect(sourceStream.chunks).toStrictEqual([ + { type: 'caip-x', data: { foo: 'bar' } }, + ]); + }); + }); +}); diff --git a/shared/modules/caip-stream.ts b/shared/modules/caip-stream.ts new file mode 100644 index 000000000000..3f13927efc27 --- /dev/null +++ b/shared/modules/caip-stream.ts @@ -0,0 +1,73 @@ +import { isObject } from '@metamask/utils'; +// eslint-disable-next-line @typescript-eslint/ban-ts-comment +// @ts-expect-error pipeline() isn't defined as part of @types/readable-stream +import { pipeline, Duplex } from 'readable-stream'; + +class Substream extends Duplex { + parent: Duplex; + + constructor(parent: Duplex) { + super({ objectMode: true }); + this.parent = parent; + } + + _read() { + return undefined; + } + + _write( + value: unknown, + _encoding: BufferEncoding, + callback: (error?: Error | null) => void, + ) { + this.parent.push({ + type: 'caip-x', + data: value, + }); + callback(); + } +} + +export class CaipStream extends Duplex { + substream: Duplex; + + constructor() { + super({ objectMode: true }); + this.substream = new Substream(this); + } + + _read() { + return undefined; + } + + _write( + value: unknown, + _encoding: BufferEncoding, + callback: (error?: Error | null) => void, + ) { + if (isObject(value) && value.type === 'caip-x') { + this.substream.push(value.data); + } + callback(); + } +} + +/** + * Creates a pipeline using a port stream meant to be consumed by the JSON-RPC engine: + * - accepts only incoming CAIP messages intended for evm providers from the port stream + * - unwraps these incoming messages into a new stream that the JSON-RPC engine should operate off + * - wraps the outgoing messages from the new stream back into the CAIP message format + * - writes these messages back to the port stream + * + * @param portStream - The source and sink duplex stream + * @returns a new duplex stream that should be operated on instead of the original portStream + */ +export const createCaipStream = (portStream: Duplex): Duplex => { + const caipStream = new CaipStream(); + + pipeline(portStream, caipStream, portStream, (err: Error) => + console.log('MetaMask CAIP stream', err), + ); + + return caipStream.substream; +};