fluencelabs/fluence-js
1
0
Fork 0
mirror of https://github.com/fluencelabs/fluence-js.git synced 2024-12-04 18:00:18 +00:00
Code Issues Projects Releases Wiki Activity

Tetraplets (#1)

Browse Source
This commit is contained in:
coder11 2020-12-23 17:24:22 +03:00 committed by GitHub
parent 2876417554
commit f6fd95ce77
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
26 changed files with 739 additions and 504 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
.eslintrc.js Normal file
View File

@ -0,0 +1,23 @@
module.exports = {
parser: '@typescript-eslint/parser',
parserOptions: {
ecmaVersion: 12,
sourceType: 'module', // Allows for the use of imports
},
env: {
browser: true,
es2021: true,
},
extends: [
'airbnb-base',
'plugin:@typescript-eslint/eslint-recommended',
'plugin:@typescript-eslint/recommended',
// Enables eslint-plugin-prettier and eslint-config-prettier. This will display prettier errors as ESLint errors. Make sure this is always the last configuration in the extends array.
'plugin:prettier/recommended',
],
plugins: ['@typescript-eslint', 'prettier'],
rules: {},
settings: {
'import/extensions': ['.js', '.ts'],
},
};

15
.gitignore vendored Normal file
View File

@ -0,0 +1,15 @@
# Logs
logs
*.log
npm-debug.log*
yarn-debug.log*
yarn-error.log*
lerna-debug.log*
# Compiled binary addons (https://nodejs.org/api/addons.html)
build/Release
bundle/
# Dependency directories
node_modules/
jspm_packages/

8
.prettierrc.js Normal file
View File

@ -0,0 +1,8 @@
module.exports = {
semi: true,
trailingComma: "all",
singleQuote: true,
printWidth: 120,
tabWidth: 4,
useTabs: false
};

8
package-lock.json generated
View File

@ -1,6 +1,6 @@
{
"name": "fluence",
"version": "0.7.101",
"version": "0.7.102",
"lockfileVersion": 1,
"requires": true,
"dependencies": {
@ -25,9 +25,9 @@
}
},
"@fluencelabs/aquamarine-stepper": {
"version": "0.0.21",
"resolved": "https://registry.npmjs.org/@fluencelabs/aquamarine-stepper/-/aquamarine-stepper-0.0.21.tgz",
"integrity": "sha512-bw0tdC5+fihzw+BxA02TrNIzMp2reuV21RqPMlDUExh2tbSzHYKBXKOxGsIY10j3QYWpHQZK9N341VnA3nw6Sw=="
"version": "0.0.27",
"resolved": "https://registry.npmjs.org/@fluencelabs/aquamarine-stepper/-/aquamarine-stepper-0.0.27.tgz",
"integrity": "sha512-UT25immkpJ79/1cBunAr8owEuaLfrhy3njw3BLfonF316gCX6pFBihlivOsvQlvw8/cL5RJDwlkzBLYAf6Lexw=="
},
"@sinonjs/commons": {
"version": "1.7.2",

4
package.json
View File

@ -1,6 +1,6 @@
{
"name": "fluence",
"version": "0.7.101",
"version": "0.7.102",
"description": "the browser js-libp2p client for the Fluence network",
"main": "./dist/fluence.js",
"typings": "./dist/fluence.d.ts",
@ -16,7 +16,7 @@
"author": "Fluence Labs",
"license": "Apache-2.0",
"dependencies": {
"@fluencelabs/aquamarine-stepper": "0.0.21",
"@fluencelabs/aquamarine-stepper": "0.0.27",
"async": "3.2.0",
"base64-js": "1.3.1",
"bs58": "4.0.1",

26
src/dataStorage.ts
View File

@ -1,27 +1,27 @@
import {getCurrentParticleId, registerService} from "./globalState";
import {ServiceMultiple} from "./service";
import log from "loglevel";
import { getCurrentParticleId, registerService } from './globalState';
import { ServiceMultiple } from './service';
import log from 'loglevel';
let storage: Map<string, Map<string, any>> = new Map();
export function addData(particleId: string, data: Map<string, any>, ttl: number) {
storage.set(particleId, data)
storage.set(particleId, data);
setTimeout(() => {
log.debug(`data for ${particleId} is deleted`)
storage.delete(particleId)
}, ttl)
log.debug(`data for ${particleId} is deleted`);
storage.delete(particleId);
}, ttl);
}
export const storageService = new ServiceMultiple("")
storageService.registerFunction("load", (args: any[]) => {
export const storageService = new ServiceMultiple('');
storageService.registerFunction('load', (args: any[]) => {
let current = getCurrentParticleId();
let data = storage.get(current)
let data = storage.get(current);
if (data) {
return data.get(args[0])
return data.get(args[0]);
} else {
return {}
return {};
}
})
});
registerService(storageService);

25
src/fluence.ts
View File

@ -14,17 +14,16 @@
* limitations under the License.
*/
import * as PeerId from "peer-id";
import Multiaddr from "multiaddr"
import {FluenceClient} from "./fluenceClient";
import * as log from "loglevel";
import {LogLevelDesc} from "loglevel";
import {parseAstClosure} from "./stepper";
import * as PeerId from 'peer-id';
import Multiaddr from 'multiaddr';
import { FluenceClient } from './fluenceClient';
import * as log from 'loglevel';
import { LogLevelDesc } from 'loglevel';
import { parseAstClosure } from './stepper';
log.setLevel('info')
log.setLevel('info');
export default class Fluence {
static setLogLevel(level: LogLevelDesc): void {
log.setLevel(level);
}
@ -33,7 +32,7 @@ export default class Fluence {
* Generates new peer id with Ed25519 private key.
*/
static async generatePeerId(): Promise<PeerId> {
return await PeerId.create({keyType: "Ed25519"});
return await PeerId.create({ keyType: 'Ed25519' });
}
/**
@ -43,10 +42,10 @@ export default class Fluence {
*/
static async local(peerId?: PeerId): Promise<FluenceClient> {
if (!peerId) {
peerId = await Fluence.generatePeerId()
peerId = await Fluence.generatePeerId();
}
let client = new FluenceClient(peerId);
let client = new FluenceClient(peerId);
await client.instantiateInterpreter();
return client;
@ -69,7 +68,7 @@ export default class Fluence {
/// NOTE & TODO: interpreter is instantiated every time, make it a lazy constant?
static async parseAIR(script: string): Promise<string> {
let closure = await parseAstClosure();
return closure(script)
return closure(script);
}
}
@ -79,6 +78,6 @@ declare global {
}
}
if (typeof window !== "undefined") {
if (typeof window !== 'undefined') {
window.Fluence = Fluence;
}

256
src/fluenceClient.ts
View File

@ -14,22 +14,21 @@
* limitations under the License.
*/
import { build, Particle } from './particle';
import { StepperOutcome } from './stepperOutcome';
import * as PeerId from 'peer-id';
import Multiaddr from 'multiaddr';
import { FluenceConnection } from './fluenceConnection';
import { Subscriptions } from './subscriptions';
import { enqueueParticle, getCurrentParticleId, popParticle, setCurrentParticleId } from './globalState';
import { instantiateInterpreter, InterpreterInvoke } from './stepper';
import log from 'loglevel';
import { waitService } from './helpers/waitService';
import { ModuleConfig } from './moduleConfig';
import {build, Particle} from "./particle";
import {StepperOutcome} from "./stepperOutcome";
import * as PeerId from "peer-id";
import Multiaddr from "multiaddr"
import {FluenceConnection} from "./fluenceConnection";
import {Subscriptions} from "./subscriptions";
import {enqueueParticle, getCurrentParticleId, popParticle, setCurrentParticleId} from "./globalState";
import {instantiateInterpreter, InterpreterInvoke} from "./stepper";
import log from "loglevel";
import {waitService} from "./helpers/waitService";
import {ModuleConfig} from "./moduleConfig";
const bs58 = require('bs58');
const bs58 = require('bs58')
const INFO_LOG_LEVEL = 2
const INFO_LOG_LEVEL = 2;
export class FluenceClient {
readonly selfPeerId: PeerId;
@ -50,22 +49,21 @@ export class FluenceClient {
* Pass a particle to a interpreter and send a result to other services.
*/
private async handleParticle(particle: Particle): Promise<void> {
// if a current particle is processing, add new particle to the queue
if (getCurrentParticleId() !== undefined && getCurrentParticleId() !== particle.id) {
enqueueParticle(particle);
} else {
if (this.interpreter === undefined) {
throw new Error("Undefined. Interpreter is not initialized. Use 'Fluence.connect' to create a client.")
throw new Error("Undefined. Interpreter is not initialized. Use 'Fluence.connect' to create a client.");
}
// start particle processing if queue is empty
try {
setCurrentParticleId(particle.id)
setCurrentParticleId(particle.id);
// check if a particle is relevant
let now = Date.now();
let actualTtl = particle.timestamp + particle.ttl - now;
if (actualTtl <= 0) {
log.info(`Particle expired. Now: ${now}, ttl: ${particle.ttl}, ts: ${particle.timestamp}`)
log.info(`Particle expired. Now: ${now}, ttl: ${particle.ttl}, ts: ${particle.timestamp}`);
} else {
// if there is no subscription yet, previous data is empty
let prevData = [];
@ -73,35 +71,40 @@ export class FluenceClient {
if (prevParticle) {
prevData = prevParticle.data;
// update a particle in a subscription
this.subscriptions.update(particle)
this.subscriptions.update(particle);
} else {
// set a particle with actual ttl
this.subscriptions.subscribe(particle, actualTtl)
this.subscriptions.subscribe(particle, actualTtl);
}
let stepperOutcomeStr = this.interpreter(particle.init_peer_id, particle.script, JSON.stringify(prevData), JSON.stringify(particle.data))
let stepperOutcomeStr = this.interpreter(
particle.init_peer_id,
particle.script,
JSON.stringify(prevData),
JSON.stringify(particle.data),
);
let stepperOutcome: StepperOutcome = JSON.parse(stepperOutcomeStr);
if (log.getLevel() <= INFO_LOG_LEVEL) {
log.info("inner interpreter outcome:");
let so = {...stepperOutcome}
log.info('inner interpreter outcome:');
let so = { ...stepperOutcome };
try {
so.data = JSON.parse(Buffer.from(so.data).toString("utf8"));
so.data = JSON.parse(Buffer.from(so.data).toString('utf8'));
log.info(so);
} catch (e) {
log.info("cannot parse StepperOutcome data as JSON: ", e);
log.info('cannot parse StepperOutcome data as JSON: ', e);
}
}
// update data after aquamarine execution
let newParticle: Particle = {...particle};
newParticle.data = stepperOutcome.data
let newParticle: Particle = { ...particle };
newParticle.data = stepperOutcome.data;
this.subscriptions.update(newParticle)
this.subscriptions.update(newParticle);
// do nothing if there is no `next_peer_pks` or if client isn't connected to the network
if (stepperOutcome.next_peer_pks.length > 0 && this.connection) {
await this.connection.sendParticle(newParticle).catch((reason) => {
console.error(`Error on sending particle with id ${particle.id}: ${reason}`)
console.error(`Error on sending particle with id ${particle.id}: ${reason}`);
});
}
}
@ -112,7 +115,7 @@ export class FluenceClient {
if (nextParticle) {
// update current particle
setCurrentParticleId(nextParticle.id);
await this.handleParticle(nextParticle)
await this.handleParticle(nextParticle);
} else {
// wait for a new call (do nothing) if there is no new particle in a queue
setCurrentParticleId(undefined);
@ -125,21 +128,20 @@ export class FluenceClient {
* Handle incoming particle from a relay.
*/
private handleExternalParticle(): (particle: Particle) => Promise<void> {
let _this = this;
return async (particle: Particle) => {
let data = particle.data;
let error: any = data["protocol!error"]
let error: any = data['protocol!error'];
if (error !== undefined) {
log.error("error in external particle: ")
log.error(error)
log.error('error in external particle: ');
log.error(error);
} else {
log.info("handle external particle: ")
log.info(particle)
log.info('handle external particle: ');
log.info(particle);
await _this.handleParticle(particle);
}
}
};
}
async disconnect(): Promise<void> {
@ -162,13 +164,13 @@ export class FluenceClient {
multiaddr = Multiaddr(multiaddr);
if (!this.interpreter) {
throw Error("you must call 'instantiateInterpreter' before 'connect'")
throw Error("you must call 'instantiateInterpreter' before 'connect'");
}
let nodePeerId = multiaddr.getPeerId();
this.nodePeerIdStr = nodePeerId;
if (!nodePeerId) {
throw Error("'multiaddr' did not contain a valid peer id")
throw Error("'multiaddr' did not contain a valid peer id");
}
let firstConnection: boolean = true;
@ -186,7 +188,7 @@ export class FluenceClient {
async sendParticle(particle: Particle): Promise<string> {
await this.handleParticle(particle);
return particle.id
return particle.id;
}
async executeParticle(particle: Particle) {
@ -194,21 +196,29 @@ export class FluenceClient {
}
nodeIdentityCall(): string {
return `(call "${this.nodePeerIdStr}" ("op" "identity") [] void[])`
return `(call "${this.nodePeerIdStr}" ("op" "identity") [] void[])`;
}
async requestResponse<T>(name: string, call: (nodeId: string) => string, returnValue: string, data: Map<string, any>, handleResponse: (args: any[]) => T, nodeId?: string, ttl?: number): Promise<T> {
async requestResponse<T>(
name: string,
call: (nodeId: string) => string,
returnValue: string,
data: Map<string, any>,
handleResponse: (args: any[]) => T,
nodeId?: string,
ttl?: number,
): Promise<T> {
if (!ttl) {
ttl = 10000
ttl = 10000;
}
if (!nodeId) {
nodeId = this.nodePeerIdStr
nodeId = this.nodePeerIdStr;
}
let serviceCall = call(nodeId)
let serviceCall = call(nodeId);
let namedPromise = waitService(name, handleResponse, ttl)
let namedPromise = waitService(name, handleResponse, ttl);
let script = `(seq
${this.nodeIdentityCall()}
@ -220,18 +230,24 @@ export class FluenceClient {
(call "${this.selfPeerIdStr}" ("${namedPromise.name}" "") [${returnValue}] void[])
)
)
`
`;
let particle = await build(this.selfPeerId, script, data, ttl)
let particle = await build(this.selfPeerId, script, data, ttl);
await this.sendParticle(particle);
return namedPromise.promise
return namedPromise.promise;
}
/**
* Send a script to add module to a relay. Waiting for a response from a relay.
*/
async addModule(name: string, moduleBase64: string, config?: ModuleConfig, nodeId?: string, ttl?: number): Promise<void> {
async addModule(
name: string,
moduleBase64: string,
config?: ModuleConfig,
nodeId?: string,
ttl?: number,
): Promise<void> {
if (!config) {
config = {
name: name,
@ -239,122 +255,166 @@ export class FluenceClient {
logger_enabled: true,
wasi: {
envs: {},
preopened_files: ["/tmp"],
preopened_files: ['/tmp'],
mapped_dirs: {},
}
}
},
};
}
let data = new Map()
data.set("module_bytes", moduleBase64)
data.set("module_config", config)
let data = new Map();
data.set('module_bytes', moduleBase64);
data.set('module_config', config);
let call = (nodeId: string) => `(call "${nodeId}" ("dist" "add_module") [module_bytes module_config] void[])`
let call = (nodeId: string) => `(call "${nodeId}" ("dist" "add_module") [module_bytes module_config] void[])`;
return this.requestResponse("addModule", call, "", data, () => {}, nodeId, ttl)
return this.requestResponse('addModule', call, '', data, () => {}, nodeId, ttl);
}
/**
* Send a script to add module to a relay. Waiting for a response from a relay.
*/
async addBlueprint(name: string, dependencies: string[], blueprintId?: string, nodeId?: string, ttl?: number): Promise<string> {
let returnValue = "blueprint_id";
let call = (nodeId: string) => `(call "${nodeId}" ("dist" "add_blueprint") [blueprint] ${returnValue})`
async addBlueprint(
name: string,
dependencies: string[],
blueprintId?: string,
nodeId?: string,
ttl?: number,
): Promise<string> {
let returnValue = 'blueprint_id';
let call = (nodeId: string) => `(call "${nodeId}" ("dist" "add_blueprint") [blueprint] ${returnValue})`;
let data = new Map()
data.set("blueprint", { name: name, dependencies: dependencies, id: blueprintId })
let data = new Map();
data.set('blueprint', { name: name, dependencies: dependencies, id: blueprintId });
return this.requestResponse("addBlueprint", call, returnValue, data, (args: any[]) => args[0] as string, nodeId, ttl)
return this.requestResponse(
'addBlueprint',
call,
returnValue,
data,
(args: any[]) => args[0] as string,
nodeId,
ttl,
);
}
/**
* Send a script to create a service to a relay. Waiting for a response from a relay.
*/
async createService(blueprintId: string, nodeId?: string, ttl?: number): Promise<string> {
let returnValue = "service_id";
let call = (nodeId: string) => `(call "${nodeId}" ("srv" "create") [blueprint_id] ${returnValue})`
let returnValue = 'service_id';
let call = (nodeId: string) => `(call "${nodeId}" ("srv" "create") [blueprint_id] ${returnValue})`;
let data = new Map()
data.set("blueprint_id", blueprintId)
let data = new Map();
data.set('blueprint_id', blueprintId);
return this.requestResponse("createService", call, returnValue, data, (args: any[]) => args[0] as string, nodeId, ttl)
return this.requestResponse(
'createService',
call,
returnValue,
data,
(args: any[]) => args[0] as string,
nodeId,
ttl,
);
}
/**
* Get all available modules hosted on a connected relay.
*/
async getAvailableModules(nodeId?: string, ttl?: number): Promise<string[]> {
let returnValue = "modules";
let call = (nodeId: string) => `(call "${nodeId}" ("dist" "get_modules") [] ${returnValue})`
let returnValue = 'modules';
let call = (nodeId: string) => `(call "${nodeId}" ("dist" "get_modules") [] ${returnValue})`;
return this.requestResponse("getAvailableModules", call, returnValue, new Map(), (args: any[]) => args[0] as string[], nodeId, ttl)
return this.requestResponse(
'getAvailableModules',
call,
returnValue,
new Map(),
(args: any[]) => args[0] as string[],
nodeId,
ttl,
);
}
/**
* Get all available blueprints hosted on a connected relay.
*/
async getBlueprints(nodeId: string, ttl?: number): Promise<string[]> {
let returnValue = "blueprints";
let call = (nodeId: string) => `(call "${nodeId}" ("dist" "get_blueprints") [] ${returnValue})`
let returnValue = 'blueprints';
let call = (nodeId: string) => `(call "${nodeId}" ("dist" "get_blueprints") [] ${returnValue})`;
return this.requestResponse("getBlueprints", call, returnValue, new Map(), (args: any[]) => args[0] as string[], nodeId, ttl)
return this.requestResponse(
'getBlueprints',
call,
returnValue,
new Map(),
(args: any[]) => args[0] as string[],
nodeId,
ttl,
);
}
/**
* Add a provider to DHT network to neighborhood around a key.
*/
async addProvider(key: Buffer, providerPeer: string, providerServiceId?: string, nodeId?: string, ttl?: number): Promise<void> {
let call = (nodeId: string) => `(call "${nodeId}" ("dht" "add_provider") [key provider] void[])`
async addProvider(
key: Buffer,
providerPeer: string,
providerServiceId?: string,
nodeId?: string,
ttl?: number,
): Promise<void> {
let call = (nodeId: string) => `(call "${nodeId}" ("dht" "add_provider") [key provider] void[])`;
key = bs58.encode(key)
key = bs58.encode(key);
let provider = {
peer: providerPeer,
service_id: providerServiceId
}
service_id: providerServiceId,
};
let data = new Map()
data.set("key", key)
data.set("provider", provider)
let data = new Map();
data.set('key', key);
data.set('provider', provider);
return this.requestResponse("addProvider", call, "", data, () => {}, nodeId, ttl)
return this.requestResponse('addProvider', call, '', data, () => {}, nodeId, ttl);
}
/**
* Get a provider from DHT network from neighborhood around a key..
*/
async getProviders(key: Buffer, nodeId?: string, ttl?: number): Promise<any> {
key = bs58.encode(key)
key = bs58.encode(key);
let returnValue = "providers"
let call = (nodeId: string) => `(call "${nodeId}" ("dht" "get_providers") [key] providers[])`
let returnValue = 'providers';
let call = (nodeId: string) => `(call "${nodeId}" ("dht" "get_providers") [key] providers[])`;
let data = new Map()
data.set("key", key)
let data = new Map();
data.set('key', key);
return this.requestResponse("getProviders", call, returnValue, data, (args) => args[0], nodeId, ttl)
return this.requestResponse('getProviders', call, returnValue, data, (args) => args[0], nodeId, ttl);
}
/**
* Get relays neighborhood
*/
async neighborhood(node: string, ttl?: number): Promise<string[]> {
let returnValue = "neighborhood"
let call = (nodeId: string) => `(call "${nodeId}" ("dht" "neighborhood") [node] ${returnValue})`
let returnValue = 'neighborhood';
let call = (nodeId: string) => `(call "${nodeId}" ("dht" "neighborhood") [node] ${returnValue})`;
let data = new Map()
data.set("node", node)
let data = new Map();
data.set('node', node);
return this.requestResponse("neighborhood", call, returnValue, data, (args) => args[0] as string[], node, ttl)
return this.requestResponse('neighborhood', call, returnValue, data, (args) => args[0] as string[], node, ttl);
}
/**
* Call relays 'identity' method. It should return passed 'fields'
*/
async relayIdentity(fields: string[], data: Map<string, any>, nodeId?: string, ttl?: number): Promise<any> {
let returnValue = "id";
let call = (nodeId: string) => `(call "${nodeId}" ("op" "identity") [${fields.join(" ")}] ${returnValue})`
let returnValue = 'id';
let call = (nodeId: string) => `(call "${nodeId}" ("op" "identity") [${fields.join(' ')}] ${returnValue})`;
return this.requestResponse("getIdentity", call, returnValue, data, (args: any[]) => args[0], nodeId, ttl)
return this.requestResponse('getIdentity', call, returnValue, data, (args: any[]) => args[0], nodeId, ttl);
}
}

68
src/fluenceConnection.ts
View File

@ -14,27 +14,26 @@
* limitations under the License.
*/
import Websockets from "libp2p-websockets";
import Mplex from "libp2p-mplex";
import SECIO from "libp2p-secio";
import Peer from "libp2p";
import {decode, encode} from "it-length-prefixed";
import pipe from "it-pipe";
import Multiaddr from "multiaddr";
import PeerId from "peer-id";
import Websockets from 'libp2p-websockets';
import Mplex from 'libp2p-mplex';
import SECIO from 'libp2p-secio';
import Peer from 'libp2p';
import { decode, encode } from 'it-length-prefixed';
import pipe from 'it-pipe';
import Multiaddr from 'multiaddr';
import PeerId from 'peer-id';
import * as log from 'loglevel';
import {build, parseParticle, Particle, stringifyParticle} from "./particle";
import { build, parseParticle, Particle, stringifyParticle } from './particle';
export const PROTOCOL_NAME = '/fluence/faas/1.0.0';
enum Status {
Initializing = "Initializing",
Connected = "Connected",
Disconnected = "Disconnected"
Initializing = 'Initializing',
Connected = 'Connected',
Disconnected = 'Disconnected',
}
export class FluenceConnection {
private readonly selfPeerId: PeerId;
private node: LibP2p;
private readonly address: Multiaddr;
@ -59,7 +58,7 @@ export class FluenceConnection {
transport: [Websockets],
streamMuxer: [Mplex],
connEncryption: [SECIO],
peerDiscovery: []
peerDiscovery: [],
},
});
@ -67,7 +66,7 @@ export class FluenceConnection {
}
isConnected() {
return this.status === Status.Connected
return this.status === Status.Connected;
}
// connection status. If `Disconnected`, it cannot be reconnected
@ -77,29 +76,25 @@ export class FluenceConnection {
if (this.status === Status.Initializing) {
await this.node.start();
log.debug("dialing to the node with address: " + this.node.peerId.toB58String());
log.debug('dialing to the node with address: ' + this.node.peerId.toB58String());
await this.node.dial(this.address);
let _this = this;
this.node.handle([PROTOCOL_NAME], async ({connection, stream}) => {
pipe(
stream.source,
decode(),
async function (source: AsyncIterable<string>) {
for await (const msg of source) {
try {
let particle = parseParticle(msg);
log.debug("Particle is received:");
log.debug(JSON.stringify(particle, undefined, 2));
_this.handleCall(particle);
} catch(e) {
log.error("error on handling a new incoming message: " + e);
}
this.node.handle([PROTOCOL_NAME], async ({ connection, stream }) => {
pipe(stream.source, decode(), async function (source: AsyncIterable<string>) {
for await (const msg of source) {
try {
let particle = parseParticle(msg);
log.debug('Particle is received:');
log.debug(JSON.stringify(particle, undefined, 2));
_this.handleCall(particle);
} catch (e) {
log.error('error on handling a new incoming message: ' + e);
}
}
)
});
});
this.status = Status.Connected;
@ -110,7 +105,7 @@ export class FluenceConnection {
private checkConnectedOrThrow() {
if (this.status !== Status.Connected) {
throw Error(`connection is in ${this.status} state`)
throw Error(`connection is in ${this.status} state`);
}
}
@ -120,17 +115,20 @@ export class FluenceConnection {
}
async buildParticle(script: string, data: Map<string, any>, ttl?: number): Promise<Particle> {
return build(this.selfPeerId, script, data, ttl)
return build(this.selfPeerId, script, data, ttl);
}
async sendParticle(particle: Particle): Promise<void> {
this.checkConnectedOrThrow();
let particleStr = stringifyParticle(particle);
log.debug("send particle: \n" + JSON.stringify(particle, undefined, 2));
log.debug('send particle: \n' + JSON.stringify(particle, undefined, 2));
// create outgoing substream
const conn = await this.node.dialProtocol(this.address, PROTOCOL_NAME) as {stream: Stream; protocol: string};
const conn = (await this.node.dialProtocol(this.address, PROTOCOL_NAME)) as {
stream: Stream;
protocol: string;
};
pipe(
[Buffer.from(particleStr, 'utf8')],

10
src/globalState.ts
View File

@ -14,8 +14,8 @@
* limitations under the License.
*/
import {Service} from "./service";
import {Particle} from "./particle";
import { Service } from './service';
import { Particle } from './particle';
// TODO put state with wasm file in each created FluenceClient
let services: Map<string, Service> = new Map();
@ -39,13 +39,13 @@ export function popParticle(): Particle | undefined {
}
export function registerService(service: Service) {
services.set(service.serviceId, service)
services.set(service.serviceId, service);
}
export function deleteService(serviceId: string): boolean {
return services.delete(serviceId)
return services.delete(serviceId);
}
export function getService(serviceId: string): Service | undefined {
return services.get(serviceId)
return services.get(serviceId);
}

40
src/helpers/waitService.ts
View File

@ -1,15 +1,15 @@
/**
* Creates service that will wait for a response from external peers.
*/
import {genUUID} from "../particle";
import log from "loglevel";
import {ServiceMultiple} from "../service";
import {deleteService, registerService} from "../globalState";
import {delay} from "../utils";
import { genUUID } from '../particle';
import log from 'loglevel';
import { ServiceMultiple } from '../service';
import { deleteService, registerService } from '../globalState';
import { delay } from '../utils';
interface NamedPromise<T> {
promise: Promise<T>,
name: string
promise: Promise<T>;
name: string;
}
/**
@ -19,28 +19,28 @@ interface NamedPromise<T> {
* @param ttl
*/
export function waitResult(ttl: number): NamedPromise<any[]> {
return waitService(genUUID(), (args: any[]) => args, ttl)
return waitService(genUUID(), (args: any[]) => args, ttl);
}
export function waitService<T>(functionName: string, func: (args: any[]) => T, ttl: number): NamedPromise<T> {
let serviceName = `${functionName}-${genUUID()}`;
log.info(`Create waiting service '${serviceName}'`)
let service = new ServiceMultiple(serviceName)
registerService(service)
log.info(`Create waiting service '${serviceName}'`);
let service = new ServiceMultiple(serviceName);
registerService(service);
let promise: Promise<T> = new Promise(function (resolve) {
service.registerFunction("", (args: any[]) => {
resolve(func(args))
return {}
})
})
service.registerFunction('', (args: any[]) => {
resolve(func(args));
return {};
});
});
let timeout = delay<T>(ttl, "Timeout on waiting " + serviceName)
let timeout = delay<T>(ttl, 'Timeout on waiting ' + serviceName);
return {
name: serviceName,
promise: Promise.race([promise, timeout]).finally(() => {
deleteService(serviceName)
})
}
deleteService(serviceName);
}),
};
}

16
src/moduleConfig.ts
View File

@ -15,15 +15,15 @@
*/
export interface ModuleConfig {
name: string,
mem_pages_count?: number,
logger_enabled?: boolean,
wasi?: Wasi,
mounted_binaries?: object
name: string;
mem_pages_count?: number;
logger_enabled?: boolean;
wasi?: Wasi;
mounted_binaries?: object;
}
export interface Wasi {
envs?: object,
preopened_files?: string[],
mapped_dirs?: object,
envs?: object;
preopened_files?: string[];
mapped_dirs?: object;
}

56
src/particle.ts
View File

@ -15,31 +15,31 @@
*/
import { v4 as uuidv4 } from 'uuid';
import PeerId from "peer-id";
import {encode} from "bs58";
import {addData} from "./dataStorage";
import PeerId from 'peer-id';
import { encode } from 'bs58';
import { addData } from './dataStorage';
const DEFAULT_TTL = 7000;
export interface Particle {
id: string,
init_peer_id: string,
timestamp: number,
ttl: number,
script: string,
id: string;
init_peer_id: string;
timestamp: number;
ttl: number;
script: string;
// sign upper fields
signature: string,
data: any
signature: string;
data: any;
}
function wrapScript(selfPeerId: string, script: string, fields: string[]): string {
fields.forEach((v) => {
script = `
script = `
(seq
(call %init_peer_id% ("" "load") ["${v}"] ${v})
${script}
)
`
`;
});
return script;
@ -47,14 +47,14 @@ function wrapScript(selfPeerId: string, script: string, fields: string[]): strin
export async function build(peerId: PeerId, script: string, data: Map<string, any>, ttl?: number): Promise<Particle> {
let id = genUUID();
let currentTime = (new Date()).getTime();
let currentTime = new Date().getTime();
if (ttl === undefined) {
ttl = DEFAULT_TTL
ttl = DEFAULT_TTL;
}
addData(id, data, ttl);
script = wrapScript(peerId.toB58String(), script, Array.from(data.keys()))
script = wrapScript(peerId.toB58String(), script, Array.from(data.keys()));
let particle: Particle = {
id: id,
@ -62,9 +62,9 @@ export async function build(peerId: PeerId, script: string, data: Map<string, an
timestamp: currentTime,
ttl: ttl,
script: script,
signature: "",
data: []
}
signature: '',
data: [],
};
particle.signature = await signParticle(peerId, particle);
@ -75,16 +75,15 @@ export async function build(peerId: PeerId, script: string, data: Map<string, an
* Copies a particle and stringify it.
*/
export function stringifyParticle(call: Particle): string {
let obj: any = {...call};
obj.action = "Particle"
let obj: any = { ...call };
obj.action = 'Particle';
// delete it after signatures will be implemented on nodes
obj.signature = []
obj.signature = [];
return JSON.stringify(obj)
return JSON.stringify(obj);
}
export function parseParticle(str: string): Particle {
let json = JSON.parse(str);
@ -95,8 +94,8 @@ export function parseParticle(str: string): Particle {
ttl: json.ttl,
script: json.script,
signature: json.signature,
data: json.data
}
data: json.data,
};
}
export function canonicalBytes(particle: Particle) {
@ -119,12 +118,11 @@ export function canonicalBytes(particle: Particle) {
/**
* Sign a particle with a private key from peerId.
*/
export async function signParticle(peerId: PeerId,
particle: Particle): Promise<string> {
export async function signParticle(peerId: PeerId, particle: Particle): Promise<string> {
let bufToSign = canonicalBytes(particle);
let signature = await peerId.privKey.sign(bufToSign)
return encode(signature)
let signature = await peerId.privKey.sign(bufToSign);
return encode(signature);
}
export function genUUID() {

25
src/securityTetraplet.ts Normal file
View File

@ -0,0 +1,25 @@
/*
* Copyright 2020 Fluence Labs Limited
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
export interface ResolvedTriplet {
peer_pk: string;
service_id: string;
function_name: string;
}
export interface SecurityTetraplet extends ResolvedTriplet {
json_path: string;
}

12
src/seed.ts
View File

@ -14,9 +14,9 @@
* limitations under the License.
*/
import * as PeerId from "peer-id";
import {decode, encode} from "bs58"
import {keys} from "libp2p-crypto";
import * as PeerId from 'peer-id';
import { decode, encode } from 'bs58';
import { keys } from 'libp2p-crypto';
/**
* @param seed 32 bytes
@ -24,11 +24,11 @@ import {keys} from "libp2p-crypto";
export async function seedToPeerId(seed: string): Promise<PeerId> {
let seedArr = decode(seed);
let privateK = await keys.generateKeyPairFromSeed("Ed25519", Uint8Array.from(seedArr), 256);
let privateK = await keys.generateKeyPairFromSeed('Ed25519', Uint8Array.from(seedArr), 256);
return await PeerId.createFromPrivKey(privateK.bytes);
}
export function peerIdToSeed(peerId: PeerId): string {
let seedBuf = peerId.privKey.marshal().subarray(0, 32)
return encode(seedBuf)
let seedBuf = peerId.privKey.marshal().subarray(0, 32);
return encode(seedBuf);
}

114
src/service.ts
View File

@ -14,111 +14,145 @@
* limitations under the License.
*/
import {getService} from "./globalState";
import { getService } from './globalState';
import { SecurityTetraplet } from './securityTetraplet';
export interface CallServiceResult {
ret_code: number,
result: string
ret_code: number;
result: string;
}
export abstract class Service {
serviceId: string;
abstract call(fnName: string, args: any[]): CallServiceResult
/**
* Calls the function from local client
* @param fnName - name of the function to call
* @param args - arguments to be passed to the function
* @param tetraplets - array of arrays of tetraplets. First index corresponds to argument number.
* If the argument is not an array the second array will always contain exactly one element.
* If the argument is an array the second index will correspond to the index of element in argument's array
*/
abstract call(fnName: string, args: any[], tetraplets: SecurityTetraplet[][]): CallServiceResult;
}
/**
* Creates one function for all function names.
*/
export class ServiceOne implements Service {
serviceId: string;
fn: (fnName: string, args: any[]) => object
fn: (fnName: string, args: any[], tetraplets: SecurityTetraplet[][]) => object;
constructor(serviceId: string, fn: (fnName: string, args: any[]) => object) {
constructor(serviceId: string, fn: (fnName: string, args: any[], tetraplets: SecurityTetraplet[][]) => object) {
this.serviceId = serviceId;
this.fn = fn;
}
call(fnName: string, args: any[]): CallServiceResult {
/**
* Calls the function from local client
* @param fnName - name of the function to call
* @param args - arguments to be passed to the function
* @param tetraplets - array of arrays of tetraplets. First index corresponds to argument number.
* If the argument is not an array the second array will always contain exactly one element.
* If the argument is an array the second index will correspond to the index of element in argument's array
*/
call(fnName: string, args: any[], tetraplets: SecurityTetraplet[][]): CallServiceResult {
try {
let result = this.fn(fnName, args)
let result = this.fn(fnName, args, tetraplets);
return {
ret_code: 0,
result: JSON.stringify(result)
}
result: JSON.stringify(result),
};
} catch (err) {
return {
ret_code: 1,
result: JSON.stringify(err)
}
result: JSON.stringify(err),
};
}
}
}
/**
* Creates function per function name. Returns an error when call a name without registered function.
*/
export class ServiceMultiple implements Service {
serviceId: string;
functions: Map<string, (args: any[]) => object> = new Map();
functions: Map<string, (args: any[], tetraplets: SecurityTetraplet[][]) => object> = new Map();
constructor(serviceId: string) {
this.serviceId = serviceId;
}
registerFunction(fnName: string, fn: (args: any[]) => object) {
/**
* Registers a callback function into Aquamarine
* @param fnName - the function name to be registered
* @param fn - callback function which will be called from Aquamarine.
* The callback function has the following parameters:
* args - arguments to be passed to the function
* tetraplets - array of arrays of tetraplets. First index corresponds to argument number.
* If the argument is not an array the second array will always contain exactly one element.
* If the argument is an array the second index will correspond to the index of element in argument's array
*/
registerFunction(fnName: string, fn: (args: any[], tetraplets: SecurityTetraplet[][]) => object) {
this.functions.set(fnName, fn);
}
call(fnName: string, args: any[]): CallServiceResult {
let fn = this.functions.get(fnName)
/**
* Calls the function from local client
* @param fnName - name of the function to call
* @param args - arguments to be passed to the function
* @param tetraplets - array of arrays of tetraplets. First index corresponds to argument number.
* If the argument is not an array the second array will always contain exactly one element.
* If the argument is an array the second index will correspond to the index of element in argument's array
*/
call(fnName: string, args: any[], tetraplets: SecurityTetraplet[][]): CallServiceResult {
let fn = this.functions.get(fnName);
if (fn) {
try {
let result = fn(args)
let result = fn(args, tetraplets);
return {
ret_code: 0,
result: JSON.stringify(result)
}
result: JSON.stringify(result),
};
} catch (err) {
return {
ret_code: 1,
result: JSON.stringify(err)
}
result: JSON.stringify(err),
};
}
} else {
let errorMsg = `Error. There is no function ${fnName}`
let errorMsg = `Error. There is no function ${fnName}`;
return {
ret_code: 1,
result: JSON.stringify(errorMsg)
}
result: JSON.stringify(errorMsg),
};
}
}
}
export function service(service_id: string, fn_name: string, args: string): CallServiceResult {
export function service(service_id: string, fn_name: string, args: string, tetraplets: string): CallServiceResult {
try {
let argsObject = JSON.parse(args)
let argsObject = JSON.parse(args);
if (!Array.isArray(argsObject)) {
throw new Error("args is not an array")
throw new Error('args is not an array');
}
let service = getService(service_id)
let tetrapletsObject: SecurityTetraplet[][] = JSON.parse(tetraplets);
let service = getService(service_id);
if (service) {
return service.call(fn_name, argsObject)
return service.call(fn_name, argsObject, tetrapletsObject);
} else {
return {
result: JSON.stringify(`Error. There is no service: ${service_id}`),
ret_code: 0
}
ret_code: 0,
};
}
} catch (err) {
console.error("Cannot parse arguments: " + JSON.stringify(err))
console.error('Cannot parse arguments: ' + JSON.stringify(err));
return {
result: JSON.stringify("Cannot parse arguments: " + JSON.stringify(err)),
ret_code: 1
}
result: JSON.stringify('Cannot parse arguments: ' + JSON.stringify(err)),
ret_code: 1,
};
}
}

104
src/stepper.ts
View File

@ -14,31 +14,33 @@
* limitations under the License.
*/
import {toByteArray} from "base64-js";
import * as aqua from "./aqua"
import {return_current_peer_id, return_call_service_result, getStringFromWasm0, free} from "./aqua"
import { toByteArray } from 'base64-js';
import * as aqua from './aqua';
import { return_current_peer_id, return_call_service_result, getStringFromWasm0, free } from './aqua';
import {service} from "./service";
import PeerId from "peer-id";
import log from "loglevel";
import {wasmBs64} from "@fluencelabs/aquamarine-stepper";
import { service } from './service';
import PeerId from 'peer-id';
import log from 'loglevel';
import { wasmBs64 } from '@fluencelabs/aquamarine-stepper';
import Instance = WebAssembly.Instance;
import Exports = WebAssembly.Exports;
import ExportValue = WebAssembly.ExportValue;
export type InterpreterInvoke = (init_user_id: string, script: string, prev_data: string, data: string) => string
export type InterpreterInvoke = (init_user_id: string, script: string, prev_data: string, data: string) => string;
type ImportObject = {
"./aquamarine_client_bg.js": {
__wbg_callserviceimpl_7d3cf77a2722659e: (arg0: any, arg1: any, arg2: any, arg3: any, arg4: any, arg5: any, arg6: any) => void;
'./aquamarine_client_bg.js': {
// fn call_service_impl(service_id: String, fn_name: String, args: String, security_tetraplets: String) -> String;
// prettier-ignore
__wbg_callserviceimpl_7d3cf77a2722659e: (arg0: any, arg1: any, arg2: any, arg3: any, arg4: any, arg5: any, arg6: any, arg7: any, arg8: any, ) => void;
__wbg_getcurrentpeeridimpl_154ce1848a306ff5: (arg0: any) => void;
__wbindgen_throw: (arg: any) => void;
};
host: LogImport
host: LogImport;
};
type LogImport = {
log_utf8_string: (level: any, target: any, offset: any, size: any) => void
}
log_utf8_string: (level: any, target: any, offset: any, size: any) => void;
};
class HostImportsConfig {
exports: Exports | undefined;
@ -54,7 +56,7 @@ class HostImportsConfig {
}
}
const interpreter_wasm = toByteArray(wasmBs64)
const interpreter_wasm = toByteArray(wasmBs64);
/// Instantiates WebAssembly runtime with AIR interpreter module
async function interpreterInstance(cfg: HostImportsConfig): Promise<Instance> {
@ -77,10 +79,10 @@ async function interpreterInstance(cfg: HostImportsConfig): Promise<Instance> {
/// If export is a function, call it. Otherwise log a warning.
/// NOTE: any here is unavoidable, see Function interface definition
function call_export(f: ExportValue, ...argArray: any[]): any {
if (typeof f === "function") {
if (typeof f === 'function') {
return f();
} else {
log.warn(`can't call export ${f}: it is not a function, but ${typeof f}`)
log.warn(`can't call export ${f}: it is not a function, but ${typeof f}`);
}
}
@ -93,25 +95,25 @@ function log_import(cfg: HostImportsConfig): LogImport {
switch (level) {
case 1:
log.error(str)
log.error(str);
break;
case 2:
log.warn(str)
log.warn(str);
break;
case 3:
log.info(str)
log.info(str);
break;
case 4:
log.debug(str)
log.debug(str);
break;
case 5:
// we don't want a trace in trace logs
log.debug(str)
log.debug(str);
break;
}
} finally {
}
}
},
};
}
@ -120,20 +122,23 @@ function newImportObject(cfg: HostImportsConfig, peerId: PeerId): ImportObject {
return {
// __wbg_callserviceimpl_c0ca292e3c8c0c97 this is a function generated by bindgen. Could be changed.
// If so, an error with a new name will be occurred after wasm initialization.
"./aquamarine_client_bg.js": {
__wbg_callserviceimpl_7d3cf77a2722659e: (arg0: any, arg1: any, arg2: any, arg3: any, arg4: any, arg5: any, arg6: any) => {
'./aquamarine_client_bg.js': {
// prettier-ignore
__wbg_callserviceimpl_7d3cf77a2722659e: (arg0: any, arg1: any, arg2: any, arg3: any, arg4: any, arg5: any, arg6: any, arg7: any, arg8: any) => {
let wasm = cfg.exports;
try {
let serviceId = getStringFromWasm0(wasm, arg1, arg2)
let fnName = getStringFromWasm0(wasm, arg3, arg4)
let serviceId = getStringFromWasm0(wasm, arg1, arg2);
let fnName = getStringFromWasm0(wasm, arg3, arg4);
let args = getStringFromWasm0(wasm, arg5, arg6);
let serviceResult = service(serviceId, fnName, args);
let resultStr = JSON.stringify(serviceResult)
let tetraplets = getStringFromWasm0(wasm, arg7, arg8);
let serviceResult = service(serviceId, fnName, args, tetraplets);
let resultStr = JSON.stringify(serviceResult);
return_call_service_result(wasm, resultStr, arg0);
} finally {
free(wasm, arg1, arg2)
free(wasm, arg3, arg4)
free(wasm, arg5, arg6)
free(wasm, arg1, arg2);
free(wasm, arg3, arg4);
free(wasm, arg5, arg6);
free(wasm, arg7, arg8);
}
},
__wbg_getcurrentpeeridimpl_154ce1848a306ff5: (arg0: any) => {
@ -143,19 +148,19 @@ function newImportObject(cfg: HostImportsConfig, peerId: PeerId): ImportObject {
},
__wbindgen_throw: (arg: any) => {
console.log(`wbindgen throw: ${JSON.stringify(arg)}`);
}
},
},
host: log_import(cfg)
host: log_import(cfg),
};
}
function newLogImport(cfg: HostImportsConfig): ImportObject {
return {
host: log_import(cfg),
"./aquamarine_client_bg.js": {
__wbg_callserviceimpl_7d3cf77a2722659e: _ => {},
__wbg_getcurrentpeeridimpl_154ce1848a306ff5: _ => {},
__wbindgen_throw: _ => {}
'./aquamarine_client_bg.js': {
__wbg_callserviceimpl_7d3cf77a2722659e: (_) => {},
__wbg_getcurrentpeeridimpl_154ce1848a306ff5: (_) => {},
__wbindgen_throw: (_) => {},
},
};
}
@ -165,29 +170,28 @@ function newLogImport(cfg: HostImportsConfig): ImportObject {
export async function instantiateInterpreter(peerId: PeerId): Promise<InterpreterInvoke> {
let cfg = new HostImportsConfig((cfg) => {
return newImportObject(cfg, peerId);
})
});
let instance = await interpreterInstance(cfg);
return (init_user_id: string, script: string, prev_data: string, data: string) => {
let logLevel = log.getLevel()
let logLevelStr = "info"
let logLevel = log.getLevel();
let logLevelStr = 'info';
if (logLevel === 0) {
logLevelStr = "trace"
logLevelStr = 'trace';
} else if (logLevel === 1) {
logLevelStr = "debug"
logLevelStr = 'debug';
} else if (logLevel === 2) {
logLevelStr = "info"
logLevelStr = 'info';
} else if (logLevel === 3) {
logLevelStr = "warn"
logLevelStr = 'warn';
} else if (logLevel === 4) {
logLevelStr = "error"
logLevelStr = 'error';
} else if (logLevel === 5) {
logLevelStr = "off"
logLevelStr = 'off';
}
return aqua.invoke(instance.exports, init_user_id, script, prev_data, data, logLevelStr)
}
return aqua.invoke(instance.exports, init_user_id, script, prev_data, data, logLevelStr);
};
}
/// Instantiate AIR interpreter with host imports containing only logger, but not call_service
@ -197,6 +201,6 @@ export async function parseAstClosure(): Promise<(script: string) => string> {
let instance = await interpreterInstance(cfg);
return (script: string) => {
return aqua.ast(instance.exports, script)
return aqua.ast(instance.exports, script);
};
}

6
src/stepperOutcome.ts
View File

@ -15,7 +15,7 @@
*/
export interface StepperOutcome {
ret_code: number,
data: number[],
next_peer_pks: string[]
ret_code: number;
data: number[];
next_peer_pks: string[];
}

14
src/subscriptions.ts
View File

@ -14,8 +14,8 @@
* limitations under the License.
*/
import {Particle} from "./particle";
import log from "loglevel";
import { Particle } from './particle';
import log from 'loglevel';
export class Subscriptions {
private subscriptions: Map<string, Particle> = new Map();
@ -31,9 +31,9 @@ export class Subscriptions {
subscribe(particle: Particle, ttl: number) {
let _this = this;
setTimeout(() => {
_this.subscriptions.delete(particle.id)
log.info(`Particle with id ${particle.id} deleted by timeout`)
}, ttl)
_this.subscriptions.delete(particle.id);
log.info(`Particle with id ${particle.id} deleted by timeout`);
}, ttl);
this.subscriptions.set(particle.id, particle);
}
@ -47,10 +47,10 @@ export class Subscriptions {
}
get(id: string): Particle | undefined {
return this.subscriptions.get(id)
return this.subscriptions.get(id);
}
hasSubscription(particle: Particle): boolean {
return this.subscriptions.has(particle.id)
return this.subscriptions.has(particle.id);
}
}

162
src/test/air.spec.ts
View File

@ -1,98 +1,133 @@
import 'mocha';
import Fluence from "../fluence";
import {build} from "../particle";
import {ServiceMultiple} from "../service";
import {registerService} from "../globalState";
import {expect} from "chai";
import Fluence from '../fluence';
import { build } from '../particle';
import { ServiceMultiple } from '../service';
import { registerService } from '../globalState';
import { expect } from 'chai';
import { SecurityTetraplet } from '../securityTetraplet';
function registerPromiseService<T>(serviceId: string, fnName: string, f: (args: any[]) => T): Promise<T> {
function registerPromiseService<T>(
serviceId: string,
fnName: string,
f: (args: any[]) => T,
): Promise<[T, SecurityTetraplet[][]]> {
let service = new ServiceMultiple(serviceId);
registerService(service);
return new Promise((resolve, reject) => {
service.registerFunction(fnName, (args: any[]) => {
resolve(f(args))
service.registerFunction(fnName, (args: any[], tetraplets: SecurityTetraplet[][]) => {
resolve([f(args), tetraplets]);
return {result: f(args)}
})
})
return { result: f(args) };
});
});
}
describe("== AIR suite", () => {
it("check init_peer_id", async function () {
let serviceId = "init_peer"
let fnName = "id"
let checkPromise = registerPromiseService(serviceId, fnName, (args) => args[0])
describe('== AIR suite', () => {
it('check init_peer_id', async function () {
let serviceId = 'init_peer';
let fnName = 'id';
let checkPromise = registerPromiseService(serviceId, fnName, (args) => args[0]);
let client = await Fluence.local();
let script = `(call %init_peer_id% ("${serviceId}" "${fnName}") [%init_peer_id%])`
let script = `(call %init_peer_id% ("${serviceId}" "${fnName}") [%init_peer_id%])`;
let particle = await build(client.selfPeerId, script, new Map())
let particle = await build(client.selfPeerId, script, new Map());
await client.executeParticle(particle);
expect(await checkPromise).to.be.equal(client.selfPeerIdStr)
})
let args = (await checkPromise)[0];
expect(args).to.be.equal(client.selfPeerIdStr);
});
it("call local function", async function () {
let serviceId = "console"
let fnName = "log"
let checkPromise = registerPromiseService(serviceId, fnName, (args) => args[0])
it('call local function', async function () {
let serviceId = 'console';
let fnName = 'log';
let checkPromise = registerPromiseService(serviceId, fnName, (args) => args[0]);
let client = await Fluence.local();
let arg = "hello"
let script = `(call %init_peer_id% ("${serviceId}" "${fnName}") ["${arg}"])`
let arg = 'hello';
let script = `(call %init_peer_id% ("${serviceId}" "${fnName}") ["${arg}"])`;
// Wrap script into particle, so it can be executed by local WASM runtime
let particle = await build(client.selfPeerId, script, new Map())
let particle = await build(client.selfPeerId, script, new Map());
await client.executeParticle(particle);
expect(await checkPromise).to.be.equal(arg)
})
let [args, tetraplets] = await checkPromise;
expect(args).to.be.equal(arg);
});
it("check particle arguments", async function () {
let serviceId = "check"
let fnName = "args"
let checkPromise = registerPromiseService(serviceId, fnName, (args) => args[0])
it('check particle arguments', async function () {
let serviceId = 'check';
let fnName = 'args';
let checkPromise = registerPromiseService(serviceId, fnName, (args) => args[0]);
let client = await Fluence.local();
let arg = "arg1"
let value = "hello"
let script = `(call %init_peer_id% ("${serviceId}" "${fnName}") [${arg}])`
let arg = 'arg1';
let value = 'hello';
let script = `(call %init_peer_id% ("${serviceId}" "${fnName}") [${arg}])`;
let data = new Map()
data.set("arg1", value)
let particle = await build(client.selfPeerId, script, data)
let data = new Map();
data.set('arg1', value);
let particle = await build(client.selfPeerId, script, data);
await client.executeParticle(particle);
expect(await checkPromise).to.be.equal(value)
})
let [args, tetraplets] = await checkPromise;
expect(args).to.be.equal(value);
});
it("check chain of services work properly", async function () {
it('check security tetraplet', async function () {
let makeDataPromise = registerPromiseService('make_data_service', 'make_data', (args) => {
field: 42;
});
let getDataPromise = registerPromiseService('get_data_service', 'get_data', (args) => args[0]);
let client = await Fluence.local();
let script = `
(seq
(call %init_peer_id% ("make_data_service" "make_data") [] result)
(call %init_peer_id% ("get_data_service" "get_data") [result.$.field])
)`;
let particle = await build(client.selfPeerId, script, new Map());
await client.executeParticle(particle);
await makeDataPromise;
let [args, tetraplets] = await getDataPromise;
let tetraplet = tetraplets[0][0];
expect(tetraplet).to.contain({
service_id: 'make_data_service',
function_name: 'make_data',
json_path: '$.field',
});
});
it('check chain of services work properly', async function () {
this.timeout(5000);
let serviceId1 = "check1"
let fnName1 = "fn1"
let checkPromise1 = registerPromiseService(serviceId1, fnName1, (args) => args[0])
let serviceId1 = 'check1';
let fnName1 = 'fn1';
let checkPromise1 = registerPromiseService(serviceId1, fnName1, (args) => args[0]);
let serviceId2 = "check2"
let fnName2 = "fn2"
let checkPromise2 = registerPromiseService(serviceId2, fnName2, (args) => args[0])
let serviceId2 = 'check2';
let fnName2 = 'fn2';
let checkPromise2 = registerPromiseService(serviceId2, fnName2, (args) => args[0]);
let serviceId3 = "check3"
let fnName3 = "fn3"
let checkPromise3 = registerPromiseService(serviceId3, fnName3, (args) => args)
let serviceId3 = 'check3';
let fnName3 = 'fn3';
let checkPromise3 = registerPromiseService(serviceId3, fnName3, (args) => args);
let client = await Fluence.local();
let arg1 = "arg1"
let arg2 = "arg2"
let arg1 = 'arg1';
let arg2 = 'arg2';
// language=Clojure
let script = `(seq
@ -100,16 +135,19 @@ describe("== AIR suite", () => {
(call %init_peer_id% ("${serviceId1}" "${fnName1}") ["${arg1}"] result1)
(call %init_peer_id% ("${serviceId2}" "${fnName2}") ["${arg2}"] result2))
(call %init_peer_id% ("${serviceId3}" "${fnName3}") [result1 result2]))
`
`;
let particle = await build(client.selfPeerId, script, new Map())
let particle = await build(client.selfPeerId, script, new Map());
await client.executeParticle(particle);
expect(await checkPromise1).to.be.equal(arg1)
expect(await checkPromise2).to.be.equal(arg2)
let args1 = (await checkPromise1)[0];
expect(args1).to.be.equal(arg1);
expect(await checkPromise3).to.be.deep.equal([{result: arg1}, {result: arg2}])
})
})
let args2 = (await checkPromise2)[0];
expect(args2).to.be.equal(arg2);
let args3 = (await checkPromise3)[0];
expect(args3).to.be.deep.equal([{ result: arg1 }, { result: arg2 }]);
});
});

21
src/test/ast.spec.ts
View File

@ -1,13 +1,22 @@
import { expect } from 'chai';
import 'mocha';
import Fluence from "../fluence";
import Fluence from '../fluence';
describe("== AST parsing suite", () => {
it("parse simple script and return ast", async function () {
describe('== AST parsing suite', () => {
it('parse simple script and return ast', async function () {
let ast = await Fluence.parseAIR(`
(call node ("service" "function") [1 2 3 arg] output)
`);
console.log(ast);
})
})
ast = JSON.parse(ast);
expect(ast).to.deep.equal({
Call: {
peer_part: { PeerPk: { Variable: 'node' } },
function_part: { ServiceIdWithFuncName: [{ Literal: 'service' }, { Literal: 'function' }] },
args: [{ Variable: '1' }, { Variable: '2' }, { Variable: '3' }, { Variable: 'arg' }],
output: { Scalar: 'output' },
},
});
});
});

101
src/test/client.spec.ts
View File

@ -1,28 +1,28 @@
import {expect} from 'chai';
import { expect } from 'chai';
import 'mocha';
import {encode} from "bs58"
import Fluence from "../fluence";
import {certificateFromString, certificateToString, issue} from "../trust/certificate";
import {TrustGraph} from "../trust/trust_graph";
import {nodeRootCert} from "../trust/misc";
import {peerIdToSeed, seedToPeerId} from "../seed";
import {build} from "../particle";
import {Service, ServiceOne} from "../service";
import {registerService} from "../globalState";
import {waitResult} from "../helpers/waitService";
import { encode } from 'bs58';
import Fluence from '../fluence';
import { certificateFromString, certificateToString, issue } from '../trust/certificate';
import { TrustGraph } from '../trust/trust_graph';
import { nodeRootCert } from '../trust/misc';
import { peerIdToSeed, seedToPeerId } from '../seed';
import { build } from '../particle';
import { Service, ServiceOne } from '../service';
import { registerService } from '../globalState';
import { waitResult } from '../helpers/waitService';
describe("Typescript usage suite", () => {
it("should create private key from seed and back", async function () {
describe('Typescript usage suite', () => {
it('should create private key from seed and back', async function () {
// prettier-ignore
let seed = [46, 188, 245, 171, 145, 73, 40, 24, 52, 233, 215, 163, 54, 26, 31, 221, 159, 179, 126, 106, 27, 199, 189, 194, 80, 133, 235, 42, 42, 247, 80, 201];
let seedStr = encode(seed)
console.log("SEED STR: " + seedStr)
let pid = await seedToPeerId(seedStr)
expect(peerIdToSeed(pid)).to.be.equal(seedStr)
})
let seedStr = encode(seed);
console.log('SEED STR: ' + seedStr);
let pid = await seedToPeerId(seedStr);
expect(peerIdToSeed(pid)).to.be.equal(seedStr);
});
it("should serialize and deserialize certificate correctly", async function () {
it('should serialize and deserialize certificate correctly', async function () {
let cert = `11
1111
5566Dn4ZXXbBK5LJdUsE7L3pG9qdAzdPY47adjzkhEx9
@ -32,7 +32,7 @@ describe("Typescript usage suite", () => {
2EvoZAZaGjKWFVdr36F1jphQ5cW7eK3yM16mqEHwQyr7
4UAJQWzB3nTchBtwARHAhsn7wjdYtqUHojps9xV6JkuLENV8KRiWM3BhQByx5KijumkaNjr7MhHjouLawmiN1A4d
1590061123504
1589974723504`
1589974723504`;
let deser = await certificateFromString(cert);
let ser = certificateToString(deser);
@ -41,54 +41,63 @@ describe("Typescript usage suite", () => {
});
// delete `.skip` and run `npm run test` to check service's and certificate's api with Fluence nodes
it.skip("test certs", async function () {
it.skip('test certs', async function () {
this.timeout(15000);
await testCerts();
});
it.skip("", async function () {
let pid = await Fluence.generatePeerId()
let cl = await Fluence.connect("/ip4/138.197.177.2/tcp/9001/ws/p2p/12D3KooWEXNUbCXooUwHrHBbrmjsrpHXoEphPwbjQXEGyzbqKnE9", pid)
it.skip('', async function () {
let pid = await Fluence.generatePeerId();
let cl = await Fluence.connect(
'/ip4/138.197.177.2/tcp/9001/ws/p2p/12D3KooWEXNUbCXooUwHrHBbrmjsrpHXoEphPwbjQXEGyzbqKnE9',
pid,
);
let service = new ServiceOne("test", (fnName: string, args: any[]) => {
console.log("called: " + args)
return {}
let service = new ServiceOne('test', (fnName: string, args: any[]) => {
console.log('called: ' + args);
return {};
});
registerService(service);
let namedPromise = waitResult(30000)
let namedPromise = waitResult(30000);
let script = `
(seq (
(call ( "${pid.toB58String()}" ("test" "test") (a b c d) result))
(call ( "${pid.toB58String()}" ("${namedPromise.name}" "") (d c b a) void[]))
))
`
`;
let data: Map<string, any> = new Map();
data.set("a", "some a")
data.set("b", "some b")
data.set("c", "some c")
data.set("d", "some d")
data.set('a', 'some a');
data.set('b', 'some b');
data.set('c', 'some c');
data.set('d', 'some d');
let particle = await build(pid, script, data, 30000)
let particle = await build(pid, script, data, 30000);
await cl.sendParticle(particle)
await cl.sendParticle(particle);
let res = await namedPromise.promise
expect(res).to.be.equal(["some d", "some c", "some b", "some a"])
})
let res = await namedPromise.promise;
expect(res).to.be.equal(['some d', 'some c', 'some b', 'some a']);
});
});
const delay = (ms: number) => new Promise(res => setTimeout(res, ms));
const delay = (ms: number) => new Promise((res) => setTimeout(res, ms));
export async function testCerts() {
let key1 = await Fluence.generatePeerId();
let key2 = await Fluence.generatePeerId();
// connect to two different nodes
let cl1 = await Fluence.connect("/dns4/134.209.186.43/tcp/9003/ws/p2p/12D3KooWBUJifCTgaxAUrcM9JysqCcS4CS8tiYH5hExbdWCAoNwb", key1);
let cl2 = await Fluence.connect("/ip4/134.209.186.43/tcp/9002/ws/p2p/12D3KooWHk9BjDQBUqnavciRPhAYFvqKBe4ZiPPvde7vDaqgn5er", key2);
let cl1 = await Fluence.connect(
'/dns4/134.209.186.43/tcp/9003/ws/p2p/12D3KooWBUJifCTgaxAUrcM9JysqCcS4CS8tiYH5hExbdWCAoNwb',
key1,
);
let cl2 = await Fluence.connect(
'/ip4/134.209.186.43/tcp/9002/ws/p2p/12D3KooWHk9BjDQBUqnavciRPhAYFvqKBe4ZiPPvde7vDaqgn5er',
key2,
);
let trustGraph1 = new TrustGraph(cl1);
let trustGraph2 = new TrustGraph(cl2);
@ -109,10 +118,10 @@ export async function testCerts() {
let certs = await trustGraph2.getCertificates(key2.toB58String());
// root certificate could be different because nodes save trusts with bigger `expiresAt` date and less `issuedAt` date
expect(certs[0].chain[1].issuedFor.toB58String()).to.be.equal(extended.chain[1].issuedFor.toB58String())
expect(certs[0].chain[1].signature).to.be.equal(extended.chain[1].signature)
expect(certs[0].chain[1].expiresAt).to.be.equal(extended.chain[1].expiresAt)
expect(certs[0].chain[1].issuedAt).to.be.equal(extended.chain[1].issuedAt)
expect(certs[0].chain[1].issuedFor.toB58String()).to.be.equal(extended.chain[1].issuedFor.toB58String());
expect(certs[0].chain[1].signature).to.be.equal(extended.chain[1].signature);
expect(certs[0].chain[1].expiresAt).to.be.equal(extended.chain[1].expiresAt);
expect(certs[0].chain[1].issuedAt).to.be.equal(extended.chain[1].issuedAt);
await cl1.disconnect();
await cl2.disconnect();

61
src/trust/certificate.ts
View File

@ -14,28 +14,28 @@
* limitations under the License.
*/
import {createTrust, Trust, trustFromString, trustToString} from "./trust";
import * as PeerId from "peer-id";
import { createTrust, Trust, trustFromString, trustToString } from './trust';
import * as PeerId from 'peer-id';
const FORMAT = "11";
const VERSION = "1111";
const FORMAT = '11';
const VERSION = '1111';
// TODO verify certificate
// Chain of trusts started from self-signed root trust.
export interface Certificate {
chain: Trust[]
chain: Trust[];
}
export function certificateToString(cert: Certificate): string {
let certStr = cert.chain.map(t => trustToString(t)).join("\n");
return `${FORMAT}\n${VERSION}\n${certStr}`
let certStr = cert.chain.map((t) => trustToString(t)).join('\n');
return `${FORMAT}\n${VERSION}\n${certStr}`;
}
export async function certificateFromString(str: string): Promise<Certificate> {
let lines = str.split("\n");
let lines = str.split('\n');
// last line could be empty
if (!lines[lines.length - 1]) {
lines.pop()
lines.pop();
}
// TODO do match different formats and versions
@ -44,27 +44,28 @@ export async function certificateFromString(str: string): Promise<Certificate> {
// every trust is 4 lines, certificate lines number without format and version should be divided by 4
if ((lines.length - 2) % 4 !== 0) {
throw Error("Incorrect format of the certificate:\n" + str);
throw Error('Incorrect format of the certificate:\n' + str);
}
let chain: Trust[] = [];
let i;
for(i = 2; i < lines.length; i = i + 4) {
chain.push(await trustFromString(lines[i], lines[i+1], lines[i+2], lines[i+3]))
for (i = 2; i < lines.length; i = i + 4) {
chain.push(await trustFromString(lines[i], lines[i + 1], lines[i + 2], lines[i + 3]));
}
return {chain};
return { chain };
}
// Creates new certificate with root trust (self-signed public key) from a key pair.
export async function issueRoot(issuedBy: PeerId,
forPk: PeerId,
expiresAt: number,
issuedAt: number,
export async function issueRoot(
issuedBy: PeerId,
forPk: PeerId,
expiresAt: number,
issuedAt: number,
): Promise<Certificate> {
if (expiresAt < issuedAt) {
throw Error("Expiration time should be greater then issued time.")
throw Error('Expiration time should be greater then issued time.');
}
let maxDate = new Date(158981172690500).getTime();
@ -74,24 +75,26 @@ export async function issueRoot(issuedBy: PeerId,
let chain = [rootTrust, trust];
return {
chain: chain
}
chain: chain,
};
}
// Adds a new trust into chain of trust in certificate.
export async function issue(issuedBy: PeerId,
forPk: PeerId,
extendCert: Certificate,
expiresAt: number,
issuedAt: number): Promise<Certificate> {
export async function issue(
issuedBy: PeerId,
forPk: PeerId,
extendCert: Certificate,
expiresAt: number,
issuedAt: number,
): Promise<Certificate> {
if (expiresAt < issuedAt) {
throw Error("Expiration time should be greater then issued time.")
throw Error('Expiration time should be greater then issued time.');
}
let lastTrust = extendCert.chain[extendCert.chain.length - 1];
if (lastTrust.issuedFor !== issuedBy) {
throw Error("`issuedFor` should be equal to `issuedBy` in the last trust of the chain.")
throw Error('`issuedFor` should be equal to `issuedBy` in the last trust of the chain.');
}
let trust = await createTrust(forPk, issuedBy, expiresAt, issuedAt);
@ -99,6 +102,6 @@ export async function issue(issuedBy: PeerId,
chain.push(trust);
return {
chain: chain
}
chain: chain,
};
}

9
src/trust/misc.ts
View File

@ -14,17 +14,18 @@
* limitations under the License.
*/
import * as PeerId from "peer-id";
import {keys} from "libp2p-crypto";
import {Certificate, issueRoot} from "./certificate";
import * as PeerId from 'peer-id';
import { keys } from 'libp2p-crypto';
import { Certificate, issueRoot } from './certificate';
/**
* Generate root certificate with one of the Fluence trusted key for one day.
*/
export async function nodeRootCert(issuedFor: PeerId): Promise<Certificate> {
// prettier-ignore
let seed = [46, 188, 245, 171, 145, 73, 40, 24, 52, 233, 215, 163, 54, 26, 31, 221, 159, 179, 126, 106, 27, 199, 189, 194, 80, 133, 235, 42, 42, 247, 80, 201];
let privateK = await keys.generateKeyPairFromSeed("Ed25519", Uint8Array.from(seed), 256);
let privateK = await keys.generateKeyPairFromSeed('Ed25519', Uint8Array.from(seed), 256);
let peerId = await PeerId.createFromPrivKey(privateK.bytes);
let issuedAt = new Date();

38
src/trust/trust.ts
View File

@ -14,24 +14,29 @@
* limitations under the License.
*/
import * as PeerId from "peer-id";
import {decode, encode} from "bs58"
import * as PeerId from 'peer-id';
import { decode, encode } from 'bs58';
import crypto from 'libp2p-crypto';
const ed25519 = crypto.keys.supportedKeys.ed25519;
// One element in chain of trust in a certificate.
export interface Trust {
issuedFor: PeerId,
expiresAt: number,
signature: string,
issuedAt: number
issuedFor: PeerId;
expiresAt: number;
signature: string;
issuedAt: number;
}
export function trustToString(trust: Trust): string {
return `${encode(trust.issuedFor.pubKey.marshal())}\n${trust.signature}\n${trust.expiresAt}\n${trust.issuedAt}`
return `${encode(trust.issuedFor.pubKey.marshal())}\n${trust.signature}\n${trust.expiresAt}\n${trust.issuedAt}`;
}
export async function trustFromString(issuedFor: string, signature: string, expiresAt: string, issuedAt: string): Promise<Trust> {
export async function trustFromString(
issuedFor: string,
signature: string,
expiresAt: string,
issuedAt: string,
): Promise<Trust> {
let pubKey = ed25519.unmarshalEd25519PublicKey(decode(issuedFor));
let peerId = await PeerId.createFromPubKey(pubKey.bytes);
@ -39,11 +44,16 @@ export async function trustFromString(issuedFor: string, signature: string, expi
issuedFor: peerId,
signature: signature,
expiresAt: parseInt(expiresAt),
issuedAt: parseInt(issuedAt)
}
issuedAt: parseInt(issuedAt),
};
}
export async function createTrust(forPk: PeerId, issuedBy: PeerId, expiresAt: number, issuedAt: number): Promise<Trust> {
export async function createTrust(
forPk: PeerId,
issuedBy: PeerId,
expiresAt: number,
issuedAt: number,
): Promise<Trust> {
let bytes = toSignMessage(forPk, expiresAt, issuedAt);
let signature = await issuedBy.privKey.sign(Buffer.from(bytes));
let signatureStr = encode(signature);
@ -52,7 +62,7 @@ export async function createTrust(forPk: PeerId, issuedBy: PeerId, expiresAt: nu
issuedFor: forPk,
expiresAt: expiresAt,
signature: signatureStr,
issuedAt: issuedAt
issuedAt: issuedAt,
};
}
@ -64,7 +74,7 @@ function toSignMessage(pk: PeerId, expiresAt: number, issuedAt: number): Uint8Ar
bytes.set(numToArray(expiresAt), 32);
bytes.set(numToArray(issuedAt), 40);
return bytes
return bytes;
}
function numToArray(n: number): number[] {
@ -72,7 +82,7 @@ function numToArray(n: number): number[] {
for (let index = 0; index < byteArray.length; index++) {
let byte = n & 0xff;
byteArray [index] = byte;
byteArray[index] = byte;
n = (n - byte) / 256;
}

21
src/trust/trust_graph.ts
View File

@ -14,14 +14,13 @@
* limitations under the License.
*/
import {FluenceClient} from "../fluenceClient";
import {Certificate, certificateFromString, certificateToString} from "./certificate";
import { FluenceClient } from '../fluenceClient';
import { Certificate, certificateFromString, certificateToString } from './certificate';
import * as log from 'loglevel';
// TODO update after 'aquamarine' implemented
// The client to interact with the Fluence trust graph API
export class TrustGraph {
client: FluenceClient;
constructor(client: FluenceClient) {
@ -42,11 +41,13 @@ export class TrustGraph {
let response: any = {};
if (response.reason) {
throw Error(response.reason)
throw Error(response.reason);
} else if (response.status) {
return response.status
return response.status;
} else {
throw Error(`Unexpected response: ${response}. Should be 'status' field for a success response or 'reason' field for an error.`)
throw Error(
`Unexpected response: ${response}. Should be 'status' field for a success response or 'reason' field for an error.`,
);
}
}
@ -57,16 +58,16 @@ export class TrustGraph {
peer_id: peerId
});*/
let certificatesRaw = resp.certificates
let certificatesRaw = resp.certificates;
if (!(certificatesRaw && Array.isArray(certificatesRaw))) {
log.error(Array.isArray(certificatesRaw))
throw Error("Unexpected. Certificates should be presented in the response as an array.")
log.error(Array.isArray(certificatesRaw));
throw Error('Unexpected. Certificates should be presented in the response as an array.');
}
let certs = [];
for (let cert of certificatesRaw) {
certs.push(await certificateFromString(cert))
certs.push(await certificateFromString(cert));
}
return certs;