Skip to content
This repository was archived by the owner on Jul 10, 2025. It is now read-only.

Commit c0b73fe

Browse files
Akimshamsartemfolex
authored
feat(js-client)!: Particle signatures [fixes DXJ-466] (#353)
* Introduce particle signatures * Fix particle id field in particle context * Fix types * Fix review comments * Remove init_peer_id from signature * Fix typo * Fix error msg * Fix async promise constructor antipattern * Refactor utils * Move text encoder outside * Use async/await * Update packages/core/js-client/src/connection/RelayConnection.ts Co-authored-by: shamsartem <[email protected]> * Hide crypto implementation beside KeyPair * Fix verify method * Comment verify method * Use particle signature instead of id * remove async/await from method * Fix type * Update packages/core/js-client/src/particle/interfaces.ts Co-authored-by: folex <[email protected]> * Update packages/core/js-client/src/particle/Particle.ts Co-authored-by: folex <[email protected]> * Fix review comment * Update pipe * set logging * try cache --------- Co-authored-by: shamsartem <[email protected]> Co-authored-by: folex <[email protected]>
1 parent 15a2c91 commit c0b73fe

File tree

21 files changed

+228
-118
lines changed

21 files changed

+228
-118
lines changed

‎.github/workflows/tests.yml

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -97,8 +97,6 @@ jobs:
9797
"@fluencelabs/marine-js": "${{ inputs.marine-js-version }}"
9898
}
9999
100-
- uses: browser-actions/setup-chrome@v1
101-
102100
- run: pnpm -r --no-frozen-lockfile i
103101
- run: pnpm -r build
104102
- run: pnpm -r test

‎.npmrc

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,2 +1,3 @@
11
auto-install-peers=true
2-
save-exact=true
2+
save-exact=true
3+
side-effects-cache=false

‎packages/core/js-client/src/clientPeer/__test__/client.spec.ts

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -11,8 +11,7 @@ describe('FluenceClient usage test suite', () => {
1111
await withClient(RELAY, {}, async (peer) => {
1212
// arrange
1313

14-
const result = await new Promise<string[]>((resolve, reject) => {
15-
const script = `
14+
const script = `
1615
(xor
1716
(seq
1817
(call %init_peer_id% ("load" "relay") [] init_relay)
@@ -26,8 +25,10 @@ describe('FluenceClient usage test suite', () => {
2625
(call %init_peer_id% ("callback" "error") [%last_error%])
2726
)
2827
)`;
29-
const particle = peer.internals.createNewParticle(script);
3028

29+
const particle = await peer.internals.createNewParticle(script);
30+
31+
const result = await new Promise<string>((resolve, reject) => {
3132
if (particle instanceof Error) {
3233
return reject(particle.message);
3334
}
@@ -92,7 +93,7 @@ describe('FluenceClient usage test suite', () => {
9293
(call "${peer2.getPeerId()}" ("test" "test") ["test"])
9394
)
9495
`;
95-
const particle = peer1.internals.createNewParticle(script);
96+
const particle = await peer1.internals.createNewParticle(script);
9697

9798
if (particle instanceof Error) {
9899
throw particle;
@@ -149,14 +150,13 @@ describe('FluenceClient usage test suite', () => {
149150

150151
it.skip('Should throw correct error when the client tries to send a particle not to the relay', async () => {
151152
await withClient(RELAY, {}, async (peer) => {
152-
const promise = new Promise((resolve, reject) => {
153-
const script = `
153+
const script = `
154154
(xor
155155
(call "incorrect_peer_id" ("any" "service") [])
156156
(call %init_peer_id% ("callback" "error") [%last_error%])
157157
)`;
158-
const particle = peer.internals.createNewParticle(script);
159-
158+
const particle = await peer.internals.createNewParticle(script);
159+
const promise = new Promise((resolve, reject) => {
160160
if (particle instanceof Error) {
161161
return reject(particle.message);
162162
}

‎packages/core/js-client/src/clientPeer/checkConnection.ts

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -28,8 +28,7 @@ const log = logger('connection');
2828
export const checkConnection = async (peer: ClientPeer, ttl?: number): Promise<boolean> => {
2929
const msg = Math.random().toString(36).substring(7);
3030

31-
const promise = new Promise<string>((resolve, reject) => {
32-
const script = `
31+
const script = `
3332
(xor
3433
(seq
3534
(call %init_peer_id% ("load" "relay") [] init_relay)
@@ -46,8 +45,9 @@ export const checkConnection = async (peer: ClientPeer, ttl?: number): Promise<b
4645
(call %init_peer_id% ("callback" "error") [%last_error%])
4746
)
4847
)`;
49-
const particle = peer.internals.createNewParticle(script, ttl);
48+
const particle = await peer.internals.createNewParticle(script, ttl);
5049

50+
const promise = new Promise<string>((resolve, reject) => {
5151
if (particle instanceof Error) {
5252
return reject(particle.message);
5353
}

‎packages/core/js-client/src/compilerSupport/callFunction.ts

Lines changed: 8 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -13,19 +13,20 @@
1313
* See the License for the specific language governing permissions and
1414
* limitations under the License.
1515
*/
16-
import { getArgumentTypes, isReturnTypeVoid, CallAquaFunctionType } from '@fluencelabs/interfaces';
16+
import { CallAquaFunctionType, getArgumentTypes, isReturnTypeVoid } from '@fluencelabs/interfaces';
1717

1818
import {
19+
errorHandlingService,
1920
injectRelayService,
21+
injectValueService,
2022
registerParticleScopeService,
2123
responseService,
22-
errorHandlingService,
2324
ServiceDescription,
2425
userHandlerService,
25-
injectValueService,
2626
} from './services.js';
2727

2828
import { logger } from '../util/logger.js';
29+
import { IParticle } from '../particle/interfaces.js';
2930

3031
const log = logger('aqua');
3132

@@ -40,13 +41,13 @@ const log = logger('aqua');
4041
* @param args - args in the form of JSON where each key corresponds to the name of the argument
4142
* @returns
4243
*/
43-
export const callAquaFunction: CallAquaFunctionType = ({ def, script, config, peer, args }) => {
44+
export const callAquaFunction: CallAquaFunctionType = async ({ def, script, config, peer, args }) => {
4445
log.trace('calling aqua function %j', { def, script, config, args });
4546
const argumentTypes = getArgumentTypes(def);
4647

47-
const promise = new Promise((resolve, reject) => {
48-
const particle = peer.internals.createNewParticle(script, config?.ttl);
48+
const particle = await peer.internals.createNewParticle(script, config?.ttl);
4949

50+
return new Promise((resolve, reject) => {
5051
if (particle instanceof Error) {
5152
return reject(particle.message);
5253
}
@@ -92,7 +93,5 @@ export const callAquaFunction: CallAquaFunctionType = ({ def, script, config, pe
9293
);
9394
}
9495
});
95-
});
96-
97-
return promise;
96+
})
9897
};

‎packages/core/js-client/src/compilerSupport/services.ts

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@ import {
2828
IFluenceInternalApi,
2929
} from '@fluencelabs/interfaces';
3030
import { CallServiceData, GenericCallServiceHandler, ResultCodes } from '../jsServiceHost/interfaces.js';
31+
import { fromUint8Array } from 'js-base64';
3132

3233
export interface ServiceDescription {
3334
serviceId: string;
@@ -177,6 +178,7 @@ const extractCallParams = (req: CallServiceData, arrow: ArrowWithoutCallbacks):
177178

178179
const callParams = {
179180
...req.particleContext,
181+
signature: fromUint8Array(req.particleContext.signature),
180182
tetraplets,
181183
};
182184

‎packages/core/js-client/src/connection/RelayConnection.ts

Lines changed: 45 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -15,16 +15,15 @@
1515
*/
1616
import { PeerIdB58 } from '@fluencelabs/interfaces';
1717
import { pipe } from 'it-pipe';
18-
import { encode, decode } from 'it-length-prefixed';
18+
import { decode, encode } from 'it-length-prefixed';
1919
import type { PeerId } from '@libp2p/interface/peer-id';
2020
import { createLibp2p, Libp2p } from 'libp2p';
2121

2222
import { noise } from '@chainsafe/libp2p-noise';
2323
import { yamux } from '@chainsafe/libp2p-yamux';
2424
import { webSockets } from '@libp2p/websockets';
2525
import { all } from '@libp2p/websockets/filters';
26-
import { multiaddr } from '@multiformats/multiaddr';
27-
import type { Multiaddr } from '@multiformats/multiaddr';
26+
import { multiaddr, type Multiaddr } from '@multiformats/multiaddr';
2827

2928
import map from 'it-map';
3029
import { fromString } from 'uint8arrays/from-string';
@@ -35,9 +34,13 @@ import { Subject } from 'rxjs';
3534
import { throwIfHasNoPeerId } from '../util/libp2pUtils.js';
3635
import { IConnection } from './interfaces.js';
3736
import { IParticle } from '../particle/interfaces.js';
38-
import { Particle, serializeToString } from '../particle/Particle.js';
37+
import { Particle, serializeToString, verifySignature } from '../particle/Particle.js';
3938
import { identifyService } from 'libp2p/identify';
4039
import { pingService } from 'libp2p/ping';
40+
import { unmarshalPublicKey } from '@libp2p/crypto/keys';
41+
import { peerIdFromString } from '@libp2p/peer-id';
42+
import { Stream } from '@libp2p/interface/connection';
43+
import { KeyPair } from '../keypair/index.js';
4144

4245
const log = logger('connection');
4346

@@ -170,6 +173,31 @@ export class RelayConnection implements IConnection {
170173
);
171174
log.trace('data written to sink');
172175
}
176+
177+
private async processIncomingMessage(msg: string, stream: Stream) {
178+
let particle: Particle | undefined;
179+
try {
180+
particle = Particle.fromString(msg);
181+
log.trace('got particle from stream with id %s and particle id %s', stream.id, particle.id);
182+
const initPeerId = peerIdFromString(particle.initPeerId);
183+
184+
if (initPeerId.publicKey === undefined) {
185+
log.error('cannot retrieve public key from init_peer_id. particle id: %s. init_peer_id: %s', particle.id, particle.initPeerId);
186+
return;
187+
}
188+
189+
const isVerified = await verifySignature(particle, initPeerId.publicKey);
190+
if (isVerified) {
191+
this.particleSource.next(particle);
192+
} else {
193+
log.trace('particle signature is incorrect. rejecting particle with id: %s', particle.id);
194+
}
195+
} catch (e) {
196+
const particleId = particle?.id;
197+
const particleIdMessage = typeof particleId === 'string' ? `. particle id: ${particleId}` : '';
198+
log.error(`error on handling an incoming message: %O%s`, e, particleIdMessage);
199+
}
200+
}
173201

174202
private async connect() {
175203
if (this.lib2p2Peer === null) {
@@ -178,30 +206,20 @@ export class RelayConnection implements IConnection {
178206

179207
await this.lib2p2Peer.handle(
180208
[PROTOCOL_NAME],
181-
async ({ connection, stream }) => {
182-
pipe(
183-
stream.source,
184-
// @ts-ignore
185-
decode(),
186-
// @ts-ignore
187-
(source) => map(source, (buf) => toString(buf.subarray())),
188-
async (source) => {
189-
try {
190-
for await (const msg of source) {
191-
try {
192-
const particle = Particle.fromString(msg);
193-
log.trace('got particle from stream with id %s and particle id %s', stream.id, particle.id);
194-
this.particleSource.next(particle);
195-
} catch (e) {
196-
log.error('error on handling a new incoming message: %j', e);
197-
}
198-
}
199-
} catch (e) {
200-
log.error('connection closed: %j', e);
209+
async ({ connection, stream }) => pipe(
210+
stream.source,
211+
decode(),
212+
(source) => map(source, (buf) => toString(buf.subarray())),
213+
async (source) => {
214+
try {
215+
for await (const msg of source) {
216+
await this.processIncomingMessage(msg, stream);
201217
}
202-
},
203-
);
204-
},
218+
} catch (e) {
219+
log.error('connection closed: %j', e);
220+
}
221+
},
222+
),
205223
{
206224
maxInboundStreams: this.config.maxInboundStreams,
207225
maxOutboundStreams: this.config.maxOutboundStreams,

‎packages/core/js-client/src/ephemeral/__test__/ephemeral.spec.ts

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -59,7 +59,7 @@ describe.skip('Ephemeral networks tests', () => {
5959
)
6060
`;
6161

62-
const particle = client.internals.createNewParticle(script);
62+
const particle = await client.internals.createNewParticle(script);
6363

6464
const promise = new Promise<string>((resolve) => {
6565
client.internals.regHandler.forParticle(particle.id, 'test', 'test', (req: CallServiceData) => {

‎packages/core/js-client/src/jsPeer/FluencePeer.ts

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,7 @@ import {
6464
ResultCodes,
6565
} from '../jsServiceHost/interfaces.js';
6666
import { JSONValue } from '../util/commonTypes.js';
67+
import { fromUint8Array } from 'js-base64';
6768

6869
const log_particle = logger('particle');
6970
const log_peer = logger('peer');
@@ -217,8 +218,8 @@ export abstract class FluencePeer {
217218
}
218219
},
219220

220-
createNewParticle: (script: string, ttl: number = this.config.defaultTtlMs): IParticle => {
221-
return Particle.createNew(script, this.keyPair.getPeerId(), ttl);
221+
createNewParticle: (script: string, ttl: number = this.config.defaultTtlMs): Promise<IParticle> => {
222+
return Particle.createNew(script, this.keyPair.getPeerId(), ttl, this.keyPair);
222223
},
223224

224225
/**
@@ -317,7 +318,7 @@ export abstract class FluencePeer {
317318
log_particle.trace('id %s. call results: %j', item.particle.id, item.callResults);
318319
}),
319320
filterExpiredParticles(this._expireParticle.bind(this)),
320-
groupBy(item => item.particle.id),
321+
groupBy(item => fromUint8Array(item.particle.signature)),
321322
mergeMap(group$ => {
322323
let prevData: Uint8Array = Buffer.from([]);
323324
let firstRun = true;

‎packages/core/js-client/src/jsPeer/__test__/avm.spec.ts

Lines changed: 18 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -5,19 +5,19 @@ import { handleTimeout } from '../../particle/Particle.js';
55
describe('Basic AVM functionality in Fluence Peer tests', () => {
66
it('Simple call', async () => {
77
await withPeer(async (peer) => {
8-
const res = await new Promise<string[]>((resolve, reject) => {
9-
const script = `
8+
const script = `
109
(call %init_peer_id% ("print" "print") ["1"])
1110
`;
12-
const particle = peer.internals.createNewParticle(script);
13-
11+
const particle = await peer.internals.createNewParticle(script);
12+
13+
const res = await new Promise<string>((resolve, reject) => {
1414
if (particle instanceof Error) {
1515
return reject(particle.message);
1616
}
1717

1818
registerHandlersHelper(peer, particle, {
1919
print: {
20-
print: (args: Array<Array<string>>) => {
20+
print: (args: Array<string>) => {
2121
const [res] = args;
2222
resolve(res);
2323
},
@@ -33,9 +33,7 @@ describe('Basic AVM functionality in Fluence Peer tests', () => {
3333

3434
it('Par call', async () => {
3535
await withPeer(async (peer) => {
36-
const res = await new Promise<string[]>((resolve, reject) => {
37-
const res: any[] = [];
38-
const script = `
36+
const script = `
3937
(seq
4038
(par
4139
(call %init_peer_id% ("print" "print") ["1"])
@@ -44,7 +42,10 @@ describe('Basic AVM functionality in Fluence Peer tests', () => {
4442
(call %init_peer_id% ("print" "print") ["2"])
4543
)
4644
`;
47-
const particle = peer.internals.createNewParticle(script);
45+
const particle = await peer.internals.createNewParticle(script);
46+
47+
const res = await new Promise<string[]>((resolve, reject) => {
48+
const res: any[] = [];
4849

4950
if (particle instanceof Error) {
5051
return reject(particle.message);
@@ -70,8 +71,7 @@ describe('Basic AVM functionality in Fluence Peer tests', () => {
7071

7172
it('Timeout in par call: race', async () => {
7273
await withPeer(async (peer) => {
73-
const res = await new Promise((resolve, reject) => {
74-
const script = `
74+
const script = `
7575
(seq
7676
(call %init_peer_id% ("op" "identity") ["slow_result"] arg)
7777
(seq
@@ -86,8 +86,9 @@ describe('Basic AVM functionality in Fluence Peer tests', () => {
8686
)
8787
)
8888
`;
89-
const particle = peer.internals.createNewParticle(script);
90-
89+
const particle = await peer.internals.createNewParticle(script);
90+
91+
const res = await new Promise((resolve, reject) => {
9192
if (particle instanceof Error) {
9293
return reject(particle.message);
9394
}
@@ -109,8 +110,7 @@ describe('Basic AVM functionality in Fluence Peer tests', () => {
109110

110111
it('Timeout in par call: wait', async () => {
111112
await withPeer(async (peer) => {
112-
const res = await new Promise((resolve, reject) => {
113-
const script = `
113+
const script = `
114114
(seq
115115
(call %init_peer_id% ("op" "identity") ["timeout_msg"] arg)
116116
(seq
@@ -136,8 +136,9 @@ describe('Basic AVM functionality in Fluence Peer tests', () => {
136136
)
137137
)
138138
`;
139-
const particle = peer.internals.createNewParticle(script);
140-
139+
const particle = await peer.internals.createNewParticle(script);
140+
141+
const res = await new Promise((resolve, reject) => {
141142
if (particle instanceof Error) {
142143
return reject(particle.message);
143144
}

0 commit comments

Comments
 (0)