diff --git a/server/package-lock.json b/server/package-lock.json index 4b9676bfa..1adf4987a 100644 --- a/server/package-lock.json +++ b/server/package-lock.json @@ -1,12 +1,12 @@ { "name": "@scrypted/server", - "version": "0.7.68", + "version": "0.7.73", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@scrypted/server", - "version": "0.7.68", + "version": "0.7.73", "license": "ISC", "dependencies": { "@mapbox/node-pre-gyp": "^1.0.10", diff --git a/server/python/plugin_remote.py b/server/python/plugin_remote.py index 3cf8013ee..3b3f23333 100644 --- a/server/python/plugin_remote.py +++ b/server/python/plugin_remote.py @@ -286,46 +286,72 @@ class PluginRemote: clusterId = options['clusterId'] clusterSecret = options['clusterSecret'] + def onProxySerialization(value: Any, proxyId: str, source: int = None): + properties: dict = rpc.RpcPeer.prepareProxyProperties(value) or {} + clusterEntry = properties.get('__cluster', None) + if not properties.get('__cluster', None): + clusterEntry = { + 'id': clusterId, + 'proxyId': proxyId, + 'port': clusterPort, + 'source': source, + } + properties['__cluster'] = clusterEntry + + # clusterEntry['proxyId'] = proxyId + # clusterEntry['source'] = source + return properties + + self.peer.onProxySerialization = onProxySerialization + + async def resolveObject(id: str, sourcePeerPort: int): + sourcePeer: rpc.RpcPeer = self.peer if not sourcePeerPort else await rpc.maybe_await(clusterPeers.get(sourcePeerPort)) + if not sourcePeer: + return + return sourcePeer.localProxyMap.get(id, None) + + clusterPeers: Mapping[int, asyncio.Future[rpc.RpcPeer]] = {} async def handleClusterClient(reader: asyncio.StreamReader, writer: asyncio.StreamWriter): + _, clusterPeerPort = writer.get_extra_info('peername') rpcTransport = rpc_reader.RpcStreamTransport(reader, writer) peer: rpc.RpcPeer peer, peerReadLoop = await rpc_reader.prepare_peer_readloop(self.loop, rpcTransport) - async def connectRPCObject(id: str, secret: str): + peer.onProxySerialization = lambda value, proxyId: onProxySerialization(value, proxyId, clusterPeerPort) + future = asyncio.Future[rpc.RpcPeer]() + future.set_result(peer) + clusterPeers[clusterPeerPort] = future + async def connectRPCObject(id: str, secret: str, sourcePeerPort: int = None): m = hashlib.sha256() m.update(bytes('%s%s' % (clusterPort, clusterSecret), 'utf8')) portSecret = m.hexdigest() if secret != portSecret: raise Exception('secret incorrect') - return self.peer.localProxyMap.get(id, None) + return await resolveObject(id, sourcePeerPort) peer.params['connectRPCObject'] = connectRPCObject try: await peerReadLoop() except: + pass + finally: + clusterPeers.pop(clusterPeerPort) + peer.kill('cluster client killed') writer.close() clusterRpcServer = await asyncio.start_server(handleClusterClient, '127.0.0.1', 0) clusterPort = clusterRpcServer.sockets[0].getsockname()[1] - clusterPeers: Mapping[int, asyncio.Future[rpc.RpcPeer]] = {} - async def connectRPCObject(value): - clusterObject = getattr(value, '__cluster') - if type(clusterObject) is not dict: - return value - - if clusterObject.get('id', None) != clusterId: - return value - - port = clusterObject['port'] - proxyId = clusterObject['proxyId'] - + def ensureClusterPeer(port: int): clusterPeerPromise = clusterPeers.get(port) if not clusterPeerPromise: async def connectClusterPeer(): reader, writer = await asyncio.open_connection( '127.0.0.1', port) + _, clusterPeerPort = writer.get_extra_info('sockname') rpcTransport = rpc_reader.RpcStreamTransport(reader, writer) peer, peerReadLoop = await rpc_reader.prepare_peer_readloop(self.loop, rpcTransport) + peer.onProxySerialization = lambda value, proxyId: onProxySerialization(value, proxyId, clusterPeerPort) + async def run_loop(): try: await peerReadLoop() @@ -337,6 +363,23 @@ class PluginRemote: return peer clusterPeerPromise = self.loop.create_task(connectClusterPeer()) clusterPeers[port] = clusterPeerPromise + return clusterPeerPromise + + async def connectRPCObject(value): + clusterObject = getattr(value, '__cluster') + if type(clusterObject) is not dict: + return value + + if clusterObject.get('id', None) != clusterId: + return value + + port = clusterObject['port'] + proxyId = clusterObject['proxyId'] + source = clusterObject.get('source', None) + if port == clusterPort: + return await resolveObject(proxyId, source) + + clusterPeerPromise = ensureClusterPeer(port) try: clusterPeer = await clusterPeerPromise @@ -344,7 +387,7 @@ class PluginRemote: m = hashlib.sha256() m.update(bytes('%s%s' % (port, clusterSecret), 'utf8')) portSecret = m.hexdigest() - newValue = await c(proxyId, portSecret) + newValue = await c(proxyId, portSecret, source) if not newValue: raise Exception('ipc object not found?') return newValue @@ -353,17 +396,6 @@ class PluginRemote: sdk.connectRPCObject = connectRPCObject - def onProxySerialization(value: Any, proxyId: str): - properties: dict = rpc.RpcPeer.prepareProxyProperties(value) or {} - properties['__cluster'] = { - 'id': clusterId, - 'proxyId': proxyId, - 'port': clusterPort, - } - return properties - - self.peer.onProxySerialization = onProxySerialization - forkMain = options and options.get('fork') if not forkMain: diff --git a/server/src/plugin/plugin-remote-worker.ts b/server/src/plugin/plugin-remote-worker.ts index 0601a11aa..50920281b 100644 --- a/server/src/plugin/plugin-remote-worker.ts +++ b/server/src/plugin/plugin-remote-worker.ts @@ -26,6 +26,15 @@ export interface StartPluginRemoteOptions { onClusterPeer(peer: RpcPeer): void; } +interface ClusterObject { + id: string; + port: number; + proxyId: string; + source: number; +} + +type ConnectRPCObject = (id: string, secret: string, sourcePeerPort: number) => Promise; + export function startPluginRemote(mainFilename: string, pluginId: string, peerSend: (message: RpcMessage, reject?: (e: Error) => void, serializationContext?: any) => void, startPluginRemoteOptions?: StartPluginRemoteOptions) { const peer = new RpcPeer('unknown', 'host', peerSend); @@ -77,39 +86,61 @@ export function startPluginRemote(mainFilename: string, pluginId: string, peerSe }, async onLoadZip(scrypted: ScryptedStatic, params: any, packageJson: any, zipData: Buffer | string, zipOptions: PluginRemoteLoadZipOptions) { const { clusterId, clusterSecret } = zipOptions; - const clusterRpcServer = net.createServer(client => { - const clusterPeer = createDuplexRpcPeer(peer.selfName, 'cluster-client', client, client); - startPluginRemoteOptions?.onClusterPeer?.(clusterPeer); - const portSecret = crypto.createHash('sha256').update(`${clusterPort}${clusterSecret}`).digest().toString('hex'); - clusterPeer.params['connectRPCObject'] = async (id: string, secret: string) => { - if (secret !== portSecret) - throw new Error('secret incorrect'); - return peer.localProxyMap.get(id); - } - client.on('close', () => clusterPeer.kill('cluster socket closed')); - }) - const clusterPort = await listenZero(clusterRpcServer); - const clusterEntry = { - id: clusterId, - port: clusterPort, - }; - peer.onProxySerialization = (value, proxyId) => { + const onProxySerialization = (value: any, proxyId: string, source?: number) => { const properties = RpcPeer.prepareProxyProperties(value) || {}; - properties.__cluster = { - ...clusterEntry, - proxyId, + let clusterEntry: ClusterObject = properties.__cluster; + + // set the cluster identity if it does not exist. + if (!clusterEntry) { + clusterEntry = { + id: clusterId, + port: clusterPort, + proxyId, + source, + }; + properties.__cluster = clusterEntry; } + // always reassign the id and source. + // if this is already a p2p object, and is passed to a different peer, + // a future p2p object must be routed to the correct p2p peer to find the object. + // clusterEntry.proxyId = proxyId; + // clusterEntry.source = source; return properties; } + peer.onProxySerialization = onProxySerialization; + const resolveObject = async (id: string, sourcePeerPort: number) => { + const sourcePeer = sourcePeerPort ? await clusterPeers.get(sourcePeerPort) : peer; + return sourcePeer?.localProxyMap.get(id); + } + + // all cluster clients, incoming and outgoing, connect with random ports which can be used as peer ids + // on the cluster server that is listening on the actual port/ + // incoming connections: use the remote random/unique port + // outgoing connections: use the local random/unique port const clusterPeers = new Map>(); - scrypted.connectRPCObject = async (value: any) => { - const clusterObject = value?.__cluster; - if (clusterObject?.id !== clusterId) - return value; - const { port, proxyId } = clusterObject; + const clusterRpcServer = net.createServer(client => { + const clusterPeer = createDuplexRpcPeer(peer.selfName, 'cluster-client', client, client); + const clusterPeerPort = client.remotePort; + clusterPeer.onProxySerialization = (value, proxyId) => onProxySerialization(value, proxyId, clusterPeerPort); + clusterPeers.set(clusterPeerPort, Promise.resolve(clusterPeer)); + startPluginRemoteOptions?.onClusterPeer?.(clusterPeer); + const portSecret = crypto.createHash('sha256').update(`${clusterPort}${clusterSecret}`).digest().toString('hex'); + const connectRPCObject: ConnectRPCObject = async (id, secret, sourcePeerPort) => { + if (secret !== portSecret) + throw new Error('secret incorrect'); + return resolveObject(id, sourcePeerPort); + } + clusterPeer.params['connectRPCObject'] = connectRPCObject; + client.on('close', () => { + clusterPeers.delete(clusterPeerPort); + clusterPeer.kill('cluster socket closed'); + }); + }) + const clusterPort = await listenZero(clusterRpcServer); + const ensureClusterPeer = (port: number) => { let clusterPeerPromise = clusterPeers.get(port); if (!clusterPeerPromise) { clusterPeerPromise = (async () => { @@ -118,7 +149,11 @@ export function startPluginRemote(mainFilename: string, pluginId: string, peerSe try { await once(socket, 'connect'); + const clusterPeerPort = (socket.address() as net.AddressInfo).port; + const ret = createDuplexRpcPeer(peer.selfName, 'cluster-server', socket, socket); + ret.onProxySerialization = (value, proxyId) => onProxySerialization(value, proxyId, clusterPeerPort); + return ret; } catch (e) { @@ -128,12 +163,23 @@ export function startPluginRemote(mainFilename: string, pluginId: string, peerSe } })(); } + return clusterPeerPromise; + }; + + scrypted.connectRPCObject = async (value: any) => { + const clusterObject: ClusterObject = value?.__cluster; + if (clusterObject?.id !== clusterId) + return value; + const { port, proxyId, source } = clusterObject; + if (port === clusterPort) + return resolveObject(proxyId, source); try { + const clusterPeerPromise = ensureClusterPeer(port); const clusterPeer = await clusterPeerPromise; - const connectRPCObject = await clusterPeer.getParam('connectRPCObject'); + const connectRPCObject: ConnectRPCObject = await clusterPeer.getParam('connectRPCObject'); const portSecret = crypto.createHash('sha256').update(`${port}${clusterSecret}`).digest().toString('hex'); - const newValue = await connectRPCObject(proxyId, portSecret); + const newValue = await connectRPCObject(proxyId, portSecret, source); if (!newValue) throw new Error('ipc object not found?'); return newValue;