ShinobiSystems
/
Shinobi
mirror of https://gitlab.com/Shinobi-Systems/Shinobi.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

begin conversion to v2 for Easy Remote Access

p2p-v2
Moe 2022-03-15 09:49:06 -07:00
parent d16a3d006a
commit c56bfb19e8
4 changed files with 455 additions and 86 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

101
libs/commander/utils.js Normal file
View File

@ -0,0 +1,101 @@
const fetch = require('node-fetch');
module.exports = (config,lang,p2pClientConnections) => {
const parseJSON = function(string){
var parsed = string
try{
parsed = JSON.parse(string)
}catch(err){
}
return parsed
}
const createQueryStringFromObject = function(obj){
var queryString = ''
var keys = Object.keys(obj)
keys.forEach(function(key){
var value = obj[key]
queryString += `&${key}=${value}`
})
return queryString
}
const doRequest = function(url,method,data,outboundMessage,callback,onDataReceived,headers){
const requestEndpoint = `${config.sslEnabled ? `https` : 'http'}://localhost:${config.sslEnabled ? config.ssl.port : config.port}` + url
const urlParts = url.split('/');
const forceStream = urlParts[2] === 'img'
const hasCallback = typeof callback === 'function' && !forceStream;
if(method === 'GET' && data){
requestEndpoint += '?' + createQueryStringFromObject(data)
}
console.log()
const theRequest = fetch(requestEndpoint,{
headers: Object.assign({ "Content-Type": "application/json" },headers),
method: method,
body: method !== 'GET' ? (data ? JSON.stringify(data) : null) : null
});
outboundMessage('httpHeaders',{
headers: headers,
})
theRequest.then((resp) => {
const buffers = []
resp.body.on('data', (chunk) => {
onDataReceived(chunk)
if(hasCallback)buffers.push(chunk)
})
if(hasCallback){
resp.body.on('end', () => {
// end of stream
const allBuffers = Buffer.concat(buffers).toString()
let newBody = allBuffers;
try{
method !== 'GET' ? (data ? data : null) : null;
newBody = JSON.parse(allBuffers);
}catch(err){
newBody = allBuffers;
}
callback(null,newBody,resp)
})
}
}).catch((err) => {
console.error(err,data)
callback(err,{})
});
// async function readOnData(){
// try {
// console.log(typeof theRequest)
// const response = await theRequest;
// for await (const chunk of response.body) {
// onDataReceived(chunk)
// }
// } catch (err) {
// console.error(err.stack);
// }
// }
// readOnData()
return theRequest
}
const createShinobiSocketConnection = (connectionId) => {
const masterConnectionToMachine = socketIOClient(`ws://localhost:${config.port}`, {transports:['websocket']})
p2pClientConnections[connectionId || p2pClientConnectionStaticName] = masterConnectionToMachine
return masterConnectionToMachine
}
const killAllClientConnections = () => {
Object.keys(p2pClientConnections).forEach((key) => {
try{
p2pClientConnections[key].disconnect()
}catch(err){
}
setTimeout(() => {
delete(p2pClientConnections[key])
},1000)
})
}
return {
parseJSON,
createQueryStringFromObject,
doRequest,
createShinobiSocketConnection,
killAllClientConnections,
}
}

25
libs/commander/utilsv1.js Normal file
View File

@ -0,0 +1,25 @@
const request = require('request')
module.exports = (s,config,lang,p2pClientConnections) => {
const doRequest = function(url,method,data,callback,onDataReceived,headers){
var requestEndpoint = `${config.sslEnabled ? `https` : 'http'}://localhost:${config.sslEnabled ? config.ssl.port : config.port}` + url
if(method === 'GET' && data){
requestEndpoint += '?' + createQueryStringFromObject(data)
}
s.debugLog(`requestEndpoint`,requestEndpoint)
const theRequest = request(requestEndpoint,{
headers: headers,
method: method,
json: method !== 'GET' ? (data ? data : null) : null,
// timeout: 3000,
}, typeof callback === 'function' ? (err,resp,body) => {
// var json = parseJSON(body)
if(err)s.debugLog(err,data)
callback(err,body,resp)
} : null)
.on('data', onDataReceived);
return theRequest
}
return {
doRequest,
}
}

177
libs/commander/worker.js
View File

@ -1,6 +1,6 @@
const { parentPort } = require('worker_threads');
const request = require('request');
const socketIOClient = require('socket.io-client');
const WebSocket = require('ws');
const bson = require('bson');
const p2pClientConnectionStaticName = 'Commander'
const p2pClientConnections = {}
const runningRequests = {}
@ -33,104 +33,120 @@ parentPort.on('message',(data) => {
})
const initialize = (config,lang) => {
const {
parseJSON,
createQueryStringFromObject,
doRequest,
createShinobiSocketConnection,
killAllClientConnections,
} = require('./utils.js')(config,lang,p2pClientConnections)
// const {
// doRequest,
// } = require('./utilsv1.js')(s,config,lang,p2pClientConnections)
const selectedP2PServerId = config.p2pServerList[config.p2pHostSelected] ? config.p2pHostSelected : Object.keys(config.p2pServerList)[0]
const selectedHost = config.p2pServerList[selectedP2PServerId].host + ':' + config.p2pServerList[selectedP2PServerId].p2pPort
const parseJSON = function(string){
var parsed = string
try{
parsed = JSON.parse(string)
}catch(err){
}
return parsed
}
const createQueryStringFromObject = function(obj){
var queryString = ''
var keys = Object.keys(obj)
keys.forEach(function(key){
var value = obj[key]
queryString += `&${key}=${value}`
})
return queryString
}
const doRequest = function(url,method,data,callback,onDataReceived){
var requestEndpoint = `${config.sslEnabled ? `https` : 'http'}://localhost:${config.sslEnabled ? config.ssl.port : config.port}` + url
if(method === 'GET' && data){
requestEndpoint += '?' + createQueryStringFromObject(data)
}
const theRequest = request(requestEndpoint,{
method: method,
json: method !== 'GET' ? (data ? data : null) : null
}, typeof callback === 'function' ? (err,resp,body) => {
// var json = parseJSON(body)
if(err)console.error(err,data)
callback(err,body,resp)
} : null)
.on('data', onDataReceived);
return theRequest
}
const createShinobiSocketConnection = (connectionId) => {
const masterConnectionToMachine = socketIOClient(`ws://localhost:${config.port}`, {transports:['websocket']})
p2pClientConnections[connectionId || p2pClientConnectionStaticName] = masterConnectionToMachine
return masterConnectionToMachine
}
const killAllClientConnections = () => {
Object.keys(p2pClientConnections).forEach((key) => {
try{
p2pClientConnections[key].disconnect()
}catch(err){
}
setTimeout(() => {
delete(p2pClientConnections[key])
},1000)
})
}
//
function startBridge(noLog){
s.debugLog('p2p',`Connecting to ${selectedHost}...`)
if(connectionToP2PServer && connectionToP2PServer.connected){
if(connectionToP2PServer){
try{
connectionToP2PServer.allowDisconnect = true;
connectionToP2PServer.disconnect()
connectionToP2PServer.close()
}catch(err){
s.debugLog(err)
}
connectionToP2PServer = socketIOClient('ws://' + selectedHost, {transports:['websocket']});
}
// connectionToP2PServer = socketIOClient('ws://' + selectedHost, {transports:['websocket']});
s.debugLog('ws://' + selectedHost + '/v2')
connectionToP2PServer = new WebSocket('ws://' + selectedHost + '/v2');
connectionToP2PServer.binaryType = 'arraybuffer';
let pendingOutboundMessages = []
connectionToP2PServer.outboundMessage = (key,data) => {
s.debugLog('queue OUTBOUND',key)
pendingOutboundMessages.push({
f: key,
data: data
})
}
connectionToP2PServer.on('open', () => {
connectionToP2PServer.outboundMessage = (key,data) => {
s.debugLog('OUTBOUND MESSAGE',key)
connectionToP2PServer.send(
bson.serialize({
f: key,
data: data
})
)
}
pendingOutboundMessages.forEach((message) => {
s.debugLog('SEND QUEUED OUTBOUND',message.f)
connectionToP2PServer.outboundMessage(message.f,message.data)
})
delete(pendingOutboundMessages)
})
const allMessageHandlers = []
connectionToP2PServer.onmessage = function(event){
const data = bson.deserialize(Buffer.from(event.data))
allMessageHandlers.forEach((handler) => {
if(data.f === handler.key){
handler.callback(data.data)
}
})
}
function onIncomingMessage(key,callback){
allMessageHandlers.push({
key: key,
callback: callback,
})
}
connectionToP2PServer.onInboundMessage = onIncomingMessage
// connectionToP2PServer.on('open', heartbeat);
// connectionToP2PServer.on('ping', heartbeat);
if(!config.p2pApiKey){
if(!noLog)s.systemLog('p2p',`Please fill 'p2pApiKey' in your conf.json.`)
}
// if(!config.p2pGroupId){
// if(!noLog)s.systemLog('p2p',`Please fill 'p2pGroupId' in your conf.json.`)
// }
connectionToP2PServer.on('connect', () => {
connectionToP2PServer.on('open', () => {
if(!noLog)s.systemLog('p2p',`Connected ${selectedHost}!`);
connectionToP2PServer.emit('initMachine',{
setTimeout(() => {
connectionToP2PServer.outboundMessage('initMachine',{
port: config.port,
apiKey: config.p2pApiKey,
groupId: config.p2pGroupId,
targetUserId: config.p2pTargetUserId,
targetGroupId: config.p2pTargetGroupId
})
},3000)
})
connectionToP2PServer.on('httpClose',(requestId) => {
connectionToP2PServer.onInboundMessage('httpClose',(requestId) => {
if(runningRequests[requestId] && runningRequests[requestId].abort){
runningRequests[requestId].abort()
delete(runningRequests[requestId])
}
})
connectionToP2PServer.on('http',(rawRequest) => {
connectionToP2PServer.onInboundMessage('http',(rawRequest) => {
const urlParts = rawRequest.url.split('/');
const forceStream = urlParts[2] === 'img'
const hasCallback = !forceStream && rawRequest.focus !== 'mp4' && rawRequest.focus !== 'flv' && rawRequest.focus !== 'mjpeg' && rawRequest.focus !== 'h264';
s.debugLog('doRequest hasCallback',hasCallback,'rawRequest.bodyOnEnd',rawRequest.bodyOnEnd)
runningRequests[rawRequest.rid] = doRequest(
rawRequest.url,
rawRequest.method,
rawRequest.data,
rawRequest.focus !== 'mp4' && rawRequest.focus !== 'flv' && rawRequest.focus !== 'mjpeg' && rawRequest.focus !== 'h264' ? function(err,json,resp){
connectionToP2PServer.emit('httpResponse',{
connectionToP2PServer.outboundMessage,
hasCallback ? function(err,json,resp){
connectionToP2PServer.outboundMessage('httpResponse',{
err: err,
json: rawRequest.bodyOnEnd ? json : null,
headers: resp.headers,
headers: resp.headers || {},
rid: rawRequest.rid
})
} : null,
(data) => {
if(!rawRequest.bodyOnEnd)connectionToP2PServer.emit('httpResponseChunk',{
if(forceStream || !rawRequest.bodyOnEnd)connectionToP2PServer.outboundMessage('httpResponseChunk',{
data: data,
rid: rawRequest.rid
})
@ -146,10 +162,10 @@ const initialize = (config,lang) => {
// })
// })
// masterConnectionToMachine.on('f',(data) => {
// connectionToP2PServer.emit('f',data)
// connectionToP2PServer.outboundMessage('f',data)
// })
connectionToP2PServer.on('wsInit',(rawRequest) => {
connectionToP2PServer.onInboundMessage('wsInit',(rawRequest) => {
const user = rawRequest.user
const clientConnectionToMachine = createShinobiSocketConnection(rawRequest.cnid)
connectedUserWebSockets[user.auth_token] = user;
@ -166,11 +182,11 @@ const initialize = (config,lang) => {
([
'f',
]).forEach((target) => {
connectionToP2PServer.on(target,(data) => {
connectionToP2PServer.onInboundMessage(target,(data) => {
clientConnectionToMachine.emit(target,data)
})
clientConnectionToMachine.on(target,(data) => {
connectionToP2PServer.emit(target,{data: data, cnid: rawRequest.cnid})
connectionToP2PServer.outboundMessage(target,{data: data, cnid: rawRequest.cnid})
})
})
});
@ -181,16 +197,16 @@ const initialize = (config,lang) => {
'e',
'super',
]).forEach((target) => {
connectionToP2PServer.on(target,(data) => {
connectionToP2PServer.onInboundMessage(target,(data) => {
var clientConnectionToMachine
if(data.f === 'init'){
clientConnectionToMachine = createShinobiSocketConnection(data.cnid)
clientConnectionToMachine.on('connect', () => {
clientConnectionToMachine.on(target,(fromData) => {
connectionToP2PServer.emit(target,{data: fromData, cnid: data.cnid})
connectionToP2PServer.outboundMessage(target,{data: fromData, cnid: data.cnid})
})
clientConnectionToMachine.on('f',(fromData) => {
connectionToP2PServer.emit('f',{data: fromData, cnid: data.cnid})
connectionToP2PServer.outboundMessage('f',{data: fromData, cnid: data.cnid})
})
clientConnectionToMachine.emit(target,data)
});
@ -202,36 +218,36 @@ const initialize = (config,lang) => {
});
config.workerStreamOutHandlers.forEach((target) => {
connectionToP2PServer.on(target,(initData) => {
connectionToP2PServer.onInboundMessage(target,(initData) => {
if(connectedUserWebSockets[initData.auth]){
const clientConnectionToMachine = createShinobiSocketConnection(initData.auth + initData.ke + initData.id)
clientConnectionToMachine.on('connect', () => {
clientConnectionToMachine.emit(target,initData)
});
clientConnectionToMachine.on('data',(data) => {
connectionToP2PServer.emit('data',{data: data, cnid: initData.cnid})
connectionToP2PServer.outboundMessage('data',{data: data, cnid: initData.cnid})
});
}else{
s.debugLog('disconnect now!')
}
})
});
connectionToP2PServer.on('wsDestroyStream',(clientKey) => {
connectionToP2PServer.onInboundMessage('wsDestroyStream',(clientKey) => {
if(p2pClientConnections[clientKey]){
p2pClientConnections[clientKey].disconnect();
}
delete(p2pClientConnections[clientKey])
});
connectionToP2PServer.on('wsDestroy',(rawRequest) => {
connectionToP2PServer.onInboundMessage('wsDestroy',(rawRequest) => {
if(p2pClientConnections[rawRequest.cnid]){
p2pClientConnections[rawRequest.cnid].disconnect();
}
delete(p2pClientConnections[rawRequest.cnid])
});
connectionToP2PServer.on('allowDisconnect',(bool) => {
connectionToP2PServer.onInboundMessage('allowDisconnect',(bool) => {
connectionToP2PServer.allowDisconnect = true;
connectionToP2PServer.disconnect()
connectionToP2PServer.close()
s.debugLog('p2p','Server Forced Disconnection')
});
const onDisconnect = () => {
@ -248,11 +264,14 @@ const initialize = (config,lang) => {
}
}
connectionToP2PServer.on('error',onDisconnect)
connectionToP2PServer.on('disconnect',onDisconnect)
connectionToP2PServer.on('close',onDisconnect)
}
startBridge()
setInterval(function(){
if(!connectionToP2PServer || !connectionToP2PServer.connected){
if(
(!connectionToP2PServer || !connectionToP2PServer.connected) &&
!connectionToP2PServer.allowDisconnect
){
connectionToP2PServer.connect()
}
},1000 * 60 * 15)

224
libs/commander/workerv1.js Normal file
View File

@ -0,0 +1,224 @@
const { parentPort } = require('worker_threads');
const request = require('request');
const socketIOClient = require('socket.io-client');
const p2pClientConnectionStaticName = 'Commander'
const p2pClientConnections = {}
const runningRequests = {}
const connectedUserWebSockets = {}
let connectionToP2PServer
const s = {
debugLog: (...args) => {
parentPort.postMessage({
f: 'debugLog',
data: args
})
},
systemLog: (...args) => {
parentPort.postMessage({
f: 'systemLog',
data: args
})
},
}
parentPort.on('message',(data) => {
switch(data.f){
case'init':
initialize(data.config,data.lang)
break;
case'exit':
s.debugLog('Closing P2P Connection...')
process.exit(0)
break;
}
})
const initialize = (config,lang) => {
const {
parseJSON,
createQueryStringFromObject,
createShinobiSocketConnection,
killAllClientConnections,
} = require('./utils.js')(config,lang,p2pClientConnections)
const {
doRequest,
} = require('./utilsv1.js')(config,lang)
const selectedP2PServerId = config.p2pServerList[config.p2pHostSelected] ? config.p2pHostSelected : Object.keys(config.p2pServerList)[0]
const selectedHost = config.p2pServerList[selectedP2PServerId].host + ':' + config.p2pServerList[selectedP2PServerId].p2pPort
//
function startBridge(noLog){
s.debugLog('p2p',`Connecting to ${selectedHost}...`)
if(connectionToP2PServer){
try{
connectionToP2PServer.allowDisconnect = true;
connectionToP2PServer.disconnect()
}catch(err){
console.log(err)
}
}
connectionToP2PServer = socketIOClient('ws://' + selectedHost, {transports:['websocket']});
if(!config.p2pApiKey){
if(!noLog)s.systemLog('p2p',`Please fill 'p2pApiKey' in your conf.json.`)
}
// if(!config.p2pGroupId){
// if(!noLog)s.systemLog('p2p',`Please fill 'p2pGroupId' in your conf.json.`)
// }
connectionToP2PServer.on('connect', () => {
if(!noLog)s.systemLog('p2p',`Connected ${selectedHost}!`);
connectionToP2PServer.emit('initMachine',{
port: config.port,
apiKey: config.p2pApiKey,
groupId: config.p2pGroupId,
targetUserId: config.p2pTargetUserId,
targetGroupId: config.p2pTargetGroupId
})
})
connectionToP2PServer.on('httpClose',(requestId) => {
if(runningRequests[requestId] && runningRequests[requestId].abort){
runningRequests[requestId].abort()
delete(runningRequests[requestId])
}
})
connectionToP2PServer.on('http',(rawRequest) => {
runningRequests[rawRequest.rid] = doRequest(
rawRequest.url,
rawRequest.method,
rawRequest.data,
rawRequest.focus !== 'mp4' && rawRequest.focus !== 'flv' && rawRequest.focus !== 'mjpeg' && rawRequest.focus !== 'h264' ? function(err,json,resp){
connectionToP2PServer.emit('httpResponse',{
err: err,
json: rawRequest.bodyOnEnd ? json : null,
headers: resp.headers,
rid: rawRequest.rid
})
} : null,
(data) => {
if(!rawRequest.bodyOnEnd)connectionToP2PServer.emit('httpResponseChunk',{
data: data,
rid: rawRequest.rid
})
})
})
// const masterConnectionToMachine = createShinobiSocketConnection()
// masterConnectionToMachine.on('connect', () => {
// masterConnectionToMachine.emit('f',{
// f: 'init',
// auth: config.p2pTargetAuth,
// ke: config.p2pTargetGroupId,
// uid: config.p2pTargetUserId
// })
// })
// masterConnectionToMachine.on('f',(data) => {
// connectionToP2PServer.emit('f',data)
// })
connectionToP2PServer.on('wsInit',(rawRequest) => {
const user = rawRequest.user
const clientConnectionToMachine = createShinobiSocketConnection(rawRequest.cnid)
connectedUserWebSockets[user.auth_token] = user;
clientConnectionToMachine.on('connect', () => {
s.debugLog('init',user.auth_token)
clientConnectionToMachine.emit('f',{
f: 'init',
auth: user.auth_token,
ke: user.ke,
uid: user.uid,
ipAddress: rawRequest.ipAddress
})
});
([
'f',
]).forEach((target) => {
connectionToP2PServer.on(target,(data) => {
clientConnectionToMachine.emit(target,data)
})
clientConnectionToMachine.on(target,(data) => {
connectionToP2PServer.emit(target,{data: data, cnid: rawRequest.cnid})
})
})
});
([
'a',
'r',
'gps',
'e',
'super',
]).forEach((target) => {
connectionToP2PServer.on(target,(data) => {
var clientConnectionToMachine
if(data.f === 'init'){
clientConnectionToMachine = createShinobiSocketConnection(data.cnid)
clientConnectionToMachine.on('connect', () => {
clientConnectionToMachine.on(target,(fromData) => {
connectionToP2PServer.emit(target,{data: fromData, cnid: data.cnid})
})
clientConnectionToMachine.on('f',(fromData) => {
connectionToP2PServer.emit('f',{data: fromData, cnid: data.cnid})
})
clientConnectionToMachine.emit(target,data)
});
}else{
clientConnectionToMachine = p2pClientConnections[data.cnid]
clientConnectionToMachine.emit(target,data)
}
})
});
config.workerStreamOutHandlers.forEach((target) => {
connectionToP2PServer.on(target,(initData) => {
if(connectedUserWebSockets[initData.auth]){
const clientConnectionToMachine = createShinobiSocketConnection(initData.auth + initData.ke + initData.id)
clientConnectionToMachine.on('connect', () => {
clientConnectionToMachine.emit(target,initData)
});
clientConnectionToMachine.on('data',(data) => {
connectionToP2PServer.emit('data',{data: data, cnid: initData.cnid})
});
}else{
s.debugLog('disconnect now!')
}
})
});
connectionToP2PServer.on('wsDestroyStream',(clientKey) => {
if(p2pClientConnections[clientKey]){
p2pClientConnections[clientKey].disconnect();
}
delete(p2pClientConnections[clientKey])
});
connectionToP2PServer.on('wsDestroy',(rawRequest) => {
if(p2pClientConnections[rawRequest.cnid]){
p2pClientConnections[rawRequest.cnid].disconnect();
}
delete(p2pClientConnections[rawRequest.cnid])
});
connectionToP2PServer.on('allowDisconnect',(bool) => {
connectionToP2PServer.allowDisconnect = true;
connectionToP2PServer.disconnect()
s.debugLog('p2p','Server Forced Disconnection')
});
const onDisconnect = () => {
if(!noLog)s.systemLog('p2p','P2P Disconnected');
killAllClientConnections()
if(!connectionToP2PServer.allowDisconnect){
if(!noLog)s.systemLog('p2p','Attempting Reconnection...');
setTimeout(() => {
startBridge()
},3000)
}else{
if(!noLog)s.systemLog('p2p','Closing Process');
process.exit()
}
}
connectionToP2PServer.on('error',onDisconnect)
connectionToP2PServer.on('disconnect',onDisconnect)
}
startBridge()
setInterval(function(){
if(
(!connectionToP2PServer || !connectionToP2PServer.connected) &&
!connectionToP2PServer.allowDisconnect
){
connectionToP2PServer.connect()
}
},1000 * 60 * 15)
}