diff --git a/server/.vscode/launch.json b/server/.vscode/launch.json index 12cb7c7ef..e9e10f1ae 100644 --- a/server/.vscode/launch.json +++ b/server/.vscode/launch.json @@ -39,5 +39,67 @@ // "DEBUG": "*", } }, + + { + "autoAttachChildProcesses": false, + "type": "node", + "request": "launch", + "name": "Launch Cluster Server", + "skipFiles": [ + "/**" + ], + "preLaunchTask": "npm: build", + "program": "${workspaceFolder}/bin/scrypted-serve", + "runtimeArgs": [ + "--trace-warnings", + "--nolazy", + ], + "sourceMaps": true, + "resolveSourceMapLocations": [ + "${workspaceFolder}/**", + "!**/node_modules/**" + ], + "outFiles": [ + "${workspaceFolder}/**/*.js" + ], + "env": { + "SCRYPTED_CLUSTER_MODE": "server", + "SCRYPTED_CLUSTER_SERVER": "192.168.2.124", + "SCRYPTED_CLUSTER_SECRET": "swordfish", + "SCRYPTED_CAN_RESTART": "true", + "SCRYPTED_VOLUME": "/Users/koush/.scrypted-cluster/volume-server", + } + }, + + { + "autoAttachChildProcesses": false, + "type": "node", + "request": "launch", + "name": "Launch Cluster Client", + "skipFiles": [ + "/**" + ], + "preLaunchTask": "npm: build", + "program": "${workspaceFolder}/bin/scrypted-serve", + "runtimeArgs": [ + "--trace-warnings", + "--nolazy", + ], + "sourceMaps": true, + "resolveSourceMapLocations": [ + "${workspaceFolder}/**", + "!**/node_modules/**" + ], + "outFiles": [ + "${workspaceFolder}/**/*.js" + ], + "env": { + "SCRYPTED_CLUSTER_MODE": "client", + "SCRYPTED_CLUSTER_SERVER": "192.168.2.124", + "SCRYPTED_CLUSTER_SECRET": "swordfish", + "SCRYPTED_CAN_RESTART": "true", + "SCRYPTED_VOLUME": "/Users/koush/.scrypted-cluster/volume-client", + } + }, ] } \ No newline at end of file diff --git a/server/src/ip.ts b/server/src/ip.ts index 1492443b1..d5687dc73 100644 --- a/server/src/ip.ts +++ b/server/src/ip.ts @@ -1,10 +1,11 @@ import os from 'os'; import net from 'net'; +export const loopbackList = new net.BlockList(); +loopbackList.addSubnet('127.0.0.0', 8); +loopbackList.addAddress('::1', 'ipv6'); + const unusableList = new net.BlockList(); -// loopback -unusableList.addSubnet('127.0.0.0', 8); -unusableList.addAddress('::1', 'ipv6'); // link local unusableList.addSubnet('169.254.0.0', 16); @@ -56,7 +57,7 @@ function isUsableNetworkAddress(address: string) { if (type === 'ipv6' && privateList.check(address, type)) return false; - return !unusableList.check(address, type); + return !unusableList.check(address, type) && !loopbackList.check(address, type); } catch (e) { return false; diff --git a/server/src/runtime.ts b/server/src/runtime.ts index d1bea7216..5cf330fc4 100644 --- a/server/src/runtime.ts +++ b/server/src/runtime.ts @@ -43,6 +43,7 @@ import { getNpmPackageInfo, PluginComponent } from './services/plugin'; import { ServiceControl } from './services/service-control'; import { UsersService } from './services/users'; import { getState, ScryptedStateManager, setState } from './state'; +import { ClusterWorker } from './scrypted-cluster'; interface DeviceProxyPair { handler: PluginDeviceProxyHandler; @@ -59,7 +60,8 @@ interface HttpPluginData { export class ScryptedRuntime extends PluginHttp { clusterId = crypto.randomBytes(3).toString('hex'); - clusterSecret = crypto.randomBytes(16).toString('hex'); + clusterSecret = process.env.SCRYPTED_CLUSTER_SECRET || crypto.randomBytes(16).toString('hex'); + clusterWorkers = new Set(); plugins: { [id: string]: PluginHost } = {}; pluginDevices: { [id: string]: PluginDevice } = {}; devices: { [id: string]: DeviceProxyPair } = {}; diff --git a/server/src/scrypted-cluster-main.ts b/server/src/scrypted-cluster-main.ts new file mode 100644 index 000000000..00cff48d4 --- /dev/null +++ b/server/src/scrypted-cluster-main.ts @@ -0,0 +1,12 @@ +import { install as installSourceMapSupport } from 'source-map-support'; +import { startClusterClient } from './scrypted-cluster'; + +installSourceMapSupport({ + environment: 'node', +}); + +async function start(mainFilename: string) { + startClusterClient(mainFilename); +} + +export default start; diff --git a/server/src/scrypted-cluster.ts b/server/src/scrypted-cluster.ts new file mode 100644 index 000000000..9e34beb20 --- /dev/null +++ b/server/src/scrypted-cluster.ts @@ -0,0 +1,151 @@ +import os from 'os'; +import net from 'net'; +import tls from 'tls'; +import type { createSelfSignedCertificate } from './cert'; +import { computeClusterObjectHash } from './cluster/cluster-hash'; +import { ClusterObject } from './cluster/connect-rpc-object'; +import { loopbackList } from './ip'; +import { RpcPeer } from './rpc'; +import { createRpcDuplexSerializer } from './rpc-serializer'; +import type { ScryptedRuntime } from './runtime'; +import { SCRYPTED_CLUSTER_WORKERS } from './server-settings'; +import { sleep } from './sleep'; +import { once } from 'events'; + +export interface ClusterWorkerProperties { + labels: string[]; +} + +export interface ClusterWorker extends ClusterWorkerProperties { + peer: RpcPeer; +} + +export function getScryptedClusterMode(): ['server' | 'client', string, number] { + const mode = process.env.SCRYPTED_CLUSTER_MODE as 'server' | 'client'; + if (!mode) + return; + + if (!['server', 'client'].includes(mode)) + throw new Error('SCRYPTED_CLUSTER_MODE must be set to either "server" or "client".'); + + const [server, sport] = process.env.SCRYPTED_CLUSTER_SERVER?.split(':') || []; + const port = parseInt(sport) || 10556; + if (!net.isIP(server)) { + if (server) + throw new Error('SCRYPTED_CLUSTER_SERVER is set but is not a valid IP address.'); + if (process.env.SCRYPTED_CLUSTER_SECRET) + throw new Error('SCRYPTED_CLUSTER_SECRET is set but SCRYPTED_CLUSTER_SERVER is not set.'); + return; + } + if (!process.env.SCRYPTED_CLUSTER_SECRET) + throw new Error('SCRYPTED_CLUSTER_SERVER is set but SCRYPTED_CLUSTER_SECRET is not set.'); + return [mode, server, port]; +} + +export function startClusterClient(mainFilename: string) { + let labels = process.env.SCRYPTED_CLUSTER_LABELS?.split(',') || []; + labels.push(process.arch, process.platform, os.hostname()); + labels = [...new Set(labels)]; + + const secret = process.env.SCRYPTED_CLUSTER_SECRET; + const clusterMode = getScryptedClusterMode(); + const [, host, port] = clusterMode; + for (let i = 0; i < SCRYPTED_CLUSTER_WORKERS; i++) { + (async () => { + while (true) { + const backoff = sleep(10000); + try { + const client = tls.connect({ + host, + port, + rejectUnauthorized: false, + }); + + const serializer = createRpcDuplexSerializer(client); + const peer = new RpcPeer('cluster-remote', 'cluster-host', (message, reject, serializationContext) => { + serializer.sendMessage(message, reject, serializationContext); + }); + serializer.setupRpcPeer(peer); + client.on('data', data => serializer.onData(data)); + client.on('error', e => { + peer.kill(e); + }); + client.on('close', () => { + peer.kill('cluster server closed'); + }); + const connectForkWorker = await peer.getParam('connectForkWorker'); + const auth: ClusterObject = { + address: client.localAddress, + port: client.localPort, + id: undefined, + proxyId: undefined, + sourceKey: undefined, + sha256: undefined, + }; + auth.sha256 = computeClusterObjectHash(auth, secret); + + const properties: ClusterWorkerProperties = { + labels, + }; + + await connectForkWorker(auth, properties); + console.warn('worker ready'); + } + catch (e) { + } + await backoff; + } + })(); + } +} + +export function createClusterServer(runtime: ScryptedRuntime, certificate: ReturnType) { + const server = tls.createServer({ + key: certificate.serviceKey, + cert: certificate.certificate, + }, (socket) => { + const serializer = createRpcDuplexSerializer(socket); + const peer = new RpcPeer('cluster-host', 'cluster-remote', (message, reject, serializationContext) => { + serializer.sendMessage(message, reject, serializationContext); + }); + serializer.setupRpcPeer(peer); + socket.on('data', data => serializer.onData(data)); + socket.on('error', e => { + peer.kill(e); + }); + socket.on('close', () => { + peer.kill('cluster client closed'); + }); + peer.killed.then(() => { + socket.destroy(); + }); + + peer.params['connectForkWorker'] = async (auth: ClusterObject, properties: ClusterWorkerProperties) => { + try { + const sha256 = computeClusterObjectHash(auth, runtime.clusterSecret); + if (sha256 !== auth.sha256) + throw new Error('cluster object hash mismatch'); + // the remote address may be ipv6 prefixed so use a fuzzy match. + // eg ::ffff:192.168.2.124 + if (auth.port !== socket.remotePort || !socket.remoteAddress.endsWith(auth.address)) + throw new Error('cluster object address mismatch'); + const worker: ClusterWorker = { + ...properties, + peer, + }; + runtime.clusterWorkers.add(worker); + peer.killed.then(() => { + runtime.clusterWorkers.delete(worker); + }); + socket.on('close', () => { + runtime.clusterWorkers.delete(worker); + }); + } + catch (e) { + peer.kill(e); + } + } + }); + + return server; +} \ No newline at end of file diff --git a/server/src/scrypted-main-exports.ts b/server/src/scrypted-main-exports.ts index e5ebb570a..48b3e2d17 100644 --- a/server/src/scrypted-main-exports.ts +++ b/server/src/scrypted-main-exports.ts @@ -8,10 +8,10 @@ import v8 from 'v8'; import vm from 'vm'; import { PluginError } from './plugin/plugin-error'; import { getScryptedVolume } from './plugin/plugin-volume'; +import { isNodePluginWorkerProcess } from './plugin/runtime/node-fork-worker'; import { RPCResultError, startPeriodicGarbageCollection } from './rpc'; import type { Runtime } from './scrypted-server-main'; -import { isNodePluginWorkerProcess } from './plugin/runtime/node-fork-worker'; - +import { getScryptedClusterMode } from './scrypted-cluster'; function start(mainFilename: string, options?: { onRuntimeCreated?: (runtime: Runtime) => Promise, @@ -28,8 +28,8 @@ function start(mainFilename: string, options?: { globalThis.gc = vm.runInNewContext("gc"); } - if (!semver.gte(process.version, '16.0.0')) { - throw new Error('"node" version out of date. Please update node to v16 or higher.') + if (!semver.gte(process.version, '18.0.0')) { + throw new Error('"node" version out of date. Please update node to v18 or higher.') } // Node 17 changes the dns resolution order to return the record order. @@ -53,20 +53,26 @@ function start(mainFilename: string, options?: { const start = require('./scrypted-plugin-main').default; return start(mainFilename); } + + // unhandled rejections are allowed if they are from a rpc/plugin call. + process.on('unhandledRejection', error => { + if (error?.constructor !== RPCResultError && error?.constructor !== PluginError) { + console.error('fatal error', error); + throw error; + } + console.warn('unhandled rejection of RPC Result', error); + }); + + dotenv.config({ + path: path.join(getScryptedVolume(), '.env'), + }); + + const clusterMode = getScryptedClusterMode(); + if (clusterMode?.[0] === 'client') { + const start = require('./scrypted-cluster-main').default; + return start(mainFilename); + } else { - // unhandled rejections are allowed if they are from a rpc/plugin call. - process.on('unhandledRejection', error => { - if (error?.constructor !== RPCResultError && error?.constructor !== PluginError) { - console.error('fatal error', error); - throw error; - } - console.warn('unhandled rejection of RPC Result', error); - }); - - dotenv.config({ - path: path.join(getScryptedVolume(), '.env'), - }); - const start = require('./scrypted-server-main').default; return start(mainFilename, options); } diff --git a/server/src/scrypted-server-main.ts b/server/src/scrypted-server-main.ts index f2f4d26b1..eb7b0b9b0 100644 --- a/server/src/scrypted-server-main.ts +++ b/server/src/scrypted-server-main.ts @@ -11,38 +11,24 @@ import net from 'net'; import os from 'os'; import path from 'path'; import process from 'process'; -import semver from 'semver'; import { install as installSourceMapSupport } from 'source-map-support'; import { createSelfSignedCertificate, CURRENT_SELF_SIGNED_CERTIFICATE_VERSION } from './cert'; import { Plugin, ScryptedUser, Settings } from './db-types'; import { getUsableNetworkAddresses } from './ip'; import Level from './level'; -import { PluginError } from './plugin/plugin-error'; import { getScryptedVolume } from './plugin/plugin-volume'; -import { RPCResultError } from './rpc'; import { ScryptedRuntime } from './runtime'; import { SCRYPTED_DEBUG_PORT, SCRYPTED_INSECURE_PORT, SCRYPTED_SECURE_PORT } from './server-settings'; import { getNpmPackageInfo } from './services/plugin'; import { setScryptedUserPassword, UsersService } from './services/users'; import { sleep } from './sleep'; import { ONE_DAY_MILLISECONDS, UserToken } from './usertoken'; +import { createClusterServer, getScryptedClusterMode } from './scrypted-cluster'; export type Runtime = ScryptedRuntime; -if (!semver.gte(process.version, '18.0.0')) { - throw new Error('"node" version out of date. Please update node to v18 or higher.') -} - -process.on('unhandledRejection', error => { - if (error?.constructor !== RPCResultError && error?.constructor !== PluginError) { - console.error('pending crash', error); - throw error; - } - console.warn('unhandled rejection of RPC Result', error); -}); - -async function listenServerPort(env: string, port: number, server: any) { - server.listen(port); +async function listenServerPort(env: string, port: number, server: http.Server | https.Server | net.Server, hostname?: string) { + server.listen(port, hostname); try { await once(server, 'listening'); } @@ -474,7 +460,7 @@ async function start(mainFilename: string, options?: { debugServer.on('connection', resolve); }); - waitDebug.catch(() => {}); + waitDebug.catch(() => { }); workerInspectPort = Math.round(Math.random() * 10000) + 30000; try { @@ -737,6 +723,12 @@ async function start(mainFilename: string, options?: { await listenServerPort('SCRYPTED_SECURE_PORT', SCRYPTED_SECURE_PORT, secure); await listenServerPort('SCRYPTED_INSECURE_PORT', SCRYPTED_INSECURE_PORT, insecure); + const clusterMode = getScryptedClusterMode(); + if (clusterMode?.[0] === 'server') { + const clusterServer = createClusterServer(scrypted, keyPair); + await listenServerPort('SCRYPTED_CLUSTER_SERVER', clusterMode[2], clusterServer); + } + console.log('#######################################################'); console.log(`Scrypted Volume : ${volumeDir}`); console.log(`Scrypted Server (Local) : https://localhost:${SCRYPTED_SECURE_PORT}/`); @@ -746,13 +738,13 @@ async function start(mainFilename: string, options?: { console.log(`Version: : ${await scrypted.info.getVersion()}`); console.log('#######################################################'); console.log('Scrypted insecure http service port:', SCRYPTED_INSECURE_PORT); - console.log('Ports can be changed with environment variables.') - console.log('https: $SCRYPTED_SECURE_PORT') - console.log('http : $SCRYPTED_INSECURE_PORT') - console.log('Certificate can be modified via tls.createSecureContext options in') + console.log('Ports can be changed with environment variables.'); + console.log('https: $SCRYPTED_SECURE_PORT'); + console.log('http : $SCRYPTED_INSECURE_PORT'); + console.log('Certificate can be modified via tls.createSecureContext options in'); console.log('JSON file located at SCRYPTED_HTTPS_OPTIONS_FILE environment variable:'); console.log('export SCRYPTED_HTTPS_OPTIONS_FILE=/path/to/options.json'); - console.log('https://nodejs.org/api/tls.html#tlscreatesecurecontextoptions') + console.log('https://nodejs.org/api/tls.html#tlscreatesecurecontextoptions'); console.log('#######################################################'); return scrypted; diff --git a/server/src/server-settings.ts b/server/src/server-settings.ts index d613db785..acb0b5cca 100644 --- a/server/src/server-settings.ts +++ b/server/src/server-settings.ts @@ -3,6 +3,7 @@ import { getUsableNetworkAddresses } from './ip'; export const SCRYPTED_INSECURE_PORT = parseInt(process.env.SCRYPTED_INSECURE_PORT) || 11080; export const SCRYPTED_SECURE_PORT = parseInt(process.env.SCRYPTED_SECURE_PORT) || 10443; export const SCRYPTED_DEBUG_PORT = parseInt(process.env.SCRYPTED_DEBUG_PORT) || 10081; +export const SCRYPTED_CLUSTER_WORKERS = parseInt(process.env.SCRYPTED_CLUSTER_WORKERS) || 32; export function getIpAddress(): string { return getUsableNetworkAddresses()[0];