Compare commits

...

38 Commits

Author SHA1 Message Date
a3bb60e50b chore: release version v0.30.0-rc.1 2020-12-11 18:28:22 +01:00
29212ea61d chore: update contributors 2020-12-11 18:28:22 +01:00
216eb9730e fix: types from ipfs integration (#832) 2020-12-11 18:24:38 +01:00
9472fefc4e chore: remove secio from packages table (#833) 2020-12-11 12:49:58 +01:00
0f451659df chore: release version v0.30.0-rc.0 2020-12-10 15:50:04 +01:00
eaf4fb7ac6 chore: update contributors 2020-12-10 15:50:04 +01:00
4bbd915a6d docs: production guide base setup (#804) 2020-12-10 15:36:49 +01:00
64124f17ed chore: add typedefs (#802) 2020-12-10 15:36:07 +01:00
117f2b97b0 chore: auto relay configuration example with noise (#828) 2020-12-10 15:23:26 +01:00
bbdaae17ae chore: update pubsub example by disabled emit self (#823) 2020-12-10 15:23:26 +01:00
7461d2a642 docs: migration 0.29 to 0.30 (#808) 2020-12-10 15:23:26 +01:00
f82a972ab2 chore: use set-delayed-interval module on circuit (#809) 2020-12-10 15:23:26 +01:00
b73106eba2 feat: discover and connect to closest peers (#798) 2020-12-10 15:23:26 +01:00
6b434aeaf7 chore: update websockets (#806)
* chore: update websockets
2020-12-10 15:23:26 +01:00
44f07033fe docs: auto relay example (#795)
* chore: auto relay example

* chore: update examples to use process arguments

* chore: add test setup for node tests and test for auto-relay

* chore: apply suggestions from code review

* chore: do not use promise for multiaddrs event on example
2020-12-10 15:23:26 +01:00
91b15b6790 feat: custom dialer addr sorter (#792)
* feat: custom dialer addr sorter

* chore: use libp2p utils sorter via addressBook getMultiaddrsForPeer

* chore: use new libp2p utils

* chore: apply suggestions from code review

Co-authored-by: Jacob Heun <jacobheun@gmail.com>

Co-authored-by: Jacob Heun <jacobheun@gmail.com>
2020-12-10 15:23:26 +01:00
9205fce34d chore: update pubsub (#801)
BREAKING CHANGE: pubsub signing policy properties were changed according to libp2p-interfaces changes to a single property. The emitSelf option default value was also modified to match the routers value
2020-12-10 15:23:26 +01:00
a4e858cbf5 test: custom announce filter 2020-12-10 15:23:26 +01:00
3f1dc20caf fix: remove test/dialing/utils extra file 2020-12-10 15:23:26 +01:00
8351adff2b chore: update address manager readme 2020-12-10 15:23:26 +01:00
22e41f5047 chore: remove noAnnounce from address manager 2020-12-10 15:23:26 +01:00
48476d504a feat: custom announce filter 2020-12-10 15:23:26 +01:00
186f9b758e chore: store self protocols in protobook (#760) 2020-12-10 15:23:26 +01:00
20d9d98b3a chore: improve logging for auto relay active listen 2020-12-10 15:20:37 +01:00
87dcd2ec47 chore: sort relay addresses to listen for public first 2020-12-10 15:20:37 +01:00
b87d42c074 chore: lint issues fixed 2020-12-10 15:20:37 +01:00
df5fe0e3ab chore: apply suggestions from code review
Co-authored-by: Jacob Heun <jacobheun@gmail.com>
2020-12-10 15:20:37 +01:00
5d79a137ee chore: add configuration docs for auto relay and hop service 2020-12-10 15:20:37 +01:00
8b6c738937 chore: address review 2020-12-10 15:20:37 +01:00
9faf1bfcf6 feat: auto relay network query for new relays 2020-12-10 15:20:37 +01:00
f9235e2ff6 chore: lint issue fixed 0.30 2020-12-10 15:20:12 +01:00
f9786c8ab8 chore: address review 2020-12-10 15:20:12 +01:00
b45ef6aa54 chore: add identify test for multiaddr change 2020-12-10 15:19:38 +01:00
d79ef8420e chore: create signed peer record on new listen addresses in transport manager 2020-12-10 15:19:06 +01:00
fba9457011 chore: use listening events to create self peer record on updates 2020-12-10 15:18:27 +01:00
c22508c3e4 chore: _isStarted is false when stop starts 2020-12-10 15:17:01 +01:00
0ac8252b4c chore: auto relay multiaddr update push 2020-12-10 15:17:01 +01:00
65ec267e7f feat: auto relay (#723)
* feat: auto relay

* fix: leverage protoBook events to ask relay peers if they support hop

* chore: refactor disconnect

* chore: do not listen on a relayed conn

* chore: tweaks

* chore: improve _listenOnAvailableHopRelays logic

* chore: default value of 1 to maxListeners on auto-relay
2020-12-10 15:17:01 +01:00
121 changed files with 4450 additions and 1375 deletions

67
.github/workflows/main.yml vendored Normal file
View File

@ -0,0 +1,67 @@
name: ci
on:
push:
branches:
- master
pull_request:
branches:
- '**'
jobs:
check:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- run: yarn
- run: yarn lint
- uses: gozala/typescript-error-reporter-action@v1.0.8
- run: yarn build
- run: yarn aegir dep-check
- uses: ipfs/aegir/actions/bundle-size@master
name: size
with:
github_token: ${{ secrets.GITHUB_TOKEN }}
test-node:
needs: check
runs-on: ${{ matrix.os }}
strategy:
matrix:
os: [windows-latest, ubuntu-latest, macos-latest]
node: [12, 14]
fail-fast: true
steps:
- uses: actions/checkout@v2
- uses: actions/setup-node@v1
with:
node-version: ${{ matrix.node }}
- run: yarn
- run: npx nyc --reporter=lcov aegir test -t node -- --bail
- uses: codecov/codecov-action@v1
test-chrome:
needs: check
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- run: yarn
- run: npx aegir test -t browser -t webworker --bail
test-firefox:
needs: check
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- run: yarn
- run: npx aegir test -t browser -t webworker --bail -- --browsers FirefoxHeadless
test-interop:
needs: check
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- run: yarn
- run: cd node_modules/interop-libp2p && yarn && LIBP2P_JS=${GITHUB_WORKSPACE}/src/index.js npx aegir test -t node --bail
test-auto-relay-example:
needs: check
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- run: yarn
- run: cd examples && yarn && npm run test -- auto-relay

View File

@ -1,50 +0,0 @@
language: node_js
cache: npm
stages:
- check
- test
- cov
node_js:
- 'lts/*'
- '14'
os:
- linux
- osx
script: npx nyc -s npm run test:node -- --bail
after_success: npx nyc report --reporter=text-lcov > coverage.lcov && npx codecov
jobs:
include:
- stage: check
script:
- npx aegir build --bundlesize
# Remove pull libs once ping is async
- npx aegir dep-check -- -i pull-handshake -i pull-stream
- npm run lint
- stage: test
name: chrome
addons:
chrome: stable
script:
- npx aegir test -t browser -t webworker
- stage: test
name: firefox
addons:
firefox: latest
script:
- npx aegir test -t browser -t webworker -- --browsers FirefoxHeadless
- stage: test
name: interop
script:
- cd node_modules/interop-libp2p
- npm install
- LIBP2P_JS=${TRAVIS_BUILD_DIR}/src/index.js npx aegir test -t node --bail
notifications:
email: false

View File

@ -1,3 +1,40 @@
# [0.30.0-rc.1](https://github.com/libp2p/js-libp2p/compare/v0.30.0-rc.0...v0.30.0-rc.1) (2020-12-11)
### Bug Fixes
* types from ipfs integration ([#832](https://github.com/libp2p/js-libp2p/issues/832)) ([216eb97](https://github.com/libp2p/js-libp2p/commit/216eb9730ef473f73a974c3dbaf306ecdc815c8b))
# [0.30.0-rc.0](https://github.com/libp2p/js-libp2p/compare/v0.29.4...v0.30.0-rc.0) (2020-12-10)
### Bug Fixes
* remove test/dialing/utils extra file ([3f1dc20](https://github.com/libp2p/js-libp2p/commit/3f1dc20caf1c80078f403deb9174cd06d08567ab))
### chore
* update pubsub ([#801](https://github.com/libp2p/js-libp2p/issues/801)) ([9205fce](https://github.com/libp2p/js-libp2p/commit/9205fce34d0cd8dd5d32988be34c110fc0a5b6e2))
### Features
* auto relay ([#723](https://github.com/libp2p/js-libp2p/issues/723)) ([65ec267](https://github.com/libp2p/js-libp2p/commit/65ec267e7f4826caacd042213c3fbacce589ab5b))
* auto relay network query for new relays ([9faf1bf](https://github.com/libp2p/js-libp2p/commit/9faf1bfcf61581acc715b9be78b71dc14501835a))
* custom announce filter ([48476d5](https://github.com/libp2p/js-libp2p/commit/48476d504a98b7b51b3e2dc64eab93670fde0c7b))
* custom dialer addr sorter ([#792](https://github.com/libp2p/js-libp2p/issues/792)) ([91b15b6](https://github.com/libp2p/js-libp2p/commit/91b15b6790952b4db11264961d9c6f2a96d1fe43))
* discover and connect to closest peers ([#798](https://github.com/libp2p/js-libp2p/issues/798)) ([b73106e](https://github.com/libp2p/js-libp2p/commit/b73106eba2d559621f427f7aa788e9b0ef47d135))
### BREAKING CHANGES
* pubsub signing policy properties were changed according to libp2p-interfaces changes to a single property. The emitSelf option default value was also modified to match the routers value
<a name="0.29.4"></a>
## [0.29.4](https://github.com/libp2p/js-libp2p/compare/v0.29.3...v0.29.4) (2020-12-09)

View File

@ -147,7 +147,6 @@ List of packages currently in existence for libp2p
| [`libp2p-websockets`](//github.com/libp2p/js-libp2p-websockets) | [![npm](https://img.shields.io/npm/v/libp2p-websockets.svg?maxAge=86400&style=flat-square)](//github.com/libp2p/js-libp2p-websockets/releases) | [![Deps](https://david-dm.org/libp2p/js-libp2p-websockets.svg?style=flat-square)](https://david-dm.org/libp2p/js-libp2p-websockets) | [![Travis CI](https://flat.badgen.net/travis/libp2p/js-libp2p-websockets/master)](https://travis-ci.com/libp2p/js-libp2p-websockets) | [![codecov](https://codecov.io/gh/libp2p/js-libp2p-websockets/branch/master/graph/badge.svg?style=flat-square)](https://codecov.io/gh/libp2p/js-libp2p-websockets) | [Jacob Heun](mailto:jacobheun@gmail.com) |
| **secure channels** |
| [`libp2p-noise`](//github.com/NodeFactoryIo/js-libp2p-noise) | [![npm](https://img.shields.io/npm/v/libp2p-noise.svg?maxAge=86400&style=flat-square)](//github.com/NodeFactoryIo/js-libp2p-noise/releases) | [![Deps](https://david-dm.org/NodeFactoryIo/js-libp2p-noise.svg?style=flat-square)](https://david-dm.org/NodeFactoryIo/js-libp2p-noise) | [![Travis CI](https://flat.badgen.net/travis/NodeFactoryIo/js-libp2p-noise/master)](https://travis-ci.com/NodeFactoryIo/js-libp2p-noise) | [![codecov](https://codecov.io/gh/NodeFactoryIo/js-libp2p-noise/branch/master/graph/badge.svg?style=flat-square)](https://codecov.io/gh/NodeFactoryIo/js-libp2p-noise) | N/A |
| [`libp2p-secio`](//github.com/libp2p/js-libp2p-secio) | [![npm](https://img.shields.io/npm/v/libp2p-secio.svg?maxAge=86400&style=flat-square)](//github.com/libp2p/js-libp2p-secio/releases) | [![Deps](https://david-dm.org/libp2p/js-libp2p-secio.svg?style=flat-square)](https://david-dm.org/libp2p/js-libp2p-secio) | [![Travis CI](https://flat.badgen.net/travis/libp2p/js-libp2p-secio/master)](https://travis-ci.com/libp2p/js-libp2p-secio) | [![codecov](https://codecov.io/gh/libp2p/js-libp2p-secio/branch/master/graph/badge.svg?style=flat-square)](https://codecov.io/gh/libp2p/js-libp2p-secio) | [Friedel Ziegelmayer](mailto:dignifiedquire@gmail.com) |
| **stream multiplexers** |
| [`libp2p-mplex`](//github.com/libp2p/js-libp2p-mplex) | [![npm](https://img.shields.io/npm/v/libp2p-mplex.svg?maxAge=86400&style=flat-square)](//github.com/libp2p/js-libp2p-mplex/releases) | [![Deps](https://david-dm.org/libp2p/js-libp2p-mplex.svg?style=flat-square)](https://david-dm.org/libp2p/js-libp2p-mplex) | [![Travis CI](https://flat.badgen.net/travis/libp2p/js-libp2p-mplex/master)](https://travis-ci.com/libp2p/js-libp2p-mplex) | [![codecov](https://codecov.io/gh/libp2p/js-libp2p-mplex/branch/master/graph/badge.svg?style=flat-square)](https://codecov.io/gh/libp2p/js-libp2p-mplex) | [Vasco Santos](mailto:vasco.santos@moxy.studio) |
| **peer discovery** |

View File

@ -13,14 +13,14 @@
* [`ping`](#ping)
* [`multiaddrs`](#multiaddrs)
* [`addressManager.getListenAddrs`](#addressmanagergetlistenaddrs)
* [`addressmger.getAnnounceAddrs`](#addressmanagergetannounceaddrs)
* [`addressManager.getNoAnnounceAddrs`](#addressmanagergetnoannounceaddrs)
* [`addressManager.getAnnounceAddrs`](#addressmanagergetannounceaddrs)
* [`contentRouting.findProviders`](#contentroutingfindproviders)
* [`contentRouting.provide`](#contentroutingprovide)
* [`contentRouting.put`](#contentroutingput)
* [`contentRouting.get`](#contentroutingget)
* [`contentRouting.getMany`](#contentroutinggetmany)
* [`peerRouting.findPeer`](#peerroutingfindpeer)
* [`peerRouting.getClosestPeers`](#peerroutinggetclosestpeers)
* [`peerStore.addressBook.add`](#peerstoreaddressbookadd)
* [`peerStore.addressBook.delete`](#peerstoreaddressbookdelete)
* [`peerStore.addressBook.get`](#peerstoreaddressbookget)
@ -37,6 +37,7 @@
* [`peerStore.protoBook.add`](#peerstoreprotobookadd)
* [`peerStore.protoBook.delete`](#peerstoreprotobookdelete)
* [`peerStore.protoBook.get`](#peerstoreprotobookget)
* [`peerStore.protoBook.remove`](#peerstoreprotobookremove)
* [`peerStore.protoBook.set`](#peerstoreprotobookset)
* [`peerStore.delete`](#peerstoredelete)
* [`peerStore.get`](#peerstoreget)
@ -90,7 +91,7 @@ Creates an instance of Libp2p.
|------|------|-------------|
| options | `object` | libp2p options |
| options.modules | [`Array<object>`](./CONFIGURATION.md#modules) | libp2p [modules](./CONFIGURATION.md#modules) to use |
| [options.addresses] | `{ listen: Array<string>, announce: Array<string>, noAnnounce: Array<string> }` | Addresses for transport listening and to advertise to the network |
| [options.addresses] | `{ listen: Array<string>, announce: Array<string>, announceFilter: (ma: Array<multiaddr>) => Array<multiaddr> }` | Addresses for transport listening and to advertise to the network |
| [options.config] | `object` | libp2p modules configuration and core configuration |
| [options.host] | `{ agentVersion: string }` | libp2p host options |
| [options.connectionManager] | [`object`](./CONFIGURATION.md#configuring-connection-manager) | libp2p Connection Manager [configuration](./CONFIGURATION.md#configuring-connection-manager) |
@ -100,6 +101,7 @@ Creates an instance of Libp2p.
| [options.keychain] | [`object`](./CONFIGURATION.md#setup-with-keychain) | keychain [configuration](./CONFIGURATION.md#setup-with-keychain) |
| [options.metrics] | [`object`](./CONFIGURATION.md#configuring-metrics) | libp2p Metrics [configuration](./CONFIGURATION.md#configuring-metrics) |
| [options.peerId] | [`PeerId`][peer-id] | peerId instance (it will be created if not provided) |
| [options.peerRouting] | [`object`](./CONFIGURATION.md#setup-with-content-and-peer-routing) | libp2p Peer routing service [configuration](./CONFIGURATION.md#setup-with-content-and-peer-routing) |
| [options.peerStore] | [`object`](./CONFIGURATION.md#configuring-peerstore) | libp2p PeerStore [configuration](./CONFIGURATION.md#configuring-peerstore) |
For Libp2p configurations and modules details read the [Configuration Document](./CONFIGURATION.md).
@ -514,26 +516,6 @@ const announceMa = libp2p.addressManager.getAnnounceAddrs()
// [ <Multiaddr 047f00000106f9ba - /dns4/peer.io/...> ]
```
### addressManager.getNoAnnounceAddrs
Get the multiaddrs that were provided to not announce to the network.
`libp2p.addressManager.getNoAnnounceAddrs()`
#### Returns
| Type | Description |
|------|-------------|
| `Array<Multiaddr>` | Provided noAnnounce multiaddrs |
#### Example
```js
// ...
const noAnnounceMa = libp2p.addressManager.getNoAnnounceAddrs()
// [ <Multiaddr 047f00000106f9ba - /ip4/127.0.0.1/tcp/63930> ]
```
### transportManager.getAddrs
Get the multiaddrs that libp2p transports are using to listen on.
@ -727,6 +709,36 @@ Iterates over all peer routers in series to find the given peer. If the DHT is e
const peer = await libp2p.peerRouting.findPeer(peerId, options)
```
### peerRouting.getClosestPeers
Iterates over all content routers in series to get the closest peers of the given key.
Once a content router succeeds, the iteration will stop. If the DHT is enabled, it will be queried first.
`libp2p.peerRouting.getClosestPeers(cid, options)`
#### Parameters
| Name | Type | Description |
|------|------|-------------|
| key | `Uint8Array` | A CID like key |
| options | `object` | operation options |
| options.timeout | `number` | How long the query can take (ms). |
#### Returns
| Type | Description |
|------|-------------|
| `AsyncIterable<{ id: PeerId, multiaddrs: Multiaddr[] }` | Async iterator for peer data |
#### Example
```js
// Iterate over the closest peers found for the given key
for await (const peer of libp2p.peerRouting.getClosestPeers(key)) {
console.log(peer.id, peer.multiaddrs)
}
```
### peerStore.addressBook.add
Adds known `multiaddrs` of a given peer. If the peer is not known, it will be set with the provided multiaddrs.
@ -875,32 +887,6 @@ Consider using `addressBook.add()` if you're not sure this is what you want to d
peerStore.addressBook.add(peerId, multiaddr)
```
### peerStore.protoBook.add
Add known `protocols` of a given peer.
`peerStore.protoBook.add(peerId, protocols)`
#### Parameters
| Name | Type | Description |
|------|------|-------------|
| peerId | [`PeerId`][peer-id] | peerId to set |
| protocols | `Array<string>` | protocols to add |
#### Returns
| Type | Description |
|------|-------------|
| `ProtoBook` | Returns the Proto Book component |
#### Example
```js
peerStore.protoBook.add(peerId, protocols)
```
### peerStore.keyBook.delete
Delete the provided peer from the book.
@ -1123,6 +1109,31 @@ Set known metadata of a given `peerId`.
peerStore.metadataBook.set(peerId, 'location', uint8ArrayFromString('Berlin'))
```
### peerStore.protoBook.add
Add known `protocols` of a given peer.
`peerStore.protoBook.add(peerId, protocols)`
#### Parameters
| Name | Type | Description |
|------|------|-------------|
| peerId | [`PeerId`][peer-id] | peerId to set |
| protocols | `Array<string>` | protocols to add |
#### Returns
| Type | Description |
|------|-------------|
| `ProtoBook` | Returns the Proto Book component |
#### Example
```js
peerStore.protoBook.add(peerId, protocols)
```
### peerStore.protoBook.delete
Delete the provided peer from the book.
@ -1179,6 +1190,31 @@ peerStore.protoBook.get(peerId)
// [ '/proto/1.0.0', '/proto/1.1.0' ]
```
### peerStore.protoBook.remove
Remove given `protocols` of a given peer.
`peerStore.protoBook.remove(peerId, protocols)`
#### Parameters
| Name | Type | Description |
|------|------|-------------|
| peerId | [`PeerId`][peer-id] | peerId to set |
| protocols | `Array<string>` | protocols to remove |
#### Returns
| Type | Description |
|------|-------------|
| `ProtoBook` | Returns the Proto Book component |
#### Example
```js
peerStore.protoBook.remove(peerId, protocols)
```
### peerStore.protoBook.set
Set known `protocols` of a given peer.

View File

@ -20,6 +20,7 @@
- [Customizing DHT](#customizing-dht)
- [Setup with Content and Peer Routing](#setup-with-content-and-peer-routing)
- [Setup with Relay](#setup-with-relay)
- [Setup with Auto Relay](#setup-with-auto-relay)
- [Setup with Keychain](#setup-with-keychain)
- [Configuring Dialing](#configuring-dialing)
- [Configuring Connection Manager](#configuring-connection-manager)
@ -210,10 +211,10 @@ Besides the `modules` and `config`, libp2p allows other internal options and con
- This is used in modules such as the DHT. If it is not provided, `js-libp2p` will use an in memory datastore.
- `peerId`: the identity of the node, an instance of [libp2p/js-peer-id](https://github.com/libp2p/js-peer-id).
- This is particularly useful if you want to reuse the same `peer-id`, as well as for modules like `libp2p-delegated-content-routing`, which need a `peer-id` in their instantiation.
- `addresses`: an object containing `listen`, `announce` and `noAnnounce` properties with `Array<string>`:
- `addresses`: an object containing `listen`, `announce` and `announceFilter`:
- `listen` addresses will be provided to the libp2p underlying transports for listening on them.
- `announce` addresses will be used to compute the advertises that the node should advertise to the network.
- `noAnnounce` addresses will be used as a filter to compute the advertises that the node should advertise to the network.
- `announceFilter`: filter function used to filter announced addresses programmatically: `(ma: Array<multiaddr>) => Array<multiaddr>`. Default: returns all addresses. [`libp2p-utils`](https://github.com/libp2p/js-libp2p-utils) provides useful [multiaddr utilities](https://github.com/libp2p/js-libp2p-utils/blob/master/API.md#multiaddr-isloopbackma) to create your filters.
### Examples
@ -321,6 +322,8 @@ const MPLEX = require('libp2p-mplex')
const { NOISE } = require('libp2p-noise')
const GossipSub = require('libp2p-gossipsub')
const { SignaturePolicy } = require('libp2p-interfaces/src/pubsub/signature-policy')
const node = await Libp2p.create({
modules: {
transport: [TCP],
@ -331,9 +334,8 @@ const node = await Libp2p.create({
config: {
pubsub: { // The pubsub options (and defaults) can be found in the pubsub router documentation
enabled: true,
emitSelf: true, // whether the node should emit to self on publish
signMessages: true, // if messages should be signed
strictSigning: true // if message signing should be required
emitSelf: false, // whether the node should emit to self on publish
globalSignaturePolicy: SignaturePolicy.StrictSign // message signing policy
}
}
})
@ -395,7 +397,14 @@ const node = await Libp2p.create({
new DelegatedPeerRouter()
],
},
peerId
peerId,
peerRouting: { // Peer routing configuration
refreshManager: { // Refresh known and connected closest peers
enabled: true, // Should find the closest peers.
interval: 6e5, // Interval for getting the new for closest peers of 10min
bootDelay: 10e3 // Delay for the initial query for closest peers
}
}
})
```
@ -419,6 +428,37 @@ const node = await Libp2p.create({
hop: {
enabled: true, // Allows you to be a relay for other peers
active: true // You will attempt to dial destination peers if you are not connected to them
},
advertise: {
bootDelay: 15 * 60 * 1000, // Delay before HOP relay service is advertised on the network
enabled: true, // Allows you to disable the advertise of the Hop service
ttl: 30 * 60 * 1000 // Delay Between HOP relay service advertisements on the network
}
}
}
})
```
#### Setup with Auto Relay
```js
const Libp2p = require('libp2p')
const TCP = require('libp2p-tcp')
const MPLEX = require('libp2p-mplex')
const { NOISE } = require('libp2p-noise')
const node = await Libp2p.create({
modules: {
transport: [TCP],
streamMuxer: [MPLEX],
connEncryption: [NOISE]
},
config: {
relay: { // Circuit Relay options (this config is part of libp2p core configurations)
enabled: true, // Allows you to dial and accept relayed connections. Does not make you a relay.
autoRelay: {
enabled: true, // Allows you to bind to relays with HOP enabled for improving node dialability
maxListeners: 2 // Configure maximum number of HOP relays to use
}
}
}
@ -466,6 +506,7 @@ Dialing in libp2p can be configured to limit the rate of dialing, and how long d
| maxDialsPerPeer | `number` | How many multiaddrs we can dial per peer, in parallel. |
| dialTimeout | `number` | Second dial timeout per peer in ms. |
| resolvers | `object` | Dial [Resolvers](https://github.com/multiformats/js-multiaddr/blob/master/src/resolvers/index.js) for resolving multiaddrs |
| addressSorter | `(Array<Address>) => Array<Address>` | Sort the known addresses of a peer before trying to dial. |
The below configuration example shows how the dialer should be configured, with the current defaults:
@ -476,6 +517,7 @@ const MPLEX = require('libp2p-mplex')
const { NOISE } = require('libp2p-noise')
const { dnsaddrResolver } = require('multiaddr/src/resolvers')
const { publicAddressesFirst } = require('libp2p-utils/src/address-sort')
const node = await Libp2p.create({
modules: {
@ -489,7 +531,8 @@ const node = await Libp2p.create({
dialTimeout: 30e3,
resolvers: {
dnsaddr: dnsaddrResolver
}
},
addressSorter: publicAddressesFirst
}
```

View File

@ -0,0 +1,185 @@
<!--Specify versions for migration below-->
# Migrating to libp2p@30
A migration guide for refactoring your application code from libp2p v0.29.x to v0.30.0.
## Table of Contents
- [API](#api)
- [Development and Testing](#development-and-testing)
- [Module Updates](#module-updates)
## API
### Pubsub
`js-libp2p` nodes prior to this version were emitting to self on publish by default.
This default value was changed on the pubsub router layer in the past, but we kept it overwritten in libp2p to avoid an upstream breaking change.
Now `js-libp2p` does not overwrite the pubsub router options anymore. Upstream projects that want this feature should enable it on their libp2p configuration.
**Before**
```js
const Gossipsub = require('libp2p-gossipsub')
const Libp2p = require('libp2p')
const libp2p = await Libp2p.create({
modules: {
// ... Add required modules according to the Configuration docs
pubsub: Gossipsub
}
})
```
**After**
```js
const Gossipsub = require('libp2p-gossipsub')
const Libp2p = require('libp2p')
const libp2p = await Libp2p.create({
modules: {
// ... Add required modules according to the Configuration docs
pubsub: Gossipsub
},
config: {
pubsub: {
emitSelf: true
}
}
})
```
The [Pubsub interface](https://github.com/libp2p/js-libp2p-interfaces/tree/master/src/pubsub) was updated on its message signing properties, taking into account the Gossipsub spec updates on [libp2p/specs#294](https://github.com/libp2p/specs/pull/294) and [libp2p/specs#299](https://github.com/libp2p/specs/pull/299)
The signing property is now based on a `globalSignaturePolicy` option instead of the previous `signMessages` and `strictSigning` options. The default to strict signing pubsub messages was kept, but if you would like to disable it, the properties should be changed as follows:
**Before**
```js
const Gossipsub = require('libp2p-gossipsub')
const Libp2p = require('libp2p')
const libp2p = await Libp2p.create({
modules: {
// ... Add required modules according to the Configuration docs
pubsub: Gossipsub
},
config: {
pubsub: {
signMessages: false,
strictSigning: false
}
}
})
```
**After**
```js
const Gossipsub = require('libp2p-gossipsub')
const { SignaturePolicy } = require('libp2p-interfaces/src/pubsub/signature-policy')
const Libp2p = require('libp2p')
const libp2p = await Libp2p.create({
modules: {
// ... Add required modules according to the Configuration docs
pubsub: Gossipsub
},
config: {
pubsub: {
globalSignaturePolicy: SignaturePolicy.StrictNoSign
}
}
})
```
### Addresses
Libp2p has supported `noAnnounce` addresses configuration for some time now. However, it did not provide the best developer experience. In this release, we dropped the `noAnnounce` configuration property in favor of an `announceFilter` property function.
**Before**
```js
const Libp2p = require('libp2p')
const libp2p = await Libp2p.create({
addresses: {
listen: ['/ip4/127.0.0.1/tcp/8000/ws'],
noAnnounce: ['/ip4/127.0.0.1/tcp/8000/ws'],
},
// ... additional configuration per the Configuration docs
})
```
**After**
```js
const Libp2p = require('libp2p')
// Libp2p utils has several multiaddr utils you can leverage
const isPrivate = require('libp2p-utils/src/multiaddr/is-private')
const libp2p = await Libp2p.create({
addresses: {
listen: ['/ip4/127.0.0.1/tcp/8000/ws'],
// Filter function: (ma: Array<multiaddr>) => Array<multiaddr>
announceFilter: (multiaddrs) => multiaddrs.filter(m => !isPrivate(m))
},
// ... additional configuration per the Configuration docs
})
```
It is important pointing out another change regarding address advertising. This is not an API breaking change, but it might have influence on your libp2p setup.
Previously, when using the addresses `announce` property, its multiaddrs were concatenated with the `listen` multiaddrs and then they were filtered out by the `noAnnounce` multiaddrs, in order to create the list of multiaddrs to advertise.
In `libp2p@0.30` the logic now operates as follows:
- If `announce` addresses are provided, only they will be announced (no filters are applied)
- If `announce` is not provided, the transport addresses will be filtered (if a filter is provided)
- if the `announceFilter` is provide it will be passed the transport addresses
## Development and Testing
While this is not an API breaking change, there was a behavioral breaking change on the Websockets transport when in a browser environment. This change might create issues on local test setups.
`libp2p-websockets` has allowed `TCP` and `DNS` addresses, both with `ws` or `wss` to be used for dial purposes. Taking into account security (and browser policies), we are now restricting addresses to `DNS` + `wss` in the browser
With this new behavior, if you need to use non DNS addresses, you can configure your libp2p node as follows:
```js
const Websockets = require('libp2p-websockets')
const filters = require('libp2p-websockets/src/filters')
const Libp2p = require('libp2p')
const transportKey = Websockets.prototype[Symbol.toStringTag]
const libp2p = await Libp2p.create({
modules: {
transport: [Websockets]
// ... Add required modules according to the Configuration docs
},
config: {
transport: {
[transportKey]: {
filter: filters.all
}
}
}
})
```
## Module Updates
With this release you should update the following libp2p modules if you are relying on them:
<!--Specify module versions in JSON for migration below.
It's recommended to check package.json changes for this:
`git diff <release> <prev> -- package.json`
-->
```json
"libp2p-delegated-content-routing": "^0.8.0",
"libp2p-delegated-peer-routing": "^0.8.0",
"libp2p-floodsub": "^0.24.0",
"libp2p-gossipsub": "^0.7.0",
"libp2p-websockets": "^0.15.0",
```
Note that some of them do not need to be updated for this libp2p version to work as expected, but we suggest you to keep them updated as part of this release.

View File

@ -0,0 +1,3 @@
# Delegate Nodes
[TODO](https://github.com/libp2p/js-libp2p/pull/718)

65
doc/production/README.md Normal file
View File

@ -0,0 +1,65 @@
# Production
Nowadays, you can run JavaScript code in several different environments, some of them with their own particularities. Moreover, you can use `js-libp2p` for a wide range of use cases. Different environments and different use cases mean different configurations and challenges in the network.
Libp2p nodes can vary from nodes behind an application, to infrastructure nodes that enable the network to operate and to be efficient. In this context, the Libp2p project provides public infrastructure to boost the network, enable nodes connectivity and improve constrained nodes performance. This public infrastructure should be leveraged for learning the concepts and experimenting. When an application on top of libp2p aims to move into production, its own infrastructure should be setup as the public nodes will be intensively used by others and its availability is not guaranteed.
This guide aims to guide you from using the public infrastructure into setting up your own.
## Table of Contents
* [Joining the Network](#joining-the-network)
* [Connecting to Nodes with connectivity limitations](#connecting-to-nodes-with-connectivity-limitations)
* [`webrtc-star` servers](#webrtc-star-servers)
* [Circuit Relay](#circuit-relay)
* [Querying the network from the browser](#querying-the-network-from-the-browser)
* [Others](#others)
* [SSL](#ssl)
## Joining the Network
Once a libp2p node stars, it will need to connect to a set of peers in order to establish its overlay network.
Currently `js-libp2p` is not the best choice for being a bootstrap node. Its DHT needs to be improved, in order to become an effective server to enable other nodes to properly bootstrap their network.
Setting up a fleet of [`go-libp2p`](https://github.com/libp2p/go-libp2p) nodes is the recommended way to proceed here.
## Connecting to Nodes with connectivity limitations
While the libp2p core codebase aims to work in multiple environments, there are some limitations that are not possible to overcome at the time of writing. These limitations include browser nodes, nodes behind NAT, reverse proxies, firewalls, or lack of compatible transports.
In the browser, libp2p supports two transports: `websockets` and `webrtc-star`. Nowadays, browsers do not support listening for connections, but only to dial known addresses. `webrtc-star` servers can be used to enable libp2p nodes to discover other nodes running on the browser and to help them establish a connection.
For nodes that cannot be dialed (including browser), circuit relay nodes should be used.
### `webrtc-star` servers
Regarding `webRTC` connections, a set of star servers are needed to act as a rendezvous point, where peers can learn about other peers (`peer-discovery`), as well as exchange their SDP offers (signaling data).
You can read on how to setup your own star servers in [libp2p/js-libp2p-webrtc-star/DEPLOYMENT.md](https://github.com/libp2p/js-libp2p-webrtc-star/blob/master/DEPLOYMENT.md).
It is worth pointing out that with new discovery protocols on the way, as well as support for distributed signaling, the star servers should be deprecated on the long run.
### Circuit Relay
Libp2p nodes acting as circuit relay aim to establish connectivity between libp2p nodes (e.g. IPFS nodes) that wouldn't otherwise be able to establish a direct connection to each other.
A relay is needed in situations where nodes are behind NAT, reverse proxies, firewalls and/or simply don't support the same transports (e.g. go-libp2p vs. browser-libp2p). The circuit relay protocol exists to overcome those scenarios. Nodes with the `auto-relay` feature enabled can automatically bind themselves on a relay to listen for connections on their behalf.
You can use [libp2p/js-libp2p-relay-server](https://github.com/libp2p/js-libp2p-relay-server) to setup your own relay server. This also includes an easy to customize Docker setup for a HOP Relay.
## Querying the network from the browser
Libp2p nodes in scenarios such as browser environment and constrained devices will not be an efficient node in the libp2p DHT overlay, as a consequence of their known limitations regarding connectivity and performance.
Aiming to support these type of nodes to find other peers and content in the network, delegate nodes can be setup. With a set of well known IPFS delegate nodes, nodes with limitations in the network can leverage them to perform peer and content routing queries.
Currently, delegate nodes must be IPFS nodes as the IPFS HTTP API is leveraged by them to make routing queries.
You can read on how to setup your own set of delegated nodes in [DELEGATE_NODES.md](./DELEGATE_NODES.md).
## Others
### SSL
TODO

View File

@ -0,0 +1,192 @@
# Auto relay
Auto Relay enables libp2p nodes to dynamically find and bind to relays on the network. Once binding (listening) is done, the node can and should advertise its addresses on the network, allowing any other node to dial it over its bound relay(s).
While direct connections to nodes are preferable, it's not always possible to do so due to NATs or browser limitations.
## 0. Setup the example
Before moving into the examples, you should run `npm install` on the top level `js-libp2p` folder, in order to install all the dependencies needed for this example. Once the install finishes, you should move into the example folder with `cd examples/auto-relay`.
This example comes with 3 main files. A `relay.js` file to be used in the first step, a `listener.js` file to be used in the second step and a `dialer.js` file to be used on the third step. All of these scripts will run their own libp2p node, which will interact with the previous ones. All nodes must be running in order for you to proceed.
## 1. Set up a relay node
In the first step of this example, we need to configure and run a relay node in order for our target node to bind to for accepting inbound connections.
The relay node will need to have its relay subsystem enabled, as well as its HOP capability. It can be configured as follows:
```js
const Libp2p = require('libp2p')
const Websockets = require('libp2p-websockets')
const { NOISE } = require('libp2p-noise')
const MPLEX = require('libp2p-mplex')
const node = await Libp2p.create({
modules: {
transport: [Websockets],
connEncryption: [NOISE],
streamMuxer: [MPLEX]
},
addresses: {
listen: ['/ip4/0.0.0.0/tcp/0/ws']
// TODO check "What is next?" section
// announce: ['/dns4/auto-relay.libp2p.io/tcp/443/wss/p2p/QmWDn2LY8nannvSWJzruUYoLZ4vV83vfCBwd8DipvdgQc3']
},
config: {
relay: {
enabled: true,
hop: {
enabled: true
},
advertise: {
enabled: true,
}
}
}
})
await node.start()
console.log(`Node started with id ${node.peerId.toB58String()}`)
console.log('Listening on:')
node.multiaddrs.forEach((ma) => console.log(`${ma.toString()}/p2p/${node.peerId.toB58String()}`))
```
The Relay HOP advertise functionality is **NOT** required to be enabled. However, if you are interested in advertising on the network that this node is available to be used as a HOP Relay you can enable it. A content router module or Rendezvous needs to be configured to leverage this option.
You should now run the following to start the relay node:
```sh
node relay.js
```
This should print out something similar to the following:
```sh
Node started with id QmWDn2LY8nannvSWJzruUYoLZ4vV83vfCBwd8DipvdgQc3
Listening on:
/ip4/127.0.0.1/tcp/61592/ws/p2p/QmWDn2LY8nannvSWJzruUYoLZ4vV83vfCBwd8DipvdgQc3
/ip4/192.168.1.120/tcp/61592/ws/p2p/QmWDn2LY8nannvSWJzruUYoLZ4vV83vfCBwd8DipvdgQc3
```
## 2. Set up a listener node with Auto Relay Enabled
One of the typical use cases for Auto Relay is nodes behind a NAT or browser nodes due to their inability to expose a public address. For running a libp2p node that automatically binds itself to connected HOP relays, you can see the following:
```js
const Libp2p = require('libp2p')
const Websockets = require('libp2p-websockets')
const { NOISE } = require('libp2p-noise')
const MPLEX = require('libp2p-mplex')
const relayAddr = process.argv[2]
if (!relayAddr) {
throw new Error('the relay address needs to be specified as a parameter')
}
const node = await Libp2p.create({
modules: {
transport: [Websockets],
connEncryption: [NOISE],
streamMuxer: [MPLEX]
},
config: {
relay: {
enabled: true,
autoRelay: {
enabled: true,
maxListeners: 2
}
}
}
})
await node.start()
console.log(`Node started with id ${node.peerId.toB58String()}`)
const conn = await node.dial(relayAddr)
// Wait for connection and relay to be bind for the example purpose
await new Promise((resolve) => {
node.peerStore.on('change:multiaddrs', ({ peerId }) => {
// Updated self multiaddrs?
if (peerId.equals(node.peerId)) {
resolve()
}
})
})
console.log(`Connected to the HOP relay ${conn.remotePeer.toString()}`)
console.log(`Advertising with a relay address of ${node.multiaddrs[0].toString()}/p2p/${node.peerId.toB58String()}`)
```
As you can see in the code, we need to provide the relay address, `relayAddr`, as a process argument. This node will dial the provided relay address and automatically bind to it.
You should now run the following to start the node running Auto Relay:
```sh
node listener.js /ip4/192.168.1.120/tcp/58941/ws/p2p/QmQKCBm87HQMbFqy14oqC85pMmnRrj6iD46ggM6reqNpsd
```
This should print out something similar to the following:
```sh
Node started with id QmerrWofKF358JE6gv3z74cEAyL7z1KqhuUoVfGEynqjRm
Connected to the HOP relay QmWDn2LY8nannvSWJzruUYoLZ4vV83vfCBwd8DipvdgQc3
Advertising with a relay address of /ip4/192.168.1.120/tcp/61592/ws/p2p/QmWDn2LY8nannvSWJzruUYoLZ4vV83vfCBwd8DipvdgQc3/p2p-circuit/p2p/QmerrWofKF358JE6gv3z74cEAyL7z1KqhuUoVfGEynqjRm
```
Per the address, it is possible to verify that the auto relay node is listening on the circuit relay node address.
Instead of dialing this relay manually, you could set up this node with the Bootstrap module and provide it in the bootstrap list. Moreover, you can use other `peer-discovery` modules to discover peers in the network and the node will automatically bind to the relays that support HOP until reaching the maximum number of listeners.
## 3. Set up a dialer node for testing connectivity
Now that you have a relay node and a node bound to that relay, you can test connecting to the auto relay node via the relay.
```js
const Libp2p = require('libp2p')
const Websockets = require('libp2p-websockets')
const { NOISE } = require('libp2p-noise')
const MPLEX = require('libp2p-mplex')
const autoRelayNodeAddr = process.argv[2]
if (!autoRelayNodeAddr) {
throw new Error('the auto relay node address needs to be specified')
}
const node = await Libp2p.create({
modules: {
transport: [Websockets],
connEncryption: [NOISE],
streamMuxer: [MPLEX]
}
})
await node.start()
console.log(`Node started with id ${node.peerId.toB58String()}`)
const conn = await node.dial(autoRelayNodeAddr)
console.log(`Connected to the auto relay node via ${conn.remoteAddr.toString()}`)
```
You should now run the following to start the relay node using the listen address from step 2:
```sh
node dialer.js /ip4/192.168.1.120/tcp/58941/ws/p2p/QmQKCBm87HQMbFqy14oqC85pMmnRrj6iD46ggM6reqNpsd
```
Once you start your test node, it should print out something similar to the following:
```sh
Node started: Qme7iEzDxFoFhhkrsrkHkMnM11aPYjysaehP4NZeUfVMKG
Connected to the auto relay node via /ip4/192.168.1.120/tcp/61592/ws/p2p/QmWDn2LY8nannvSWJzruUYoLZ4vV83vfCBwd8DipvdgQc3/p2p-circuit/p2p/QmerrWofKF358JE6gv3z74cEAyL7z1KqhuUoVfGEynqjRm
```
As you can see from the output, the remote address of the established connection uses the relayed connection.
## 4. What is next?
Before moving into production, there are a few things that you should take into account.
A relay node should not advertise its private address in a real world scenario, as the node would not be reachable by others. You should provide an array of public addresses in the libp2p `addresses.announce` option. If you are using websockets, bear in mind that due to browsers security policies you cannot establish unencrypted connection from secure context. The simplest solution is to setup SSL with nginx and proxy to the node and setup a domain name for the certificate.

View File

@ -0,0 +1,29 @@
'use strict'
const Libp2p = require('libp2p')
const Websockets = require('libp2p-websockets')
const { NOISE } = require('libp2p-noise')
const MPLEX = require('libp2p-mplex')
async function main () {
const autoRelayNodeAddr = process.argv[2]
if (!autoRelayNodeAddr) {
throw new Error('the auto relay node address needs to be specified')
}
const node = await Libp2p.create({
modules: {
transport: [Websockets],
connEncryption: [NOISE],
streamMuxer: [MPLEX]
}
})
await node.start()
console.log(`Node started with id ${node.peerId.toB58String()}`)
const conn = await node.dial(autoRelayNodeAddr)
console.log(`Connected to the auto relay node via ${conn.remoteAddr.toString()}`)
}
main()

View File

@ -0,0 +1,47 @@
'use strict'
const Libp2p = require('libp2p')
const Websockets = require('libp2p-websockets')
const { NOISE } = require('libp2p-noise')
const MPLEX = require('libp2p-mplex')
async function main () {
const relayAddr = process.argv[2]
if (!relayAddr) {
throw new Error('the relay address needs to be specified as a parameter')
}
const node = await Libp2p.create({
modules: {
transport: [Websockets],
connEncryption: [NOISE],
streamMuxer: [MPLEX]
},
config: {
relay: {
enabled: true,
autoRelay: {
enabled: true,
maxListeners: 2
}
}
}
})
await node.start()
console.log(`Node started with id ${node.peerId.toB58String()}`)
const conn = await node.dial(relayAddr)
console.log(`Connected to the HOP relay ${conn.remotePeer.toString()}`)
// Wait for connection and relay to be bind for the example purpose
node.peerStore.on('change:multiaddrs', ({ peerId }) => {
// Updated self multiaddrs?
if (peerId.equals(node.peerId)) {
console.log(`Advertising with a relay address of ${node.multiaddrs[0].toString()}/p2p/${node.peerId.toB58String()}`)
}
})
}
main()

View File

@ -0,0 +1,40 @@
'use strict'
const Libp2p = require('libp2p')
const Websockets = require('libp2p-websockets')
const { NOISE } = require('libp2p-noise')
const MPLEX = require('libp2p-mplex')
async function main () {
const node = await Libp2p.create({
modules: {
transport: [Websockets],
connEncryption: [NOISE],
streamMuxer: [MPLEX]
},
addresses: {
listen: ['/ip4/0.0.0.0/tcp/0/ws']
// TODO check "What is next?" section
// announce: ['/dns4/auto-relay.libp2p.io/tcp/443/wss/p2p/QmWDn2LY8nannvSWJzruUYoLZ4vV83vfCBwd8DipvdgQc3']
},
config: {
relay: {
enabled: true,
hop: {
enabled: true
},
advertise: {
enabled: true,
}
}
}
})
await node.start()
console.log(`Node started with id ${node.peerId.toB58String()}`)
console.log('Listening on:')
node.multiaddrs.forEach((ma) => console.log(`${ma.toString()}/p2p/${node.peerId.toB58String()}`))
}
main()

View File

@ -0,0 +1,94 @@
'use strict'
const path = require('path')
const execa = require('execa')
const pDefer = require('p-defer')
const uint8ArrayToString = require('uint8arrays/to-string')
function startProcess (name, args = []) {
return execa('node', [path.join(__dirname, name), ...args], {
cwd: path.resolve(__dirname),
all: true
})
}
async function test () {
let output1 = ''
let output2 = ''
let output3 = ''
let relayAddr
let autoRelayAddr
const proc1Ready = pDefer()
const proc2Ready = pDefer()
// Step 1 process
process.stdout.write('relay.js\n')
const proc1 = startProcess('relay.js')
proc1.all.on('data', async (data) => {
process.stdout.write(data)
output1 += uint8ArrayToString(data)
if (output1.includes('Listening on:') && output1.includes('/p2p/')) {
relayAddr = output1.trim().split('Listening on:\n')[1].split('\n')[0]
proc1Ready.resolve()
}
})
await proc1Ready.promise
process.stdout.write('==================================================================\n')
// Step 2 process
process.stdout.write('listener.js\n')
const proc2 = startProcess('listener.js', [relayAddr])
proc2.all.on('data', async (data) => {
process.stdout.write(data)
output2 += uint8ArrayToString(data)
if (output2.includes('Advertising with a relay address of') && output2.includes('/p2p/')) {
autoRelayAddr = output2.trim().split('Advertising with a relay address of ')[1]
proc2Ready.resolve()
}
})
await proc2Ready.promise
process.stdout.write('==================================================================\n')
// Step 3 process
process.stdout.write('dialer.js\n')
const proc3 = startProcess('dialer.js', [autoRelayAddr])
proc3.all.on('data', async (data) => {
process.stdout.write(data)
output3 += uint8ArrayToString(data)
if (output3.includes('Connected to the auto relay node via')) {
const remoteAddr = output3.trim().split('Connected to the auto relay node via ')[1]
if (remoteAddr === autoRelayAddr) {
proc3.kill()
proc2.kill()
proc1.kill()
} else {
throw new Error('dialer did not dial through the relay')
}
}
})
await Promise.all([
proc1,
proc2,
proc3
]).catch((err) => {
if (err.signal !== 'SIGTERM') {
throw err
}
})
}
module.exports = test

16
examples/package.json Normal file
View File

@ -0,0 +1,16 @@
{
"name": "libp2p-examples",
"version": "1.0.0",
"description": "Examples of how to use libp2p",
"scripts": {
"test": "node ./test.js",
"test:all": "node ./test-all.js"
},
"license": "MIT",
"dependencies": {
"execa": "^2.1.0",
"fs-extra": "^8.1.0",
"p-defer": "^3.0.0",
"which": "^2.0.1"
}
}

View File

@ -43,6 +43,7 @@ const createNode = async () => {
})
await node1.pubsub.subscribe(topic)
// Will not receive own published messages by default
node2.pubsub.on(topic, (msg) => {
console.log(`node2 received: ${uint8ArrayToString(msg.data)}`)
})

View File

@ -44,7 +44,6 @@ const node2 = nodes[1]
// Add node's 2 data to the PeerStore
node1.peerStore.addressBook.set(node2.peerId, node2.multiaddrs)
await node1.dial(node2.peerId)
node1.pubsub.on(topic, (msg) => {
@ -52,6 +51,7 @@ node1.pubsub.on(topic, (msg) => {
})
await node1.pubsub.subscribe(topic)
// Will not receive own published messages by default
node2.pubsub.on(topic, (msg) => {
console.log(`node2 received: ${uint8ArrayToString(msg.data)}`)
})
@ -68,25 +68,34 @@ The output of the program should look like:
```
> node 1.js
connected to QmWpvkKm6qHLhoxpWrTswY6UMNWDyn8hN265Qp9ZYvgS82
node2 received: Bird bird bird, bird is the word!
node1 received: Bird bird bird, bird is the word!
node2 received: Bird bird bird, bird is the word!
node1 received: Bird bird bird, bird is the word!
```
You can change the pubsub `emitSelf` option if you don't want the publishing node to receive its own messages.
You can change the pubsub `emitSelf` option if you want the publishing node to receive its own messages.
```JavaScript
const defaults = {
config: {
pubsub: {
enabled: true,
emitSelf: false
emitSelf: true
}
}
}
```
The output of the program should look like:
```
> node 1.js
connected to QmWpvkKm6qHLhoxpWrTswY6UMNWDyn8hN265Qp9ZYvgS82
node1 received: Bird bird bird, bird is the word!
node2 received: Bird bird bird, bird is the word!
node1 received: Bird bird bird, bird is the word!
node2 received: Bird bird bird, bird is the word!
```
## 2. Future work
libp2p/IPFS PubSub is enabling a whole set of Distributed Real Time applications using CRDT (Conflict-Free Replicated Data Types). It is still going through heavy research (and hacking) and we invite you to join the conversation at [research-CRDT](https://github.com/ipfs/research-CRDT). Here is a list of some of the exciting examples:

View File

@ -44,6 +44,7 @@ const createNode = async () => {
//subscribe
node1.pubsub.on(topic, (msg) => {
// Will not receive own published messages by default
console.log(`node1 received: ${uint8ArrayToString(msg.data)}`)
})
await node1.pubsub.subscribe(topic)

View File

@ -97,15 +97,12 @@ Result
```
> node 1.js
############## fruit banana ##############
node1 received: banana
node2 received: banana
node3 received: banana
############## fruit apple ##############
node1 received: apple
node2 received: apple
node3 received: apple
############## fruit car ##############
node1 received: car
############## fruit orange ##############
node1 received: orange
node2 received: orange

33
examples/test-all.js Normal file
View File

@ -0,0 +1,33 @@
'use strict'
process.on('unhandedRejection', (err) => {
console.error(err)
process.exit(1)
})
const path = require('path')
const fs = require('fs')
const {
waitForOutput
} = require('./utils')
async function testAll () {
for (const dir of fs.readdirSync(__dirname)) {
if (dir === 'node_modules' || dir === 'tests_output') {
continue
}
const stats = fs.statSync(path.join(__dirname, dir))
if (!stats.isDirectory()) {
continue
}
await waitForOutput('npm info ok', 'npm', ['test', '--', dir], {
cwd: __dirname
})
}
}
testAll()

95
examples/test.js Normal file
View File

@ -0,0 +1,95 @@
'use strict'
process.env.NODE_ENV = 'test'
process.env.CI = true // needed for some "clever" build tools
const fs = require('fs-extra')
const path = require('path')
const execa = require('execa')
const dir = path.join(__dirname, process.argv[2])
testExample(dir)
.then(() => {}, (err) => {
if (err.exitCode) {
process.exit(err.exitCode)
}
console.error(err)
process.exit(1)
})
async function testExample (dir) {
await installDeps(dir)
await build(dir)
await runTest(dir)
// TODO: add browser test setup
}
async function installDeps (dir) {
if (!fs.existsSync(path.join(dir, 'package.json'))) {
console.info('Nothing to install in', dir)
return
}
if (fs.existsSync(path.join(dir, 'node_modules'))) {
console.info('Dependencies already installed in', dir)
return
}
const proc = execa.command('npm install', {
cwd: dir
})
proc.all.on('data', (data) => {
process.stdout.write(data)
})
await proc
}
async function build (dir) {
const pkgJson = path.join(dir, 'package.json')
if (!fs.existsSync(pkgJson)) {
console.info('Nothing to build in', dir)
return
}
const pkg = require(pkgJson)
let build
if (pkg.scripts.bundle) {
build = 'bundle'
}
if (pkg.scripts.build) {
build = 'build'
}
if (!build) {
console.info('No "build" or "bundle" script in', pkgJson)
return
}
const proc = execa('npm', ['run', build], {
cwd: dir
})
proc.all.on('data', (data) => {
process.stdout.write(data)
})
await proc
}
async function runTest (dir) {
console.info('Running node tests in', dir)
const testFile = path.join(dir, 'test.js')
if (!fs.existsSync(testFile)) {
console.info('Nothing to test in', dir)
return
}
const runTest = require(testFile)
await runTest()
}

61
examples/utils.js Normal file
View File

@ -0,0 +1,61 @@
'use strict'
const execa = require('execa')
const fs = require('fs-extra')
const which = require('which')
async function isExecutable (command) {
try {
await fs.access(command, fs.constants.X_OK)
return true
} catch (err) {
if (err.code === 'ENOENT') {
return isExecutable(await which(command))
}
if (err.code === 'EACCES') {
return false
}
throw err
}
}
async function waitForOutput (expectedOutput, command, args = [], opts = {}) {
if (!await isExecutable(command)) {
args.unshift(command)
command = 'node'
}
const proc = execa(command, args, opts)
let output = ''
let time = 120000
let timeout = setTimeout(() => {
throw new Error(`Did not see "${expectedOutput}" in output from "${[command].concat(args).join(' ')}" after ${time/1000}s`)
}, time)
proc.all.on('data', (data) => {
process.stdout.write(data)
output += data.toString('utf8')
if (output.includes(expectedOutput)) {
clearTimeout(timeout)
proc.kill()
}
})
try {
await proc
} catch (err) {
if (!err.killed) {
throw err
}
}
}
module.exports = {
waitForOutput
}

View File

@ -23,7 +23,6 @@
"secure channels",
["NodeFactoryIo/js-libp2p-noise", "libp2p-noise"],
["libp2p/js-libp2p-secio", "libp2p-secio"],
"stream multiplexers",
["libp2p/js-libp2p-mplex", "libp2p-mplex"],

View File

@ -1,9 +1,18 @@
{
"name": "libp2p",
"version": "0.29.4",
"version": "0.30.0-rc.1",
"description": "JavaScript implementation of libp2p, a modular peer to peer network stack",
"leadMaintainer": "Jacob Heun <jacobheun@gmail.com>",
"main": "src/index.js",
"types": "dist/src/index.d.ts",
"typesVersions": {
"*": {
"src/*": [
"dist/src/*",
"dist/src/*/index"
]
}
},
"files": [
"dist",
"src"
@ -14,6 +23,7 @@
"test": "npm run test:node && npm run test:browser",
"test:node": "aegir test -t node -f \"./test/**/*.{node,spec}.js\"",
"test:browser": "aegir test -t browser",
"test:examples": "cd examples && npm run test:all",
"release": "aegir release -t node -t browser",
"release-minor": "aegir release --type minor -t node -t browser",
"release-major": "aegir release --type major -t node -t browser",
@ -45,13 +55,14 @@
"aggregate-error": "^3.0.1",
"any-signal": "^1.1.0",
"bignumber.js": "^9.0.0",
"cids": "^1.0.0",
"class-is": "^1.1.0",
"debug": "^4.1.1",
"err-code": "^2.0.0",
"events": "^3.1.0",
"hashlru": "^2.3.0",
"interface-datastore": "^2.0.0",
"ipfs-utils": "^2.2.0",
"ipfs-utils": "^5.0.1",
"it-all": "^1.0.1",
"it-buffer": "^0.1.2",
"it-handshake": "^1.0.1",
@ -59,13 +70,14 @@
"it-pipe": "^1.1.0",
"it-protocol-buffers": "^0.2.0",
"libp2p-crypto": "^0.18.0",
"libp2p-interfaces": "^0.5.1",
"libp2p-utils": "^0.2.0",
"libp2p-interfaces": "^0.8.1",
"libp2p-utils": "^0.2.2",
"mafmt": "^8.0.0",
"merge-options": "^2.0.0",
"moving-average": "^1.0.0",
"multiaddr": "^8.1.0",
"multicodec": "^2.0.0",
"multihashing-async": "^2.0.1",
"multistream-select": "^1.0.0",
"mutable-proxy": "^1.0.0",
"node-forge": "^0.9.1",
@ -76,6 +88,7 @@
"protons": "^2.0.0",
"retimer": "^2.0.0",
"sanitize-filename": "^1.6.3",
"set-delayed-interval": "^1.0.0",
"streaming-iterables": "^5.0.2",
"timeout-abort-controller": "^1.1.1",
"varint": "^5.0.0",
@ -84,25 +97,22 @@
"devDependencies": {
"@nodeutils/defaults-deep": "^1.1.0",
"abortable-iterator": "^3.0.0",
"aegir": "^27.0.0",
"chai": "^4.2.0",
"chai-as-promised": "^7.1.1",
"aegir": "^29.2.0",
"chai-bytes": "^0.1.2",
"chai-string": "^1.5.0",
"cids": "^1.0.0",
"delay": "^4.3.0",
"dirty-chai": "^2.0.1",
"interop-libp2p": "^0.3.0",
"into-stream": "^6.0.0",
"ipfs-http-client": "^47.0.1",
"it-concat": "^1.0.0",
"it-pair": "^1.0.0",
"it-pushable": "^1.4.0",
"libp2p": ".",
"libp2p-bootstrap": "^0.12.0",
"libp2p-delegated-content-routing": "^0.7.0",
"libp2p-delegated-peer-routing": "^0.7.0",
"libp2p-floodsub": "^0.23.0",
"libp2p-gossipsub": "^0.6.0",
"libp2p-delegated-content-routing": "^0.8.0",
"libp2p-delegated-peer-routing": "^0.8.0",
"libp2p-floodsub": "^0.24.0",
"libp2p-gossipsub": "^0.7.0",
"libp2p-kad-dht": "^0.20.0",
"libp2p-mdns": "^0.15.0",
"libp2p-mplex": "^0.10.1",
@ -110,7 +120,7 @@
"libp2p-secio": "^0.13.1",
"libp2p-tcp": "^0.15.1",
"libp2p-webrtc-star": "^0.20.0",
"libp2p-websockets": "^0.14.0",
"libp2p-websockets": "^0.15.0",
"multihashes": "^3.0.1",
"nock": "^13.0.3",
"p-defer": "^3.0.0",

View File

@ -1,6 +1,6 @@
# Address Manager
The Address manager is responsible for keeping an updated register of the peer's addresses. It includes 3 different types of Addresses: `Listen Addresses`, `Announce Addresses` and `No Announce Addresses`.
The Address manager is responsible for keeping an updated register of the peer's addresses. It includes 2 different types of Addresses: `Listen Addresses` and `Announce Addresses`.
These Addresses should be specified in your libp2p [configuration](../../doc/CONFIGURATION.md) when you create your node.
@ -20,17 +20,11 @@ Scenarios for Announce Addresses include:
- when you setup a libp2p node in your private network at home, but you need to announce your public IP Address to the outside world;
- when you want to announce a DNS address, which maps to your public IP Address.
## No Announce Addresses
While we need to add Announce Addresses to enable peers' connectivity, we should also avoid announcing addresses that will not be reachable. No Announce Addresses should be specified so that they are filtered from the advertised multiaddrs.
As stated in the Listen Addresses section, Listen Addresses might be modified by libp2p transports after the successfully bind to those addresses. Libp2p should also take these changes into account so that they can be matched when No Announce Addresses are being filtered out of the advertised multiaddrs.
## Implementation
When a libp2p node is created, the Address Manager will be populated from the provided addresses through the libp2p configuration. Once the node is started, the Transport Manager component will gather the listen addresses from the Address Manager, so that the libp2p transports can attempt to bind to them.
Libp2p will use the the Address Manager as the source of truth when advertising the peers addresses. After all transports are ready, other libp2p components/subsystems will kickoff, namely the Identify Service and the DHT. Both of them will announce the node addresses to the other peers in the network. The announce and noAnnounce addresses will have an important role here and will be gathered by libp2p to compute its current addresses to advertise everytime it is needed.
Libp2p will use the the Address Manager as the source of truth when advertising the peers addresses. After all transports are ready, other libp2p components/subsystems will kickoff, namely the Identify Service and the DHT. Both of them will announce the node addresses to the other peers in the network. The announce addresses will have an important role here and will be gathered by libp2p to compute its current addresses to advertise everytime it is needed.
## Future Considerations

View File

@ -1,36 +1,35 @@
'use strict'
const debug = require('debug')
const log = debug('libp2p:addresses')
log.error = debug('libp2p:addresses:error')
const multiaddr = require('multiaddr')
/**
* Responsible for managing this peers addresses.
* Peers can specify their listen, announce and noAnnounce addresses.
* The listen addresses will be used by the libp2p transports to listen for new connections,
* while the announce an noAnnounce addresses will be combined with the listen addresses for
* address adverstising to other peers in the network.
* @typedef {import('multiaddr')} Multiaddr
*/
/**
* @typedef {Object} AddressManagerOptions
* @property {string[]} [listen = []] - list of multiaddrs string representation to listen.
* @property {string[]} [announce = []] - list of multiaddrs string representation to announce.
*/
class AddressManager {
/**
* Responsible for managing the peer addresses.
* Peers can specify their listen and announce addresses.
* The listen addresses will be used by the libp2p transports to listen for new connections,
* while the announce addresses will be used for the peer addresses' to other peers in the network.
*
* @class
* @param {object} [options]
* @param {Array<string>} [options.listen = []] - list of multiaddrs string representation to listen.
* @param {Array<string>} [options.announce = []] - list of multiaddrs string representation to announce.
* @param {Array<string>} [options.noAnnounce = []] - list of multiaddrs string representation to not announce.
* @param {AddressManagerOptions} [options]
*/
constructor ({ listen = [], announce = [], noAnnounce = [] } = {}) {
constructor ({ listen = [], announce = [] } = {}) {
this.listen = new Set(listen)
this.announce = new Set(announce)
this.noAnnounce = new Set(noAnnounce)
}
/**
* Get peer listen multiaddrs.
*
* @returns {Array<Multiaddr>}
* @returns {Multiaddr[]}
*/
getListenAddrs () {
return Array.from(this.listen).map((a) => multiaddr(a))
@ -39,20 +38,11 @@ class AddressManager {
/**
* Get peer announcing multiaddrs.
*
* @returns {Array<Multiaddr>}
* @returns {Multiaddr[]}
*/
getAnnounceAddrs () {
return Array.from(this.announce).map((a) => multiaddr(a))
}
/**
* Get peer noAnnouncing multiaddrs.
*
* @returns {Array<Multiaddr>}
*/
getNoAnnounceAddrs () {
return Array.from(this.noAnnounce).map((a) => multiaddr(a))
}
}
module.exports = AddressManager

268
src/circuit/auto-relay.js Normal file
View File

@ -0,0 +1,268 @@
'use strict'
const debug = require('debug')
const log = Object.assign(debug('libp2p:auto-relay'), {
error: debug('libp2p:auto-relay:err')
})
const uint8ArrayFromString = require('uint8arrays/from-string')
const uint8ArrayToString = require('uint8arrays/to-string')
const multiaddr = require('multiaddr')
const PeerId = require('peer-id')
const { relay: multicodec } = require('./multicodec')
const { canHop } = require('./circuit/hop')
const { namespaceToCid } = require('./utils')
const {
CIRCUIT_PROTO_CODE,
HOP_METADATA_KEY,
HOP_METADATA_VALUE,
RELAY_RENDEZVOUS_NS
} = require('./constants')
/**
* @typedef {import('libp2p-interfaces/src/connection').Connection} Connection
* @typedef {import('../peer-store/address-book').Address} Address
*/
/**
* @typedef {Object} AutoRelayProperties
* @property {import('../')} libp2p
*
* @typedef {Object} AutoRelayOptions
* @property {number} [maxListeners = 1] - maximum number of relays to listen.
*/
class AutoRelay {
/**
* Creates an instance of AutoRelay.
*
* @class
* @param {AutoRelayProperties & AutoRelayOptions} props
*/
constructor ({ libp2p, maxListeners = 1 }) {
this._libp2p = libp2p
this._peerId = libp2p.peerId
this._peerStore = libp2p.peerStore
this._connectionManager = libp2p.connectionManager
this._transportManager = libp2p.transportManager
this._addressSorter = libp2p.dialer.addressSorter
this.maxListeners = maxListeners
/**
* @type {Set<string>}
*/
this._listenRelays = new Set()
this._onProtocolChange = this._onProtocolChange.bind(this)
this._onPeerDisconnected = this._onPeerDisconnected.bind(this)
this._peerStore.on('change:protocols', this._onProtocolChange)
this._connectionManager.on('peer:disconnect', this._onPeerDisconnected)
}
/**
* Check if a peer supports the relay protocol.
* If the protocol is not supported, check if it was supported before and remove it as a listen relay.
* If the protocol is supported, check if the peer supports **HOP** and add it as a listener if
* inside the threshold.
*
* @param {Object} props
* @param {PeerId} props.peerId
* @param {string[]} props.protocols
* @returns {Promise<void>}
*/
async _onProtocolChange ({ peerId, protocols }) {
const id = peerId.toB58String()
// Check if it has the protocol
const hasProtocol = protocols.find(protocol => protocol === multicodec)
// If no protocol, check if we were keeping the peer before as a listenRelay
if (!hasProtocol && this._listenRelays.has(id)) {
this._removeListenRelay(id)
return
} else if (!hasProtocol || this._listenRelays.has(id)) {
return
}
// If protocol, check if can hop, store info in the metadataBook and listen on it
try {
const connection = this._connectionManager.get(peerId)
if (!connection) {
return
}
// Do not hop on a relayed connection
if (connection.remoteAddr.protoCodes().includes(CIRCUIT_PROTO_CODE)) {
log(`relayed connection to ${id} will not be used to hop on`)
return
}
const supportsHop = await canHop({ connection })
if (supportsHop) {
this._peerStore.metadataBook.set(peerId, HOP_METADATA_KEY, uint8ArrayFromString(HOP_METADATA_VALUE))
await this._addListenRelay(connection, id)
}
} catch (err) {
log.error(err)
}
}
/**
* Peer disconnects.
*
* @param {Connection} connection - connection to the peer
* @returns {void}
*/
_onPeerDisconnected (connection) {
const peerId = connection.remotePeer
const id = peerId.toB58String()
// Not listening on this relay
if (!this._listenRelays.has(id)) {
return
}
this._removeListenRelay(id)
}
/**
* Attempt to listen on the given relay connection.
*
* @private
* @param {Connection} connection - connection to the peer
* @param {string} id - peer identifier string
* @returns {Promise<void>}
*/
async _addListenRelay (connection, id) {
// Check if already listening on enough relays
if (this._listenRelays.size >= this.maxListeners) {
return
}
// Get peer known addresses and sort them per public addresses first
const remoteAddrs = this._peerStore.addressBook.getMultiaddrsForPeer(
connection.remotePeer, this._addressSorter
)
if (!remoteAddrs || !remoteAddrs.length) {
return
}
const listenAddr = `${remoteAddrs[0].toString()}/p2p-circuit`
this._listenRelays.add(id)
// Attempt to listen on relay
try {
await this._transportManager.listen([multiaddr(listenAddr)])
// Announce multiaddrs will update on listen success by TransportManager event being triggered
} catch (err) {
log.error(err)
this._listenRelays.delete(id)
}
}
/**
* Remove listen relay.
*
* @private
* @param {string} id - peer identifier string.
* @returns {void}
*/
_removeListenRelay (id) {
if (this._listenRelays.delete(id)) {
// TODO: this should be responsibility of the connMgr
this._listenOnAvailableHopRelays([id])
}
}
/**
* Try to listen on available hop relay connections.
* The following order will happen while we do not have enough relays.
* 1. Check the metadata store for known relays, try to listen on the ones we are already connected.
* 2. Dial and try to listen on the peers we know that support hop but are not connected.
* 3. Search the network.
*
* @param {string[]} [peersToIgnore]
* @returns {Promise<void>}
*/
async _listenOnAvailableHopRelays (peersToIgnore = []) {
// TODO: The peer redial issue on disconnect should be handled by connection gating
// Check if already listening on enough relays
if (this._listenRelays.size >= this.maxListeners) {
return
}
const knownHopsToDial = []
// Check if we have known hop peers to use and attempt to listen on the already connected
for (const [id, metadataMap] of this._peerStore.metadataBook.data.entries()) {
// Continue to next if listening on this or peer to ignore
if (this._listenRelays.has(id) || peersToIgnore.includes(id)) {
continue
}
const supportsHop = metadataMap.get(HOP_METADATA_KEY)
// Continue to next if it does not support Hop
if (!supportsHop || uint8ArrayToString(supportsHop) !== HOP_METADATA_VALUE) {
continue
}
const peerId = PeerId.createFromCID(id)
const connection = this._connectionManager.get(peerId)
// If not connected, store for possible later use.
if (!connection) {
knownHopsToDial.push(peerId)
continue
}
await this._addListenRelay(connection, id)
// Check if already listening on enough relays
if (this._listenRelays.size >= this.maxListeners) {
return
}
}
// Try to listen on known peers that are not connected
for (const peerId of knownHopsToDial) {
const connection = await this._libp2p.dial(peerId)
await this._addListenRelay(connection, peerId.toB58String())
// Check if already listening on enough relays
if (this._listenRelays.size >= this.maxListeners) {
return
}
}
// Try to find relays to hop on the network
try {
const cid = await namespaceToCid(RELAY_RENDEZVOUS_NS)
for await (const provider of this._libp2p.contentRouting.findProviders(cid)) {
if (!provider.multiaddrs.length) {
continue
}
const peerId = provider.id
this._peerStore.addressBook.add(peerId, provider.multiaddrs)
const connection = await this._libp2p.dial(peerId)
await this._addListenRelay(connection, peerId.toB58String())
// Check if already listening on enough relays
if (this._listenRelays.size >= this.maxListeners) {
return
}
}
} catch (err) {
log.error(err)
}
}
}
module.exports = AutoRelay

View File

@ -1,22 +1,42 @@
'use strict'
const debug = require('debug')
const log = debug('libp2p:circuit:hop')
log.error = debug('libp2p:circuit:hop:error')
const log = Object.assign(debug('libp2p:circuit:hop'), {
error: debug('libp2p:circuit:hop:err')
})
const errCode = require('err-code')
const PeerId = require('peer-id')
const { validateAddrs } = require('./utils')
const StreamHandler = require('./stream-handler')
const { CircuitRelay: CircuitPB } = require('../protocol')
const pipe = require('it-pipe')
const errCode = require('err-code')
const { pipe } = require('it-pipe')
const { codes: Errors } = require('../../errors')
const { stop } = require('./stop')
const multicodec = require('./../multicodec')
module.exports.handleHop = async function handleHop ({
/**
* @typedef {import('../../types').CircuitRequest} CircuitRequest
* @typedef {import('libp2p-interfaces/src/connection').Connection} Connection
* @typedef {import('./stream-handler')<CircuitRequest>} StreamHandlerT
* @typedef {import('../transport')} Transport
*/
/**
* @typedef {Object} HopRequest
* @property {Connection} connection
* @property {CircuitRequest} request
* @property {StreamHandlerT} streamHandler
* @property {Transport} circuit
*/
/**
* @param {HopRequest} options
* @returns {Promise<void>}
*/
async function handleHop ({
connection,
request,
streamHandler,
@ -51,6 +71,9 @@ module.exports.handleHop = async function handleHop ({
}
// TODO: Handle being an active relay
if (!destinationConnection) {
return
}
// Handle the incoming HOP request by performing a STOP request
const stopRequest = {
@ -63,8 +86,7 @@ module.exports.handleHop = async function handleHop ({
try {
destinationStream = await stop({
connection: destinationConnection,
request: stopRequest,
circuit
request: stopRequest
})
} catch (err) {
return log.error(err)
@ -91,10 +113,10 @@ module.exports.handleHop = async function handleHop ({
*
* @param {object} options
* @param {Connection} options.connection - Connection to the relay
* @param {*} options.request
* @param {CircuitRequest} options.request
* @returns {Promise<Connection>}
*/
module.exports.hop = async function hop ({
async function hop ({
connection,
request
}) {
@ -116,16 +138,44 @@ module.exports.hop = async function hop ({
throw errCode(new Error(`HOP request failed with code ${response.code}`), Errors.ERR_HOP_REQUEST_FAILED)
}
/**
* Performs a CAN_HOP request to a relay peer, in order to understand its capabilities.
*
* @param {object} options
* @param {Connection} options.connection - Connection to the relay
* @returns {Promise<boolean>}
*/
async function canHop ({
connection
}) {
// Create a new stream to the relay
const { stream } = await connection.newStream([multicodec.relay])
// Send the HOP request
const streamHandler = new StreamHandler({ stream })
streamHandler.write({
type: CircuitPB.Type.CAN_HOP
})
const response = await streamHandler.read()
await streamHandler.close()
if (response.code !== CircuitPB.Status.SUCCESS) {
return false
}
return true
}
/**
* Creates an unencoded CAN_HOP response based on the Circuits configuration
*
* @param {Object} options
* @param {Connection} options.connection
* @param {StreamHandler} options.streamHandler
* @param {Circuit} options.circuit
* @param {StreamHandlerT} options.streamHandler
* @param {Transport} options.circuit
* @private
*/
module.exports.handleCanHop = function handleCanHop ({
function handleCanHop ({
connection,
streamHandler,
circuit
@ -137,3 +187,10 @@ module.exports.handleCanHop = function handleCanHop ({
code: canHop ? CircuitPB.Status.SUCCESS : CircuitPB.Status.HOP_CANT_SPEAK_RELAY
})
}
module.exports = {
handleHop,
hop,
canHop,
handleCanHop
}

View File

@ -1,23 +1,31 @@
'use strict'
const debug = require('debug')
const log = Object.assign(debug('libp2p:circuit:stop'), {
error: debug('libp2p:circuit:stop:err')
})
const { CircuitRelay: CircuitPB } = require('../protocol')
const multicodec = require('../multicodec')
const StreamHandler = require('./stream-handler')
const { validateAddrs } = require('./utils')
const debug = require('debug')
const log = debug('libp2p:circuit:stop')
log.error = debug('libp2p:circuit:stop:error')
/**
* @typedef {import('libp2p-interfaces/src/connection').Connection} Connection
* @typedef {import('libp2p-interfaces/src/stream-muxer/types').MuxedStream} MuxedStream
* @typedef {import('../../types').CircuitRequest} CircuitRequest
* @typedef {import('./stream-handler')<CircuitRequest>} StreamHandlerT
*/
/**
* Handles incoming STOP requests
*
* @private
* @param {*} options
* @param {Object} options
* @param {Connection} options.connection
* @param {*} options.request - The CircuitRelay protobuf request (unencoded)
* @param {StreamHandler} options.streamHandler
* @returns {Promise<*>} Resolves a duplex iterable
* @param {CircuitRequest} options.request - The CircuitRelay protobuf request (unencoded)
* @param {StreamHandlerT} options.streamHandler
* @returns {Promise<MuxedStream>|void} Resolves a duplex iterable
*/
module.exports.handleStop = function handleStop ({
connection,
@ -44,10 +52,10 @@ module.exports.handleStop = function handleStop ({
* Creates a STOP request
*
* @private
* @param {*} options
* @param {Object} options
* @param {Connection} options.connection
* @param {*} options.request - The CircuitRelay protobuf request (unencoded)
* @returns {Promise<*>} Resolves a duplex iterable
* @param {CircuitRequest} options.request - The CircuitRelay protobuf request (unencoded)
* @returns {Promise<MuxedStream|void>} Resolves a duplex iterable
*/
module.exports.stop = async function stop ({
connection,

View File

@ -1,20 +1,29 @@
'use strict'
const debug = require('debug')
const log = Object.assign(debug('libp2p:circuit:stream-handler'), {
error: debug('libp2p:circuit:stream-handler:err')
})
const lp = require('it-length-prefixed')
const handshake = require('it-handshake')
const { CircuitRelay: CircuitPB } = require('../protocol')
const debug = require('debug')
const log = debug('libp2p:circuit:stream-handler')
log.error = debug('libp2p:circuit:stream-handler:error')
/**
* @typedef {import('libp2p-interfaces/src/stream-muxer/types').MuxedStream} MuxedStream
*/
/**
* @template T
*/
class StreamHandler {
/**
* Create a stream handler for connection
*
* @class
* @param {object} options
* @param {*} options.stream - A duplex iterable
* @param {number} options.maxLength - max bytes length of message
* @param {MuxedStream} options.stream - A duplex iterable
* @param {number} [options.maxLength = 4096] - max bytes length of message
*/
constructor ({ stream, maxLength = 4096 }) {
this.stream = stream
@ -27,7 +36,7 @@ class StreamHandler {
* Read and decode message
*
* @async
* @returns {void}
* @returns {Promise<T|undefined>}
*/
async read () {
const msg = await this.decoder.next()
@ -45,10 +54,12 @@ class StreamHandler {
/**
* Encode and write array of buffers
*
* @param {*} msg - An unencoded CircuitRelay protobuf message
* @param {CircuitPB} msg - An unencoded CircuitRelay protobuf message
* @returns {void}
*/
write (msg) {
log('write message type %s', msg.type)
// @ts-ignore lp.encode expects type type 'Buffer | BufferList', not 'Uint8Array'
this.shake.write(lp.encode.single(CircuitPB.encode(msg)))
}

View File

@ -3,11 +3,16 @@
const multiaddr = require('multiaddr')
const { CircuitRelay } = require('../protocol')
/**
* @typedef {import('./stream-handler')} StreamHandler
* @typedef {import('../../types').CircuitStatus} CircuitStatus
*/
/**
* Write a response
*
* @param {StreamHandler} streamHandler
* @param {CircuitRelay.Status} status
* @param {CircuitStatus} status
*/
function writeResponse (streamHandler, status) {
streamHandler.write({

12
src/circuit/constants.js Normal file
View File

@ -0,0 +1,12 @@
'use strict'
const minute = 60 * 1000
module.exports = {
ADVERTISE_BOOT_DELAY: 15 * minute, // Delay before HOP relay service is advertised on the network
ADVERTISE_TTL: 30 * minute, // Delay Between HOP relay service advertisements on the network
CIRCUIT_PROTO_CODE: 290, // Multicodec code
HOP_METADATA_KEY: 'hop_relay', // PeerStore metadaBook key for HOP relay service
HOP_METADATA_VALUE: 'true', // PeerStore metadaBook value for HOP relay service
RELAY_RENDEZVOUS_NS: '/libp2p/relay' // Relay HOP relay service namespace for discovery
}

View File

@ -1,187 +1,109 @@
'use strict'
const mafmt = require('mafmt')
const multiaddr = require('multiaddr')
const PeerId = require('peer-id')
const withIs = require('class-is')
const { CircuitRelay: CircuitPB } = require('./protocol')
const debug = require('debug')
const log = debug('libp2p:circuit')
log.error = debug('libp2p:circuit:error')
const toConnection = require('libp2p-utils/src/stream-to-ma-conn')
const log = Object.assign(debug('libp2p:relay'), {
error: debug('libp2p:relay:err')
})
const { relay: multicodec } = require('./multicodec')
const createListener = require('./listener')
const { handleCanHop, handleHop, hop } = require('./circuit/hop')
const { handleStop } = require('./circuit/stop')
const StreamHandler = require('./circuit/stream-handler')
const {
setDelayedInterval,
clearDelayedInterval
} = require('set-delayed-interval')
class Circuit {
const AutoRelay = require('./auto-relay')
const { namespaceToCid } = require('./utils')
const {
ADVERTISE_BOOT_DELAY,
ADVERTISE_TTL,
RELAY_RENDEZVOUS_NS
} = require('./constants')
/**
* @typedef {import('../')} Libp2p
*
* @typedef {Object} RelayAdvertiseOptions
* @property {number} [bootDelay = ADVERTISE_BOOT_DELAY]
* @property {boolean} [enabled = true]
* @property {number} [ttl = ADVERTISE_TTL]
*
* @typedef {Object} HopOptions
* @property {boolean} [enabled = false]
* @property {boolean} [active = false]
*
* @typedef {Object} AutoRelayOptions
* @property {number} [maxListeners = 2] - maximum number of relays to listen.
* @property {boolean} [enabled = false]
*/
class Relay {
/**
* Creates an instance of Circuit.
* Creates an instance of Relay.
*
* @class
* @param {object} options
* @param {Libp2p} options.libp2p
* @param {Upgrader} options.upgrader
* @param {Libp2p} libp2p
*/
constructor ({ libp2p, upgrader }) {
this._dialer = libp2p.dialer
this._registrar = libp2p.registrar
this._connectionManager = libp2p.connectionManager
this._upgrader = upgrader
this._options = libp2p._config.relay
constructor (libp2p) {
this._libp2p = libp2p
this.peerId = libp2p.peerId
this._registrar.handle(multicodec, this._onProtocol.bind(this))
}
async _onProtocol ({ connection, stream, protocol }) {
const streamHandler = new StreamHandler({ stream })
const request = await streamHandler.read()
const circuit = this
let virtualConnection
switch (request.type) {
case CircuitPB.Type.CAN_HOP: {
log('received CAN_HOP request from %s', connection.remotePeer.toB58String())
await handleCanHop({ circuit, connection, streamHandler })
break
}
case CircuitPB.Type.HOP: {
log('received HOP request from %s', connection.remotePeer.toB58String())
virtualConnection = await handleHop({
connection,
request,
streamHandler,
circuit
})
break
}
case CircuitPB.Type.STOP: {
log('received STOP request from %s', connection.remotePeer.toB58String())
virtualConnection = await handleStop({
connection,
request,
streamHandler,
circuit
})
break
}
default: {
log('Request of type %s not supported', request.type)
}
this._options = {
advertise: {
bootDelay: ADVERTISE_BOOT_DELAY,
enabled: true,
ttl: ADVERTISE_TTL,
...libp2p._config.relay.advertise
},
...libp2p._config.relay
}
if (virtualConnection) {
const remoteAddr = multiaddr(request.dstPeer.addrs[0])
const localAddr = multiaddr(request.srcPeer.addrs[0])
const maConn = toConnection({
stream: virtualConnection,
remoteAddr,
localAddr
})
const type = CircuitPB.Type === CircuitPB.Type.HOP ? 'relay' : 'inbound'
log('new %s connection %s', type, maConn.remoteAddr)
// Create autoRelay if enabled
this._autoRelay = this._options.autoRelay.enabled && new AutoRelay({ libp2p, ...this._options.autoRelay })
const conn = await this._upgrader.upgradeInbound(maConn)
log('%s connection %s upgraded', type, maConn.remoteAddr)
this.handler && this.handler(conn)
this._advertiseService = this._advertiseService.bind(this)
}
/**
* Start Relay service.
*
* @returns {void}
*/
start () {
// Advertise service if HOP enabled
const canHop = this._options.hop.enabled
if (canHop && this._options.advertise.enabled) {
this._timeout = setDelayedInterval(
this._advertiseService, this._options.advertise.ttl, this._options.advertise.bootDelay
)
}
}
/**
* Dial a peer over a relay
* Stop Relay service.
*
* @param {multiaddr} ma - the multiaddr of the peer to dial
* @param {Object} options - dial options
* @param {AbortSignal} [options.signal] - An optional abort signal
* @returns {Connection} - the connection
* @returns {void}
*/
async dial (ma, options) {
// Check the multiaddr to see if it contains a relay and a destination peer
const addrs = ma.toString().split('/p2p-circuit')
const relayAddr = multiaddr(addrs[0])
const destinationAddr = multiaddr(addrs[addrs.length - 1])
const relayPeer = PeerId.createFromCID(relayAddr.getPeerId())
const destinationPeer = PeerId.createFromCID(destinationAddr.getPeerId())
let disconnectOnFailure = false
let relayConnection = this._connectionManager.get(relayPeer)
if (!relayConnection) {
relayConnection = await this._dialer.connectToPeer(relayAddr, options)
disconnectOnFailure = true
}
stop () {
clearDelayedInterval(this._timeout)
}
/**
* Advertise hop relay service in the network.
*
* @returns {Promise<void>}
*/
async _advertiseService () {
try {
const virtualConnection = await hop({
connection: relayConnection,
circuit: this,
request: {
type: CircuitPB.Type.HOP,
srcPeer: {
id: this.peerId.toBytes(),
addrs: this._libp2p.multiaddrs.map(addr => addr.bytes)
},
dstPeer: {
id: destinationPeer.toBytes(),
addrs: [multiaddr(destinationAddr).bytes]
}
}
})
const localAddr = relayAddr.encapsulate(`/p2p-circuit/p2p/${this.peerId.toB58String()}`)
const maConn = toConnection({
stream: virtualConnection,
remoteAddr: ma,
localAddr
})
log('new outbound connection %s', maConn.remoteAddr)
return this._upgrader.upgradeOutbound(maConn)
const cid = await namespaceToCid(RELAY_RENDEZVOUS_NS)
await this._libp2p.contentRouting.provide(cid)
} catch (err) {
log.error('Circuit relay dial failed', err)
disconnectOnFailure && await relayConnection.close()
throw err
if (err.code === 'NO_ROUTERS_AVAILABLE') {
log.error('a content router, such as a DHT, must be provided in order to advertise the relay service', err)
// Stop the advertise
this.stop()
} else {
log.error(err)
}
}
}
/**
* Create a listener
*
* @param {any} options
* @param {Function} handler
* @returns {listener}
*/
createListener (options, handler) {
if (typeof options === 'function') {
handler = options
options = {}
}
// Called on successful HOP and STOP requests
this.handler = handler
return createListener(this, options)
}
/**
* Filter check for all Multiaddrs that this transport can dial on
*
* @param {Array<Multiaddr>} multiaddrs
* @returns {Array<Multiaddr>}
*/
filter (multiaddrs) {
multiaddrs = Array.isArray(multiaddrs) ? multiaddrs : [multiaddrs]
return multiaddrs.filter((ma) => {
return mafmt.Circuit.matches(ma)
})
}
}
/**
* @type {Circuit}
*/
module.exports = withIs(Circuit, { className: 'Circuit', symbolName: '@libp2p/js-libp2p-circuit/circuit' })
module.exports = Relay

View File

@ -1,43 +1,36 @@
'use strict'
const EventEmitter = require('events')
const { EventEmitter } = require('events')
const multiaddr = require('multiaddr')
const debug = require('debug')
const log = debug('libp2p:circuit:listener')
log.err = debug('libp2p:circuit:error:listener')
/**
* @typedef {import('multiaddr')} Multiaddr
* @typedef {import('libp2p-interfaces/src/transport/types').Listener} Listener
*/
/**
* @param {*} circuit
* @param {import('../')} libp2p
* @returns {Listener} a transport listener
*/
module.exports = (circuit) => {
const listener = new EventEmitter()
module.exports = (libp2p) => {
const listeningAddrs = new Map()
/**
* Add swarm handler and listen for incoming connections
*
* @param {Multiaddr} addr
* @returns {void}
* @returns {Promise<void>}
*/
listener.listen = async (addr) => {
async function listen (addr) {
const addrString = String(addr).split('/p2p-circuit').find(a => a !== '')
const relayConn = await circuit._dialer.connectToPeer(multiaddr(addrString))
const relayConn = await libp2p.dial(multiaddr(addrString))
const relayedAddr = relayConn.remoteAddr.encapsulate('/p2p-circuit')
listeningAddrs.set(relayConn.remotePeer.toB58String(), relayedAddr)
listener.emit('listening')
}
/**
* TODO: Remove the peers from our topology
*
* @returns {void}
*/
listener.close = () => {}
/**
* Get fixed up multiaddrs
*
@ -54,7 +47,7 @@ module.exports = (circuit) => {
*
* @returns {Multiaddr[]}
*/
listener.getAddrs = () => {
function getAddrs () {
const addrs = []
for (const addr of listeningAddrs.values()) {
addrs.push(addr)
@ -62,5 +55,22 @@ module.exports = (circuit) => {
return addrs
}
/** @type Listener */
const listener = Object.assign(new EventEmitter(), {
close: () => Promise.resolve(),
listen,
getAddrs
})
// Remove listeningAddrs when a peer disconnects
libp2p.connectionManager.on('peer:disconnect', (connection) => {
const deleted = listeningAddrs.delete(connection.remotePeer.toB58String())
if (deleted) {
// Announce listen addresses change
listener.emit('close')
}
})
return listener
}

View File

@ -1,5 +1,7 @@
'use strict'
const protobuf = require('protons')
/** @type {{CircuitRelay: import('../../types').CircuitMessageProto}} */
module.exports = protobuf(`
message CircuitRelay {

218
src/circuit/transport.js Normal file
View File

@ -0,0 +1,218 @@
'use strict'
const debug = require('debug')
const log = Object.assign(debug('libp2p:circuit'), {
error: debug('libp2p:circuit:err')
})
const mafmt = require('mafmt')
const multiaddr = require('multiaddr')
const PeerId = require('peer-id')
const { CircuitRelay: CircuitPB } = require('./protocol')
const toConnection = require('libp2p-utils/src/stream-to-ma-conn')
const { relay: multicodec } = require('./multicodec')
const createListener = require('./listener')
const { handleCanHop, handleHop, hop } = require('./circuit/hop')
const { handleStop } = require('./circuit/stop')
const StreamHandler = require('./circuit/stream-handler')
const transportSymbol = Symbol.for('@libp2p/js-libp2p-circuit/circuit')
/**
* @typedef {import('multiaddr')} Multiaddr
* @typedef {import('libp2p-interfaces/src/connection').Connection} Connection
* @typedef {import('libp2p-interfaces/src/stream-muxer/types').MuxedStream} MuxedStream
* @typedef {import('../types').CircuitRequest} CircuitRequest
*/
class Circuit {
/**
* Creates an instance of the Circuit Transport.
*
* @class
* @param {object} options
* @param {import('../')} options.libp2p
* @param {import('../upgrader')} options.upgrader
*/
constructor ({ libp2p, upgrader }) {
this._dialer = libp2p.dialer
this._registrar = libp2p.registrar
this._connectionManager = libp2p.connectionManager
this._upgrader = upgrader
this._options = libp2p._config.relay
this._libp2p = libp2p
this.peerId = libp2p.peerId
this._registrar.handle(multicodec, this._onProtocol.bind(this))
}
/**
* @param {Object} props
* @param {Connection} props.connection
* @param {MuxedStream} props.stream
*/
async _onProtocol ({ connection, stream }) {
/** @type {import('./circuit/stream-handler')<CircuitRequest>} */
const streamHandler = new StreamHandler({ stream })
const request = await streamHandler.read()
if (!request) {
return
}
const circuit = this
let virtualConnection
switch (request.type) {
case CircuitPB.Type.CAN_HOP: {
log('received CAN_HOP request from %s', connection.remotePeer.toB58String())
await handleCanHop({ circuit, connection, streamHandler })
break
}
case CircuitPB.Type.HOP: {
log('received HOP request from %s', connection.remotePeer.toB58String())
virtualConnection = await handleHop({
connection,
request,
streamHandler,
circuit
})
break
}
case CircuitPB.Type.STOP: {
log('received STOP request from %s', connection.remotePeer.toB58String())
virtualConnection = await handleStop({
connection,
request,
streamHandler
})
break
}
default: {
log('Request of type %s not supported', request.type)
}
}
if (virtualConnection) {
const remoteAddr = multiaddr(request.dstPeer.addrs[0])
const localAddr = multiaddr(request.srcPeer.addrs[0])
const maConn = toConnection({
stream: virtualConnection,
remoteAddr,
localAddr
})
const type = request.type === CircuitPB.Type.HOP ? 'relay' : 'inbound'
log('new %s connection %s', type, maConn.remoteAddr)
const conn = await this._upgrader.upgradeInbound(maConn)
log('%s connection %s upgraded', type, maConn.remoteAddr)
this.handler && this.handler(conn)
}
}
/**
* Dial a peer over a relay
*
* @param {Multiaddr} ma - the multiaddr of the peer to dial
* @param {Object} options - dial options
* @param {AbortSignal} [options.signal] - An optional abort signal
* @returns {Promise<Connection>} - the connection
*/
async dial (ma, options) {
// Check the multiaddr to see if it contains a relay and a destination peer
const addrs = ma.toString().split('/p2p-circuit')
const relayAddr = multiaddr(addrs[0])
const destinationAddr = multiaddr(addrs[addrs.length - 1])
const relayPeer = PeerId.createFromCID(relayAddr.getPeerId())
const destinationPeer = PeerId.createFromCID(destinationAddr.getPeerId())
let disconnectOnFailure = false
let relayConnection = this._connectionManager.get(relayPeer)
if (!relayConnection) {
relayConnection = await this._dialer.connectToPeer(relayAddr, options)
disconnectOnFailure = true
}
try {
const virtualConnection = await hop({
connection: relayConnection,
request: {
type: CircuitPB.Type.HOP,
srcPeer: {
id: this.peerId.toBytes(),
addrs: this._libp2p.multiaddrs.map(addr => addr.bytes)
},
dstPeer: {
id: destinationPeer.toBytes(),
addrs: [multiaddr(destinationAddr).bytes]
}
}
})
const localAddr = relayAddr.encapsulate(`/p2p-circuit/p2p/${this.peerId.toB58String()}`)
const maConn = toConnection({
stream: virtualConnection,
remoteAddr: ma,
localAddr
})
log('new outbound connection %s', maConn.remoteAddr)
return this._upgrader.upgradeOutbound(maConn)
} catch (err) {
log.error('Circuit relay dial failed', err)
disconnectOnFailure && await relayConnection.close()
throw err
}
}
/**
* Create a listener
*
* @param {any} options
* @param {Function} handler
* @returns {import('libp2p-interfaces/src/transport/types').Listener}
*/
createListener (options, handler) {
if (typeof options === 'function') {
handler = options
options = {}
}
// Called on successful HOP and STOP requests
this.handler = handler
return createListener(this._libp2p)
}
/**
* Filter check for all Multiaddrs that this transport can dial on
*
* @param {Multiaddr[]} multiaddrs
* @returns {Multiaddr[]}
*/
filter (multiaddrs) {
multiaddrs = Array.isArray(multiaddrs) ? multiaddrs : [multiaddrs]
return multiaddrs.filter((ma) => {
return mafmt.Circuit.matches(ma)
})
}
get [Symbol.toStringTag] () {
return 'Circuit'
}
/**
* Checks if the given value is a Transport instance.
*
* @param {any} other
* @returns {other is Transport}
*/
static isTransport (other) {
return Boolean(other && other[transportSymbol])
}
}
module.exports = Circuit

19
src/circuit/utils.js Normal file
View File

@ -0,0 +1,19 @@
'use strict'
const CID = require('cids')
const multihashing = require('multihashing-async')
const TextEncoder = require('ipfs-utils/src/text-encoder')
/**
* Convert a namespace string into a cid.
*
* @param {string} namespace
* @returns {Promise<CID>}
*/
module.exports.namespaceToCid = async (namespace) => {
const bytes = new TextEncoder('utf8').encode(namespace)
const hash = await multihashing(bytes, 'sha2-256')
return new CID(hash)
}

View File

@ -5,7 +5,9 @@ const { dnsaddrResolver } = require('multiaddr/src/resolvers')
const Constants = require('./constants')
const { AGENT_VERSION } = require('./identify/consts')
const RelayConstants = require('./circuit/constants')
const { publicAddressesFirst } = require('libp2p-utils/src/address-sort')
const { FaultTolerance } = require('./transport-manager')
const DefaultConfig = {
@ -26,7 +28,8 @@ const DefaultConfig = {
dialTimeout: Constants.DIAL_TIMEOUT,
resolvers: {
dnsaddr: dnsaddrResolver
}
},
addressSorter: publicAddressesFirst
},
host: {
agentVersion: AGENT_VERSION
@ -38,6 +41,13 @@ const DefaultConfig = {
persistence: false,
threshold: 5
},
peerRouting: {
refreshManager: {
enabled: true,
interval: 6e5,
bootDelay: 10e3
}
},
config: {
dht: {
enabled: false,
@ -53,16 +63,22 @@ const DefaultConfig = {
autoDial: true
},
pubsub: {
enabled: true,
emitSelf: true,
signMessages: true,
strictSigning: true
enabled: true
},
relay: {
enabled: true,
advertise: {
bootDelay: RelayConstants.ADVERTISE_BOOT_DELAY,
enabled: false,
ttl: RelayConstants.ADVERTISE_TTL
},
hop: {
enabled: false,
active: false
},
autoRelay: {
enabled: false,
maxListeners: 2
}
},
transport: {}

View File

@ -1,8 +1,9 @@
'use strict'
const debug = require('debug')
const log = debug('libp2p:connection-manager')
log.error = debug('libp2p:connection-manager:error')
const log = Object.assign(debug('libp2p:connection-manager'), {
error: debug('libp2p:connection-manager:err')
})
const errcode = require('err-code')
const mergeOptions = require('merge-options')
@ -14,7 +15,7 @@ const { EventEmitter } = require('events')
const PeerId = require('peer-id')
const {
ERR_INVALID_PARAMETERS
codes: { ERR_INVALID_PARAMETERS }
} = require('../errors')
const defaultOptions = {
@ -31,29 +32,39 @@ const defaultOptions = {
}
/**
* Responsible for managing known connections.
* @typedef {import('../')} Libp2p
* @typedef {import('libp2p-interfaces/src/connection').Connection} Connection
*/
/**
* @typedef {Object} ConnectionManagerOptions
* @property {number} [maxConnections = Infinity] - The maximum number of connections allowed.
* @property {number} [minConnections = 0] - The minimum number of connections to avoid pruning.
* @property {number} [maxData = Infinity] - The max data (in and out), per average interval to allow.
* @property {number} [maxSentData = Infinity] - The max outgoing data, per average interval to allow.
* @property {number} [maxReceivedData = Infinity] - The max incoming data, per average interval to allow.
* @property {number} [maxEventLoopDelay = Infinity] - The upper limit the event loop can take to run.
* @property {number} [pollInterval = 2000] - How often, in milliseconds, metrics and latency should be checked.
* @property {number} [movingAverageInterval = 60000] - How often, in milliseconds, to compute averages.
* @property {number} [defaultPeerValue = 1] - The value of the peer.
* @property {boolean} [autoDial = true] - Should preemptively guarantee connections are above the low watermark.
* @property {number} [autoDialInterval = 10000] - How often, in milliseconds, it should preemptively guarantee connections are above the low watermark.
*/
/**
*
* @fires ConnectionManager#peer:connect Emitted when a new peer is connected.
* @fires ConnectionManager#peer:disconnect Emitted when a peer is disconnected.
*/
class ConnectionManager extends EventEmitter {
/**
* Responsible for managing known connections.
*
* @class
* @param {Libp2p} libp2p
* @param {object} options
* @param {number} options.maxConnections - The maximum number of connections allowed. Default=Infinity
* @param {number} options.minConnections - The minimum number of connections to avoid pruning. Default=0
* @param {number} options.maxData - The max data (in and out), per average interval to allow. Default=Infinity
* @param {number} options.maxSentData - The max outgoing data, per average interval to allow. Default=Infinity
* @param {number} options.maxReceivedData - The max incoming data, per average interval to allow.. Default=Infinity
* @param {number} options.maxEventLoopDelay - The upper limit the event loop can take to run. Default=Infinity
* @param {number} options.pollInterval - How often, in milliseconds, metrics and latency should be checked. Default=2000
* @param {number} options.movingAverageInterval - How often, in milliseconds, to compute averages. Default=60000
* @param {number} options.defaultPeerValue - The value of the peer. Default=1
* @param {boolean} options.autoDial - Should preemptively guarantee connections are above the low watermark. Default=true
* @param {number} options.autoDialInterval - How often, in milliseconds, it should preemptively guarantee connections are above the low watermark. Default=10000
* @param {ConnectionManagerOptions} options
*/
constructor (libp2p, options) {
constructor (libp2p, options = {}) {
super()
this._libp2p = libp2p
@ -66,8 +77,6 @@ class ConnectionManager extends EventEmitter {
log('options: %j', this._options)
this._libp2p = libp2p
/**
* Map of peer identifiers to their peer value for pruning connections.
*
@ -78,7 +87,7 @@ class ConnectionManager extends EventEmitter {
/**
* Map of connections per peer
*
* @type {Map<string, Array<conn>>}
* @type {Map<string, Connection[]>}
*/
this.connections = new Map()
@ -159,15 +168,13 @@ class ConnectionManager extends EventEmitter {
*
* @param {PeerId} peerId
* @param {number} value - A number between 0 and 1
* @returns {void}
*/
setPeerValue (peerId, value) {
if (value < 0 || value > 1) {
throw new Error('value should be a number between 0 and 1')
}
if (peerId.toB58String) {
peerId = peerId.toB58String()
}
this._peerValues.set(peerId, value)
this._peerValues.set(peerId.toB58String(), value)
}
/**
@ -177,21 +184,24 @@ class ConnectionManager extends EventEmitter {
* @private
*/
_checkMetrics () {
const movingAverages = this._libp2p.metrics.global.movingAverages
const received = movingAverages.dataReceived[this._options.movingAverageInterval].movingAverage()
this._checkMaxLimit('maxReceivedData', received)
const sent = movingAverages.dataSent[this._options.movingAverageInterval].movingAverage()
this._checkMaxLimit('maxSentData', sent)
const total = received + sent
this._checkMaxLimit('maxData', total)
log('metrics update', total)
this._timer = retimer(this._checkMetrics, this._options.pollInterval)
if (this._libp2p.metrics) {
const movingAverages = this._libp2p.metrics.global.movingAverages
const received = movingAverages.dataReceived[this._options.movingAverageInterval].movingAverage()
this._checkMaxLimit('maxReceivedData', received)
const sent = movingAverages.dataSent[this._options.movingAverageInterval].movingAverage()
this._checkMaxLimit('maxSentData', sent)
const total = received + sent
this._checkMaxLimit('maxData', total)
log('metrics update', total)
this._timer = retimer(this._checkMetrics, this._options.pollInterval)
}
}
/**
* Tracks the incoming connection and check the connection limit
*
* @param {Connection} connection
* @returns {void}
*/
onConnect (connection) {
const peerId = connection.remotePeer
@ -218,6 +228,7 @@ class ConnectionManager extends EventEmitter {
* Removes the connection from tracking
*
* @param {Connection} connection
* @returns {void}
*/
onDisconnect (connection) {
const peerId = connection.remotePeer.toB58String()
@ -237,7 +248,7 @@ class ConnectionManager extends EventEmitter {
* Get a connection with a peer.
*
* @param {PeerId} peerId
* @returns {Connection}
* @returns {Connection|null}
*/
get (peerId) {
const connections = this.getAll(peerId)
@ -251,7 +262,7 @@ class ConnectionManager extends EventEmitter {
* Get all open connections with a peer.
*
* @param {PeerId} peerId
* @returns {Array<Connection>}
* @returns {Connection[]}
*/
getAll (peerId) {
if (!PeerId.isPeerId(peerId)) {

View File

@ -1,3 +1,4 @@
// @ts-nocheck
'use strict'
/**
@ -6,7 +7,7 @@
/* global window */
const globalThis = require('ipfs-utils/src/globalthis')
const EventEmitter = require('events')
const { EventEmitter } = require('events')
const VisibilityChangeEmitter = require('./visibility-change-emitter')
const debug = require('debug')('latency-monitor:LatencyMonitor')
@ -17,6 +18,12 @@ const debug = require('debug')('latency-monitor:LatencyMonitor')
* @property {number} maxMS What was the max time for a cb to be called
* @property {number} avgMs What was the average time for a cb to be called
* @property {number} lengthMs How long this interval was in ms
*
* @typedef {Object} LatencyMonitorOptions
* @property {number} [latencyCheckIntervalMs=500] - How often to add a latency check event (ms)
* @property {number} [dataEmitIntervalMs=5000] - How often to summarize latency check events. null or 0 disables event firing
* @property {Function} [asyncTestFn] - What cb-style async function to use
* @property {number} [latencyRandomPercentage=5] - What percent (+/-) of latencyCheckIntervalMs should we randomly use? This helps avoid alignment to other events.
*/
/**
@ -24,6 +31,8 @@ const debug = require('debug')('latency-monitor:LatencyMonitor')
* the asyncTestFn and timing how long it takes the callback to be called. It can also periodically emit stats about this.
* This can be disabled and stats can be pulled via setting dataEmitIntervalMs = 0.
*
* @extends {EventEmitter}
*
* The default implementation is an event loop latency monitor. This works by firing periodic events into the event loop
* and timing how long it takes to get back.
*
@ -37,11 +46,8 @@ const debug = require('debug')('latency-monitor:LatencyMonitor')
*/
class LatencyMonitor extends EventEmitter {
/**
* @param {object} [options]
* @param {number} [options.latencyCheckIntervalMs=500] - How often to add a latency check event (ms)
* @param {number} [options.dataEmitIntervalMs=5000] - How often to summarize latency check events. null or 0 disables event firing
* @param {Function} [options.asyncTestFn] - What cb-style async function to use
* @param {number} [options.latencyRandomPercentage=5] - What percent (+/-) of latencyCheckIntervalMs should we randomly use? This helps avoid alignment to other events.
* @class
* @param {LatencyMonitorOptions} [options]
*/
constructor ({ latencyCheckIntervalMs, dataEmitIntervalMs, asyncTestFn, latencyRandomPercentage } = {}) {
super()
@ -91,6 +97,7 @@ class LatencyMonitor extends EventEmitter {
// See: http://stackoverflow.com/questions/6032429/chrome-timeouts-interval-suspended-in-background-tabs
if (isBrowser()) {
that._visibilityChangeEmitter = new VisibilityChangeEmitter()
that._visibilityChangeEmitter.on('visibilityChange', (pageInFocus) => {
if (pageInFocus) {
that._startTimers()

View File

@ -1,10 +1,12 @@
// @ts-nocheck
/* global document */
/**
* This code is based on `latency-monitor` (https://github.com/mlucool/latency-monitor) by `mlucool` (https://github.com/mlucool), available under Apache License 2.0 (https://github.com/mlucool/latency-monitor/blob/master/LICENSE)
*/
'use strict'
const EventEmitter = require('events')
const { EventEmitter } = require('events')
const debug = require('debug')('latency-monitor:VisibilityChangeEmitter')
@ -29,12 +31,12 @@ const debug = require('debug')('latency-monitor:VisibilityChangeEmitter')
* });
* // To access the visibility state directly, call:
* console.log('Am I focused now? ' + myVisibilityEmitter.isVisible());
*
* @class VisibilityChangeEmitter
*/
module.exports = class VisibilityChangeEmitter extends EventEmitter {
class VisibilityChangeEmitter extends EventEmitter {
/**
* Creates a VisibilityChangeEmitter
*
* @class
*/
constructor () {
super()
@ -119,3 +121,5 @@ module.exports = class VisibilityChangeEmitter extends EventEmitter {
this.emit('visibilityChange', visible)
}
}
module.exports = VisibilityChangeEmitter

View File

@ -6,111 +6,130 @@ const { messages, codes } = require('./errors')
const all = require('it-all')
const pAny = require('p-any')
module.exports = (node) => {
const routers = node._modules.contentRouting || []
const dht = node._dht
/**
* @typedef {import('peer-id')} PeerId
* @typedef {import('multiaddr')} Multiaddr
* @typedef {import('cids')} CID
*/
// If we have the dht, make it first
if (dht) {
routers.unshift(dht)
}
/**
* @typedef {Object} GetData
* @property {PeerId} from
* @property {Uint8Array} val
*/
return {
/**
* Iterates over all content routers in series to find providers of the given key.
* Once a content router succeeds, iteration will stop.
*
* @param {CID} key - The CID key of the content to find
* @param {object} [options]
* @param {number} [options.timeout] - How long the query should run
* @param {number} [options.maxNumProviders] - maximum number of providers to find
* @returns {AsyncIterable<{ id: PeerId, multiaddrs: Multiaddr[] }>}
*/
async * findProviders (key, options) {
if (!routers.length) {
throw errCode(new Error('No content routers available'), 'NO_ROUTERS_AVAILABLE')
}
class ContentRouting {
/**
* @class
* @param {import('./')} libp2p
*/
constructor (libp2p) {
this.libp2p = libp2p
this.routers = libp2p._modules.contentRouting || []
this.dht = libp2p._dht
const result = await pAny(
routers.map(async (router) => {
const provs = await all(router.findProviders(key, options))
if (!provs || !provs.length) {
throw errCode(new Error('not found'), 'NOT_FOUND')
}
return provs
})
)
for (const peer of result) {
yield peer
}
},
/**
* Iterates over all content routers in parallel to notify it is
* a provider of the given key.
*
* @param {CID} key - The CID key of the content to find
* @returns {Promise<void>}
*/
async provide (key) { // eslint-disable-line require-await
if (!routers.length) {
throw errCode(new Error('No content routers available'), 'NO_ROUTERS_AVAILABLE')
}
return Promise.all(routers.map((router) => router.provide(key)))
},
/**
* Store the given key/value pair in the DHT.
*
* @param {Uint8Array} key
* @param {Uint8Array} value
* @param {Object} [options] - put options
* @param {number} [options.minPeers] - minimum number of peers required to successfully put
* @returns {Promise<void>}
*/
async put (key, value, options) { // eslint-disable-line require-await
if (!node.isStarted() || !dht.isStarted) {
throw errCode(new Error(messages.NOT_STARTED_YET), codes.DHT_NOT_STARTED)
}
return dht.put(key, value, options)
},
/**
* Get the value to the given key.
* Times out after 1 minute by default.
*
* @param {Uint8Array} key
* @param {Object} [options] - get options
* @param {number} [options.timeout] - optional timeout (default: 60000)
* @returns {Promise<{from: PeerId, val: Uint8Array}>}
*/
async get (key, options) { // eslint-disable-line require-await
if (!node.isStarted() || !dht.isStarted) {
throw errCode(new Error(messages.NOT_STARTED_YET), codes.DHT_NOT_STARTED)
}
return dht.get(key, options)
},
/**
* Get the `n` values to the given key without sorting.
*
* @param {Uint8Array} key
* @param {number} nVals
* @param {Object} [options] - get options
* @param {number} [options.timeout] - optional timeout (default: 60000)
* @returns {Promise<Array<{from: PeerId, val: Uint8Array}>>}
*/
async getMany (key, nVals, options) { // eslint-disable-line require-await
if (!node.isStarted() || !dht.isStarted) {
throw errCode(new Error(messages.NOT_STARTED_YET), codes.DHT_NOT_STARTED)
}
return dht.getMany(key, nVals, options)
// If we have the dht, make it first
if (this.dht) {
this.routers.unshift(this.dht)
}
}
/**
* Iterates over all content routers in series to find providers of the given key.
* Once a content router succeeds, iteration will stop.
*
* @param {CID} key - The CID key of the content to find
* @param {object} [options]
* @param {number} [options.timeout] - How long the query should run
* @param {number} [options.maxNumProviders] - maximum number of providers to find
* @returns {AsyncIterable<{ id: PeerId, multiaddrs: Multiaddr[] }>}
*/
async * findProviders (key, options) {
if (!this.routers.length) {
throw errCode(new Error('No content this.routers available'), 'NO_ROUTERS_AVAILABLE')
}
const result = await pAny(
this.routers.map(async (router) => {
const provs = await all(router.findProviders(key, options))
if (!provs || !provs.length) {
throw errCode(new Error('not found'), 'NOT_FOUND')
}
return provs
})
)
for (const peer of result) {
yield peer
}
}
/**
* Iterates over all content routers in parallel to notify it is
* a provider of the given key.
*
* @param {CID} key - The CID key of the content to find
* @returns {Promise<void>}
*/
async provide (key) {
if (!this.routers.length) {
throw errCode(new Error('No content routers available'), 'NO_ROUTERS_AVAILABLE')
}
await Promise.all(this.routers.map((router) => router.provide(key)))
}
/**
* Store the given key/value pair in the DHT.
*
* @param {Uint8Array} key
* @param {Uint8Array} value
* @param {Object} [options] - put options
* @param {number} [options.minPeers] - minimum number of peers required to successfully put
* @returns {Promise<void>}
*/
put (key, value, options) {
if (!this.libp2p.isStarted() || !this.dht.isStarted) {
throw errCode(new Error(messages.NOT_STARTED_YET), codes.DHT_NOT_STARTED)
}
return this.dht.put(key, value, options)
}
/**
* Get the value to the given key.
* Times out after 1 minute by default.
*
* @param {Uint8Array} key
* @param {Object} [options] - get options
* @param {number} [options.timeout] - optional timeout (default: 60000)
* @returns {Promise<GetData>}
*/
get (key, options) {
if (!this.libp2p.isStarted() || !this.dht.isStarted) {
throw errCode(new Error(messages.NOT_STARTED_YET), codes.DHT_NOT_STARTED)
}
return this.dht.get(key, options)
}
/**
* Get the `n` values to the given key without sorting.
*
* @param {Uint8Array} key
* @param {number} nVals
* @param {Object} [options] - get options
* @param {number} [options.timeout] - optional timeout (default: 60000)
* @returns {Promise<GetData[]>}
*/
async getMany (key, nVals, options) { // eslint-disable-line require-await
if (!this.libp2p.isStarted() || !this.dht.isStarted) {
throw errCode(new Error(messages.NOT_STARTED_YET), codes.DHT_NOT_STARTED)
}
return this.dht.getMany(key, nVals, options)
}
}
module.exports = ContentRouting

View File

@ -1,14 +1,27 @@
'use strict'
const AbortController = require('abort-controller')
const anySignal = require('any-signal')
const debug = require('debug')
const errCode = require('err-code')
const log = debug('libp2p:dialer:request')
log.error = debug('libp2p:dialer:request:error')
const AbortController = require('abort-controller').default
const anySignal = require('any-signal')
const FIFO = require('p-fifo')
const pAny = require('p-any')
/**
* @typedef {import('libp2p-interfaces/src/connection').Connection} Connection
* @typedef {import('./')} Dialer
* @typedef {import('multiaddr')} Multiaddr
*/
/**
* @typedef {Object} DialOptions
* @property {AbortSignal} signal
*
* @typedef {Object} DialRequestOptions
* @property {Multiaddr[]} addrs
* @property {(m: Multiaddr, options: DialOptions) => Promise<Connection>} dialAction
* @property {Dialer} dialer
*/
class DialRequest {
/**
* Manages running the `dialAction` on multiple provided `addrs` in parallel
@ -17,10 +30,8 @@ class DialRequest {
* started using `DialRequest.run(options)`. Once a single dial has succeeded,
* all other dials in the request will be cancelled.
*
* @param {object} options
* @param {Multiaddr[]} options.addrs
* @param {function(Multiaddr):Promise<Connection>} options.dialAction
* @param {Dialer} options.dialer
* @class
* @param {DialRequestOptions} options
*/
constructor ({
addrs,
@ -34,11 +45,11 @@ class DialRequest {
/**
* @async
* @param {object} options
* @param {AbortSignal} options.signal - An AbortController signal
* @returns {Connection}
* @param {object} [options]
* @param {AbortSignal} [options.signal] - An AbortController signal
* @returns {Promise<Connection>}
*/
async run (options) {
async run (options = {}) {
const tokens = this.dialer.getTokens(this.addrs.length)
// If no tokens are available, throw
if (tokens.length < 1) {
@ -78,4 +89,4 @@ class DialRequest {
}
}
module.exports.DialRequest = DialRequest
module.exports = DialRequest

View File

@ -1,14 +1,16 @@
'use strict'
const multiaddr = require('multiaddr')
const debug = require('debug')
const log = Object.assign(debug('libp2p:dialer'), {
error: debug('libp2p:dialer:err')
})
const errCode = require('err-code')
const multiaddr = require('multiaddr')
const TimeoutController = require('timeout-abort-controller')
const anySignal = require('any-signal')
const debug = require('debug')
const log = debug('libp2p:dialer')
log.error = debug('libp2p:dialer:error')
const { DialRequest } = require('./dial-request')
const DialRequest = require('./dial-request')
const { publicAddressesFirst } = require('libp2p-utils/src/address-sort')
const getPeer = require('../get-peer')
const { codes } = require('../errors')
@ -18,20 +20,49 @@ const {
MAX_PER_PEER_DIALS
} = require('../constants')
/**
* @typedef {import('libp2p-interfaces/src/connection').Connection} Connection
* @typedef {import('multiaddr')} Multiaddr
* @typedef {import('peer-id')} PeerId
* @typedef {import('../peer-store')} PeerStore
* @typedef {import('../peer-store/address-book').Address} Address
* @typedef {import('../transport-manager')} TransportManager
*/
/**
* @typedef {Object} DialerProperties
* @property {PeerStore} peerStore
* @property {TransportManager} transportManager
*
* @typedef {(addr:Multiaddr) => Promise<string[]>} Resolver
*
* @typedef {Object} DialerOptions
* @property {(addresses: Address[]) => Address[]} [options.addressSorter = publicAddressesFirst] - Sort the known addresses of a peer before trying to dial.
* @property {number} [concurrency = MAX_PARALLEL_DIALS] - Number of max concurrent dials.
* @property {number} [perPeerLimit = MAX_PER_PEER_DIALS] - Number of max concurrent dials per peer.
* @property {number} [timeout = DIAL_TIMEOUT] - How long a dial attempt is allowed to take.
* @property {Record<string, Resolver>} [resolvers = {}] - multiaddr resolvers to use when dialing
*
* @typedef DialTarget
* @property {string} id
* @property {Multiaddr[]} addrs
*
* @typedef PendingDial
* @property {DialRequest} dialRequest
* @property {TimeoutController} controller
* @property {Promise} promise
* @property {function():void} destroy
*/
class Dialer {
/**
* @class
* @param {object} options
* @param {TransportManager} options.transportManager
* @param {Peerstore} options.peerStore
* @param {number} [options.concurrency = MAX_PARALLEL_DIALS] - Number of max concurrent dials.
* @param {number} [options.perPeerLimit = MAX_PER_PEER_DIALS] - Number of max concurrent dials per peer.
* @param {number} [options.timeout = DIAL_TIMEOUT] - How long a dial attempt is allowed to take.
* @param {object} [options.resolvers = {}] - multiaddr resolvers to use when dialing
* @param {DialerProperties & DialerOptions} options
*/
constructor ({
transportManager,
peerStore,
addressSorter = publicAddressesFirst,
concurrency = MAX_PARALLEL_DIALS,
timeout = DIAL_TIMEOUT,
perPeerLimit = MAX_PER_PEER_DIALS,
@ -39,6 +70,7 @@ class Dialer {
}) {
this.transportManager = transportManager
this.peerStore = peerStore
this.addressSorter = addressSorter
this.concurrency = concurrency
this.timeout = timeout
this.perPeerLimit = perPeerLimit
@ -98,12 +130,6 @@ class Dialer {
}
}
/**
* @typedef DialTarget
* @property {string} id
* @property {Multiaddr[]} addrs
*/
/**
* Creates a DialTarget. The DialTarget is used to create and track
* the DialRequest to a given peer.
@ -120,7 +146,7 @@ class Dialer {
this.peerStore.addressBook.add(id, multiaddrs)
}
let knownAddrs = this.peerStore.addressBook.getMultiaddrsForPeer(id) || []
let knownAddrs = this.peerStore.addressBook.getMultiaddrsForPeer(id, this.addressSorter) || []
// If received a multiaddr to dial, it should be the first to use
// But, if we know other multiaddrs for the peer, we should try them too.
@ -141,14 +167,6 @@ class Dialer {
}
}
/**
* @typedef PendingDial
* @property {DialRequest} dialRequest
* @property {TimeoutController} controller
* @property {Promise} promise
* @property {function():void} destroy
*/
/**
* Creates a PendingDial that wraps the underlying DialRequest
*
@ -158,7 +176,7 @@ class Dialer {
* @param {AbortSignal} [options.signal] - An AbortController signal
* @returns {PendingDial}
*/
_createPendingDial (dialTarget, options) {
_createPendingDial (dialTarget, options = {}) {
const dialAction = (addr, options) => {
if (options.signal.aborted) throw errCode(new Error('already aborted'), codes.ERR_ALREADY_ABORTED)
return this.transportManager.dial(addr, options)
@ -207,7 +225,7 @@ class Dialer {
* Resolve multiaddr recursively.
*
* @param {Multiaddr} ma
* @returns {Promise<Array<Multiaddr>>}
* @returns {Promise<Multiaddr[]>}
*/
async _resolve (ma) {
// TODO: recursive logic should live in multiaddr once dns4/dns6 support is in place
@ -224,19 +242,20 @@ class Dialer {
return this._resolve(nm)
}))
return recursiveMultiaddrs.flat().reduce((array, newM) => {
const addrs = recursiveMultiaddrs.flat()
return addrs.reduce((array, newM) => {
if (!array.find(m => m.equals(newM))) {
array.push(newM)
}
return array
}, []) // Unique addresses
}, /** @type {Multiaddr[]} */([]))
}
/**
* Resolve a given multiaddr. If this fails, an empty array will be returned
*
* @param {Multiaddr} ma
* @returns {Promise<Array<Multiaddr>>}
* @returns {Promise<Multiaddr[]>}
*/
async _resolveRecord (ma) {
try {

View File

@ -6,12 +6,16 @@ const errCode = require('err-code')
const { codes } = require('./errors')
/**
* @typedef {import('multiaddr')} Multiaddr
*/
/**
* Converts the given `peer` to a `Peer` object.
* If a multiaddr is received, the addressBook is updated.
*
* @param {PeerId|Multiaddr|string} peer
* @returns {{ id: PeerId, multiaddrs: Array<Multiaddr> }}
* @returns {{ id: PeerId, multiaddrs: Multiaddr[]|undefined }}
*/
function getPeer (peer) {
if (typeof peer === 'string') {

View File

@ -1,5 +1,6 @@
'use strict'
// @ts-ignore file not listed within the file list of projects
const libp2pVersion = require('../../package.json').version
module.exports.PROTOCOL_VERSION = 'ipfs/0.1.0'

View File

@ -1,13 +1,13 @@
'use strict'
const debug = require('debug')
const log = debug('libp2p:identify')
log.error = debug('libp2p:identify:error')
const log = Object.assign(debug('libp2p:identify'), {
error: debug('libp2p:identify:err')
})
const errCode = require('err-code')
const pb = require('it-protocol-buffers')
const lp = require('it-length-prefixed')
const pipe = require('it-pipe')
const { pipe } = require('it-pipe')
const { collect, take, consume } = require('streaming-iterables')
const uint8ArrayFromString = require('uint8arrays/from-string')
@ -29,58 +29,22 @@ const {
const { codes } = require('../errors')
/**
* @typedef {import('libp2p-interfaces/src/connection').Connection} Connection
* @typedef {import('libp2p-interfaces/src/stream-muxer/types').MuxedStream} MuxedStream
*/
class IdentifyService {
/**
* Takes the `addr` and converts it to a Multiaddr if possible
*
* @param {Uint8Array | string} addr
* @returns {Multiaddr|null}
*/
static getCleanMultiaddr (addr) {
if (addr && addr.length > 0) {
try {
return multiaddr(addr)
} catch (_) {
return null
}
}
return null
}
/**
* @class
* @param {object} options
* @param {Libp2p} options.libp2p
* @param {Map<string, handler>} options.protocols - A reference to the protocols we support
* @param {Object} options
* @param {import('../')} options.libp2p
*/
constructor ({ libp2p, protocols }) {
/**
* @property {PeerStore}
*/
this.peerStore = libp2p.peerStore
/**
* @property {ConnectionManager}
*/
this.connectionManager = libp2p.connectionManager
this.connectionManager.on('peer:connect', (connection) => {
const peerId = connection.remotePeer
this.identify(connection, peerId).catch(log.error)
})
/**
* @property {PeerId}
*/
this.peerId = libp2p.peerId
/**
* @property {AddressManager}
*/
constructor ({ libp2p }) {
this._libp2p = libp2p
this._protocols = protocols
this.peerStore = libp2p.peerStore
this.connectionManager = libp2p.connectionManager
this.peerId = libp2p.peerId
this.handleMessage = this.handleMessage.bind(this)
@ -93,18 +57,36 @@ class IdentifyService {
this.peerStore.metadataBook.set(this.peerId, 'AgentVersion', uint8ArrayFromString(this._host.agentVersion))
this.peerStore.metadataBook.set(this.peerId, 'ProtocolVersion', uint8ArrayFromString(this._host.protocolVersion))
// When a new connection happens, trigger identify
this.connectionManager.on('peer:connect', (connection) => {
this.identify(connection).catch(log.error)
})
// When self multiaddrs change, trigger identify-push
this.peerStore.on('change:multiaddrs', ({ peerId }) => {
if (peerId.toString() === this.peerId.toString()) {
this.pushToPeerStore()
}
})
// When self protocols change, trigger identify-push
this.peerStore.on('change:protocols', ({ peerId }) => {
if (peerId.toString() === this.peerId.toString()) {
this.pushToPeerStore()
}
})
}
/**
* Send an Identify Push update to the list of connections
*
* @param {Array<Connection>} connections
* @returns {Promise<void>}
* @param {Connection[]} connections
* @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())
const protocols = this.peerStore.protoBook.get(this.peerId) || []
const pushes = connections.map(async connection => {
try {
@ -132,12 +114,17 @@ class IdentifyService {
/**
* Calls `push` for all peers in the `peerStore` that are connected
*
* @param {PeerStore} peerStore
* @returns {void}
*/
pushToPeerStore (peerStore) {
pushToPeerStore () {
// Do not try to push if libp2p node is not running
if (!this._libp2p.isStarted()) {
return
}
const connections = []
let connection
for (const peer of peerStore.peers.values()) {
for (const peer of this.peerStore.peers.values()) {
if (peer.protocols.includes(MULTICODEC_IDENTIFY_PUSH) && (connection = this.connectionManager.get(peer.id))) {
connections.push(connection)
}
@ -221,11 +208,11 @@ class IdentifyService {
/**
* A handler to register with Libp2p to process identify messages.
*
* @param {object} options
* @param {string} options.protocol
* @param {*} options.stream
* @param {Object} options
* @param {Connection} options.connection
* @returns {Promise<void>}
* @param {MuxedStream} options.stream
* @param {string} options.protocol
* @returns {Promise<void>|undefined}
*/
handleMessage ({ connection, stream, protocol }) {
switch (protocol) {
@ -243,9 +230,10 @@ class IdentifyService {
* to the requesting peer over the given `connection`
*
* @private
* @param {object} options
* @param {*} options.stream
* @param {Object} options
* @param {MuxedStream} options.stream
* @param {Connection} options.connection
* @returns {Promise<void>}
*/
async _handleIdentify ({ connection, stream }) {
let publicKey = new Uint8Array(0)
@ -253,7 +241,8 @@ class IdentifyService {
publicKey = this.peerId.pubKey.bytes
}
const signedPeerRecord = await this._getSelfPeerRecord()
const signedPeerRecord = await this.peerStore.addressBook.getRawEnvelope(this.peerId)
const protocols = this.peerStore.protoBook.get(this.peerId) || []
const message = Message.encode({
protocolVersion: this._host.protocolVersion,
@ -262,7 +251,7 @@ class IdentifyService {
listenAddrs: this._libp2p.multiaddrs.map((ma) => ma.bytes),
signedPeerRecord,
observedAddr: connection.remoteAddr.bytes,
protocols: Array.from(this._protocols.keys())
protocols
})
try {
@ -282,8 +271,9 @@ class IdentifyService {
*
* @private
* @param {object} options
* @param {*} options.stream
* @param {MuxedStream} options.stream
* @param {Connection} options.connection
* @returns {Promise<void>}
*/
async _handlePush ({ connection, stream }) {
let message
@ -325,42 +315,34 @@ class IdentifyService {
}
/**
* Get self signed peer record raw envelope.
* Takes the `addr` and converts it to a Multiaddr if possible
*
* @returns {Uint8Array}
* @param {Uint8Array | string} addr
* @returns {multiaddr|null}
*/
async _getSelfPeerRecord () {
const selfSignedPeerRecord = this.peerStore.addressBook.getRawEnvelope(this.peerId)
// TODO: support invalidation when dynamic multiaddrs are supported
if (selfSignedPeerRecord) {
return selfSignedPeerRecord
}
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')
static getCleanMultiaddr (addr) {
if (addr && addr.length > 0) {
try {
return multiaddr(addr)
} catch (_) {
return null
}
}
return null
}
}
module.exports.IdentifyService = IdentifyService
/**
* The protocols the IdentifyService supports
*
* @property multicodecs
*/
module.exports.multicodecs = {
const multicodecs = {
IDENTIFY: MULTICODEC_IDENTIFY,
IDENTIFY_PUSH: MULTICODEC_IDENTIFY_PUSH
}
module.exports.Message = Message
IdentifyService.multicodecs = multicodecs
IdentifyService.Messsage = Message
module.exports = IdentifyService

View File

@ -1,23 +1,25 @@
'use strict'
const { EventEmitter } = require('events')
const debug = require('debug')
const log = Object.assign(debug('libp2p'), {
error: debug('libp2p:err')
})
const { EventEmitter } = require('events')
const globalThis = require('ipfs-utils/src/globalthis')
const log = debug('libp2p')
log.error = debug('libp2p:error')
const errCode = require('err-code')
const PeerId = require('peer-id')
const peerRouting = require('./peer-routing')
const contentRouting = require('./content-routing')
const PeerRouting = require('./peer-routing')
const ContentRouting = require('./content-routing')
const getPeer = require('./get-peer')
const { validate: validateConfig } = require('./config')
const { codes, messages } = require('./errors')
const AddressManager = require('./address-manager')
const ConnectionManager = require('./connection-manager')
const Circuit = require('./circuit')
const Circuit = require('./circuit/transport')
const Relay = require('./circuit')
const Dialer = require('./dialer')
const Keychain = require('./keychain')
const Metrics = require('./metrics')
@ -28,22 +30,95 @@ const PubsubAdapter = require('./pubsub-adapter')
const PersistentPeerStore = require('./peer-store/persistent')
const Registrar = require('./registrar')
const ping = require('./ping')
const {
IdentifyService,
multicodecs: IDENTIFY_PROTOCOLS
} = require('./identify')
const IdentifyService = require('./identify')
const IDENTIFY_PROTOCOLS = IdentifyService.multicodecs
/**
* @typedef {import('multiaddr')} Multiaddr
* @typedef {import('libp2p-interfaces/src/connection').Connection} Connection
* @typedef {import('libp2p-interfaces/src/stream-muxer/types').MuxedStream} MuxedStream
* @typedef {import('libp2p-interfaces/src/transport/types').TransportFactory} TransportFactory
* @typedef {import('libp2p-interfaces/src/stream-muxer/types').MuxerFactory} MuxerFactory
* @typedef {import('libp2p-interfaces/src/crypto/types').Crypto} Crypto
* @typedef {import('libp2p-interfaces/src/pubsub')} Pubsub
*/
/**
* @typedef {Object} PeerStoreOptions
* @property {boolean} persistence
*
* @typedef {Object} PeerDiscoveryOptions
* @property {boolean} autoDial
*
* @typedef {Object} RelayOptions
* @property {boolean} enabled
* @property {import('./circuit').RelayAdvertiseOptions} advertise
* @property {import('./circuit').HopOptions} hop
* @property {import('./circuit').AutoRelayOptions} autoRelay
*
* @typedef {Object} Libp2pConfig
* @property {Object} [dht] dht module options
* @property {PeerDiscoveryOptions} [peerDiscovery]
* @property {Pubsub} [pubsub] pubsub module options
* @property {RelayOptions} [relay]
* @property {Record<string, Object>} [transport] transport options indexed by transport key
*
* @typedef {Object} Libp2pModules
* @property {TransportFactory[]} transport
* @property {MuxerFactory[]} streamMuxer
* @property {Crypto[]} connEncryption
*
* @typedef {Object} Libp2pOptions
* @property {Libp2pModules} modules libp2p modules to use
* @property {import('./address-manager').AddressManagerOptions} [addresses]
* @property {import('./connection-manager').ConnectionManagerOptions} [connectionManager]
* @property {import('./dialer').DialerOptions} [dialer]
* @property {import('./metrics').MetricsOptions} [metrics]
* @property {Object} [keychain]
* @property {import('./transport-manager').TransportManagerOptions} [transportManager]
* @property {PeerStoreOptions & import('./peer-store/persistent').PersistentPeerStoreOptions} [peerStore]
* @property {Libp2pConfig} [config]
* @property {PeerId} peerId
*
* @typedef {Object} CreateOptions
* @property {PeerId} peerId
*
* @extends {EventEmitter}
* @fires Libp2p#error Emitted when an error occurs
* @fires Libp2p#peer:discovery Emitted when a peer is discovered
*/
class Libp2p extends EventEmitter {
/**
* Like `new Libp2p(options)` except it will create a `PeerId`
* instance if one is not provided in options.
*
* @param {Libp2pOptions & CreateOptions} options - Libp2p configuration options
* @returns {Promise<Libp2p>}
*/
static async create (options) {
if (options.peerId) {
return new Libp2p(options)
}
const peerId = await PeerId.create()
options.peerId = peerId
return new Libp2p(options)
}
/**
* Libp2p node.
*
* @class
* @param {Libp2pOptions} _options
*/
constructor (_options) {
super()
// validateConfig will ensure the config is correct,
// and add default values where appropriate
this._options = validateConfig(_options)
/** @type {PeerId} */
this.peerId = this._options.peerId
this.datastore = this._options.datastore
@ -135,7 +210,8 @@ class Libp2p extends EventEmitter {
concurrency: this._options.dialer.maxParallelDials,
perPeerLimit: this._options.dialer.maxDialsPerPeer,
timeout: this._options.dialer.dialTimeout,
resolvers: this._options.dialer.resolvers
resolvers: this._options.dialer.resolvers,
addressSorter: this._options.dialer.addressSorter
})
this._modules.transport.forEach((Transport) => {
@ -145,7 +221,9 @@ class Libp2p extends EventEmitter {
})
if (this._config.relay.enabled) {
// @ts-ignore Circuit prototype
this.transportManager.add(Circuit.prototype[Symbol.toStringTag], Circuit)
this.relay = new Relay(this)
}
// Attach stream multiplexers
@ -156,10 +234,7 @@ class Libp2p extends EventEmitter {
})
// Add the identify service since we can multiplex
this.identifyService = new IdentifyService({
libp2p: this,
protocols: this.upgrader.protocols
})
this.identifyService = new IdentifyService({ libp2p: this })
this.handle(Object.values(IDENTIFY_PROTOCOLS), this.identifyService.handleMessage)
}
@ -188,13 +263,14 @@ class Libp2p extends EventEmitter {
if (this._modules.pubsub) {
const Pubsub = this._modules.pubsub
// using pubsub adapter with *DEPRECATED* handlers functionality
/** @type {Pubsub} */
this.pubsub = PubsubAdapter(Pubsub, this, this._config.pubsub)
}
// Attach remaining APIs
// peer and content routing will automatically get modules from _modules and _dht
this.peerRouting = peerRouting(this)
this.contentRouting = contentRouting(this)
this.peerRouting = new PeerRouting(this)
this.contentRouting = new ContentRouting(this)
// Mount default protocols
ping.mount(this)
@ -208,13 +284,16 @@ class Libp2p extends EventEmitter {
*
* @param {string} eventName
* @param {...any} args
* @returns {void}
* @returns {boolean}
*/
emit (eventName, ...args) {
// TODO: do we still need this?
// @ts-ignore _events does not exist in libp2p
if (eventName === 'error' && !this._events.error) {
log.error(...args)
log.error(args)
return false
} else {
super.emit(eventName, ...args)
return super.emit(eventName, ...args)
}
}
@ -248,6 +327,11 @@ class Libp2p extends EventEmitter {
log('libp2p is stopping')
try {
this._isStarted = false
this.relay && this.relay.stop()
this.peerRouting.stop()
for (const service of this._discovery.values()) {
service.removeListener('peer', this._onDiscoveryPeer)
}
@ -275,7 +359,6 @@ class Libp2p extends EventEmitter {
this.emit('error', err)
}
}
this._isStarted = false
log('libp2p has stopped')
}
@ -284,9 +367,13 @@ class Libp2p extends EventEmitter {
* Imports the private key as 'self', if needed.
*
* @async
* @returns {void}
* @returns {Promise<void>}
*/
async loadKeychain () {
if (!this.keychain) {
return
}
try {
await this.keychain.findKeyByName('self')
} catch (err) {
@ -313,12 +400,12 @@ class Libp2p extends EventEmitter {
* peer will be added to the nodes `peerStore`
*
* @param {PeerId|Multiaddr|string} peer - The peer to dial
* @param {object} options
* @param {object} [options]
* @param {AbortSignal} [options.signal]
* @returns {Promise<Connection>}
*/
dial (peer, options) {
return this.dialProtocol(peer, null, options)
return this.dialProtocol(peer, [], options)
}
/**
@ -329,7 +416,7 @@ class Libp2p extends EventEmitter {
* @async
* @param {PeerId|Multiaddr|string} peer - The peer to dial
* @param {string[]|string} protocols
* @param {object} options
* @param {object} [options]
* @param {AbortSignal} [options.signal]
* @returns {Promise<Connection|*>}
*/
@ -349,7 +436,7 @@ class Libp2p extends EventEmitter {
}
// If a protocol was provided, create a new stream
if (protocols) {
if (protocols && protocols.length) {
return connection.newStream(protocols)
}
@ -361,34 +448,24 @@ class Libp2p extends EventEmitter {
* by transports to listen with the announce addresses.
* Duplicated addresses and noAnnounce addresses are filtered out.
*
* @returns {Array<Multiaddr>}
* @returns {Multiaddr[]}
*/
get multiaddrs () {
// Filter noAnnounce multiaddrs
const filterMa = this.addressManager.getNoAnnounceAddrs()
const announceAddrs = this.addressManager.getAnnounceAddrs()
if (announceAddrs.length) {
return announceAddrs
}
const announceFilter = this._options.addresses.announceFilter || ((multiaddrs) => multiaddrs)
// Create advertising list
return this.transportManager.getAddrs()
.concat(this.addressManager.getAnnounceAddrs())
.filter((ma, index, array) => {
// Filter out if repeated
if (array.findIndex((otherMa) => otherMa.equals(ma)) !== index) {
return false
}
// Filter out if in noAnnounceMultiaddrs
if (filterMa.find((fm) => fm.equals(ma))) {
return false
}
return true
})
return announceFilter(this.transportManager.getAddrs())
}
/**
* Disconnects all connections to the given `peer`
*
* @param {PeerId|multiaddr|string} peer - the peer to close connections to
* @param {PeerId|Multiaddr|string} peer - the peer to close connections to
* @returns {Promise<void>}
*/
async hangUp (peer) {
@ -428,7 +505,7 @@ class Libp2p extends EventEmitter {
* Registers the `handler` for each protocol
*
* @param {string[]|string} protocols
* @param {function({ connection:*, stream:*, protocol:string })} handler
* @param {({ connection: Connection, stream: MuxedStream, protocol: string }) => void} handler
*/
handle (protocols, handler) {
protocols = Array.isArray(protocols) ? protocols : [protocols]
@ -436,10 +513,8 @@ class Libp2p extends EventEmitter {
this.upgrader.protocols.set(protocol, handler)
})
// Only push if libp2p is running
if (this.isStarted() && this.identifyService) {
this.identifyService.pushToPeerStore(this.peerStore)
}
// Add new protocols to self protocols in the Protobook
this.peerStore.protoBook.add(this.peerId, protocols)
}
/**
@ -454,15 +529,14 @@ class Libp2p extends EventEmitter {
this.upgrader.protocols.delete(protocol)
})
// Only push if libp2p is running
if (this.isStarted() && this.identifyService) {
this.identifyService.pushToPeerStore(this.peerStore)
}
// Remove protocols from self protocols in the Protobook
this.peerStore.protoBook.remove(this.peerId, protocols)
}
async _onStarting () {
// Listen on the provided transports
await this.transportManager.listen()
// Listen on the provided transports for the provided addresses
const addrs = this.addressManager.getListenAddrs()
await this.transportManager.listen(addrs)
// Start PeerStore
await this.peerStore.start()
@ -507,6 +581,11 @@ class Libp2p extends EventEmitter {
// Peer discovery
await this._setupPeerDiscovery()
// Relay
this.relay && this.relay.start()
this.peerRouting.start()
}
/**
@ -514,7 +593,7 @@ class Libp2p extends EventEmitter {
* Known peers may be emitted.
*
* @private
* @param {{ id: PeerId, multiaddrs: Array<Multiaddr>, protocols: Array<string> }} peer
* @param {{ id: PeerId, multiaddrs: Multiaddr[], protocols: string[] }} peer
*/
_onDiscoveryPeer (peer) {
if (peer.id.toB58String() === this.peerId.toB58String()) {
@ -592,7 +671,9 @@ class Libp2p extends EventEmitter {
// Transport modules with discovery
for (const Transport of this.transportManager.getTransports()) {
// @ts-ignore Transport interface does not include discovery
if (Transport.discovery) {
// @ts-ignore Transport interface does not include discovery
setupService(Transport.discovery)
}
}
@ -601,22 +682,4 @@ class Libp2p extends EventEmitter {
}
}
/**
* Like `new Libp2p(options)` except it will create a `PeerId`
* instance if one is not provided in options.
*
* @param {object} options - Libp2p configuration options
* @returns {Promise<Libp2p>}
*/
Libp2p.create = async function create (options = {}) {
if (options.peerId) {
return new Libp2p(options)
}
const peerId = await PeerId.create()
options.peerId = peerId
return new Libp2p(options)
}
module.exports = Libp2p

View File

@ -1,21 +1,33 @@
'use strict'
const debug = require('debug')
const log = Object.assign(debug('libp2p:plaintext'), {
error: debug('libp2p:plaintext:err')
})
const handshake = require('it-handshake')
const lp = require('it-length-prefixed')
const PeerId = require('peer-id')
const debug = require('debug')
const log = debug('libp2p:plaintext')
log.error = debug('libp2p:plaintext:error')
const { UnexpectedPeerError, InvalidCryptoExchangeError } = require('libp2p-interfaces/src/crypto/errors')
const { Exchange, KeyType } = require('./proto')
const protocol = '/plaintext/2.0.0'
/**
* @typedef {import('libp2p-interfaces/src/connection').Connection} Connection
*/
function lpEncodeExchange (exchange) {
const pb = Exchange.encode(exchange)
return lp.encode.single(pb)
}
/**
* Encrypt connection.
*
* @param {PeerId} localId
* @param {Connection} conn
* @param {PeerId} [remoteId]
*/
async function encrypt (localId, conn, remoteId) {
const shake = handshake(conn)

View File

@ -8,6 +8,8 @@ const errcode = require('err-code')
const uint8ArrayFromString = require('uint8arrays/from-string')
const uint8ArrayToString = require('uint8arrays/to-string')
const privates = new WeakMap()
/**
* Cryptographic Message Syntax (aka PKCS #7)
*
@ -21,14 +23,16 @@ class CMS {
/**
* Creates a new instance with a keychain
*
* @param {Keychain} keychain - the available keys
* @param {import('./index')} keychain - the available keys
* @param {string} dek
*/
constructor (keychain) {
constructor (keychain, dek) {
if (!keychain) {
throw errcode(new Error('keychain is required'), 'ERR_KEYCHAIN_REQUIRED')
}
this.keychain = keychain
privates.set(this, { dek })
}
/**
@ -38,7 +42,7 @@ class CMS {
*
* @param {string} name - The local key name.
* @param {Uint8Array} plain - The data to encrypt.
* @returns {undefined}
* @returns {Promise<Uint8Array>}
*/
async encrypt (name, plain) {
if (!(plain instanceof Uint8Array)) {
@ -47,7 +51,9 @@ class CMS {
const key = await this.keychain.findKeyByName(name)
const pem = await this.keychain._getPrivateKey(name)
const privateKey = forge.pki.decryptRsaPrivateKey(pem, this.keychain._())
/** @type {string} */
const dek = privates.get(this).dek
const privateKey = forge.pki.decryptRsaPrivateKey(pem, dek)
const certificate = await certificateForKey(key, privateKey)
// create a p7 enveloped message
@ -68,7 +74,7 @@ class CMS {
* exists, an Error is returned with the property 'missingKeys'. It is array of key ids.
*
* @param {Uint8Array} cmsData - The CMS encrypted data to decrypt.
* @returns {undefined}
* @returns {Promise<Uint8Array>}
*/
async decrypt (cmsData) {
if (!(cmsData instanceof Uint8Array)) {
@ -114,8 +120,14 @@ class CMS {
}
const key = await this.keychain.findKeyById(r.keyId)
if (!key) {
throw errcode(new Error('No key available to decrypto'), 'ERR_NO_KEY')
}
const pem = await this.keychain._getPrivateKey(key.name)
const privateKey = forge.pki.decryptRsaPrivateKey(pem, this.keychain._())
const dek = privates.get(this).dek
const privateKey = forge.pki.decryptRsaPrivateKey(pem, dek)
cms.decrypt(r.recipient, privateKey)
return uint8ArrayFromString(cms.content.getBytes(), 'ascii')
}

View File

@ -4,7 +4,7 @@
const sanitize = require('sanitize-filename')
const mergeOptions = require('merge-options')
const crypto = require('libp2p-crypto')
const DS = require('interface-datastore')
const Datastore = require('interface-datastore')
const CMS = require('./cms')
const errcode = require('err-code')
const { Number } = require('ipfs-utils/src/globalthis')
@ -13,8 +13,14 @@ const uint8ArrayFromString = require('uint8arrays/from-string')
require('node-forge/lib/sha512')
/**
* @typedef {import('peer-id')} PeerId
* @typedef {import('interface-datastore/src/key')} Key
*/
const keyPrefix = '/pkcs8/'
const infoPrefix = '/info/'
const privates = new WeakMap()
// NIST SP 800-132
const NIST = {
@ -45,7 +51,8 @@ function validateKeyName (name) {
* This assumes than an error indicates that the keychain is under attack. Delay returning an
* error to make brute force attacks harder.
*
* @param {string | Error} err - The error
* @param {string|Error} err - The error
* @returns {Promise<never>}
* @private
*/
async function throwDelayed (err) {
@ -61,29 +68,28 @@ async function throwDelayed (err) {
* Converts a key name into a datastore name.
*
* @param {string} name
* @returns {DS.Key}
* @returns {Key}
* @private
*/
function DsName (name) {
return new DS.Key(keyPrefix + name)
return new Datastore.Key(keyPrefix + name)
}
/**
* Converts a key name into a datastore info name.
*
* @param {string} name
* @returns {DS.Key}
* @returns {Key}
* @private
*/
function DsInfoName (name) {
return new DS.Key(infoPrefix + name)
return new Datastore.Key(infoPrefix + name)
}
/**
* Information about a key.
*
* @typedef {Object} KeyInfo
*
* @property {string} id - The universally unique key id.
* @property {string} name - The local key name.
*/
@ -100,8 +106,9 @@ class Keychain {
/**
* Creates a new instance of a key chain.
*
* @param {DS} store - where the key are.
* @param {object} options - ???
* @param {Datastore} store - where the key are.
* @param {object} options
* @class
*/
constructor (store, options) {
if (!store) {
@ -132,7 +139,7 @@ class Keychain {
this.opts.dek.keyLength,
this.opts.dek.hash) : ''
Object.defineProperty(this, '_', { value: () => dek })
privates.set(this, { dek })
}
/**
@ -146,13 +153,13 @@ class Keychain {
* @returns {CMS}
*/
get cms () {
return new CMS(this)
return new CMS(this, privates.get(this).dek)
}
/**
* Generates the options for a keychain. A random salt is produced.
*
* @returns {object}
* @returns {Object}
*/
static generateOptions () {
const options = Object.assign({}, defaultOptions)
@ -165,7 +172,7 @@ class Keychain {
* Gets an object that can encrypt/decrypt protected data.
* The default options for a keychain.
*
* @returns {object}
* @returns {Object}
*/
static get options () {
return defaultOptions
@ -176,10 +183,10 @@ class Keychain {
*
* @param {string} name - The local key name; cannot already exist.
* @param {string} type - One of the key types; 'rsa'.
* @param {int} [size] - The key size in bits. Used for rsa keys only.
* @returns {KeyInfo}
* @param {number} [size = 2048] - The key size in bits. Used for rsa keys only.
* @returns {Promise<KeyInfo>}
*/
async createKey (name, type, size) {
async createKey (name, type, size = 2048) {
const self = this
if (!validateKeyName(name) || name === 'self') {
@ -206,9 +213,12 @@ class Keychain {
let keyInfo
try {
// @ts-ignore Differences between several crypto return types need to be fixed in libp2p-crypto
const keypair = await crypto.keys.generateKeyPair(type, size)
const kid = await keypair.id()
const pem = await keypair.export(this._())
/** @type {string} */
const dek = privates.get(this).dek
const pem = await keypair.export(dek)
keyInfo = {
name: name,
id: kid
@ -228,7 +238,7 @@ class Keychain {
/**
* List all the keys.
*
* @returns {KeyInfo[]}
* @returns {Promise<KeyInfo[]>}
*/
async listKeys () {
const self = this
@ -248,7 +258,7 @@ class Keychain {
* Find a key by it's id.
*
* @param {string} id - The universally unique key identifier.
* @returns {KeyInfo}
* @returns {Promise<KeyInfo|undefined>}
*/
async findKeyById (id) {
try {
@ -263,7 +273,7 @@ class Keychain {
* Find a key by it's name.
*
* @param {string} name - The local key name.
* @returns {KeyInfo}
* @returns {Promise<KeyInfo>}
*/
async findKeyByName (name) {
if (!validateKeyName(name)) {
@ -283,7 +293,7 @@ class Keychain {
* Remove an existing key.
*
* @param {string} name - The local key name; must already exist.
* @returns {KeyInfo}
* @returns {Promise<KeyInfo>}
*/
async removeKey (name) {
const self = this
@ -304,7 +314,7 @@ class Keychain {
*
* @param {string} oldName - The old local key name; must already exist.
* @param {string} newName - The new local key name; must not already exist.
* @returns {KeyInfo}
* @returns {Promise<KeyInfo>}
*/
async renameKey (oldName, newName) {
const self = this
@ -345,7 +355,7 @@ class Keychain {
*
* @param {string} name - The local key name; must already exist.
* @param {string} password - The password
* @returns {string}
* @returns {Promise<string>}
*/
async exportKey (name, password) {
if (!validateKeyName(name)) {
@ -359,7 +369,9 @@ class Keychain {
try {
const res = await this.store.get(dsname)
const pem = uint8ArrayToString(res)
const privateKey = await crypto.keys.import(pem, this._())
/** @type {string} */
const dek = privates.get(this).dek
const privateKey = await crypto.keys.import(pem, dek)
return privateKey.export(password)
} catch (err) {
return throwDelayed(err)
@ -372,7 +384,7 @@ class Keychain {
* @param {string} name - The local key name; must not already exist.
* @param {string} pem - The PEM encoded PKCS #8 string
* @param {string} password - The password.
* @returns {KeyInfo}
* @returns {Promise<KeyInfo>}
*/
async importKey (name, pem, password) {
const self = this
@ -396,7 +408,9 @@ class Keychain {
let kid
try {
kid = await privateKey.id()
pem = await privateKey.export(this._())
/** @type {string} */
const dek = privates.get(this).dek
pem = await privateKey.export(dek)
} catch (err) {
return throwDelayed(err)
}
@ -413,6 +427,13 @@ class Keychain {
return keyInfo
}
/**
* Import a peer key
*
* @param {string} name - The local key name; must not already exist.
* @param {PeerId} peer - The PEM encoded PKCS #8 string
* @returns {Promise<KeyInfo>}
*/
async importPeer (name, peer) {
const self = this
if (!validateKeyName(name)) {
@ -429,7 +450,9 @@ class Keychain {
try {
const kid = await privateKey.id()
const pem = await privateKey.export(this._())
/** @type {string} */
const dek = privates.get(this).dek
const pem = await privateKey.export(dek)
const keyInfo = {
name: name,
id: kid
@ -448,8 +471,7 @@ class Keychain {
* Gets the private key as PEM encoded PKCS #8 string.
*
* @param {string} name
* @returns {string}
* @private
* @returns {Promise<string>}
*/
async _getPrivateKey (name) {
if (!validateKeyName(name)) {

View File

@ -1,3 +1,4 @@
// @ts-nocheck
'use strict'
require('node-forge/lib/x509')

View File

@ -1,7 +1,8 @@
// @ts-nocheck
'use strict'
const mergeOptions = require('merge-options')
const pipe = require('it-pipe')
const { pipe } = require('it-pipe')
const { tap } = require('streaming-iterables')
const oldPeerLRU = require('./old-peers')
const { METRICS: defaultOptions } = require('../constants')
@ -17,15 +18,26 @@ const directionToEvent = {
out: 'dataSent'
}
/**
* @typedef {import('peer-id')} PeerId
* @typedef {import('libp2p-interfaces/src/transport/types').MultiaddrConnection} MultiaddrConnection
*/
/**
* @typedef MetricsProperties
* @property {import('../connection-manager')} connectionManager
*
* @typedef MetricsOptions
* @property {number} [computeThrottleMaxQueueSize = defaultOptions.computeThrottleMaxQueueSize]
* @property {number} [computeThrottleTimeout = defaultOptions.computeThrottleTimeout]
* @property {number[]} [movingAverageIntervals = defaultOptions.movingAverageIntervals]
* @property {number} [maxOldPeersRetention = defaultOptions.maxOldPeersRetention]
*/
class Metrics {
/**
*
* @param {object} options
* @param {ConnectionManager} options.connectionManager
* @param {number} options.computeThrottleMaxQueueSize
* @param {number} options.computeThrottleTimeout
* @param {Array<number>} options.movingAverageIntervals
* @param {number} options.maxOldPeersRetention
* @class
* @param {MetricsProperties & MetricsOptions} options
*/
constructor (options) {
this._options = mergeOptions(defaultOptions, options)
@ -76,7 +88,7 @@ class Metrics {
/**
* Returns a list of `PeerId` strings currently being tracked
*
* @returns {Array<string>}
* @returns {string[]}
*/
get peers () {
return Array.from(this._peerStats.keys())
@ -97,7 +109,7 @@ class Metrics {
/**
* Returns a list of all protocol strings currently being tracked.
*
* @returns {Array<string>}
* @returns {string[]}
*/
get protocols () {
return Array.from(this._protocolStats.keys())
@ -176,6 +188,7 @@ class Metrics {
*
* @param {PeerId} placeholder - A peerId string
* @param {PeerId} peerId
* @returns {void}
*/
updatePlaceholder (placeholder, peerId) {
if (!this._running) return
@ -205,10 +218,10 @@ class Metrics {
* with the placeholder string returned from here, and the known `PeerId`.
*
* @param {Object} options
* @param {{ sink: function(*), source: function() }} options.stream - A duplex iterable stream
* @param {MultiaddrConnection} options.stream - A duplex iterable stream
* @param {PeerId} [options.remotePeer] - The id of the remote peer that's connected
* @param {string} [options.protocol] - The protocol the stream is running
* @returns {string} The peerId string or placeholder string
* @returns {MultiaddrConnection} The peerId string or placeholder string
*/
trackStream ({ stream, remotePeer, protocol }) {
const metrics = this

View File

@ -6,9 +6,10 @@ const LRU = require('hashlru')
* Creates and returns a Least Recently Used Cache
*
* @param {number} maxSize
* @returns {LRUCache}
* @returns {any}
*/
module.exports = (maxSize) => {
// @ts-ignore LRU expression is not callable
const patched = LRU(maxSize)
patched.delete = patched.remove
return patched

View File

@ -1,17 +1,19 @@
// @ts-nocheck
'use strict'
const EventEmitter = require('events')
const { EventEmitter } = require('events')
const Big = require('bignumber.js')
const MovingAverage = require('moving-average')
const retimer = require('retimer')
/**
* A queue based manager for stat processing
*
* @param {Array<string>} initialCounters
* @param {any} options
*/
class Stats extends EventEmitter {
/**
* A queue based manager for stat processing
*
* @class
* @param {string[]} initialCounters
* @param {any} options
*/
constructor (initialCounters, options) {
super()
@ -21,6 +23,7 @@ class Stats extends EventEmitter {
this._frequencyLastTime = Date.now()
this._frequencyAccumulators = {}
this._movingAverages = {}
this._update = this._update.bind(this)
@ -68,7 +71,7 @@ class Stats extends EventEmitter {
/**
* Returns a clone of the current stats.
*
* @returns {Map<string, Stat>}
* @returns {Object}
*/
get snapshot () {
return Object.assign({}, this._stats)
@ -77,7 +80,7 @@ class Stats extends EventEmitter {
/**
* Returns a clone of the internal movingAverages
*
* @returns {Array<MovingAverage>}
* @returns {MovingAverage}
*/
get movingAverages () {
return Object.assign({}, this._movingAverages)
@ -229,7 +232,7 @@ class Stats extends EventEmitter {
* will be updated or initialized if they don't already exist.
*
* @private
* @param {Array<string, number>} op
* @param {{string, number}[]} op
* @throws {InvalidNumber}
* @returns {void}
*/
@ -238,7 +241,7 @@ class Stats extends EventEmitter {
const inc = op[1]
if (typeof inc !== 'number') {
throw new Error('invalid increment number:', inc)
throw new Error(`invalid increment number: ${inc}`)
}
let n

View File

@ -1,40 +1,128 @@
'use strict'
const debug = require('debug')
const log = Object.assign(debug('libp2p:peer-routing'), {
error: debug('libp2p:peer-routing:err')
})
const errCode = require('err-code')
const all = require('it-all')
const pAny = require('p-any')
const {
setDelayedInterval,
clearDelayedInterval
} = require('set-delayed-interval')
module.exports = (node) => {
const routers = node._modules.peerRouting || []
/**
* @typedef {import('peer-id')} PeerId
* @typedef {import('multiaddr')} Multiaddr
*/
class PeerRouting {
/**
* @class
* @param {import('./')} libp2p
*/
constructor (libp2p) {
this._peerId = libp2p.peerId
this._peerStore = libp2p.peerStore
this._routers = libp2p._modules.peerRouting || []
// If we have the dht, make it first
if (node._dht) {
routers.unshift(node._dht)
// If we have the dht, make it first
if (libp2p._dht) {
this._routers.unshift(libp2p._dht)
}
this._refreshManagerOptions = libp2p._options.peerRouting.refreshManager
this._findClosestPeersTask = this._findClosestPeersTask.bind(this)
}
return {
/**
* Iterates over all peer routers in series to find the given peer.
*
* @param {string} id - The id of the peer to find
* @param {object} [options]
* @param {number} [options.timeout] - How long the query should run
* @returns {Promise<{ id: PeerId, multiaddrs: Multiaddr[] }>}
*/
findPeer: async (id, options) => { // eslint-disable-line require-await
if (!routers.length) {
throw errCode(new Error('No peer routers available'), 'NO_ROUTERS_AVAILABLE')
/**
* Start peer routing service.
*/
start () {
if (!this._routers.length || this._timeoutId || !this._refreshManagerOptions.enabled) {
return
}
this._timeoutId = setDelayedInterval(
this._findClosestPeersTask, this._refreshManagerOptions.interval, this._refreshManagerOptions.bootDelay
)
}
/**
* Recurrent task to find closest peers and add their addresses to the Address Book.
*/
async _findClosestPeersTask () {
try {
for await (const { id, multiaddrs } of this.getClosestPeers(this._peerId.id)) {
this._peerStore.addressBook.add(id, multiaddrs)
}
} catch (err) {
log.error(err)
}
}
/**
* Stop peer routing service.
*/
stop () {
clearDelayedInterval(this._timeoutId)
}
/**
* Iterates over all peer routers in series to find the given peer.
*
* @param {PeerId} id - The id of the peer to find
* @param {object} [options]
* @param {number} [options.timeout] - How long the query should run
* @returns {Promise<{ id: PeerId, multiaddrs: Multiaddr[] }>}
*/
async findPeer (id, options) { // eslint-disable-line require-await
if (!this._routers.length) {
throw errCode(new Error('No peer routers available'), 'NO_ROUTERS_AVAILABLE')
}
return pAny(this._routers.map(async (router) => {
const result = await router.findPeer(id, options)
// If we don't have a result, we need to provide an error to keep trying
if (!result || Object.keys(result).length === 0) {
throw errCode(new Error('not found'), 'NOT_FOUND')
}
return pAny(routers.map(async (router) => {
const result = await router.findPeer(id, options)
return result
}))
}
// If we don't have a result, we need to provide an error to keep trying
if (!result || Object.keys(result).length === 0) {
/**
* Attempt to find the closest peers on the network to the given key.
*
* @param {Uint8Array} key - A CID like key
* @param {Object} [options]
* @param {number} [options.timeout=30e3] - How long the query can take.
* @returns {AsyncIterable<{ id: PeerId, multiaddrs: Multiaddr[] }>}
*/
async * getClosestPeers (key, options = { timeout: 30e3 }) {
if (!this._routers.length) {
throw errCode(new Error('No peer routers available'), 'NO_ROUTERS_AVAILABLE')
}
const result = await pAny(
this._routers.map(async (router) => {
const peers = await all(router.getClosestPeers(key, options))
if (!peers || !peers.length) {
throw errCode(new Error('not found'), 'NOT_FOUND')
}
return peers
})
)
return result
}))
for (const peer of result) {
yield peer
}
}
}
module.exports = PeerRouting

View File

@ -1,9 +1,10 @@
'use strict'
const errcode = require('err-code')
const debug = require('debug')
const log = debug('libp2p:peer-store:address-book')
log.error = debug('libp2p:peer-store:address-book:error')
const log = Object.assign(debug('libp2p:peer-store:address-book'), {
error: debug('libp2p:peer-store:address-book:err')
})
const errcode = require('err-code')
const multiaddr = require('multiaddr')
const PeerId = require('peer-id')
@ -17,35 +18,31 @@ const {
const Envelope = require('../record/envelope')
/**
* The AddressBook is responsible for keeping the known multiaddrs
* of a peer.
* @typedef {import('multiaddr')} Multiaddr
* @typedef {import('./')} PeerStore
*/
/**
* @typedef {Object} Address
* @property {Multiaddr} multiaddr peer multiaddr.
* @property {boolean} isCertified obtained from a signed peer record.
*
* @typedef {Object} CertifiedRecord
* @property {Uint8Array} raw raw envelope.
* @property {number} seqNumber seq counter.
*
* @typedef {Object} Entry
* @property {Address[]} addresses peer Addresses.
* @property {CertifiedRecord} record certified peer record.
*/
/**
* @extends {Book}
*/
class AddressBook extends Book {
/**
* Address object
* The AddressBook is responsible for keeping the known multiaddrs of a peer.
*
* @typedef {Object} Address
* @property {Multiaddr} multiaddr peer multiaddr.
* @property {boolean} isCertified obtained from a signed peer record.
*/
/**
* CertifiedRecord object
*
* @typedef {Object} CertifiedRecord
* @property {Uint8Array} raw raw envelope.
* @property {number} seqNumber seq counter.
*/
/**
* Entry object for the addressBook
*
* @typedef {Object} Entry
* @property {Array<Address>} addresses peer Addresses.
* @property {CertifiedRecord} record certified peer record.
*/
/**
* @class
* @param {PeerStore} peerStore
*/
@ -70,7 +67,7 @@ class AddressBook extends Book {
/**
* Map known peers to their known Address Entries.
*
* @type {Map<string, Array<Entry>>}
* @type {Map<string, Entry>}
*/
this.data = new Map()
}
@ -105,7 +102,7 @@ class AddressBook extends Book {
const peerId = peerRecord.peerId
const id = peerId.toB58String()
const entry = this.data.get(id) || {}
const entry = this.data.get(id) || { record: undefined }
const storedRecord = entry.record
// ensure seq is greater than, or equal to, the last received
@ -151,7 +148,7 @@ class AddressBook extends Book {
* Returns undefined if no record exists.
*
* @param {PeerId} peerId
* @returns {Promise<Envelope|void>}
* @returns {Promise<Envelope|void>|undefined}
*/
getPeerRecord (peerId) {
const raw = this.getRawEnvelope(peerId)
@ -171,7 +168,7 @@ class AddressBook extends Book {
*
* @override
* @param {PeerId} peerId
* @param {Array<Multiaddr>} multiaddrs
* @param {Multiaddr[]} multiaddrs
* @returns {AddressBook}
*/
set (peerId, multiaddrs) {
@ -181,22 +178,22 @@ class AddressBook extends Book {
}
const addresses = this._toAddresses(multiaddrs)
const id = peerId.toB58String()
const entry = this.data.get(id) || {}
const rec = entry.addresses
// Not replace multiaddrs
if (!addresses.length) {
return this
}
const id = peerId.toB58String()
const entry = this.data.get(id)
// Already knows the peer
if (rec && rec.length === addresses.length) {
const intersection = rec.filter((addr) => addresses.some((newAddr) => addr.multiaddr.equals(newAddr.multiaddr)))
if (entry && entry.addresses && entry.addresses.length === addresses.length) {
const intersection = entry.addresses.filter((addr) => addresses.some((newAddr) => addr.multiaddr.equals(newAddr.multiaddr)))
// Are new addresses equal to the old ones?
// If yes, no changes needed!
if (intersection.length === rec.length) {
if (intersection.length === entry.addresses.length) {
log(`the addresses provided to store are equal to the already stored for ${id}`)
return this
}
@ -204,12 +201,12 @@ class AddressBook extends Book {
this._setData(peerId, {
addresses,
record: entry.record
record: entry && entry.record
})
log(`stored provided multiaddrs for ${id}`)
// Notify the existance of a new peer
if (!rec) {
if (!entry) {
this._ps.emit('peer', peerId)
}
@ -221,7 +218,7 @@ class AddressBook extends Book {
* If the peer is not known, it is set with the given addresses.
*
* @param {PeerId} peerId
* @param {Array<Multiaddr>} multiaddrs
* @param {Multiaddr[]} multiaddrs
* @returns {AddressBook}
*/
add (peerId, multiaddrs) {
@ -233,32 +230,33 @@ class AddressBook extends Book {
const addresses = this._toAddresses(multiaddrs)
const id = peerId.toB58String()
const entry = this.data.get(id) || {}
const rec = entry.addresses || []
const entry = this.data.get(id)
// Add recorded uniquely to the new array (Union)
rec.forEach((addr) => {
if (!addresses.find(r => r.multiaddr.equals(addr.multiaddr))) {
addresses.push(addr)
if (entry && entry.addresses) {
// Add recorded uniquely to the new array (Union)
entry.addresses.forEach((addr) => {
if (!addresses.find(r => r.multiaddr.equals(addr.multiaddr))) {
addresses.push(addr)
}
})
// If the recorded length is equal to the new after the unique union
// The content is the same, no need to update.
if (entry.addresses.length === addresses.length) {
log(`the addresses provided to store are already stored for ${id}`)
return this
}
})
// If the recorded length is equal to the new after the unique union
// The content is the same, no need to update.
if (rec && rec.length === addresses.length) {
log(`the addresses provided to store are already stored for ${id}`)
return this
}
this._setData(peerId, {
addresses,
record: entry.record
record: entry && entry.record
})
log(`added provided multiaddrs for ${id}`)
// Notify the existance of a new peer
if (!entry.addresses) {
if (!(entry && entry.addresses)) {
this._ps.emit('peer', peerId)
}
@ -270,7 +268,7 @@ class AddressBook extends Book {
*
* @override
* @param {PeerId} peerId
* @returns {Array<data>}
* @returns {Address[]|undefined}
*/
get (peerId) {
if (!PeerId.isPeerId(peerId)) {
@ -286,9 +284,9 @@ class AddressBook extends Book {
* Transforms received multiaddrs into Address.
*
* @private
* @param {Array<Multiaddr>} multiaddrs
* @param {Multiaddr[]} multiaddrs
* @param {boolean} [isCertified]
* @returns {Array<Address>}
* @returns {Address[]}
*/
_toAddresses (multiaddrs, isCertified = false) {
if (!multiaddrs) {
@ -319,20 +317,22 @@ class AddressBook extends Book {
* Returns `undefined` if there are no known multiaddrs for the given peer.
*
* @param {PeerId} peerId
* @returns {Array<Multiaddr>|undefined}
* @param {(addresses: Address[]) => Address[]} [addressSorter]
* @returns {Multiaddr[]|undefined}
*/
getMultiaddrsForPeer (peerId) {
getMultiaddrsForPeer (peerId, addressSorter = (ms) => ms) {
if (!PeerId.isPeerId(peerId)) {
throw errcode(new Error('peerId must be an instance of peer-id'), ERR_INVALID_PARAMETERS)
}
const entry = this.data.get(peerId.toB58String())
if (!entry || !entry.addresses) {
return undefined
}
return entry.addresses.map((address) => {
return addressSorter(
entry.addresses || []
).map((address) => {
const multiaddr = address.multiaddr
const idString = multiaddr.getPeerId()

View File

@ -10,16 +10,19 @@ const {
const passthrough = data => data
/**
* The Book is the skeleton for the PeerStore books.
* @typedef {import('./')} PeerStore
*/
class Book {
/**
* The Book is the skeleton for the PeerStore books.
*
* @class
* @param {Object} properties
* @param {PeerStore} properties.peerStore - PeerStore instance.
* @param {string} properties.eventName - Name of the event to emit by the PeerStore.
* @param {string} properties.eventProperty - Name of the property to emit by the PeerStore.
* @param {Function} [properties.eventTransformer] - Transformer function of the provided data for being emitted.
* @param {(data: any) => any[]} [properties.eventTransformer] - Transformer function of the provided data for being emitted.
*/
constructor ({ peerStore, eventName, eventProperty, eventTransformer = passthrough }) {
this._ps = peerStore
@ -30,7 +33,7 @@ class Book {
/**
* Map known peers to their data.
*
* @type {Map<string, Array<Data>}
* @type {Map<string, any[]|any>}
*/
this.data = new Map()
}
@ -39,7 +42,7 @@ class Book {
* Set known data of a provided peer.
*
* @param {PeerId} peerId
* @param {Array<Data>|Data} data
* @param {any[]|any} data
*/
set (peerId, data) {
throw errcode(new Error('set must be implemented by the subclass'), 'ERR_NOT_IMPLEMENTED')
@ -48,9 +51,9 @@ class Book {
/**
* Set data into the datastructure, persistence and emit it using the provided transformers.
*
* @private
* @protected
* @param {PeerId} peerId - peerId of the data to store
* @param {*} data - data to store.
* @param {any} data - data to store.
* @param {Object} [options] - storing options.
* @param {boolean} [options.emit = true] - emit the provided data.
* @returns {void}
@ -68,9 +71,9 @@ class Book {
/**
* Emit data.
*
* @private
* @protected
* @param {PeerId} peerId
* @param {*} data
* @param {any} [data]
*/
_emit (peerId, data) {
this._ps.emit(this.eventName, {
@ -84,7 +87,7 @@ class Book {
* Returns `undefined` if there is no available data for the given peer.
*
* @param {PeerId} peerId
* @returns {Array<Data>|undefined}
* @returns {any[]|any|undefined}
*/
get (peerId) {
if (!PeerId.isPeerId(peerId)) {
@ -93,6 +96,7 @@ class Book {
const rec = this.data.get(peerId.toB58String())
// @ts-ignore
return rec ? [...rec] : undefined
}

View File

@ -1,9 +1,6 @@
'use strict'
const errcode = require('err-code')
const debug = require('debug')
const log = debug('libp2p:peer-store')
log.error = debug('libp2p:peer-store:error')
const { EventEmitter } = require('events')
const PeerId = require('peer-id')
@ -14,11 +11,15 @@ const MetadataBook = require('./metadata-book')
const ProtoBook = require('./proto-book')
const {
ERR_INVALID_PARAMETERS
codes: { ERR_INVALID_PARAMETERS }
} = require('../errors')
/**
* Responsible for managing known peers, as well as their addresses, protocols and metadata.
* @typedef {import('./address-book').Address} Address
*/
/**
* @extends {EventEmitter}
*
* @fires PeerStore#peer Emitted when a new peer is added.
* @fires PeerStore#change:protocols Emitted when a known peer supports a different set of protocols.
@ -32,12 +33,14 @@ class PeerStore extends EventEmitter {
*
* @typedef {Object} Peer
* @property {PeerId} id peer's peer-id instance.
* @property {Array<Address>} addresses peer's addresses containing its multiaddrs and metadata.
* @property {Array<string>} protocols peer's supported protocols.
* @property {Map<string, Buffer>} metadata peer's metadata map.
* @property {Address[]} addresses peer's addresses containing its multiaddrs and metadata.
* @property {string[]} protocols peer's supported protocols.
* @property {Map<string, Uint8Array>|undefined} metadata peer's metadata map.
*/
/**
* Responsible for managing known peers, as well as their addresses, protocols and metadata.
*
* @param {object} options
* @param {PeerId} options.peerId
* @class
@ -121,7 +124,7 @@ class PeerStore extends EventEmitter {
* Get the stored information of a given peer.
*
* @param {PeerId} peerId
* @returns {Peer}
* @returns {Peer|undefined}
*/
get (peerId) {
if (!PeerId.isPeerId(peerId)) {

View File

@ -1,9 +1,10 @@
'use strict'
const errcode = require('err-code')
const debug = require('debug')
const log = debug('libp2p:peer-store:key-book')
log.error = debug('libp2p:peer-store:key-book:error')
const log = Object.assign(debug('libp2p:peer-store:key-book'), {
error: debug('libp2p:peer-store:key-book:err')
})
const errcode = require('err-code')
const PeerId = require('peer-id')
@ -14,10 +15,17 @@ const {
} = require('../errors')
/**
* The KeyBook is responsible for keeping the known public keys of a peer.
* @typedef {import('./')} PeerStore
* @typedef {import('libp2p-crypto').PublicKey} PublicKey
*/
/**
* @extends {Book}
*/
class KeyBook extends Book {
/**
* The KeyBook is responsible for keeping the known public keys of a peer.
*
* @class
* @param {PeerStore} peerStore
*/
@ -42,7 +50,7 @@ class KeyBook extends Book {
*
* @override
* @param {PeerId} peerId
* @param {RsaPublicKey|Ed25519PublicKey|Secp256k1PublicKey} publicKey
* @param {PublicKey} publicKey
* @returns {KeyBook}
*/
set (peerId, publicKey) {
@ -72,7 +80,7 @@ class KeyBook extends Book {
*
* @override
* @param {PeerId} peerId
* @returns {RsaPublicKey|Ed25519PublicKey|Secp256k1PublicKey}
* @returns {PublicKey | undefined}
*/
get (peerId) {
if (!PeerId.isPeerId(peerId)) {

View File

@ -1,9 +1,10 @@
'use strict'
const errcode = require('err-code')
const debug = require('debug')
const log = debug('libp2p:peer-store:proto-book')
log.error = debug('libp2p:peer-store:proto-book:error')
const log = Object.assign(debug('libp2p:peer-store:proto-book'), {
error: debug('libp2p:peer-store:proto-book:err')
})
const errcode = require('err-code')
const uint8ArrayEquals = require('uint8arrays/equals')
const PeerId = require('peer-id')
@ -15,13 +16,19 @@ const {
} = require('../errors')
/**
* The MetadataBook is responsible for keeping the known supported
* protocols of a peer.
* @typedef {import('./')} PeerStore
*/
/**
* @extends {Book}
*
* @fires MetadataBook#change:metadata
*/
class MetadataBook extends Book {
/**
* The MetadataBook is responsible for keeping the known supported
* protocols of a peer.
*
* @class
* @param {PeerStore} peerStore
*/
@ -51,8 +58,9 @@ class MetadataBook extends Book {
* @param {PeerId} peerId
* @param {string} key - metadata key
* @param {Uint8Array} value - metadata value
* @returns {ProtoBook}
* @returns {MetadataBook}
*/
// @ts-ignore override with more then the parameters expected in Book
set (peerId, key, value) {
if (!PeerId.isPeerId(peerId)) {
log.error('peerId must be an instance of peer-id to store data')
@ -95,7 +103,7 @@ class MetadataBook extends Book {
* Get the known data of a provided peer.
*
* @param {PeerId} peerId
* @returns {Map<string, Uint8Array>}
* @returns {Map<string, Uint8Array>|undefined}
*/
get (peerId) {
if (!PeerId.isPeerId(peerId)) {
@ -110,7 +118,7 @@ class MetadataBook extends Book {
*
* @param {PeerId} peerId
* @param {string} key
* @returns {Uint8Array}
* @returns {Uint8Array | undefined}
*/
getValue (peerId, key) {
if (!PeerId.isPeerId(peerId)) {

View File

@ -1,9 +1,9 @@
'use strict'
const debug = require('debug')
const log = debug('libp2p:persistent-peer-store')
log.error = debug('libp2p:persistent-peer-store:error')
const log = Object.assign(debug('libp2p:persistent-peer-store'), {
error: debug('libp2p:persistent-peer-store:err')
})
const { Key } = require('interface-datastore')
const multiaddr = require('multiaddr')
const PeerId = require('peer-id')
@ -21,16 +21,22 @@ const {
const Addresses = require('./pb/address-book.proto')
const Protocols = require('./pb/proto-book.proto')
/**
* @typedef {Object} PersistentPeerStoreProperties
* @property {PeerId} peerId
* @property {any} datastore
*
* @typedef {Object} PersistentPeerStoreOptions
* @property {number} [threshold = 5] - Number of dirty peers allowed before commit data.
*/
/**
* Responsible for managing the persistence of data in the PeerStore.
*/
class PersistentPeerStore extends PeerStore {
/**
* @class
* @param {Object} properties
* @param {PeerId} properties.peerId
* @param {Datastore} properties.datastore - Datastore to persist data.
* @param {number} [properties.threshold = 5] - Number of dirty peers allowed before commit data.
* @param {PersistentPeerStoreProperties & PersistentPeerStoreOptions} properties
*/
constructor ({ peerId, datastore, threshold = 5 }) {
super({ peerId })
@ -340,6 +346,7 @@ class PersistentPeerStore extends PeerStore {
case 'addrs':
decoded = Addresses.decode(value)
// @ts-ignore protected function
this.addressBook._setData(
peerId,
{
@ -357,6 +364,7 @@ class PersistentPeerStore extends PeerStore {
case 'keys':
decoded = await PeerId.createFromPubKey(value)
// @ts-ignore protected function
this.keyBook._setData(
decoded,
decoded,
@ -372,6 +380,7 @@ class PersistentPeerStore extends PeerStore {
case 'protos':
decoded = Protocols.decode(value)
// @ts-ignore protected function
this.protoBook._setData(
peerId,
new Set(decoded.protocols),

View File

@ -1,10 +1,10 @@
'use strict'
const errcode = require('err-code')
const debug = require('debug')
const log = debug('libp2p:peer-store:proto-book')
log.error = debug('libp2p:peer-store:proto-book:error')
const log = Object.assign(debug('libp2p:peer-store:proto-book'), {
error: debug('libp2p:peer-store:proto-book:err')
})
const errcode = require('err-code')
const PeerId = require('peer-id')
const Book = require('./book')
@ -14,13 +14,19 @@ const {
} = require('../errors')
/**
* The ProtoBook is responsible for keeping the known supported
* protocols of a peer.
* @typedef {import('./')} PeerStore
*/
/**
* @extends {Book}
*
* @fires ProtoBook#change:protocols
*/
class ProtoBook extends Book {
/**
* The ProtoBook is responsible for keeping the known supported
* protocols of a peer.
*
* @class
* @param {PeerStore} peerStore
*/
@ -50,7 +56,7 @@ class ProtoBook extends Book {
*
* @override
* @param {PeerId} peerId
* @param {Array<string>} protocols
* @param {string[]} protocols
* @returns {ProtoBook}
*/
set (peerId, protocols) {
@ -88,7 +94,7 @@ class ProtoBook extends Book {
* If the peer was not known before, it will be added.
*
* @param {PeerId} peerId
* @param {Array<string>} protocols
* @param {string[]} protocols
* @returns {ProtoBook}
*/
add (peerId, protocols) {
@ -112,13 +118,50 @@ class ProtoBook extends Book {
return this
}
protocols = [...newSet]
this._setData(peerId, newSet)
log(`added provided protocols for ${id}`)
return this
}
/**
* Removes known protocols of a provided peer.
* If the protocols did not exist before, nothing will be done.
*
* @param {PeerId} peerId
* @param {string[]} protocols
* @returns {ProtoBook}
*/
remove (peerId, protocols) {
if (!PeerId.isPeerId(peerId)) {
log.error('peerId must be an instance of peer-id to store data')
throw errcode(new Error('peerId must be an instance of peer-id'), ERR_INVALID_PARAMETERS)
}
if (!protocols) {
log.error('protocols must be provided to store data')
throw errcode(new Error('protocols must be provided'), ERR_INVALID_PARAMETERS)
}
const id = peerId.toB58String()
const recSet = this.data.get(id)
if (recSet) {
const newSet = new Set([
...recSet
].filter((p) => !protocols.includes(p)))
// Any protocol removed?
if (recSet.size === newSet.size) {
return this
}
this._setData(peerId, newSet)
log(`removed provided protocols for ${id}`)
}
return this
}
}
module.exports = ProtoBook

View File

@ -1,30 +1,39 @@
'use strict'
const debug = require('debug')
const log = debug('libp2p-ping')
log.error = debug('libp2p-ping:error')
const log = Object.assign(debug('libp2p:ping'), {
error: debug('libp2p:ping:err')
})
const errCode = require('err-code')
const crypto = require('libp2p-crypto')
const pipe = require('it-pipe')
const { pipe } = require('it-pipe')
const { toBuffer } = require('it-buffer')
const { collect, take } = require('streaming-iterables')
const equals = require('uint8arrays/equals')
const { PROTOCOL, PING_LENGTH } = require('./constants')
/**
* @typedef {import('../')} Libp2p
* @typedef {import('multiaddr')} Multiaddr
* @typedef {import('peer-id')} PeerId
*/
/**
* Ping a given peer and wait for its response, getting the operation latency.
*
* @param {Libp2p} node
* @param {PeerId|multiaddr} peer
* @param {PeerId|Multiaddr} peer
* @returns {Promise<number>}
*/
async function ping (node, peer) {
// @ts-ignore multiaddr might not have toB58String
log('dialing %s to %s', PROTOCOL, peer.toB58String ? peer.toB58String() : peer)
const { stream } = await node.dialProtocol(peer, PROTOCOL)
const start = new Date()
const start = Date.now()
const data = crypto.randomBytes(PING_LENGTH)
const [result] = await pipe(
@ -36,7 +45,7 @@ async function ping (node, peer) {
)
const end = Date.now()
if (!data.equals(result)) {
if (!equals(data, result)) {
throw errCode(new Error('Received wrong ping ack'), 'ERR_WRONG_PING_ACK')
}

View File

@ -1,16 +1,17 @@
'use strict'
const debug = require('debug')
const log = Object.assign(debug('libp2p:pnet'), {
trace: debug('libp2p:pnet:trace'),
error: debug('libp2p:pnet:err')
})
const Errors = require('./errors')
const xsalsa20 = require('xsalsa20')
const KEY_LENGTH = require('./key-generator').KEY_LENGTH
const uint8ArrayFromString = require('uint8arrays/from-string')
const uint8ArrayToString = require('uint8arrays/to-string')
const log = debug('libp2p:pnet')
log.trace = debug('libp2p:pnet:trace')
log.error = debug('libp2p:pnet:err')
/**
* Creates a stream iterable to encrypt messages in a private network
*

View File

@ -1,12 +1,16 @@
'use strict'
const pipe = require('it-pipe')
const debug = require('debug')
const log = Object.assign(debug('libp2p:pnet'), {
error: debug('libp2p:pnet:err')
})
const { pipe } = require('it-pipe')
const errcode = require('err-code')
const duplexPair = require('it-pair/duplex')
const crypto = require('libp2p-crypto')
const Errors = require('./errors')
const {
ERR_INVALID_PARAMETERS
codes: { ERR_INVALID_PARAMETERS }
} = require('../errors')
const {
createBoxStream,
@ -15,16 +19,16 @@ const {
} = require('./crypto')
const handshake = require('it-handshake')
const { NONCE_LENGTH } = require('./key-generator')
const debug = require('debug')
const log = debug('libp2p:pnet')
log.error = debug('libp2p:pnet:err')
/**
* Takes a Private Shared Key (psk) and provides a `protect` method
* for wrapping existing connections in a private encryption stream
* @typedef {import('libp2p-interfaces/src/transport/types').MultiaddrConnection} MultiaddrConnection
*/
class Protector {
/**
* Takes a Private Shared Key (psk) and provides a `protect` method
* for wrapping existing connections in a private encryption stream.
*
* @param {Uint8Array} keyBuffer - The private shared key buffer
* @class
*/
@ -39,8 +43,8 @@ class Protector {
* between its two peers from the PSK the Protector instance was
* created with.
*
* @param {Connection} connection - The connection to protect
* @returns {*} A protected duplex iterable
* @param {MultiaddrConnection} connection - The connection to protect
* @returns {Promise<MultiaddrConnection>} A protected duplex iterable
*/
async protect (connection) {
if (!connection) {

View File

@ -22,6 +22,8 @@ module.exports = generate
module.exports.NONCE_LENGTH = 24
module.exports.KEY_LENGTH = KEY_LENGTH
// @ts-ignore This condition will always return 'false' since the types 'Module | undefined'
if (require.main === module) {
// @ts-ignore
generate(process.stdout)
}

View File

@ -1,42 +1,54 @@
'use strict'
/**
* @typedef {import('libp2p-interfaces/src/pubsub').InMessage} InMessage
* @typedef {import('libp2p-interfaces/src/pubsub')} PubsubRouter
*/
// Pubsub adapter to keep API with handlers while not removed.
module.exports = (PubsubRouter, libp2p, options) => {
class Pubsub extends PubsubRouter {
/**
* Subscribes to a given topic.
*
* @override
* @param {string} topic
* @param {function(msg: InMessage)} [handler]
* @returns {void}
*/
subscribe (topic, handler) {
// Bind provided handler
handler && this.on(topic, handler)
super.subscribe(topic)
function pubsubAdapter (PubsubRouter, libp2p, options) {
const pubsub = new PubsubRouter(libp2p, options)
pubsub._subscribeAdapter = pubsub.subscribe
pubsub._unsubscribeAdapter = pubsub.unsubscribe
/**
* Subscribes to a given topic.
*
* @override
* @param {string} topic
* @param {(msg: InMessage) => void} [handler]
* @returns {void}
*/
function subscribe (topic, handler) {
// Bind provided handler
handler && pubsub.on(topic, handler)
pubsub._subscribeAdapter(topic)
}
/**
* Unsubscribe from the given topic.
*
* @override
* @param {string} topic
* @param {(msg: InMessage) => void} [handler]
* @returns {void}
*/
function unsubscribe (topic, handler) {
if (!handler) {
pubsub.removeAllListeners(topic)
} else {
pubsub.removeListener(topic, handler)
}
/**
* Unsubscribe from the given topic.
*
* @override
* @param {string} topic
* @param {function(msg: InMessage)} [handler]
* @returns {void}
*/
unsubscribe (topic, handler) {
if (!handler) {
this.removeAllListeners(topic)
} else {
this.removeListener(topic, handler)
}
if (this.listenerCount(topic) === 0) {
super.unsubscribe(topic)
}
if (pubsub.listenerCount(topic) === 0) {
pubsub._unsubscribeAdapter(topic)
}
}
return new Pubsub(libp2p, options)
pubsub.subscribe = subscribe
pubsub.unsubscribe = unsubscribe
return pubsub
}
module.exports = pubsubAdapter

View File

@ -2,7 +2,8 @@
const protons = require('protons')
const message = `
/** @type {{Envelope: import('../../types').MessageProto}} */
module.exports = protons(`
message Envelope {
// public_key is the public key of the keypair the enclosed payload was
// signed with.
@ -20,6 +21,4 @@ message Envelope {
// additional security.
bytes signature = 5;
}
`
module.exports = protons(message).Envelope
`)

View File

@ -1,8 +1,5 @@
'use strict'
const debug = require('debug')
const log = debug('libp2p:envelope')
log.error = debug('libp2p:envelope:error')
const errCode = require('err-code')
const uint8arraysConcat = require('uint8arrays/concat')
const uint8arraysFromString = require('uint8arrays/from-string')
@ -15,11 +12,14 @@ const { codes } = require('../../errors')
const Protobuf = require('./envelope.proto')
/**
* The Envelope is responsible for keeping an arbitrary signed record
* by a libp2p peer.
* @typedef {import('libp2p-interfaces/src/record/types').Record} Record
*/
class Envelope {
/**
* The Envelope is responsible for keeping an arbitrary signed record
* by a libp2p peer.
*
* @class
* @param {object} params
* @param {PeerId} params.peerId
@ -49,7 +49,7 @@ class Envelope {
const publicKey = cryptoKeys.marshalPublicKey(this.peerId.pubKey)
this._marshal = Protobuf.encode({
this._marshal = Protobuf.Envelope.encode({
public_key: publicKey,
payload_type: this.payloadType,
payload: this.payload,
@ -102,14 +102,14 @@ const formatSignaturePayload = (domain, payloadType, payload) => {
// - The length of the payload field in bytes
// - The value of the payload field
domain = uint8arraysFromString(domain)
const domainLength = varint.encode(domain.byteLength)
const domainUint8Array = uint8arraysFromString(domain)
const domainLength = varint.encode(domainUint8Array.byteLength)
const payloadTypeLength = varint.encode(payloadType.length)
const payloadLength = varint.encode(payload.length)
return uint8arraysConcat([
new Uint8Array(domainLength),
domain,
domainUint8Array,
new Uint8Array(payloadTypeLength),
payloadType,
new Uint8Array(payloadLength),
@ -124,7 +124,7 @@ const formatSignaturePayload = (domain, payloadType, payload) => {
* @returns {Promise<Envelope>}
*/
Envelope.createFromProtobuf = async (data) => {
const envelopeData = Protobuf.decode(data)
const envelopeData = Protobuf.Envelope.decode(data)
const peerId = await PeerId.createFromPubKey(envelopeData.public_key)
return new Envelope({
@ -142,7 +142,7 @@ Envelope.createFromProtobuf = async (data) => {
* @async
* @param {Record} record
* @param {PeerId} peerId
* @returns {Envelope}
* @returns {Promise<Envelope>}
*/
Envelope.seal = async (record, peerId) => {
const domain = record.domain
@ -166,7 +166,7 @@ Envelope.seal = async (record, peerId) => {
*
* @param {Uint8Array} data
* @param {string} domain
* @returns {Envelope}
* @returns {Promise<Envelope>}
*/
Envelope.openAndCertify = async (data, domain) => {
const envelope = await Envelope.createFromProtobuf(data)

View File

@ -2,7 +2,6 @@
const multiaddr = require('multiaddr')
const PeerId = require('peer-id')
const Record = require('libp2p-interfaces/src/record')
const arrayEquals = require('libp2p-utils/src/array-equals')
const Protobuf = require('./peer-record.proto')
@ -12,19 +11,28 @@ const {
} = require('./consts')
/**
* The PeerRecord is used for distributing peer routing records across the network.
* It contains the peer's reachable listen addresses.
* @typedef {import('peer-id')} PeerId
* @typedef {import('multiaddr')} Multiaddr
* @typedef {import('libp2p-interfaces/src/record/types').Record} Record
*/
class PeerRecord extends Record {
/**
* @implements {Record}
*/
class PeerRecord {
/**
* The PeerRecord is used for distributing peer routing records across the network.
* It contains the peer's reachable listen addresses.
*
* @class
* @param {object} params
* @param {Object} params
* @param {PeerId} params.peerId
* @param {Array<multiaddr>} params.multiaddrs - addresses of the associated peer.
* @param {Multiaddr[]} params.multiaddrs - addresses of the associated peer.
* @param {number} [params.seqNumber] - monotonically-increasing sequence counter that's used to order PeerRecords in time.
*/
constructor ({ peerId, multiaddrs = [], seqNumber = Date.now() }) {
super(ENVELOPE_DOMAIN_PEER_RECORD, ENVELOPE_PAYLOAD_TYPE_PEER_RECORD)
this.domain = ENVELOPE_DOMAIN_PEER_RECORD
this.codec = ENVELOPE_PAYLOAD_TYPE_PEER_RECORD
this.peerId = peerId
this.multiaddrs = multiaddrs
@ -44,7 +52,7 @@ class PeerRecord extends Record {
return this._marshal
}
this._marshal = Protobuf.encode({
this._marshal = Protobuf.PeerRecord.encode({
peer_id: this.peerId.toBytes(),
seq: this.seqNumber,
addresses: this.multiaddrs.map((m) => ({
@ -58,10 +66,14 @@ class PeerRecord extends Record {
/**
* Returns true if `this` record equals the `other`.
*
* @param {Record} other
* @param {unknown} other
* @returns {boolean}
*/
equals (other) {
if (!(other instanceof PeerRecord)) {
return false
}
// Validate PeerId
if (!this.peerId.equals(other.peerId)) {
return false
@ -89,7 +101,7 @@ class PeerRecord extends Record {
*/
PeerRecord.createFromProtobuf = (buf) => {
// Decode
const peerRecord = Protobuf.decode(buf)
const peerRecord = Protobuf.PeerRecord.decode(buf)
const peerId = PeerId.createFromBytes(peerRecord.peer_id)
const multiaddrs = (peerRecord.addresses || []).map((a) => multiaddr(a.multiaddr))

View File

@ -7,7 +7,8 @@ const protons = require('protons')
// is expected to expand to include other information in the future.
// PeerRecords are designed to be serialized to bytes and placed inside of
// SignedEnvelopes before sharing with other peers.
const message = `
/** @type {{PeerRecord: import('../../types').MessageProto}} */
module.exports = protons(`
message PeerRecord {
// AddressInfo is a wrapper around a binary multiaddr. It is defined as a
// separate message to allow us to add per-address metadata in the future.
@ -24,6 +25,4 @@ message PeerRecord {
// addresses is a list of public listen addresses for the peer.
repeated AddressInfo addresses = 3;
}
`
module.exports = protons(message).PeerRecord
`)

25
src/record/utils.js Normal file
View File

@ -0,0 +1,25 @@
'use strict'
const Envelope = require('./envelope')
const PeerRecord = require('./peer-record')
/**
* @typedef {import('../')} Libp2p
*/
/**
* Create (or update if existing) self peer record and store it in the AddressBook.
*
* @param {Libp2p} libp2p
* @returns {Promise<void>}
*/
async function updateSelfPeerRecord (libp2p) {
const peerRecord = new PeerRecord({
peerId: libp2p.peerId,
multiaddrs: libp2p.multiaddrs
})
const envelope = await Envelope.seal(peerRecord, libp2p.peerId)
libp2p.peerStore.addressBook.consumePeerRecord(envelope)
}
module.exports.updateSelfPeerRecord = updateSelfPeerRecord

View File

@ -1,15 +1,24 @@
'use strict'
const debug = require('debug')
const log = Object.assign(debug('libp2p:peer-store'), {
error: debug('libp2p:peer-store:err')
})
const errcode = require('err-code')
const log = debug('libp2p:peer-store')
log.error = debug('libp2p:peer-store:error')
const {
ERR_INVALID_PARAMETERS
codes: { ERR_INVALID_PARAMETERS }
} = require('./errors')
const Topology = require('libp2p-interfaces/src/topology')
/**
* @typedef {import('peer-id')} PeerId
* @typedef {import('./peer-store')} PeerStore
* @typedef {import('./connection-manager')} ConnectionManager
* @typedef {import('libp2p-interfaces/src/connection').Connection} Connection
* @typedef {import('libp2p-interfaces/src/topology')} Topology
*/
/**
* Responsible for notifying registered protocols of events in the network.
*/
@ -17,7 +26,7 @@ class Registrar {
/**
* @param {Object} props
* @param {PeerStore} props.peerStore
* @param {connectionManager} props.connectionManager
* @param {ConnectionManager} props.connectionManager
* @class
*/
constructor ({ peerStore, connectionManager }) {
@ -51,7 +60,7 @@ class Registrar {
* Get a connection with a peer.
*
* @param {PeerId} peerId
* @returns {Connection}
* @returns {Connection | null}
*/
getConnection (peerId) {
return this.connectionManager.get(peerId)
@ -65,11 +74,12 @@ class Registrar {
*/
register (topology) {
if (!Topology.isTopology(topology)) {
log.error('topology must be an instance of interfaces/topology')
throw errcode(new Error('topology must be an instance of interfaces/topology'), ERR_INVALID_PARAMETERS)
}
// Create topology
const id = (parseInt(Math.random() * 1e9)).toString(36) + Date.now()
const id = (Math.random() * 1e9).toString(36) + Date.now()
this.topologies.set(id, topology)

View File

@ -1,23 +1,39 @@
'use strict'
const debug = require('debug')
const log = Object.assign(debug('libp2p:transports'), {
error: debug('libp2p:transports:err')
})
const pSettle = require('p-settle')
const { codes } = require('./errors')
const errCode = require('err-code')
const debug = require('debug')
const log = debug('libp2p:transports')
log.error = debug('libp2p:transports:error')
const { updateSelfPeerRecord } = require('./record/utils')
/**
* @typedef {import('multiaddr')} Multiaddr
* @typedef {import('libp2p-interfaces/src/connection').Connection} Connection
* @typedef {import('libp2p-interfaces/src/transport/types').TransportFactory} TransportFactory
* @typedef {import('libp2p-interfaces/src/transport/types').Transport} Transport
*
* @typedef {Object} TransportManagerProperties
* @property {import('./')} libp2p
* @property {import('./upgrader')} upgrader
*
* @typedef {Object} TransportManagerOptions
* @property {number} [faultTolerance = FAULT_TOLERANCE.FATAL_ALL] - Address listen error tolerance.
*/
class TransportManager {
/**
* @class
* @param {object} options
* @param {Libp2p} options.libp2p - The Libp2p instance. It will be passed to the transports.
* @param {Upgrader} options.upgrader - The upgrader to provide to the transports
* @param {boolean} [options.faultTolerance = FAULT_TOLERANCE.FATAL_ALL] - Address listen error tolerance.
* @param {TransportManagerProperties & TransportManagerOptions} options
*/
constructor ({ libp2p, upgrader, faultTolerance = FAULT_TOLERANCE.FATAL_ALL }) {
this.libp2p = libp2p
this.upgrader = upgrader
/** @type {Map<string, Transport>} */
this._transports = new Map()
this._listeners = new Map()
this._listenerOptions = new Map()
@ -28,7 +44,7 @@ class TransportManager {
* Adds a `Transport` to the manager
*
* @param {string} key
* @param {Transport} Transport
* @param {TransportFactory} Transport
* @param {*} transportOptions - Additional options to pass to the transport
* @returns {void}
*/
@ -65,6 +81,8 @@ class TransportManager {
log('closing listeners for %s', key)
while (listeners.length) {
const listener = listeners.pop()
listener.removeAllListeners('listening')
listener.removeAllListeners('close')
tasks.push(listener.close())
}
}
@ -115,7 +133,7 @@ class TransportManager {
/**
* Returns all the transports instances.
*
* @returns {Iterator<Transport>}
* @returns {IterableIterator<Transport>}
*/
getTransports () {
return this._transports.values()
@ -139,11 +157,10 @@ class TransportManager {
* Starts listeners for each listen Multiaddr.
*
* @async
* @param {Multiaddr[]} addrs - addresses to attempt to listen on
*/
async listen () {
const addrs = this.libp2p.addressManager.getListenAddrs()
if (addrs.length === 0) {
async listen (addrs) {
if (!addrs || addrs.length === 0) {
log('no addresses were provided for listening, this node is dial only')
return
}
@ -156,9 +173,13 @@ class TransportManager {
// For each supported multiaddr, create a listener
for (const addr of supportedAddrs) {
log('creating listener for %s on %s', key, addr)
const listener = transport.createListener(this._listenerOptions.get(key), this.onConnection)
const listener = transport.createListener(this._listenerOptions.get(key))
this._listeners.get(key).push(listener)
// Track listen/close events
listener.on('listening', () => updateSelfPeerRecord(this.libp2p))
listener.on('close', () => updateSelfPeerRecord(this.libp2p))
// We need to attempt to listen on everything
tasks.push(listener.listen(addr))
}
@ -203,6 +224,8 @@ class TransportManager {
if (this._listeners.has(key)) {
// Close any running listeners
for (const listener of this._listeners.get(key)) {
listener.removeAllListeners('listening')
listener.removeAllListeners('close')
await listener.close()
}
}

84
src/types.ts Normal file
View File

@ -0,0 +1,84 @@
// Insecure Message types
export enum KeyType {
RSA = 0,
Ed25519 = 1,
Secp256k1 = 2,
ECDSA = 3
}
// Protobufs
export type MessageProto = {
encode(value: any): Uint8Array
decode(bytes: Uint8Array): any
}
export type SUCCESS = 100;
export type HOP_SRC_ADDR_TOO_LONG = 220;
export type HOP_DST_ADDR_TOO_LONG = 221;
export type HOP_SRC_MULTIADDR_INVALID = 250;
export type HOP_DST_MULTIADDR_INVALID = 251;
export type HOP_NO_CONN_TO_DST = 260;
export type HOP_CANT_DIAL_DST = 261;
export type HOP_CANT_OPEN_DST_STREAM = 262;
export type HOP_CANT_SPEAK_RELAY = 270;
export type HOP_CANT_RELAY_TO_SELF = 280;
export type STOP_SRC_ADDR_TOO_LONG = 320;
export type STOP_DST_ADDR_TOO_LONG = 321;
export type STOP_SRC_MULTIADDR_INVALID = 350;
export type STOP_DST_MULTIADDR_INVALID = 351;
export type STOP_RELAY_REFUSED = 390;
export type MALFORMED_MESSAGE = 400;
export type CircuitStatus = SUCCESS | HOP_SRC_ADDR_TOO_LONG | HOP_DST_ADDR_TOO_LONG
| HOP_SRC_MULTIADDR_INVALID | HOP_DST_MULTIADDR_INVALID | HOP_NO_CONN_TO_DST
| HOP_CANT_DIAL_DST | HOP_CANT_OPEN_DST_STREAM | HOP_CANT_SPEAK_RELAY | HOP_CANT_RELAY_TO_SELF
| STOP_SRC_ADDR_TOO_LONG | STOP_DST_ADDR_TOO_LONG | STOP_SRC_MULTIADDR_INVALID
| STOP_DST_MULTIADDR_INVALID | STOP_RELAY_REFUSED | MALFORMED_MESSAGE
export type HOP = 1;
export type STOP = 2;
export type STATUS = 3;
export type CAN_HOP = 4;
export type CircuitType = HOP | STOP | STATUS | CAN_HOP
export type CircuitPeer = {
id: Uint8Array
addrs: Uint8Array[]
}
export type CircuitRequest = {
type: CircuitType
dstPeer: CircuitPeer
srcPeer: CircuitPeer
}
export type CircuitMessageProto = {
encode(value: any): Uint8Array
decode(bytes: Uint8Array): any
Status: {
SUCCESS: SUCCESS,
HOP_SRC_ADDR_TOO_LONG: HOP_SRC_ADDR_TOO_LONG,
HOP_DST_ADDR_TOO_LONG: HOP_DST_ADDR_TOO_LONG,
HOP_SRC_MULTIADDR_INVALID: HOP_SRC_MULTIADDR_INVALID,
HOP_DST_MULTIADDR_INVALID: HOP_DST_MULTIADDR_INVALID,
HOP_NO_CONN_TO_DST: HOP_NO_CONN_TO_DST,
HOP_CANT_DIAL_DST: HOP_CANT_DIAL_DST,
HOP_CANT_OPEN_DST_STREAM: HOP_CANT_OPEN_DST_STREAM,
HOP_CANT_SPEAK_RELAY: HOP_CANT_SPEAK_RELAY,
HOP_CANT_RELAY_TO_SELF: HOP_CANT_RELAY_TO_SELF,
STOP_SRC_ADDR_TOO_LONG: STOP_SRC_ADDR_TOO_LONG,
STOP_DST_ADDR_TOO_LONG: STOP_DST_ADDR_TOO_LONG,
STOP_SRC_MULTIADDR_INVALID: STOP_SRC_MULTIADDR_INVALID,
STOP_DST_MULTIADDR_INVALID: STOP_DST_MULTIADDR_INVALID,
STOP_RELAY_REFUSED: STOP_RELAY_REFUSED,
MALFORMED_MESSAGE: MALFORMED_MESSAGE
},
Type: {
HOP: HOP,
STOP: STOP,
STATUS: STATUS,
CAN_HOP: CAN_HOP
}
}

View File

@ -1,29 +1,30 @@
'use strict'
const debug = require('debug')
const log = debug('libp2p:upgrader')
log.error = debug('libp2p:upgrader:error')
const log = Object.assign(debug('libp2p:upgrader'), {
error: debug('libp2p:upgrader:err')
})
const errCode = require('err-code')
const Multistream = require('multistream-select')
const { Connection } = require('libp2p-interfaces/src/connection')
const ConnectionStatus = require('libp2p-interfaces/src/connection/status')
const PeerId = require('peer-id')
const pipe = require('it-pipe')
const errCode = require('err-code')
const { pipe } = require('it-pipe')
const mutableProxy = require('mutable-proxy')
const { codes } = require('./errors')
/**
* @typedef MultiaddrConnection
* @property {Function} sink
* @property {AsyncIterator} source
* @property {*} conn
* @property {Multiaddr} remoteAddr
* @typedef {import('libp2p-interfaces/src/transport/types').MultiaddrConnection} MultiaddrConnection
* @typedef {import('libp2p-interfaces/src/stream-muxer/types').MuxerFactory} MuxerFactory
* @typedef {import('libp2p-interfaces/src/stream-muxer/types').Muxer} Muxer
* @typedef {import('libp2p-interfaces/src/stream-muxer/types').MuxedStream} MuxedStream
* @typedef {import('libp2p-interfaces/src/crypto/types').Crypto} Crypto
* @typedef {import('multiaddr')} Multiaddr
*/
/**
* @typedef CryptoResult
* @property {*} conn A duplex iterable
* @property {MultiaddrConnection} conn A duplex iterable
* @property {PeerId} remotePeer
* @property {string} protocol
*/
@ -32,24 +33,24 @@ class Upgrader {
/**
* @param {object} options
* @param {PeerId} options.localPeer
* @param {Metrics} options.metrics
* @param {Map<string, Crypto>} options.cryptos
* @param {Map<string, Muxer>} options.muxers
* @param {function(Connection)} options.onConnection - Called when a connection is upgraded
* @param {function(Connection)} options.onConnectionEnd
* @param {import('./metrics')} [options.metrics]
* @param {Map<string, Crypto>} [options.cryptos]
* @param {Map<string, MuxerFactory>} [options.muxers]
* @param {(Connection) => void} options.onConnection - Called when a connection is upgraded
* @param {(Connection) => void} options.onConnectionEnd
*/
constructor ({
localPeer,
metrics,
cryptos,
muxers,
cryptos = new Map(),
muxers = new Map(),
onConnectionEnd = () => {},
onConnection = () => {}
}) {
this.localPeer = localPeer
this.metrics = metrics
this.cryptos = cryptos || new Map()
this.muxers = muxers || new Map()
this.cryptos = cryptos
this.muxers = muxers
this.protector = null
this.protocols = new Map()
this.onConnection = onConnection
@ -74,7 +75,7 @@ class Upgrader {
if (this.metrics) {
({ setTarget: setPeer, proxy: proxyPeer } = mutableProxy())
const idString = (parseInt(Math.random() * 1e9)).toString(36) + Date.now()
const idString = (Math.random() * 1e9).toString(36) + Date.now()
setPeer({ toB58String: () => idString })
maConn = this.metrics.trackStream({ stream: maConn, remotePeer: proxyPeer })
}
@ -132,12 +133,7 @@ class Upgrader {
* @returns {Promise<Connection>}
*/
async upgradeOutbound (maConn) {
let remotePeerId
try {
remotePeerId = PeerId.createFromB58String(maConn.remoteAddr.getPeerId())
} catch (err) {
log.error('multiaddr did not contain a valid peer id', err)
}
const remotePeerId = PeerId.createFromB58String(maConn.remoteAddr.getPeerId())
let encryptedConn
let remotePeer
@ -149,7 +145,7 @@ class Upgrader {
if (this.metrics) {
({ setTarget: setPeer, proxy: proxyPeer } = mutableProxy())
const idString = (parseInt(Math.random() * 1e9)).toString(36) + Date.now()
const idString = (Math.random() * 1e9).toString(36) + Date.now()
setPeer({ toB58String: () => idString })
maConn = this.metrics.trackStream({ stream: maConn, remotePeer: proxyPeer })
}
@ -207,8 +203,8 @@ class Upgrader {
* @param {string} options.cryptoProtocol - The crypto protocol that was negotiated
* @param {string} options.direction - One of ['inbound', 'outbound']
* @param {MultiaddrConnection} options.maConn - The transport layer connection
* @param {*} options.upgradedConn - A duplex connection returned from multiplexer and/or crypto selection
* @param {Muxer} options.Muxer - The muxer to be used for muxing
* @param {MuxedStream | MultiaddrConnection} options.upgradedConn - A duplex connection returned from multiplexer and/or crypto selection
* @param {MuxerFactory} [options.Muxer] - The muxer to be used for muxing
* @param {PeerId} options.remotePeer - The peer the connection is with
* @returns {Connection}
*/
@ -272,7 +268,7 @@ class Upgrader {
// Wait for close to finish before notifying of the closure
(async () => {
try {
if (connection.stat.status === ConnectionStatus.OPEN) {
if (connection.stat.status === 'open') {
await connection.close()
}
} catch (err) {
@ -300,6 +296,7 @@ class Upgrader {
remotePeer: remotePeer,
stat: {
direction,
// @ts-ignore
timeline: maConn.timeline,
multiplexer: Muxer && Muxer.multicodec,
encryption: cryptoProtocol
@ -326,7 +323,7 @@ class Upgrader {
* @private
* @param {object} options
* @param {Connection} options.connection - The connection the stream belongs to
* @param {Stream} options.stream
* @param {MuxedStream} options.stream
* @param {string} options.protocol
*/
_onStream ({ connection, stream, protocol }) {
@ -342,7 +339,7 @@ class Upgrader {
* @param {PeerId} localPeer - The initiators PeerId
* @param {*} connection
* @param {Map<string, Crypto>} cryptos
* @returns {CryptoResult} An encrypted connection, remote peer `PeerId` and the protocol of the `Crypto` used
* @returns {Promise<CryptoResult>} An encrypted connection, remote peer `PeerId` and the protocol of the `Crypto` used
*/
async _encryptInbound (localPeer, connection, cryptos) {
const mss = new Multistream.Listener(connection)
@ -354,6 +351,10 @@ class Upgrader {
const crypto = cryptos.get(protocol)
log('encrypting inbound connection...')
if (!crypto) {
throw new Error(`no crypto module found for ${protocol}`)
}
return {
...await crypto.secureInbound(localPeer, stream),
protocol
@ -373,7 +374,7 @@ class Upgrader {
* @param {*} connection
* @param {PeerId} remotePeerId
* @param {Map<string, Crypto>} cryptos
* @returns {CryptoResult} An encrypted connection, remote peer `PeerId` and the protocol of the `Crypto` used
* @returns {Promise<CryptoResult>} An encrypted connection, remote peer `PeerId` and the protocol of the `Crypto` used
*/
async _encryptOutbound (localPeer, connection, remotePeerId, cryptos) {
const mss = new Multistream.Dialer(connection)
@ -385,6 +386,10 @@ class Upgrader {
const crypto = cryptos.get(protocol)
log('encrypting outbound connection to %j', remotePeerId)
if (!crypto) {
throw new Error(`no crypto module found for ${protocol}`)
}
return {
...await crypto.secureOutbound(localPeer, stream, remotePeerId),
protocol
@ -400,9 +405,9 @@ class Upgrader {
*
* @private
* @async
* @param {*} connection - A basic duplex connection to multiplex
* @param {Map<string, Muxer>} muxers - The muxers to attempt multiplexing with
* @returns {*} A muxed connection
* @param {MultiaddrConnection} connection - A basic duplex connection to multiplex
* @param {Map<string, MuxerFactory>} muxers - The muxers to attempt multiplexing with
* @returns {Promise<{ stream: MuxedStream, Muxer?: MuxerFactory}>} A muxed connection
*/
async _multiplexOutbound (connection, muxers) {
const dialer = new Multistream.Dialer(connection)
@ -424,9 +429,9 @@ class Upgrader {
*
* @private
* @async
* @param {*} connection - A basic duplex connection to multiplex
* @param {Map<string, Muxer>} muxers - The muxers to attempt multiplexing with
* @returns {*} A muxed connection
* @param {MultiaddrConnection} connection - A basic duplex connection to multiplex
* @param {Map<string, MuxerFactory>} muxers - The muxers to attempt multiplexing with
* @returns {Promise<{ stream: MuxedStream, Muxer?: MuxerFactory}>} A muxed connection
*/
async _multiplexInbound (connection, muxers) {
const listener = new Multistream.Listener(connection)

View File

@ -1,11 +1,7 @@
'use strict'
/* eslint-env mocha */
const chai = require('chai')
chai.use(require('dirty-chai'))
chai.use(require('chai-as-promised'))
const { expect } = chai
const { expect } = require('aegir/utils/chai')
const multiaddr = require('multiaddr')
const AddressManager = require('../../src/address-manager')
@ -20,7 +16,6 @@ describe('Address Manager', () => {
expect(am.listen.size).to.equal(0)
expect(am.announce.size).to.equal(0)
expect(am.noAnnounce.size).to.equal(0)
})
it('should return listen multiaddrs on get', () => {
@ -30,7 +25,6 @@ describe('Address Manager', () => {
expect(am.listen.size).to.equal(listenAddresses.length)
expect(am.announce.size).to.equal(0)
expect(am.noAnnounce.size).to.equal(0)
const listenMultiaddrs = am.getListenAddrs()
expect(listenMultiaddrs.length).to.equal(2)
@ -46,28 +40,11 @@ describe('Address Manager', () => {
expect(am.listen.size).to.equal(listenAddresses.length)
expect(am.announce.size).to.equal(announceAddreses.length)
expect(am.noAnnounce.size).to.equal(0)
const announceMultiaddrs = am.getAnnounceAddrs()
expect(announceMultiaddrs.length).to.equal(1)
expect(announceMultiaddrs[0].equals(multiaddr(announceAddreses[0]))).to.equal(true)
})
it('should return noAnnounce multiaddrs on get', () => {
const am = new AddressManager({
listen: listenAddresses,
noAnnounce: listenAddresses
})
expect(am.listen.size).to.equal(listenAddresses.length)
expect(am.announce.size).to.equal(0)
expect(am.noAnnounce.size).to.equal(listenAddresses.length)
const noAnnounceMultiaddrs = am.getNoAnnounceAddrs()
expect(noAnnounceMultiaddrs.length).to.equal(2)
expect(noAnnounceMultiaddrs[0].equals(multiaddr(listenAddresses[0]))).to.equal(true)
expect(noAnnounceMultiaddrs[1].equals(multiaddr(listenAddresses[1]))).to.equal(true)
})
})
describe('libp2p.addressManager', () => {
@ -80,14 +57,12 @@ describe('libp2p.addressManager', () => {
config: {
addresses: {
listen: listenAddresses,
announce: announceAddreses,
noAnnounce: listenAddresses
announce: announceAddreses
}
}
})
expect(libp2p.addressManager.listen.size).to.equal(listenAddresses.length)
expect(libp2p.addressManager.announce.size).to.equal(announceAddreses.length)
expect(libp2p.addressManager.noAnnounce.size).to.equal(listenAddresses.length)
})
})

View File

@ -1,17 +1,17 @@
'use strict'
/* eslint-env mocha */
const chai = require('chai')
chai.use(require('dirty-chai'))
chai.use(require('chai-as-promised'))
const { expect } = chai
const { expect } = require('aegir/utils/chai')
const sinon = require('sinon')
const multiaddr = require('multiaddr')
const isLoopback = require('libp2p-utils/src/multiaddr/is-loopback')
const { AddressesOptions } = require('./utils')
const peerUtils = require('../utils/creators/peer')
const listenAddresses = ['/ip4/127.0.0.1/tcp/0', '/ip4/127.0.0.1/tcp/8000/ws']
const announceAddreses = ['/dns4/peer.io']
const announceAddreses = ['/dns4/peer.io/tcp/433/p2p/12D3KooWNvSZnPi3RrhrTwEY4LuuBeB6K6facKUCJcyWG1aoDd2p']
describe('libp2p.multiaddrs', () => {
let libp2p
@ -45,8 +45,33 @@ describe('libp2p.multiaddrs', () => {
expect(listenAddrs.has(listenAddresses[1])).to.equal(true)
})
it('should advertise all addresses if noAnnounce addresses are not provided, but with correct ports', async () => {
it('should announce transport listen addresses if announce addresses are not provided', async () => {
[libp2p] = await peerUtils.createPeer({
started: false,
config: {
...AddressesOptions,
addresses: {
listen: listenAddresses
}
}
})
await libp2p.start()
const tmListen = libp2p.transportManager.getAddrs().map((ma) => ma.toString())
// Announce 2 listen (transport)
const advertiseMultiaddrs = libp2p.multiaddrs.map((ma) => ma.toString())
expect(advertiseMultiaddrs.length).to.equal(2)
tmListen.forEach((m) => {
expect(advertiseMultiaddrs).to.include(m)
})
expect(advertiseMultiaddrs).to.not.include(listenAddresses[0]) // Random Port switch
})
it('should only announce the given announce addresses when provided', async () => {
[libp2p] = await peerUtils.createPeer({
started: false,
config: {
...AddressesOptions,
addresses: {
@ -56,71 +81,70 @@ describe('libp2p.multiaddrs', () => {
}
})
await libp2p.start()
const tmListen = libp2p.transportManager.getAddrs().map((ma) => ma.toString())
const spyAnnounce = sinon.spy(libp2p.addressManager, 'getAnnounceAddrs')
const spyNoAnnounce = sinon.spy(libp2p.addressManager, 'getNoAnnounceAddrs')
const spyListen = sinon.spy(libp2p.addressManager, 'getListenAddrs')
const spyTranspMgr = sinon.spy(libp2p.transportManager, 'getAddrs')
// Announce 1 announce addr
const advertiseMultiaddrs = libp2p.multiaddrs.map((ma) => ma.toString())
expect(spyAnnounce).to.have.property('callCount', 1)
expect(spyNoAnnounce).to.have.property('callCount', 1)
expect(spyListen).to.have.property('callCount', 0) // Listen addr should not be used
expect(spyTranspMgr).to.have.property('callCount', 1)
// Announce 2 listen (transport) + 1 announce
expect(advertiseMultiaddrs.length).to.equal(3)
tmListen.forEach((m) => {
expect(advertiseMultiaddrs).to.include(m)
expect(advertiseMultiaddrs.length).to.equal(announceAddreses.length)
advertiseMultiaddrs.forEach((m) => {
expect(tmListen).to.not.include(m)
expect(announceAddreses).to.include(m)
})
announceAddreses.forEach((m) => {
expect(advertiseMultiaddrs).to.include(m)
})
expect(advertiseMultiaddrs).to.not.include(listenAddresses[0]) // Random Port switch
})
it('should remove replicated addresses', async () => {
it('can filter out loopback addresses by the announce filter', async () => {
[libp2p] = await peerUtils.createPeer({
started: false,
config: {
...AddressesOptions,
addresses: {
listen: listenAddresses,
announce: [listenAddresses[1]]
announceFilter: (multiaddrs) => multiaddrs.filter(m => !isLoopback(m))
}
}
})
const advertiseMultiaddrs = libp2p.multiaddrs.map((ma) => ma.toString())
await libp2p.start()
// Announce 2 listen (transport), ignoring duplicated in announce
expect(advertiseMultiaddrs.length).to.equal(2)
expect(libp2p.multiaddrs.length).to.equal(0)
// Stub transportManager addresses to add a public address
const stubMa = multiaddr('/ip4/120.220.10.1/tcp/1000')
sinon.stub(libp2p.transportManager, 'getAddrs').returns([
...listenAddresses.map((a) => multiaddr(a)),
stubMa
])
const multiaddrs = libp2p.multiaddrs
expect(multiaddrs.length).to.equal(1)
expect(multiaddrs[0].equals(stubMa)).to.eql(true)
})
it('should not advertise noAnnounce addresses', async () => {
const noAnnounce = [listenAddresses[1]]
;[libp2p] = await peerUtils.createPeer({
it('can filter out loopback addresses to announced by the announce filter', async () => {
[libp2p] = await peerUtils.createPeer({
started: false,
config: {
...AddressesOptions,
addresses: {
listen: listenAddresses,
announce: announceAddreses,
noAnnounce
announceFilter: (multiaddrs) => multiaddrs.filter(m => !isLoopback(m))
}
}
})
const advertiseMultiaddrs = libp2p.multiaddrs.map((ma) => ma.toString())
const listenAddrs = libp2p.addressManager.listen
expect(listenAddrs.size).to.equal(listenAddresses.length)
expect(listenAddrs.has(listenAddresses[0])).to.equal(true)
expect(listenAddrs.has(listenAddresses[1])).to.equal(true)
// Announce 1 listen (transport) not in the noAnnounce and the announce
expect(advertiseMultiaddrs.length).to.equal(2)
await libp2p.start()
announceAddreses.forEach((m) => {
expect(advertiseMultiaddrs).to.include(m)
})
noAnnounce.forEach((m) => {
expect(advertiseMultiaddrs).to.not.include(m)
})
const multiaddrs = libp2p.multiaddrs
expect(multiaddrs.length).to.equal(announceAddreses.length)
expect(multiaddrs.includes(listenAddresses[0])).to.equal(false)
expect(multiaddrs.includes(listenAddresses[1])).to.equal(false)
})
})

View File

@ -1,10 +1,7 @@
'use strict'
/* eslint-env mocha */
const chai = require('chai')
chai.use(require('dirty-chai'))
chai.use(require('chai-as-promised'))
const { expect } = chai
const { expect } = require('aegir/utils/chai')
const sinon = require('sinon')
const delay = require('delay')

View File

@ -1,10 +1,7 @@
'use strict'
/* eslint-env mocha */
const chai = require('chai')
chai.use(require('dirty-chai'))
chai.use(require('chai-as-promised'))
const { expect } = chai
const { expect } = require('aegir/utils/chai')
const sinon = require('sinon')
const peerUtils = require('../utils/creators/peer')

View File

@ -1,10 +1,7 @@
'use strict'
/* eslint-env mocha */
const chai = require('chai')
chai.use(require('dirty-chai'))
chai.use(require('chai-as-promised'))
const { expect } = chai
const { expect } = require('aegir/utils/chai')
const nock = require('nock')
const sinon = require('sinon')

View File

@ -1,10 +1,7 @@
'use strict'
/* eslint-env mocha */
const chai = require('chai')
chai.use(require('dirty-chai'))
const { expect } = chai
const { expect } = require('aegir/utils/chai')
const mergeOptions = require('merge-options')
const { create } = require('../../../src')

View File

@ -1,10 +1,7 @@
'use strict'
/* eslint-env mocha */
const chai = require('chai')
chai.use(require('dirty-chai'))
chai.use(require('chai-as-promised'))
const { expect } = chai
const { expect } = require('aegir/utils/chai')
const Transport = require('libp2p-websockets')
const { NOISE: Crypto } = require('libp2p-noise')

View File

@ -1,9 +1,7 @@
'use strict'
/* eslint-env mocha */
const chai = require('chai')
chai.use(require('dirty-chai'))
const { expect } = chai
const { expect } = require('aegir/utils/chai')
const Transport = require('libp2p-tcp')
const { NOISE: Crypto } = require('libp2p-noise')

View File

@ -1,9 +1,7 @@
'use strict'
/* eslint-env mocha */
const chai = require('chai')
chai.use(require('dirty-chai'))
const { expect } = chai
const { expect } = require('aegir/utils/chai')
const pTimes = require('p-times')
const pipe = require('it-pipe')

View File

@ -1,10 +1,7 @@
'use strict'
/* eslint-env mocha */
const chai = require('chai')
chai.use(require('dirty-chai'))
chai.use(require('chai-as-promised'))
const { expect } = chai
const { expect } = require('aegir/utils/chai')
const sinon = require('sinon')
const { AbortError } = require('libp2p-interfaces/src/transport/errors')
@ -13,7 +10,7 @@ const AggregateError = require('aggregate-error')
const pDefer = require('p-defer')
const delay = require('delay')
const { DialRequest } = require('../../src/dialer/dial-request')
const DialRequest = require('../../src/dialer/dial-request')
const createMockConnection = require('../utils/mockConnection')
const error = new Error('dial failes')

View File

@ -42,21 +42,28 @@ describe('Dialing (direct, TCP)', () => {
let peerStore
let remoteAddr
before(async () => {
const [remotePeerId] = await Promise.all([
PeerId.createFromJSON(Peers[0])
beforeEach(async () => {
const [localPeerId, remotePeerId] = await Promise.all([
PeerId.createFromJSON(Peers[0]),
PeerId.createFromJSON(Peers[1])
])
peerStore = new PeerStore({ peerId: remotePeerId })
remoteTM = new TransportManager({
libp2p: {
addressManager: new AddressManager({ listen: [listenAddr] })
addressManager: new AddressManager({ listen: [listenAddr] }),
peerId: remotePeerId,
peerStore
},
upgrader: mockUpgrader
})
remoteTM.add(Transport.prototype[Symbol.toStringTag], Transport)
peerStore = new PeerStore({ peerId: remotePeerId })
localTM = new TransportManager({
libp2p: {},
libp2p: {
peerId: localPeerId,
peerStore: new PeerStore({ peerId: localPeerId })
},
upgrader: mockUpgrader
})
localTM.add(Transport.prototype[Symbol.toStringTag], Transport)
@ -66,7 +73,7 @@ describe('Dialing (direct, TCP)', () => {
remoteAddr = remoteTM.getAddrs()[0].encapsulate(`/p2p/${remotePeerId.toB58String()}`)
})
after(() => remoteTM.close())
afterEach(() => remoteTM.close())
afterEach(() => {
sinon.restore()
@ -112,7 +119,7 @@ describe('Dialing (direct, TCP)', () => {
peerStore
})
peerStore.addressBook.set(peerId, [remoteAddr])
peerStore.addressBook.set(peerId, remoteTM.getAddrs())
const connection = await dialer.connectToPeer(peerId)
expect(connection).to.exist()

View File

@ -1,15 +1,13 @@
'use strict'
/* eslint-env mocha */
const chai = require('chai')
chai.use(require('dirty-chai'))
chai.use(require('chai-as-promised'))
const { expect } = chai
const { expect } = require('aegir/utils/chai')
const sinon = require('sinon')
const pDefer = require('p-defer')
const pWaitFor = require('p-wait-for')
const delay = require('delay')
const Transport = require('libp2p-websockets')
const filters = require('libp2p-websockets/src/filters')
const Muxer = require('libp2p-mplex')
const { NOISE: Crypto } = require('libp2p-noise')
const multiaddr = require('multiaddr')
@ -19,6 +17,7 @@ const { AbortError } = require('libp2p-interfaces/src/transport/errors')
const { codes: ErrorCodes } = require('../../src/errors')
const Constants = require('../../src/constants')
const Dialer = require('../../src/dialer')
const addressSort = require('libp2p-utils/src/address-sort')
const PeerStore = require('../../src/peer-store')
const TransportManager = require('../../src/transport-manager')
const Libp2p = require('../../src')
@ -43,10 +42,11 @@ describe('Dialing (direct, WebSockets)', () => {
upgrader: mockUpgrader,
onConnection: () => {}
})
localTM.add(Transport.prototype[Symbol.toStringTag], Transport)
localTM.add(Transport.prototype[Symbol.toStringTag], Transport, { filter: filters.all })
})
afterEach(() => {
peerStore.delete(peerId)
sinon.restore()
})
@ -179,6 +179,37 @@ describe('Dialing (direct, WebSockets)', () => {
.and.to.have.property('code', ErrorCodes.ERR_TIMEOUT)
})
it('should sort addresses on dial', async () => {
const peerMultiaddrs = [
multiaddr('/ip4/127.0.0.1/tcp/15001/ws'),
multiaddr('/ip4/20.0.0.1/tcp/15001/ws'),
multiaddr('/ip4/30.0.0.1/tcp/15001/ws')
]
sinon.spy(addressSort, 'publicAddressesFirst')
sinon.stub(localTM, 'dial').callsFake(createMockConnection)
const dialer = new Dialer({
transportManager: localTM,
addressSorter: addressSort.publicAddressesFirst,
concurrency: 3,
peerStore
})
// Inject data in the AddressBook
peerStore.addressBook.add(peerId, peerMultiaddrs)
// Perform 3 multiaddr dials
await dialer.connectToPeer(peerId)
expect(addressSort.publicAddressesFirst.callCount).to.eql(1)
const sortedAddresses = addressSort.publicAddressesFirst(peerMultiaddrs.map((m) => ({ multiaddr: m })))
expect(localTM.dial.getCall(0).args[0].equals(sortedAddresses[0].multiaddr))
expect(localTM.dial.getCall(1).args[0].equals(sortedAddresses[1].multiaddr))
expect(localTM.dial.getCall(2).args[0].equals(sortedAddresses[2].multiaddr))
})
it('should dial to the max concurrency', async () => {
const dialer = new Dialer({
transportManager: localTM,
@ -262,6 +293,7 @@ describe('Dialing (direct, WebSockets)', () => {
})
describe('libp2p.dialer', () => {
const transportKey = Transport.prototype[Symbol.toStringTag]
let libp2p
afterEach(async () => {
@ -277,6 +309,13 @@ describe('Dialing (direct, WebSockets)', () => {
transport: [Transport],
streamMuxer: [Muxer],
connEncryption: [Crypto]
},
config: {
transport: {
[transportKey]: {
filter: filters.all
}
}
}
})
@ -300,6 +339,13 @@ describe('Dialing (direct, WebSockets)', () => {
maxParallelDials: 10,
maxDialsPerPeer: 1,
dialTimeout: 1e3 // 30 second dial timeout per peer
},
config: {
transport: {
[transportKey]: {
filter: filters.all
}
}
}
}
libp2p = await Libp2p.create(config)
@ -317,6 +363,13 @@ describe('Dialing (direct, WebSockets)', () => {
transport: [Transport],
streamMuxer: [Muxer],
connEncryption: [Crypto]
},
config: {
transport: {
[transportKey]: {
filter: filters.all
}
}
}
})
@ -340,6 +393,13 @@ describe('Dialing (direct, WebSockets)', () => {
transport: [Transport],
streamMuxer: [Muxer],
connEncryption: [Crypto]
},
config: {
transport: {
[transportKey]: {
filter: filters.all
}
}
}
})
@ -349,7 +409,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 +417,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)
})
@ -370,6 +427,13 @@ describe('Dialing (direct, WebSockets)', () => {
transport: [Transport],
streamMuxer: [Muxer],
connEncryption: [Crypto]
},
config: {
transport: {
[transportKey]: {
filter: filters.all
}
}
}
})
@ -387,6 +451,13 @@ describe('Dialing (direct, WebSockets)', () => {
transport: [Transport],
streamMuxer: [Muxer],
connEncryption: [Crypto]
},
config: {
transport: {
[transportKey]: {
filter: filters.all
}
}
}
})
@ -400,6 +471,13 @@ describe('Dialing (direct, WebSockets)', () => {
transport: [Transport],
streamMuxer: [Muxer],
connEncryption: [Crypto]
},
config: {
transport: {
[transportKey]: {
filter: filters.all
}
}
}
})

View File

@ -37,11 +37,12 @@ describe('Dialing (resolvable addresses)', () => {
[libp2p, remoteLibp2p] = await peerUtils.createPeer({
number: 2,
config: {
modules: baseOptions.modules,
...baseOptions,
addresses: {
listen: [multiaddr(`${relayAddr}/p2p-circuit`)]
},
config: {
...baseOptions.config,
peerDiscovery: {
autoDial: false
}

View File

@ -1,14 +1,10 @@
'use strict'
/* eslint-env mocha */
const chai = require('chai')
chai.use(require('dirty-chai'))
chai.use(require('chai-as-promised'))
const { expect } = chai
const { expect } = require('aegir/utils/chai')
const sinon = require('sinon')
const { EventEmitter } = require('events')
const delay = require('delay')
const PeerId = require('peer-id')
const duplexPair = require('it-pair/duplex')
const multiaddr = require('multiaddr')
@ -16,12 +12,14 @@ const pWaitFor = require('p-wait-for')
const unit8ArrayToString = require('uint8arrays/to-string')
const { codes: Errors } = require('../../src/errors')
const { IdentifyService, multicodecs } = require('../../src/identify')
const IdentifyService = require('../../src/identify')
const multicodecs = IdentifyService.multicodecs
const Peers = require('../fixtures/peers')
const Libp2p = require('../../src')
const Envelope = require('../../src/record/envelope')
const PeerStore = require('../../src/peer-store')
const baseOptions = require('../utils/base-options.browser')
const { updateSelfPeerRecord } = require('../../src/record/utils')
const pkg = require('../../package.json')
const { MULTIADDRS_WEBSOCKETS } = require('../fixtures/browser')
@ -29,18 +27,21 @@ const remoteAddr = MULTIADDRS_WEBSOCKETS[0]
const listenMaddrs = [multiaddr('/ip4/127.0.0.1/tcp/15002/ws')]
describe('Identify', () => {
let localPeer
let remotePeer
const protocols = new Map([
[multicodecs.IDENTIFY, () => {}],
[multicodecs.IDENTIFY_PUSH, () => {}]
])
let localPeer, localPeerStore
let remotePeer, remotePeerStore
const protocols = [multicodecs.IDENTIFY, multicodecs.IDENTIFY_PUSH]
before(async () => {
[localPeer, remotePeer] = (await Promise.all([
PeerId.createFromJSON(Peers[0]),
PeerId.createFromJSON(Peers[1])
]))
localPeerStore = new PeerStore({ peerId: localPeer })
localPeerStore.protoBook.set(localPeer, protocols)
remotePeerStore = new PeerStore({ peerId: remotePeer })
remotePeerStore.protoBook.set(remotePeer, protocols)
})
afterEach(() => {
@ -52,22 +53,21 @@ describe('Identify', () => {
libp2p: {
peerId: localPeer,
connectionManager: new EventEmitter(),
peerStore: new PeerStore({ peerId: localPeer }),
peerStore: localPeerStore,
multiaddrs: listenMaddrs,
isStarted: () => true,
_options: { host: {} }
},
protocols
}
})
const remoteIdentify = new IdentifyService({
libp2p: {
peerId: remotePeer,
connectionManager: new EventEmitter(),
peerStore: new PeerStore({ peerId: remotePeer }),
peerStore: remotePeerStore,
multiaddrs: listenMaddrs,
isStarted: () => true,
_options: { host: {} }
},
protocols
}
})
const observedAddr = multiaddr('/ip4/127.0.0.1/tcp/1234')
@ -80,6 +80,9 @@ describe('Identify', () => {
sinon.spy(localIdentify.peerStore.addressBook, 'consumePeerRecord')
sinon.spy(localIdentify.peerStore.protoBook, 'set')
// Transport Manager creates signed peer record
await updateSelfPeerRecord(remoteIdentify._libp2p)
// Run identify
await Promise.all([
localIdentify.identify(localConnectionMock),
@ -107,22 +110,22 @@ describe('Identify', () => {
libp2p: {
peerId: localPeer,
connectionManager: new EventEmitter(),
peerStore: new PeerStore({ peerId: localPeer }),
peerStore: localPeerStore,
multiaddrs: listenMaddrs,
isStarted: () => true,
_options: { host: {} }
},
protocols
}
})
const remoteIdentify = new IdentifyService({
libp2p: {
peerId: remotePeer,
connectionManager: new EventEmitter(),
peerStore: new PeerStore({ peerId: remotePeer }),
peerStore: remotePeerStore,
multiaddrs: listenMaddrs,
isStarted: () => true,
_options: { host: {} }
},
protocols
}
})
const observedAddr = multiaddr('/ip4/127.0.0.1/tcp/1234')
@ -168,21 +171,19 @@ describe('Identify', () => {
libp2p: {
peerId: localPeer,
connectionManager: new EventEmitter(),
peerStore: new PeerStore({ peerId: localPeer }),
peerStore: localPeerStore,
multiaddrs: [],
_options: { host: {} }
},
protocols
}
})
const remoteIdentify = new IdentifyService({
libp2p: {
peerId: remotePeer,
connectionManager: new EventEmitter(),
peerStore: new PeerStore({ peerId: remotePeer }),
peerStore: remotePeerStore,
multiaddrs: [],
_options: { host: {} }
},
protocols
}
})
const observedAddr = multiaddr('/ip4/127.0.0.1/tcp/1234')
@ -239,35 +240,40 @@ describe('Identify', () => {
describe('push', () => {
it('should be able to push identify updates to another peer', async () => {
const storedProtocols = [multicodecs.IDENTIFY, multicodecs.IDENTIFY_PUSH, '/echo/1.0.0']
const connectionManager = new EventEmitter()
connectionManager.getConnection = () => { }
const localPeerStore = new PeerStore({ peerId: localPeer })
localPeerStore.protoBook.set(localPeer, storedProtocols)
const localIdentify = new IdentifyService({
libp2p: {
peerId: localPeer,
connectionManager: new EventEmitter(),
peerStore: new PeerStore({ peerId: localPeer }),
peerStore: localPeerStore,
multiaddrs: listenMaddrs,
isStarted: () => true,
_options: { host: {} }
},
protocols: new Map([
[multicodecs.IDENTIFY],
[multicodecs.IDENTIFY_PUSH],
['/echo/1.0.0']
])
}
})
const remotePeerStore = new PeerStore({ peerId: remotePeer })
remotePeerStore.protoBook.set(remotePeer, storedProtocols)
const remoteIdentify = new IdentifyService({
libp2p: {
peerId: remotePeer,
connectionManager,
peerStore: new PeerStore({ peerId: remotePeer }),
peerStore: remotePeerStore,
multiaddrs: [],
isStarted: () => true,
_options: { host: {} }
}
})
// Setup peer protocols and multiaddrs
const localProtocols = new Set([multicodecs.IDENTIFY, multicodecs.IDENTIFY_PUSH, '/echo/1.0.0'])
const localProtocols = new Set(storedProtocols)
const localConnectionMock = { newStream: () => { } }
const remoteConnectionMock = { remotePeer: localPeer }
@ -277,6 +283,10 @@ describe('Identify', () => {
sinon.spy(remoteIdentify.peerStore.addressBook, 'consumePeerRecord')
sinon.spy(remoteIdentify.peerStore.protoBook, 'set')
// Transport Manager creates signed peer record
await updateSelfPeerRecord(localIdentify._libp2p)
await updateSelfPeerRecord(remoteIdentify._libp2p)
// Run identify
await Promise.all([
localIdentify.push([localConnectionMock]),
@ -287,7 +297,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)
@ -302,35 +312,40 @@ describe('Identify', () => {
// LEGACY
it('should be able to push identify updates to another peer with no certified peer records support', async () => {
const storedProtocols = [multicodecs.IDENTIFY, multicodecs.IDENTIFY_PUSH, '/echo/1.0.0']
const connectionManager = new EventEmitter()
connectionManager.getConnection = () => { }
const localPeerStore = new PeerStore({ peerId: localPeer })
localPeerStore.protoBook.set(localPeer, storedProtocols)
const localIdentify = new IdentifyService({
libp2p: {
peerId: localPeer,
connectionManager: new EventEmitter(),
peerStore: new PeerStore({ peerId: localPeer }),
peerStore: localPeerStore,
multiaddrs: listenMaddrs,
isStarted: () => true,
_options: { host: {} }
},
protocols: new Map([
[multicodecs.IDENTIFY],
[multicodecs.IDENTIFY_PUSH],
['/echo/1.0.0']
])
}
})
const remotePeerStore = new PeerStore({ peerId: remotePeer })
remotePeerStore.protoBook.set(remotePeer, storedProtocols)
const remoteIdentify = new IdentifyService({
libp2p: {
peerId: remotePeer,
connectionManager,
peerStore: new PeerStore({ peerId: remotePeer }),
multiaddrs: [],
_options: { host: {} }
_options: { host: {} },
isStarted: () => true
}
})
// Setup peer protocols and multiaddrs
const localProtocols = new Set([multicodecs.IDENTIFY, multicodecs.IDENTIFY_PUSH, '/echo/1.0.0'])
const localProtocols = new Set(storedProtocols)
const localConnectionMock = { newStream: () => {} }
const remoteConnectionMock = { remotePeer: localPeer }
@ -399,8 +414,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
@ -421,8 +436,6 @@ describe('Identify', () => {
const connection = await libp2p.dialer.connectToPeer(remoteAddr)
expect(connection).to.exist()
// Wait for nextTick to trigger the identify call
await delay(1)
// Wait for identify to finish
await libp2p.identifyService.identify.firstCall.returnValue
@ -462,5 +475,39 @@ describe('Identify', () => {
expect(agentVersion).to.eql(unit8ArrayToString(storedAgentVersion))
expect(storedProtocolVersion).to.exist()
})
it('should push multiaddr updates to an already connected peer', async () => {
libp2p = new Libp2p({
...baseOptions,
peerId
})
await libp2p.start()
sinon.spy(libp2p.identifyService, 'identify')
sinon.spy(libp2p.identifyService, 'push')
const connection = await libp2p.dialer.connectToPeer(remoteAddr)
expect(connection).to.exist()
// Wait for identify to finish
await libp2p.identifyService.identify.firstCall.returnValue
sinon.stub(libp2p, 'isStarted').returns(true)
libp2p.peerStore.addressBook.add(libp2p.peerId, [multiaddr('/ip4/180.0.0.1/tcp/15001/ws')])
// Verify the remote peer is notified of change
expect(libp2p.identifyService.push.callCount).to.equal(1)
for (const call of libp2p.identifyService.push.getCalls()) {
const [connections] = call.args
expect(connections.length).to.equal(1)
expect(connections[0].remotePeer.toB58String()).to.equal(remoteAddr.getPeerId())
const results = await call.returnValue
expect(results.length).to.equal(1)
}
// Verify the streams close
await pWaitFor(() => connection.streams.length === 0)
})
})
})

View File

@ -1,9 +1,7 @@
'use strict'
/* eslint-env mocha */
const chai = require('chai')
chai.use(require('dirty-chai'))
const { expect } = chai
const { expect } = require('aegir/utils/chai')
const sinon = require('sinon')
const PeerId = require('peer-id')

View File

@ -2,10 +2,7 @@
/* eslint-env mocha */
'use strict'
const chai = require('chai')
const dirtyChai = require('dirty-chai')
const expect = chai.expect
chai.use(dirtyChai)
const { chai, expect } = require('aegir/utils/chai')
chai.use(require('chai-string'))
const uint8ArrayFromString = require('uint8arrays/from-string')
const uint8ArrayToString = require('uint8arrays/to-string')

View File

@ -73,9 +73,9 @@ describe('keychain', () => {
it('can find a key without a password', async () => {
const keychain = new Keychain(datastore2)
const keychainWithPassword = new Keychain(datastore2, { passPhrase: `hello-${Date.now()}-${Date.now()}` })
const id = `key-${Math.random()}`
const name = `key-${Math.random()}`
await keychainWithPassword.createKey(id, 'ed25519')
const { id } = await keychainWithPassword.createKey(name, 'ed25519')
await expect(keychain.findKeyById(id)).to.eventually.be.ok()
})

View File

@ -1,10 +1,7 @@
/* eslint-env mocha */
'use strict'
const chai = require('chai')
const dirtyChai = require('dirty-chai')
const expect = chai.expect
chai.use(dirtyChai)
const { expect } = require('aegir/utils/chai')
const PeerId = require('peer-id')
const multihash = require('multihashes')
const crypto = require('libp2p-crypto')

View File

@ -1,10 +1,7 @@
'use strict'
/* eslint-env mocha */
const chai = require('chai')
chai.use(require('dirty-chai'))
chai.use(require('chai-as-promised'))
const { expect } = chai
const { expect } = require('aegir/utils/chai')
const sinon = require('sinon')
const { randomBytes } = require('libp2p-crypto')

View File

@ -1,10 +1,7 @@
'use strict'
/* eslint-env mocha */
const chai = require('chai')
chai.use(require('dirty-chai'))
chai.use(require('chai-as-promised'))
const { expect } = chai
const { expect } = require('aegir/utils/chai')
const sinon = require('sinon')
const { EventEmitter } = require('events')

View File

@ -1,9 +1,7 @@
'use strict'
/* eslint-env mocha */
const chai = require('chai')
chai.use(require('dirty-chai'))
const { expect } = chai
const { expect } = require('aegir/utils/chai')
const sinon = require('sinon')
const defer = require('p-defer')
const mergeOptions = require('merge-options')

View File

@ -1,9 +1,7 @@
'use strict'
/* eslint-env mocha */
const chai = require('chai')
chai.use(require('dirty-chai'))
const { expect } = chai
const { expect } = require('aegir/utils/chai')
const sinon = require('sinon')
const defer = require('p-defer')

View File

@ -1,17 +1,20 @@
'use strict'
/* eslint-env mocha */
const chai = require('chai')
chai.use(require('dirty-chai'))
const { expect } = chai
const { expect } = require('aegir/utils/chai')
const nock = require('nock')
const sinon = require('sinon')
const intoStream = require('into-stream')
const delay = require('delay')
const pDefer = require('p-defer')
const pWaitFor = require('p-wait-for')
const mergeOptions = require('merge-options')
const ipfsHttpClient = require('ipfs-http-client')
const DelegatedPeerRouter = require('libp2p-delegated-peer-routing')
const multiaddr = require('multiaddr')
const PeerId = require('peer-id')
const peerUtils = require('../utils/creators/peer')
const { baseOptions, routingOptions } = require('./utils')
@ -31,6 +34,16 @@ describe('peer-routing', () => {
.to.eventually.be.rejected()
.and.to.have.property('code', 'NO_ROUTERS_AVAILABLE')
})
it('.getClosestPeers should return an error', async () => {
try {
for await (const _ of node.peerRouting.getClosestPeers('a cid')) { } // eslint-disable-line
throw new Error('.getClosestPeers should return an error')
} catch (err) {
expect(err).to.exist()
expect(err.code).to.equal('NO_ROUTERS_AVAILABLE')
}
})
})
describe('via dht router', () => {
@ -66,6 +79,19 @@ describe('peer-routing', () => {
nodes[0].peerRouting.findPeer()
return deferred.promise
})
it('should use the nodes dht to get the closest peers', async () => {
const deferred = pDefer()
sinon.stub(nodes[0]._dht, 'getClosestPeers').callsFake(function * () {
deferred.resolve()
yield
})
await nodes[0].peerRouting.getClosestPeers().next()
return deferred.promise
})
})
describe('via delegate router', () => {
@ -112,6 +138,19 @@ describe('peer-routing', () => {
return deferred.promise
})
it('should use the delegate router to get the closest peers', async () => {
const deferred = pDefer()
sinon.stub(delegate, 'getClosestPeers').callsFake(function * () {
deferred.resolve()
yield
})
await node.peerRouting.getClosestPeers().next()
return deferred.promise
})
it('should be able to find a peer', async () => {
const peerKey = 'QmTp9VkYvnHyrqKQuFPiuZkiX9gPcqj6x5LJ1rmWuSySnL'
const mockApi = nock('http://0.0.0.0:60197')
@ -156,6 +195,60 @@ describe('peer-routing', () => {
expect(mockApi.isDone()).to.equal(true)
})
it('should be able to get the closest peers', async () => {
const peerId = await PeerId.create({ keyType: 'ed25519' })
const closest1 = '12D3KooWLewYMMdGWAtuX852n4rgCWkK7EBn4CWbwwBzhsVoKxk3'
const closest2 = '12D3KooWDtoQbpKhtnWddfj72QmpFvvLDTsBLTFkjvgQm6cde2AK'
const mockApi = nock('http://0.0.0.0:60197')
.post('/api/v0/dht/query')
.query(true)
.reply(200,
() => intoStream([
`{"extra":"","id":"${closest1}","responses":[{"ID":"${closest1}","Addrs":["/ip4/127.0.0.1/tcp/63930","/ip4/127.0.0.1/tcp/63930"]}],"type":1}\n`,
`{"extra":"","id":"${closest2}","responses":[{"ID":"${closest2}","Addrs":["/ip4/127.0.0.1/tcp/63506","/ip4/127.0.0.1/tcp/63506"]}],"type":1}\n`,
`{"Extra":"","ID":"${closest2}","Responses":[],"Type":2}\n`,
`{"Extra":"","ID":"${closest1}","Responses":[],"Type":2}\n`
]),
[
'Content-Type', 'application/json',
'X-Chunked-Output', '1'
])
const closestPeers = []
for await (const peer of node.peerRouting.getClosestPeers(peerId.id, { timeout: 1000 })) {
closestPeers.push(peer)
}
expect(closestPeers).to.have.length(2)
expect(closestPeers[0].id.toB58String()).to.equal(closest2)
expect(closestPeers[0].multiaddrs).to.have.lengthOf(2)
expect(closestPeers[1].id.toB58String()).to.equal(closest1)
expect(closestPeers[1].multiaddrs).to.have.lengthOf(2)
expect(mockApi.isDone()).to.equal(true)
})
it('should handle errors when getting the closest peers', async () => {
const peerId = await PeerId.create({ keyType: 'ed25519' })
const mockApi = nock('http://0.0.0.0:60197')
.post('/api/v0/dht/query')
.query(true)
.reply(502, 'Bad Gateway', [
'X-Chunked-Output', '1'
])
try {
for await (const _ of node.peerRouting.getClosestPeers(peerId.id)) { } // eslint-disable-line
throw new Error('should handle errors when getting the closest peers')
} catch (err) {
expect(err).to.exist()
}
expect(mockApi.isDone()).to.equal(true)
})
})
describe('via dht and delegate routers', () => {
@ -210,5 +303,148 @@ describe('peer-routing', () => {
const peer = await node.peerRouting.findPeer('a peer id')
expect(peer).to.eql(results)
})
it('should only use the dht if it gets the closest peers', async () => {
const results = [true]
sinon.stub(node._dht, 'getClosestPeers').callsFake(function * () {
yield results[0]
})
sinon.stub(delegate, 'getClosestPeers').callsFake(function * () { // eslint-disable-line require-yield
throw new Error('the delegate should not have been called')
})
const closest = []
for await (const peer of node.peerRouting.getClosestPeers('a cid')) {
closest.push(peer)
}
expect(closest).to.have.length.above(0)
expect(closest).to.eql(results)
})
it('should use the delegate if the dht fails to get the closest peer', async () => {
const results = [true]
sinon.stub(node._dht, 'getClosestPeers').callsFake(function * () { })
sinon.stub(delegate, 'getClosestPeers').callsFake(function * () {
yield results[0]
})
const closest = []
for await (const peer of node.peerRouting.getClosestPeers('a cid')) {
closest.push(peer)
}
expect(closest).to.have.length.above(0)
expect(closest).to.eql(results)
})
})
describe('peer routing refresh manager service', () => {
let node
let peerIds
before(async () => {
peerIds = await peerUtils.createPeerId({ number: 2 })
})
afterEach(() => {
sinon.restore()
return node && node.stop()
})
it('should be enabled and start by default', async () => {
const results = [
{ id: peerIds[0], multiaddrs: [multiaddr('/ip4/30.0.0.1/tcp/2000')] },
{ id: peerIds[1], multiaddrs: [multiaddr('/ip4/32.0.0.1/tcp/2000')] }
]
;[node] = await peerUtils.createPeer({
config: mergeOptions(routingOptions, {
peerRouting: {
refreshManager: {
bootDelay: 100
}
}
}),
started: false
})
sinon.spy(node.peerStore.addressBook, 'add')
sinon.stub(node._dht, 'getClosestPeers').callsFake(function * () {
yield results[0]
yield results[1]
})
await node.start()
await pWaitFor(() => node._dht.getClosestPeers.callCount === 1)
await pWaitFor(() => node.peerStore.addressBook.add.callCount === results.length)
const call0 = node.peerStore.addressBook.add.getCall(0)
expect(call0.args[0].equals(results[0].id))
call0.args[1].forEach((m, index) => {
expect(m.equals(results[0].multiaddrs[index]))
})
const call1 = node.peerStore.addressBook.add.getCall(1)
expect(call1.args[0].equals(results[1].id))
call0.args[1].forEach((m, index) => {
expect(m.equals(results[1].multiaddrs[index]))
})
})
it('should support being disabled', async () => {
[node] = await peerUtils.createPeer({
config: mergeOptions(routingOptions, {
peerRouting: {
refreshManager: {
bootDelay: 100,
enabled: false
}
}
}),
started: false
})
sinon.stub(node._dht, 'getClosestPeers').callsFake(function * () {
yield
throw new Error('should not be called')
})
await node.start()
await delay(100)
expect(node._dht.getClosestPeers.callCount === 0)
})
it('should start and run recurrently on interval', async () => {
[node] = await peerUtils.createPeer({
config: mergeOptions(routingOptions, {
peerRouting: {
refreshManager: {
interval: 500,
bootDelay: 200
}
}
}),
started: false
})
sinon.stub(node._dht, 'getClosestPeers').callsFake(function * () {
yield { id: peerIds[0], multiaddrs: [multiaddr('/ip4/30.0.0.1/tcp/2000')] }
})
await node.start()
await delay(300)
expect(node._dht.getClosestPeers.callCount).to.eql(1)
await delay(500)
expect(node._dht.getClosestPeers.callCount).to.eql(2)
})
})
})

Some files were not shown because too many files have changed in this diff Show More