fluencelabs/js-libp2p
1
0
Fork 0
mirror of https://github.com/fluencelabs/js-libp2p synced 2025-04-25 10:32:14 +00:00
Code Issues Projects Releases Wiki Activity

chore: create signed peer record on new listen addresses in transport manager

Browse Source
This commit is contained in:
Vasco Santos 2020-09-23 18:45:01 +02:00 committed by Vasco Santos
parent 7b93ece7f2
commit 43eda43f06
7 changed files with 80 additions and 82 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
doc/API.md
View File

@ -73,7 +73,6 @@
* [`libp2p`](#libp2p)
* [`libp2p.connectionManager`](#libp2pconnectionmanager)
* [`libp2p.peerStore`](#libp2ppeerStore)
* [`libp2p.transportManager`](#libp2ptransportmanager)
* [Types](#types)
* [`Stats`](#stats)
@ -2020,14 +2019,6 @@ This event will be triggered anytime we are disconnected from another peer, rega
- `peerId`: instance of [`PeerId`][peer-id]
- `protocols`: array of known, supported protocols for the peer (string identifiers)
### libp2p.transportManager
#### Listening addresses change
This event will be triggered anytime the listening addresses change.
`libp2p.transportManager.on('listening', () => {})`
## Types
### Stats

2
src/circuit/listener.js
View File

@ -21,7 +21,7 @@ module.exports = (libp2p) => {
if (deleted) {
// Announce listen addresses change
listener.emit('listening')
listener.emit('close')
}
})

52
src/identify/index.js
View File

@ -64,11 +64,6 @@ class IdentifyService {
*/
this.connectionManager = libp2p.connectionManager
/**
* @property {TransportManager}
*/
this.transportManager = libp2p.transportManager
/**
* @property {PeerId}
*/
@ -96,10 +91,11 @@ class IdentifyService {
this.identify(connection).catch(log.error)
})
// When new addresses are used for listening, update self peer record
this.transportManager.on('listening', async () => {
await this._createSelfPeerRecord()
this.pushToPeerStore()
// When self multiaddrs change, trigger identify-push
this.peerStore.on('change:multiaddrs', ({ peerId }) => {
if (peerId.toString() === this.peerId.toString()) {
this.pushToPeerStore()
}
})
}
@ -110,7 +106,7 @@ class IdentifyService {
* @returns {Promise<void>}
*/
async push (connections) {
const signedPeerRecord = await this._getSelfPeerRecord()
const signedPeerRecord = await this.peerStore.addressBook.getRawEnvelope(this.peerId)
const listenAddrs = this._libp2p.multiaddrs.map((ma) => ma.bytes)
const protocols = Array.from(this._protocols.keys())
@ -260,7 +256,7 @@ class IdentifyService {
publicKey = this.peerId.pubKey.bytes
}
const signedPeerRecord = await this._getSelfPeerRecord()
const signedPeerRecord = await this.peerStore.addressBook.getRawEnvelope(this.peerId)
const message = Message.encode({
protocolVersion: this._host.protocolVersion,
@ -330,40 +326,6 @@ class IdentifyService {
// Update the protocols
this.peerStore.protoBook.set(id, message.protocols)
}
/**
* Get self signed peer record raw envelope.
* @return {Promise<Uint8Array>}
*/
_getSelfPeerRecord () {
const selfSignedPeerRecord = this.peerStore.addressBook.getRawEnvelope(this.peerId)
if (selfSignedPeerRecord) {
return selfSignedPeerRecord
}
return this._createSelfPeerRecord()
}
/**
* Create self signed peer record raw envelope.
* @return {Uint8Array}
*/
async _createSelfPeerRecord () {
try {
const peerRecord = new PeerRecord({
peerId: this.peerId,
multiaddrs: this._libp2p.multiaddrs
})
const envelope = await Envelope.seal(peerRecord, this.peerId)
this.peerStore.addressBook.consumePeerRecord(envelope)
return this.peerStore.addressBook.getRawEnvelope(this.peerId)
} catch (err) {
log.error('failed to get self peer record')
}
return null
}
}
module.exports.IdentifyService = IdentifyService

39
src/transport-manager.js
View File

@ -1,6 +1,5 @@
'use strict'
const { EventEmitter } = require('events')
const pSettle = require('p-settle')
const { codes } = require('./errors')
const errCode = require('err-code')
@ -8,11 +7,10 @@ const debug = require('debug')
const log = debug('libp2p:transports')
log.error = debug('libp2p:transports:error')
/**
* Responsible for managing the transports and their listeners.
* @fires TransportManager#listening Emitted when listening addresses change.
*/
class TransportManager extends EventEmitter {
const Envelope = require('./record/envelope')
const PeerRecord = require('./record/peer-record')
class TransportManager {
/**
* @class
* @param {object} options
@ -21,8 +19,6 @@ class TransportManager extends EventEmitter {
* @param {boolean} [options.faultTolerance = FAULT_TOLERANCE.FATAL_ALL] - Address listen error tolerance.
*/
constructor ({ libp2p, upgrader, faultTolerance = FAULT_TOLERANCE.FATAL_ALL }) {
super()
this.libp2p = libp2p
this.upgrader = upgrader
this._transports = new Map()
@ -73,6 +69,7 @@ class TransportManager extends EventEmitter {
while (listeners.length) {
const listener = listeners.pop()
listener.removeAllListeners('listening')
listener.removeAllListeners('close')
tasks.push(listener.close())
}
}
@ -166,8 +163,9 @@ class TransportManager extends EventEmitter {
const listener = transport.createListener(this._listenerOptions.get(key), this.onConnection)
this._listeners.get(key).push(listener)
// Track listen events
listener.on('listening', () => this.emit('listening'))
// Track listen/close events
listener.on('listening', () => this._createSelfPeerRecord())
listener.on('close', () => this._createSelfPeerRecord())
// We need to attempt to listen on everything
tasks.push(listener.listen(addr))
@ -214,6 +212,7 @@ class TransportManager extends EventEmitter {
// Close any running listeners
for (const listener of this._listeners.get(key)) {
listener.removeAllListeners('listening')
listener.removeAllListeners('close')
await listener.close()
}
}
@ -236,6 +235,26 @@ class TransportManager extends EventEmitter {
await Promise.all(tasks)
}
/**
* Create self signed peer record raw envelope.
* @return {Uint8Array}
*/
async _createSelfPeerRecord () {
try {
const peerRecord = new PeerRecord({
peerId: this.libp2p.peerId,
multiaddrs: this.libp2p.multiaddrs
})
const envelope = await Envelope.seal(peerRecord, this.libp2p.peerId)
this.libp2p.peerStore.addressBook.consumePeerRecord(envelope)
return this.libp2p.peerStore.addressBook.getRawEnvelope(this.libp2p.peerId)
} catch (err) {
log.error('failed to get self peer record')
}
return null
}
}
/**

3
test/dialing/direct.spec.js
View File

@ -349,7 +349,6 @@ describe('Dialing (direct, WebSockets)', () => {
const connection = await libp2p.dial(remoteAddr)
expect(connection).to.exist()
sinon.spy(libp2p.peerStore.addressBook, 'consumePeerRecord')
sinon.spy(libp2p.peerStore.protoBook, 'set')
// Wait for onConnection to be called
@ -358,8 +357,6 @@ describe('Dialing (direct, WebSockets)', () => {
expect(libp2p.identifyService.identify.callCount).to.equal(1)
await libp2p.identifyService.identify.firstCall.returnValue
// Self + New peer
expect(libp2p.peerStore.addressBook.consumePeerRecord.callCount).to.equal(2)
expect(libp2p.peerStore.protoBook.set.callCount).to.equal(1)
})

39
test/identify/index.spec.js
View File

@ -20,6 +20,7 @@ const { IdentifyService, multicodecs } = require('../../src/identify')
const Peers = require('../fixtures/peers')
const Libp2p = require('../../src')
const Envelope = require('../../src/record/envelope')
const PeerRecord = require('../../src/record/peer-record')
const PeerStore = require('../../src/peer-store')
const baseOptions = require('../utils/base-options.browser')
const pkg = require('../../package.json')
@ -52,7 +53,6 @@ describe('Identify', () => {
libp2p: {
peerId: localPeer,
connectionManager: new EventEmitter(),
transportManager: new EventEmitter(),
peerStore: new PeerStore({ peerId: localPeer }),
multiaddrs: listenMaddrs,
_options: { host: {} }
@ -64,7 +64,6 @@ describe('Identify', () => {
libp2p: {
peerId: remotePeer,
connectionManager: new EventEmitter(),
transportManager: new EventEmitter(),
peerStore: new PeerStore({ peerId: remotePeer }),
multiaddrs: listenMaddrs,
_options: { host: {} }
@ -82,6 +81,9 @@ describe('Identify', () => {
sinon.spy(localIdentify.peerStore.addressBook, 'consumePeerRecord')
sinon.spy(localIdentify.peerStore.protoBook, 'set')
// Transport Manager creates signed peer record
await _createSelfPeerRecord(remoteIdentify._libp2p)
// Run identify
await Promise.all([
localIdentify.identify(localConnectionMock),
@ -109,7 +111,6 @@ describe('Identify', () => {
libp2p: {
peerId: localPeer,
connectionManager: new EventEmitter(),
transportManager: new EventEmitter(),
peerStore: new PeerStore({ peerId: localPeer }),
multiaddrs: listenMaddrs,
_options: { host: {} }
@ -121,7 +122,6 @@ describe('Identify', () => {
libp2p: {
peerId: remotePeer,
connectionManager: new EventEmitter(),
transportManager: new EventEmitter(),
peerStore: new PeerStore({ peerId: remotePeer }),
multiaddrs: listenMaddrs,
_options: { host: {} }
@ -172,7 +172,6 @@ describe('Identify', () => {
libp2p: {
peerId: localPeer,
connectionManager: new EventEmitter(),
transportManager: new EventEmitter(),
peerStore: new PeerStore({ peerId: localPeer }),
multiaddrs: [],
_options: { host: {} }
@ -183,7 +182,6 @@ describe('Identify', () => {
libp2p: {
peerId: remotePeer,
connectionManager: new EventEmitter(),
transportManager: new EventEmitter(),
peerStore: new PeerStore({ peerId: remotePeer }),
multiaddrs: [],
_options: { host: {} }
@ -252,7 +250,6 @@ describe('Identify', () => {
libp2p: {
peerId: localPeer,
connectionManager: new EventEmitter(),
transportManager: new EventEmitter(),
peerStore: new PeerStore({ peerId: localPeer }),
multiaddrs: listenMaddrs,
_options: { host: {} }
@ -267,7 +264,6 @@ describe('Identify', () => {
libp2p: {
peerId: remotePeer,
connectionManager,
transportManager: new EventEmitter(),
peerStore: new PeerStore({ peerId: remotePeer }),
multiaddrs: [],
_options: { host: {} }
@ -285,6 +281,10 @@ describe('Identify', () => {
sinon.spy(remoteIdentify.peerStore.addressBook, 'consumePeerRecord')
sinon.spy(remoteIdentify.peerStore.protoBook, 'set')
// Transport Manager creates signed peer record
await _createSelfPeerRecord(localIdentify._libp2p)
await _createSelfPeerRecord(remoteIdentify._libp2p)
// Run identify
await Promise.all([
localIdentify.push([localConnectionMock]),
@ -295,7 +295,7 @@ describe('Identify', () => {
})
])
expect(remoteIdentify.peerStore.addressBook.consumePeerRecord.callCount).to.equal(1)
expect(remoteIdentify.peerStore.addressBook.consumePeerRecord.callCount).to.equal(2)
expect(remoteIdentify.peerStore.protoBook.set.callCount).to.equal(1)
const addresses = localIdentify.peerStore.addressBook.get(localPeer)
@ -317,7 +317,6 @@ describe('Identify', () => {
libp2p: {
peerId: localPeer,
connectionManager: new EventEmitter(),
transportManager: new EventEmitter(),
peerStore: new PeerStore({ peerId: localPeer }),
multiaddrs: listenMaddrs,
_options: { host: {} }
@ -332,7 +331,6 @@ describe('Identify', () => {
libp2p: {
peerId: remotePeer,
connectionManager,
transportManager: new EventEmitter(),
peerStore: new PeerStore({ peerId: remotePeer }),
multiaddrs: [],
_options: { host: {} }
@ -409,8 +407,8 @@ describe('Identify', () => {
expect(connection).to.exist()
// Wait for peer store to be updated
// Dialer._createDialTarget (add), Identify (consume), Create self (consume)
await pWaitFor(() => peerStoreSpyConsumeRecord.callCount === 2 && peerStoreSpyAdd.callCount === 1)
// Dialer._createDialTarget (add), Identify (consume)
await pWaitFor(() => peerStoreSpyConsumeRecord.callCount === 1 && peerStoreSpyAdd.callCount === 1)
expect(libp2p.identifyService.identify.callCount).to.equal(1)
// The connection should have no open streams
@ -474,3 +472,18 @@ describe('Identify', () => {
})
})
})
// Self peer record creating on Transport Manager simulation
const _createSelfPeerRecord = async (libp2p) => {
try {
const peerRecord = new PeerRecord({
peerId: libp2p.peerId,
multiaddrs: libp2p.multiaddrs
})
const envelope = await Envelope.seal(peerRecord, libp2p.peerId)
libp2p.peerStore.addressBook.consumePeerRecord(envelope)
return libp2p.peerStore.addressBook.getRawEnvelope(libp2p.peerId)
} catch (_) {}
return null
}

18
test/transports/transport-manager.node.js
View File

@ -4,13 +4,17 @@
const chai = require('chai')
chai.use(require('dirty-chai'))
const { expect } = chai
const sinon = require('sinon')
const AddressManager = require('../../src/address-manager')
const TransportManager = require('../../src/transport-manager')
const PeerStore = require('../../src/peer-store')
const Transport = require('libp2p-tcp')
const PeerId = require('peer-id')
const multiaddr = require('multiaddr')
const mockUpgrader = require('../utils/mockUpgrader')
const sinon = require('sinon')
const Peers = require('../fixtures/peers')
const addrs = [
multiaddr('/ip4/127.0.0.1/tcp/0'),
multiaddr('/ip4/127.0.0.1/tcp/0')
@ -18,11 +22,17 @@ const addrs = [
describe('Transport Manager (TCP)', () => {
let tm
let localPeer
before(async () => {
localPeer = await PeerId.createFromJSON(Peers[0])
})
before(() => {
tm = new TransportManager({
libp2p: {
addressManager: new AddressManager({ listen: addrs })
addressManager: new AddressManager({ listen: addrs }),
PeerStore: new PeerStore({ peerId: localPeer })
},
upgrader: mockUpgrader,
onConnection: () => {}
@ -41,12 +51,18 @@ describe('Transport Manager (TCP)', () => {
})
it('should be able to listen', async () => {
sinon.spy(tm, '_createSelfPeerRecord')
tm.add(Transport.prototype[Symbol.toStringTag], Transport, { listenerOptions: { listen: 'carefully' } })
const transport = tm._transports.get(Transport.prototype[Symbol.toStringTag])
const spyListener = sinon.spy(transport, 'createListener')
await tm.listen()
expect(tm._listeners).to.have.key(Transport.prototype[Symbol.toStringTag])
expect(tm._listeners.get(Transport.prototype[Symbol.toStringTag])).to.have.length(addrs.length)
// Created Self Peer record on new listen address
expect(tm._createSelfPeerRecord.callCount).to.equal(addrs.length)
// Ephemeral ip addresses may result in multiple listeners
expect(tm.getAddrs().length).to.equal(addrs.length)
await tm.close()