Commit f8ccea14 authored by Axel Chalon's avatar Axel Chalon Committed by Amaury Martiny
Browse files

Use IPC instead of WebSockets (#559)

* Use IPC instead of WebSockets

* Lint

* Don't log responses

* Disable HTTP/WS JSON-RPC

* Fix IPC path

* Launch Parity with all APIs enabled for ipc

* Grumbles

* Add --ipc-path option to connect to already running node

* Fix tests
parent ff62c5e4
Pipeline #53512 passed with stages
in 10 minutes and 19 seconds
......@@ -29,8 +29,8 @@ cli
DEFAULT_CHAIN
)
.option(
'--no-run-parity',
`${productName} will not attempt to run the locally installed parity.`
'--ipc-path <path>',
`${productName} will not attempt to run the bundled Parity Ethereum, and will connect to the specified IPC socket instead. All IPC APIs must be enabled.`
)
// We want to ignore some flags that are sometimes passed to Fether, but not
// officially recognized by Fether:
......
// Copyright 2015-2019 Parity Technologies (UK) Ltd.
// This file is part of Parity.
//
// SPDX-License-Identifier: BSD-3-Clause
import EventEmitter from 'eventemitter3';
import net from 'net';
import Pino from '../utils/pino';
const pino = Pino();
/*
* IpcChannel to send messages to and receive messages from the node via ipc.
*
* Usage:
* ipcChannel.init('path/to/ipc');
* ipcChannel.send('{"jsonrpc":"2.0", ...}')
* ipcChannel.on('message', x => { console.log(x) });
*/
class IpcChannel extends EventEmitter {
_queued = [];
init (path) {
return new Promise((resolve, reject) => {
const socket = net.createConnection(path);
socket.on('connect', () => {
pino.info('Connected to IPC socket.');
this._socket = socket;
this._sendQueued();
resolve(true);
});
socket.on('data', data_ => {
const data = data_.toString();
// Sometimes we receive multiple messages at once
const messages = data.split(/\r?\n/).filter(Boolean);
messages.forEach(data => {
this.emit('message', data);
});
});
socket.on('error', err => {
pino.error('Error connecting to IPC socket.', err);
reject(err);
});
});
}
send (message) {
if (this._socket) this._socket.write(message + '\r\n');
else this._queued.push(message);
}
_sendQueued () {
this._queued.forEach(message => this.send(message));
}
}
export default new IpcChannel();
// Copyright 2015-2019 Parity Technologies (UK) Ltd.
// This file is part of Parity.
//
// SPDX-License-Identifier: BSD-3-Clause
/* eslint-env jest */
import ipcChannel from './index.js';
describe('ipcChannel', () => {
test.skip('should send & receive msgs', done => {
ipcChannel.init('/var/tmp/ipc.ipc').then(r => {
ipcChannel.send('{"jsonrpc":"2.0","id":1,"method":"parity_versionInfo"}');
});
ipcChannel.on('message', x => {
console.log('Got message', x);
done();
});
});
});
......@@ -3,16 +3,17 @@
//
// SPDX-License-Identifier: BSD-3-Clause
import { checkClockSync, signerNewToken } from '@parity/electron';
import { checkClockSync } from '@parity/electron';
import settings from 'electron-settings';
import { bundledParityPath } from '../utils/paths';
import Pino from '../utils/pino';
import setupParityEthereum from '../methods/setupParityEthereum';
import { DEFAULT_WS_PORT, TRUSTED_LOOPBACK } from '../constants';
import ipcChannel from '../ipcChannel';
const pino = Pino();
var rpcResponsesSetUp = false;
/**
* Handle all asynchronous messages from renderer to main.
*/
......@@ -58,31 +59,20 @@ export default async (fetherApp, event, data) => {
});
break;
}
case 'SIGNER_NEW_TOKEN_REQUEST': {
const token = await signerNewToken({ parityPath: bundledParityPath });
// Send back the token to the renderer process
event.sender.send('send-to-renderer', {
action: 'SIGNER_NEW_TOKEN_RESPONSE',
from: 'fether:electron',
payload: token
});
break;
}
case 'WS_INTERFACE_REQUEST': {
event.sender.send('send-to-renderer', {
action: 'WS_INTERFACE_RESPONSE',
from: 'fether:electron',
payload: TRUSTED_LOOPBACK
});
case 'RPC_REQUEST': {
if (!rpcResponsesSetUp) {
ipcChannel.on('message', data => {
event.sender.send('send-to-renderer', {
action: 'RPC_RESPONSE',
from: 'fether:electron',
payload: data
});
});
break;
}
case 'WS_PORT_REQUEST': {
event.sender.send('send-to-renderer', {
action: 'WS_PORT_RESPONSE',
from: 'fether:electron',
payload: DEFAULT_WS_PORT
});
rpcResponsesSetUp = true;
}
ipcChannel.send(data.payload);
break;
}
......
......@@ -3,58 +3,53 @@
//
// SPDX-License-Identifier: BSD-3-Clause
import { isParityRunning, runParity } from '@parity/electron';
import { runParity } from '@parity/electron';
import { bundledParityPath } from '../utils/paths';
import { bundledParityPath, BUNDLED_IPC_PATH } from '../utils/paths';
import handleError from '../utils/handleError';
import cli from '../cli';
import ipcChannel from '../ipcChannel';
import Pino from '../utils/pino';
const pino = Pino();
class ParityEthereum {
constructor () {
/*
* - If an instance of Parity Ethereum is already running, we connect to it
* and then check in fether-react if the parity_versionInfo RPC returns
* a compatible version; otherwise, we error out.
* - If no instance of Parity Ethereum is running, we run the bundled Parity
* Ethereum binary.
*
* `parity signer new-token` is run on the bundled binary in any case. We
* don't use the $PATH anymore.
*/
// Run the bundled Parity Ethereum if needed and wanted
return new Promise(async (resolve, reject) => {
// Parity Ethereum is already running: don't run the bundled binary
if (await this.isRunning()) {
resolve(true);
return;
}
// User ran Fether with --no-run-parity: don't run the bundled binary
if (!cli.runParity) {
resolve(false);
return;
}
// Parity Ethereum isn't running: run the bundled binary
await this.run();
pino.info('Running Parity Ethereum');
resolve(true);
}).catch(handleError);
if (cli.ipcPath) {
pino.info('--ipc-path provided; connecting to', cli.ipcPath);
return ipcChannel.init(cli.ipcPath).catch(handleError);
}
pino.info('Running Parity Ethereum');
// Run the bundled Parity Ethereum
return this.run()
.then(
_ =>
new Promise((resolve, reject) => {
setTimeout(resolve, 1000); // delay is needed to give time for the ipc file to be set up
})
)
.then(() => ipcChannel.init(BUNDLED_IPC_PATH))
.catch(handleError);
}
isRunning = async () => {
return isParityRunning();
};
// Run the bundled Parity Ethereum binary
run = async () => {
return runParity({
parityPath: bundledParityPath,
flags: ['--light', '--chain', cli.chain],
flags: [
'--light',
'--no-jsonrpc',
'--no-ws',
'--ipc-path',
BUNDLED_IPC_PATH,
'--ipc-apis',
'all', // we need to enable personal to use personal_signTransaction
'--chain',
cli.chain
],
onParityError: err =>
handleError(err, 'An error occured with Parity Ethereum.')
});
......
......@@ -10,6 +10,15 @@ const { app } = electron;
const IS_TEST = !app;
const IS_PACKAGED = !IS_TEST && app.isPackaged;
const BUNDLED_IPC_PATH =
process.platform === 'win32'
? path.join(
'\\\\?\\pipe',
electron.app.getPath('userData'),
'fether-parity-ipc.ipc'
)
: path.join(electron.app.getPath('userData'), 'fether-parity-ipc.ipc');
/**
* Get the path to the `static` folder.
*
......@@ -27,4 +36,4 @@ const bundledParityPath =
? path.join(staticPath, 'parity.exe')
: path.join(staticPath, 'parity');
export { IS_PACKAGED, bundledParityPath, staticPath };
export { IS_PACKAGED, BUNDLED_IPC_PATH, bundledParityPath, staticPath };
......@@ -14,22 +14,27 @@ import { name } from '../../../../package.json';
const pretty = Pino.pretty();
pretty.pipe(process.stdout);
// Create userData folder if it doesn't exist
try {
fs.statSync(app.getPath('userData'));
} catch (e) {
fs.mkdirSync(app.getPath('userData'));
}
const streams = [];
// Create 2 output streams:
// In production, create 2 output streams:
// - fether.log file (raw JSON)
// - stdout (prettified output)
const streams = [
{
const IS_TEST = !app;
if (!IS_TEST) {
// Create userData folder if it doesn't exist
try {
fs.statSync(app.getPath('userData'));
} catch (e) {
fs.mkdirSync(app.getPath('userData'));
}
streams.push({
level: 'info',
stream: fs.createWriteStream(`${app.getPath('userData')}/${name}.log`)
},
{ level: 'info', stream: pretty }
];
});
}
streams.push({ level: 'info', stream: pretty });
export default opts => Pino({ name, ...opts }, multistream(streams));
......@@ -3,20 +3,12 @@
//
// SPDX-License-Identifier: BSD-3-Clause
import { action, observable } from 'mobx';
import { observable } from 'mobx';
import Api from '@parity/api';
import { distinctUntilChanged, map, take, tap } from 'rxjs/operators';
import { distinctUntilChanged, map } from 'rxjs/operators';
import light from '@parity/light.js';
import { of, timer, zip } from 'rxjs';
import store from 'store';
import Debug from '../utils/debug';
import LS_PREFIX from './utils/lsPrefix';
import * as postMessage from '../utils/postMessage';
const debug = Debug('parityStore');
const LS_KEY = `${LS_PREFIX}::secureToken`;
import { timer } from 'rxjs';
import PostMessageProvider from '../utils/PostMessageProvider';
export class ParityStore {
// TODO This is not working
......@@ -32,38 +24,9 @@ export class ParityStore {
api = undefined;
constructor () {
// Request WS port and interface from electron
postMessage.send('WS_INTERFACE_REQUEST');
postMessage.send('WS_PORT_REQUEST');
const lsToken = store.get(LS_KEY);
const token$ = lsToken
? of(lsToken).pipe(tap(() => debug('Got token from localStorage.')))
: this.requestNewToken$();
zip(
token$,
postMessage.listen$('WS_INTERFACE_RESPONSE'),
postMessage.listen$('WS_PORT_RESPONSE')
)
.pipe(take(1))
.subscribe(([token, wsInterface, wsPort]) =>
this.connectToApi(token, wsInterface, wsPort)
);
}
connectToApi (token, wsInterface, wsPort) {
// Get the provider, optionally from --ws-interface and --ws-port flags
let provider = `ws://${wsInterface}:${wsPort}`;
const provider = new PostMessageProvider();
debug(`Connecting to ${provider}.`);
const api = new Api(
// FIXME - change to WsSecure when implement `wss` and security certificates
new Api.Provider.Ws(
provider,
token.replace(/[^a-zA-Z0-9]/g, '') // Sanitize token
)
);
const api = new Api(provider);
// Initialize the light.js lib
light.setApi(api);
......@@ -71,21 +34,6 @@ export class ParityStore {
// Also set api as member for React Components to use it if needed
this.api = api;
}
requestNewToken$ () {
// Request new token from Electron
debug('Requesting new token.');
postMessage.send('SIGNER_NEW_TOKEN_REQUEST');
return postMessage.listen$('SIGNER_NEW_TOKEN_RESPONSE').pipe(
tap(() => debug('Successfully received new token.')),
tap(this.updateLS)
);
}
@action
updateLS = token => {
store.set(LS_KEY, token);
};
}
export default new ParityStore();
// Copyright 2015-2019 Parity Technologies (UK) Ltd.
// This file is part of Parity.
// Parity is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// Parity is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
// You should have received a copy of the GNU General Public License
// along with Parity. If not, see <http://www.gnu.org/licenses/>.
import * as postMessage from './postMessage';
import EventEmitter from 'eventemitter3';
export default class PostMessageProvider extends EventEmitter {
constructor (destination, source) {
super();
this._destination = destination || window.parent;
this._id = 1;
this._messages = {};
this._subscriptionsToId = {};
this._receiveMessage = this._receiveMessage.bind(this);
this.subscribe = this.subscribe.bind(this);
this.unsubscribe = this.unsubscribe.bind(this);
postMessage.listen$('RPC_RESPONSE').subscribe(this._receiveMessage);
}
_constructMessage (id, data) {
return Object.assign({}, data, {
id,
to: 'shell'
});
}
_send (message, meta) {
const { method, params } = message;
const id = this._id++;
const raw = JSON.stringify({
jsonrpc: '2.0',
id,
method,
params
});
postMessage.send('RPC_REQUEST', raw);
this._messages[id] = meta;
}
send (method, params, callback) {
return new Promise((resolve, reject) => {
this._send(
{
method,
params
},
{
resolve: (...argz) => {
resolve(...argz);
callback(null, ...argz);
},
reject: err => {
reject(err);
callback(err);
}
}
);
});
}
_methodsFromApi (api) {
if (api.subscription) {
const { subscribe, unsubscribe, subscription } = api;
return {
method: subscribe,
uMethod: unsubscribe,
subscription
};
}
return {
method: `${api}_subscribe`,
uMethod: `${api}_unsubscribe`,
subscription: `${api}_subscription`
};
}
subscribe (api, callback, params) {
return new Promise((resolve, reject) => {
const mf = this._methodsFromApi(api);
this._send(
{
method: mf.method,
params
},
{
method: mf.method,
uMethod: mf.uMethod,
subscription: mf.subscription,
resolve,
reject,
callback
}
);
});
}
unsubscribe (subId) {
return new Promise((resolve, reject) => {
this._send(
{
method: this._messages[subId].uMethod,
params: [subId]
},
{
resolve: v => {
delete this._messages[subId];
resolve(v);
},
reject
}
);
});
}
_receiveMessage (raw) {
const parsed = JSON.parse(raw);
const { id, error } = parsed;
const subscription = parsed.params && parsed.params.subscription;
if (subscription) {
// subscription notification
const result = parsed.params.result;
let messageId = this._subscriptionsToId[subscription];
this._messages[messageId].callback(error && new Error(error), result);
} else {
// request response
const result = parsed.result;
if (error) {
this._messages[id].reject(new Error(error));
} else {
this._messages[id].resolve(result);
// if this is a response to a subscription request, we store the mapping
// between request id & subscription id
if (this._messages[id].subscription) {
this._subscriptionsToId[result] = id;
} else {
delete this._messages[id];
}
}
}
}
}
......@@ -6,10 +6,6 @@
import { filter, map, publish } from 'rxjs/operators';
import { Observable } from 'rxjs';
import Debug from '../utils/debug';
const debug = Debug('postMessage');
const RENDERER_ORIGIN =
process.env.NODE_ENV === 'production' ? 'file://' : 'http://localhost:3000';
......@@ -34,8 +30,6 @@ const messages$ = Observable.create(observer => {
return;
}
debug(`Received post message from ${from ? `${from}` : ''}`, data);
observer.next(data);
};
......