mirror of
https://github.com/fluencelabs/js-libp2p
synced 2025-07-31 16:32:01 +00:00
refactor: add core modules to libp2p (#400)
* refactor: add js-libp2p-connection-manager to repo Co-authored-by: David Dias <daviddias.p@gmail.com> Co-authored-by: Jacob Heun <jacobheun@gmail.com> Co-authored-by: Pedro Teixeira <i@pgte.me> Co-authored-by: Vasco Santos <vasco.santos@ua.pt> * test(conn-mgr): only run in node * refactor: add js-libp2p-identify to repo Co-authored-by: David Dias <daviddias.p@gmail.com> Co-authored-by: Friedel Ziegelmayer <dignifiedquire@gmail.com> Co-authored-by: Hugo Dias <hugomrdias@gmail.com> Co-authored-by: Jacob Heun <jacobheun@gmail.com> Co-authored-by: Maciej Krüger <mkg20001@gmail.com> Co-authored-by: Richard Littauer <richard.littauer@gmail.com> Co-authored-by: Vasco Santos <vasco.santos@moxy.studio> Co-authored-by: Yusef Napora <yusef@protocol.ai> Co-authored-by: ᴠɪᴄᴛᴏʀ ʙᴊᴇʟᴋʜᴏʟᴍ <victorbjelkholm@gmail.com> * refactor: add libp2p-pnet to repo Co-authored-by: Jacob Heun <jacobheun@gmail.com> Co-authored-by: Vasco Santos <vasco.santos@moxy.studio> * refactor: add libp2p-ping to repo Co-authored-by: David Dias <daviddias.p@gmail.com> Co-authored-by: Francisco Baio Dias <xicombd@gmail.com> Co-authored-by: Friedel Ziegelmayer <dignifiedquire@gmail.com> Co-authored-by: Hugo Dias <mail@hugodias.me> Co-authored-by: Jacob Heun <jacobheun@gmail.com> Co-authored-by: João Antunes <j.goncalo.antunes@gmail.com> Co-authored-by: Richard Littauer <richard.littauer@gmail.com> Co-authored-by: Vasco Santos <vasco.santos@moxy.studio> Co-authored-by: Vasco Santos <vasco.santos@ua.pt> Co-authored-by: ᴠɪᴄᴛᴏʀ ʙᴊᴇʟᴋʜᴏʟᴍ <victorbjelkholm@gmail.com> * refactor: add libp2p-circuit to repo Co-authored-by: David Dias <daviddias.p@gmail.com> Co-authored-by: Dmitriy Ryajov <dryajov@gmail.com> Co-authored-by: Friedel Ziegelmayer <dignifiedquire@gmail.com> Co-authored-by: Hugo Dias <mail@hugodias.me> Co-authored-by: Jacob Heun <jacobheun@gmail.com> Co-authored-by: Maciej Krüger <mkg20001@gmail.com> Co-authored-by: Oli Evans <oli@tableflip.io> Co-authored-by: Pedro Teixeira <i@pgte.me> Co-authored-by: Vasco Santos <vasco.santos@ua.pt> Co-authored-by: Victor Bjelkholm <victorbjelkholm@gmail.com> Co-authored-by: Yusef Napora <yusef@napora.org> Co-authored-by: dirkmc <dirk@mccormick.cx> * test(switch): avoid using instanceof * chore(switch): update bignumber dep * refactor(circuit): clean up tests * refactor(switch): consolidate get peer utils * test(identify): do deep checks of addresses * test(identify): bump timeout for identify test * test(switch): tidy up limit dialer test * refactor(switch): remove redundant circuit tests * chore: add coverage script * refactor(circuit): consolidate get peer info * docs: reference original repositories in each sub readme * docs: fix comment * refactor: clean up sub package.json files and readmes
This commit is contained in:
Jacob Heun
GitHub
128
src/circuit/IMPLEMENTATION_NOTES.md
Normal file
128
src/circuit/IMPLEMENTATION_NOTES.md
Normal file
@@ -0,0 +1,128 @@
|
||||
EDIT: This document is outdated and here only for historical purposes
|
||||
|
||||
NOTE: This document is structured in an `if-then/else[if]-then` manner, each line is a precondition for following lines with a higher number of indentation
|
||||
|
||||
Example:
|
||||
|
||||
- if there are apples
|
||||
- eat them
|
||||
- if not, check for pears
|
||||
- then eat them
|
||||
- if not, check for cherries
|
||||
- then eat them
|
||||
|
||||
Or,
|
||||
|
||||
- if there are apples
|
||||
- eat them
|
||||
- if not
|
||||
- check for pears
|
||||
- then eat them
|
||||
- if not
|
||||
- check for cherries
|
||||
- then eat them
|
||||
|
||||
In order to minimize nesting, the first example is preferred
|
||||
|
||||
# Relay flow
|
||||
|
||||
## Relay transport (dialer/listener)
|
||||
|
||||
- ### Dial over a relay
|
||||
- See if there is a relay that's already connected to the destination peer, if not
|
||||
- Ask all the peer's known relays to dial the destination peer until an active relay (one that can dial on behalf of other peers), or a relay that may have recently acquired a connection to the destination peer is successful.
|
||||
- If successful
|
||||
- Write the `/ipfs/relay/circuit/1.0.0` header to the relay, followed by the destination address
|
||||
- e.g. `/ipfs/relay/circuit/1.0.0\n/p2p-circuit/ipfs/QmDest`.
|
||||
- If no relays could connect, fail the same way a regular transport would
|
||||
- Once the connection has been established, the swarm should treat it as a regular connection,
|
||||
- i.e. muxing, encrypt, etc should all be performed on the relayed connection
|
||||
|
||||
- ### Listen for relayed connections
|
||||
- Peer mounts the `/ipfs/relay/circuit/1.0.0` proto and listens for relayed connections
|
||||
- A connection arrives
|
||||
- read the address of the source peer from the incoming connection stream
|
||||
- if valid, create a PeerInfo object for that peer and add the incoming address to its multiaddresses list
|
||||
- pass the connection to `protocolMuxer(swarm.protocols, conn)` to have it go through the regular muxing/encryption flow
|
||||
|
||||
- ### Relay discovery and static relay addresses in swarm config
|
||||
|
||||
- #### Relay address in swarm config
|
||||
- A peer has relay addresses in its swarm config section
|
||||
- On node startup, connect to the relays in swarm config
|
||||
- if successful add address to swarms PeerInfo's multiaddresses
|
||||
- `identify` should take care of announcing that the peer is reachable over the listed relays
|
||||
|
||||
- #### Passive relay discovery
|
||||
- A peer that can dial over `/ipfs/relay/circuit/1.0.0` listens for the `peer-mux-established` swarm event, every time a new muxed connection arrives, it checks if the incoming peer is a relay. (How would this work? Some way of discovering if its a relay is required.)
|
||||
- *Useful in cases when the peer/node doesn't know of any relays on startup and also, to learn of as many additional relays in the network as possible*
|
||||
- *Useful during startup, when connecting to bootstrap nodes. It allows us to implicitly learn if its a relay without having to explicitly add `/p2p-circuit` addresses to the bootstrap list*
|
||||
- *Also useful if the relay communicates its capabilities upon connecting to it, as to avoid additional unnecessary requests/queries. I.e. if it supports weather its able to forward connections and weather it supports the `ls` or other commands.*
|
||||
- *Should it be possible to disable passive relay discovery?*
|
||||
- This could be useful when the peer wants to be reachable **only** over the listed relays
|
||||
- If the incoming peer is a relay, send an `ls` and record its peers
|
||||
|
||||
## Relay Nodes
|
||||
|
||||
- ### Passive relay node
|
||||
- *A passive relay does not explicitly dial into any requested peer, only those that it's swarm already has connections to.*
|
||||
- When the relay gets a request, read the the destination peer's multiaddr from the connection stream and if its a valid address and peer id
|
||||
- check its swarm's peerbook(?) see if its a known peer, if it is
|
||||
- use the swarms existing connection and
|
||||
- send the multistream header and the source peer address to the dest peer
|
||||
- e.g. `/ipfs/relay/circuit/1.0.0\n/p2p-circuit/ipfs/QmSource`
|
||||
- circuit the source and dest connections
|
||||
- if couldn't dial, or the connection/stream to the dest peer closed prematurelly
|
||||
- close the src stream
|
||||
|
||||
|
||||
- ### Active relay node
|
||||
- *An active relay node can dial other peers even if its swarm doesnt know about those peers*
|
||||
- When the relay gets a request, read the the destination peer's multiaddr from the connection stream and if its a valid address and peer id
|
||||
- use the swarm to dial to the dest node
|
||||
- send the multistream header and the source peer address to the dest peer
|
||||
- e.g. `/ipfs/relay/circuit/1.0.0\n/p2p-circuit/ipfs/QmSource`
|
||||
- circuit the source and dest connections
|
||||
- if couldn't dial, or the connection/stream to the dest peer closed prematurely
|
||||
- close the src stream
|
||||
|
||||
- ### `ls` command
|
||||
- *A relay node can allow the peers known to it's swarm to be listed*
|
||||
- *this should be possible to enable/disable from the config*
|
||||
- when a relay gets the `ls` request
|
||||
- if enabled, get its swarm's peerbook's known peers and return their ids and multiaddrs
|
||||
- e.g `[{id: /ipfs/QmPeerId, addrs: ['ma1', 'ma2', 'ma3']}, ...]`
|
||||
- if disabled, respond with `na`
|
||||
|
||||
|
||||
## Relay Implementation notes
|
||||
|
||||
- ### Relay transport
|
||||
- Currently I've implemented the dialer and listener parts of the relay as a transport, meaning that it *tries* to implement the `interface-transport` interface as closely as possible. This seems to work pretty well and it's makes the dialer/listener parts really easy to plug in into the swarm. I think this is the cleanest solution.
|
||||
|
||||
- ### `circuit-relay`
|
||||
- This is implemented as a separate piece (not a transport), and it can be enabled/disabled with a config. The transport listener however, will do the initial parsing of the incoming header and figure out weather it's a connection that's needs to be handled by the circuit-relay, or its a connection that is being relayed from a circuit-relay.
|
||||
|
||||
## Relay swarm integration
|
||||
|
||||
- The relay transport is mounted explicitly by calling the `swarm.connection.relay(config.relay)` from libp2p
|
||||
- Swarm will register the dialer and listener using the swarm `transport.add` and `transport.listen` methods
|
||||
|
||||
- ### Listener
|
||||
- the listener registers itself as a multistream handler on the `/ipfs/relay/circuit/1.0.0` proto
|
||||
- if `circuit-relay` is enabled, the listener will delegate connections to it if appropriate
|
||||
- when the listener receives a connection, it will read the multiaddr and determine if its a connection that needs to be relayed, or its a connection that is being relayed
|
||||
|
||||
- ### Dialer
|
||||
- When the swarm attempts to dial to a peer, it will filter the protocols that the peer can be reached on
|
||||
- *The relay will be used in two cases*
|
||||
- If the peer has an explicit relay address that it can be reached on
|
||||
- no other transport is available
|
||||
- The relay will attempt to dial the peer over that relay
|
||||
- If no explicit relay address is provided
|
||||
- no other transport is available
|
||||
- A generic circuit address will be added to the peers multiaddr list
|
||||
- i.e. `/p2p-circuit/ipfs/QmDest`
|
||||
- If another transport is available, then use that instead of the relay
|
||||
|
||||
|
||||
156
src/circuit/README.md
Normal file
156
src/circuit/README.md
Normal file
@@ -0,0 +1,156 @@
|
||||
# js-libp2p-circuit
|
||||
|
||||
> Node.js implementation of the Circuit module that libp2p uses, which implements the [interface-connection](https://github.com/libp2p/interface-connection) interface for dial/listen.
|
||||
|
||||
**Note**: git history prior to merging into js-libp2p can be found in the original repository, https://github.com/libp2p/js-libp2p-circuit.
|
||||
|
||||
`libp2p-circuit` implements the circuit-relay mechanism that allows nodes that don't speak the same protocol to communicate using a third _relay_ node.
|
||||
|
||||
This module uses [pull-streams](https://pull-stream.github.io) for all stream based interfaces.
|
||||
|
||||
### Why?
|
||||
|
||||
`circuit-relaying` uses additional nodes in order to transfer traffic between two otherwise unreachable nodes. This allows nodes that don't speak the same protocols or are running in limited environments, e.g. browsers and IoT devices, to communicate, which would otherwise be impossible given the fact that for example browsers don't have any socket support and as such cannot be directly dialed.
|
||||
|
||||
The use of circuit-relaying is not limited to routing traffic between browser nodes, other uses include:
|
||||
- routing traffic between private nets and circumventing NAT layers
|
||||
- route mangling for better privacy (matreshka/shallot dialing).
|
||||
|
||||
It's also possible to use it for clients that implement exotic transports such as devices that only have bluetooth radios to be reachable over bluetooth enabled relays and become full p2p nodes.
|
||||
|
||||
### libp2p-circuit and IPFS
|
||||
|
||||
Prior to `libp2p-circuit` there was a rift in the IPFS network, were IPFS nodes could only access content from nodes that speak the same protocol, for example TCP only nodes could only dial to other TCP only nodes, same for any other protocol combination. In practice, this limitation was most visible in JS-IPFS browser nodes, since they can only dial out but not be dialed in over WebRTC or WebSockets, hence any content that the browser node held was not reachable by the rest of the network even through it was announced on the DHT. Non browser IPFS nodes would usually speak more than one protocol such as TCP, WebSockets and/or WebRTC, this made the problem less severe outside of the browser. `libp2p-circuit` solves this problem completely, as long as there are `relay nodes` capable of routing traffic between those nodes their content should be available to the rest of the IPFS network.
|
||||
|
||||
## Table of Contents
|
||||
|
||||
- [Install](#install)
|
||||
- [npm](#npm)
|
||||
- [Usage](#usage)
|
||||
- [Example](#example)
|
||||
- [This module uses `pull-streams`](#this-module-uses-pull-streams)
|
||||
- [Converting `pull-streams` to Node.js Streams](#converting-pull-streams-to-nodejs-streams)
|
||||
- [API](#api)
|
||||
- [Contribute](#contribute)
|
||||
- [License](#license)
|
||||
|
||||
## Usage
|
||||
|
||||
### Example
|
||||
|
||||
#### Create dialer/listener
|
||||
|
||||
```js
|
||||
const Circuit = require('libp2p-circuit')
|
||||
const multiaddr = require('multiaddr')
|
||||
const pull = require('pull-stream')
|
||||
|
||||
const mh1 = multiaddr('/p2p-circuit/ipfs/QmHash') // dial /ipfs/QmHash over any circuit
|
||||
|
||||
const circuit = new Circuit(swarmInstance, options) // pass swarm instance and options
|
||||
|
||||
const listener = circuit.createListener(mh1, (connection) => {
|
||||
console.log('new connection opened')
|
||||
pull(
|
||||
pull.values(['hello']),
|
||||
socket
|
||||
)
|
||||
})
|
||||
|
||||
listener.listen(() => {
|
||||
console.log('listening')
|
||||
|
||||
pull(
|
||||
circuit.dial(mh1),
|
||||
pull.log,
|
||||
pull.onEnd(() => {
|
||||
circuit.close()
|
||||
})
|
||||
)
|
||||
})
|
||||
```
|
||||
|
||||
Outputs:
|
||||
|
||||
```sh
|
||||
listening
|
||||
new connection opened
|
||||
hello
|
||||
```
|
||||
|
||||
#### Create `relay`
|
||||
|
||||
```js
|
||||
const Relay = require('libp2p-circuit').Relay
|
||||
|
||||
const relay = new Relay(options)
|
||||
|
||||
relay.mount(swarmInstance) // start relaying traffic
|
||||
```
|
||||
|
||||
### This module uses `pull-streams`
|
||||
|
||||
We expose a streaming interface based on `pull-streams`, rather then on the Node.js core streams implementation (aka Node.js streams). `pull-streams` offers us a better mechanism for error handling and flow control guarantees. If you would like to know more about why we did this, see the discussion at this [issue](https://github.com/ipfs/js-ipfs/issues/362).
|
||||
|
||||
You can learn more about pull-streams at:
|
||||
|
||||
- [The history of Node.js streams, nodebp April 2014](https://www.youtube.com/watch?v=g5ewQEuXjsQ)
|
||||
- [The history of streams, 2016](http://dominictarr.com/post/145135293917/history-of-streams)
|
||||
- [pull-streams, the simple streaming primitive](http://dominictarr.com/post/149248845122/pull-streams-pull-streams-are-a-very-simple)
|
||||
- [pull-streams documentation](https://pull-stream.github.io/)
|
||||
|
||||
#### Converting `pull-streams` to Node.js Streams
|
||||
|
||||
If you are a Node.js streams user, you can convert a pull-stream to a Node.js stream using the module [`pull-stream-to-stream`](https://github.com/dominictarr/pull-stream-to-stream), giving you an instance of a Node.js stream that is linked to the pull-stream. For example:
|
||||
|
||||
```js
|
||||
const pullToStream = require('pull-stream-to-stream')
|
||||
|
||||
const nodeStreamInstance = pullToStream(pullStreamInstance)
|
||||
// nodeStreamInstance is an instance of a Node.js Stream
|
||||
```
|
||||
|
||||
To learn more about this utility, visit https://pull-stream.github.io/#pull-stream-to-stream.
|
||||
|
||||
## API
|
||||
|
||||
[](https://github.com/libp2p/interface-transport)
|
||||
|
||||
`libp2p-circuit` accepts Circuit addresses for both IPFS and non IPFS encapsulated addresses, i.e:
|
||||
|
||||
`/p2p-circuit/ip4/127.0.0.1/tcp/4001/ipfs/QmHash`
|
||||
|
||||
Both for dialing and listening.
|
||||
|
||||
### Implementation rational
|
||||
|
||||
This module is not a transport, however it implements `interface-transport` interface in order to allow circuit to be plugged with `libp2p-swarm`. The rational behind it is that, `libp2p-circuit` has a dial and listen flow, which fits nicely with other transports, moreover, it requires the _raw_ connection to be encrypted and muxed just as a regular transport's connection does. All in all, `interface-transport` ended up being the correct level of abstraction for circuit, as well as allowed us to reuse existing integration points in `libp2p-swarm` and `libp2p` without adding any ad-hoc logic. All parts of `interface-transport` are used, including `.getAddr` which returns a list of `/p2p-circuit` addresses that circuit is currently listening.
|
||||
|
||||
```
|
||||
libp2p libp2p-circuit (transport)
|
||||
+-------------------------------------------------+ +--------------------------+
|
||||
| +---------------------------------+ | | |
|
||||
| | | | | +------------------+ |
|
||||
| | | | circuit-relay listens for the HOP | | | |
|
||||
| | libp2p-swarm <------------------------------------------------| circuit-relay | |
|
||||
| | | | message to handle incomming relay | | | |
|
||||
| | | | requests from other nodes | +------------------+ |
|
||||
| +---------------------------------+ | | |
|
||||
| ^ ^ ^ ^ ^ ^ | | +------------------+ |
|
||||
| | | | | | | | | | +-------------+ | |
|
||||
| | | | | | | | dialer uses libp2p-swarm to dial | | | | | |
|
||||
| | | | +----------------------------------------------------------------------> dialer | | |
|
||||
| | | transports | | to a circuit-relay node using the | | | | | |
|
||||
| | | | | | | HOP message | | +-------------+ | |
|
||||
| | | | | | | | | | |
|
||||
| v v | v v | | | | |
|
||||
|+------------------|----------------------------+| | | +-------------+ | |
|
||||
|| | | | | || | | | | | |
|
||||
||libp2p-tcp |libp2p-ws | .... |libp2p-circuit || listener handles STOP messages from| | | listener | | |
|
||||
|| | +--------------------------------------------------------------------------> | | |
|
||||
|| | | |plugs in just || circuit-relay nodes | | +-------------+ | |
|
||||
|| | | |as any other || | | | |
|
||||
|| | | |transport || | +------------------+ |
|
||||
|+-----------------------------------------------+| | |
|
||||
+-------------------------------------------------+ +--------------------------+
|
||||
```
|
||||
126
src/circuit/circuit.js
Normal file
126
src/circuit/circuit.js
Normal file
@@ -0,0 +1,126 @@
|
||||
'use strict'
|
||||
|
||||
const mafmt = require('mafmt')
|
||||
const multiaddr = require('multiaddr')
|
||||
|
||||
const CircuitDialer = require('./circuit/dialer')
|
||||
const utilsFactory = require('./circuit/utils')
|
||||
|
||||
const debug = require('debug')
|
||||
const log = debug('libp2p:circuit:transportdialer')
|
||||
log.err = debug('libp2p:circuit:error:transportdialer')
|
||||
|
||||
const createListener = require('./listener')
|
||||
|
||||
class Circuit {
|
||||
static get tag () {
|
||||
return 'Circuit'
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates an instance of Dialer.
|
||||
*
|
||||
* @param {Swarm} swarm - the swarm
|
||||
* @param {any} options - config options
|
||||
*
|
||||
* @memberOf Dialer
|
||||
*/
|
||||
constructor (swarm, options) {
|
||||
this.options = options || {}
|
||||
|
||||
this.swarm = swarm
|
||||
this.dialer = null
|
||||
this.utils = utilsFactory(swarm)
|
||||
this.peerInfo = this.swarm._peerInfo
|
||||
this.relays = this.filter(this.peerInfo.multiaddrs.toArray())
|
||||
|
||||
// if no explicit relays, add a default relay addr
|
||||
if (this.relays.length === 0) {
|
||||
this.peerInfo
|
||||
.multiaddrs
|
||||
.add(`/p2p-circuit/ipfs/${this.peerInfo.id.toB58String()}`)
|
||||
}
|
||||
|
||||
this.dialer = new CircuitDialer(swarm, options)
|
||||
|
||||
this.swarm.on('peer-mux-established', (peerInfo) => {
|
||||
this.dialer.canHop(peerInfo)
|
||||
})
|
||||
this.swarm.on('peer-mux-closed', (peerInfo) => {
|
||||
this.dialer.relayPeers.delete(peerInfo.id.toB58String())
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Dial the relays in the Addresses.Swarm config
|
||||
*
|
||||
* @param {Array} relays
|
||||
* @return {void}
|
||||
*/
|
||||
_dialSwarmRelays () {
|
||||
// if we have relay addresses in swarm config, then dial those relays
|
||||
this.relays.forEach((relay) => {
|
||||
const relaySegments = relay
|
||||
.toString()
|
||||
.split('/p2p-circuit')
|
||||
.filter(segment => segment.length)
|
||||
|
||||
relaySegments.forEach((relaySegment) => {
|
||||
const ma = this.utils.peerInfoFromMa(multiaddr(relaySegment))
|
||||
this.dialer._dialRelay(ma)
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Dial a peer over a relay
|
||||
*
|
||||
* @param {multiaddr} ma - the multiaddr of the peer to dial
|
||||
* @param {Object} options - dial options
|
||||
* @param {Function} cb - a callback called once dialed
|
||||
* @returns {Connection} - the connection
|
||||
*
|
||||
* @memberOf Dialer
|
||||
*/
|
||||
dial (ma, options, cb) {
|
||||
return this.dialer.dial(ma, options, cb)
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a listener
|
||||
*
|
||||
* @param {any} options
|
||||
* @param {Function} handler
|
||||
* @return {listener}
|
||||
*/
|
||||
createListener (options, handler) {
|
||||
if (typeof options === 'function') {
|
||||
handler = options
|
||||
options = this.options || {}
|
||||
}
|
||||
|
||||
const listener = createListener(this.swarm, options, handler)
|
||||
listener.on('listen', this._dialSwarmRelays.bind(this))
|
||||
return listener
|
||||
}
|
||||
|
||||
/**
|
||||
* Filter check for all multiaddresses
|
||||
* that this transport can dial on
|
||||
*
|
||||
* @param {any} multiaddrs
|
||||
* @returns {Array<multiaddr>}
|
||||
*
|
||||
* @memberOf Dialer
|
||||
*/
|
||||
filter (multiaddrs) {
|
||||
if (!Array.isArray(multiaddrs)) {
|
||||
multiaddrs = [multiaddrs]
|
||||
}
|
||||
return multiaddrs.filter((ma) => {
|
||||
return mafmt.Circuit.matches(ma)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = Circuit
|
||||
275
src/circuit/circuit/dialer.js
Normal file
275
src/circuit/circuit/dialer.js
Normal file
@@ -0,0 +1,275 @@
|
||||
'use strict'
|
||||
|
||||
const once = require('once')
|
||||
const PeerId = require('peer-id')
|
||||
const waterfall = require('async/waterfall')
|
||||
const setImmediate = require('async/setImmediate')
|
||||
const multiaddr = require('multiaddr')
|
||||
|
||||
const Connection = require('interface-connection').Connection
|
||||
|
||||
const utilsFactory = require('./utils')
|
||||
const StreamHandler = require('./stream-handler')
|
||||
|
||||
const debug = require('debug')
|
||||
const log = debug('libp2p:circuit:dialer')
|
||||
log.err = debug('libp2p:circuit:error:dialer')
|
||||
|
||||
const multicodec = require('../multicodec')
|
||||
const proto = require('../protocol')
|
||||
|
||||
class Dialer {
|
||||
/**
|
||||
* Creates an instance of Dialer.
|
||||
* @param {Swarm} swarm - the swarm
|
||||
* @param {any} options - config options
|
||||
*
|
||||
* @memberOf Dialer
|
||||
*/
|
||||
constructor (swarm, options) {
|
||||
this.swarm = swarm
|
||||
this.relayPeers = new Map()
|
||||
this.relayConns = new Map()
|
||||
this.options = options
|
||||
this.utils = utilsFactory(swarm)
|
||||
}
|
||||
|
||||
/**
|
||||
* Helper that returns a relay connection
|
||||
*
|
||||
* @param {*} relay
|
||||
* @param {*} callback
|
||||
* @returns {Function} - callback
|
||||
*/
|
||||
_dialRelayHelper (relay, callback) {
|
||||
if (this.relayConns.has(relay.id.toB58String())) {
|
||||
return callback(null, this.relayConns.get(relay.id.toB58String()))
|
||||
}
|
||||
|
||||
return this._dialRelay(relay, callback)
|
||||
}
|
||||
|
||||
/**
|
||||
* Dial a peer over a relay
|
||||
*
|
||||
* @param {multiaddr} ma - the multiaddr of the peer to dial
|
||||
* @param {Function} cb - a callback called once dialed
|
||||
* @returns {Connection} - the connection
|
||||
*
|
||||
*/
|
||||
dial (ma, cb) {
|
||||
cb = cb || (() => { })
|
||||
const strMa = ma.toString()
|
||||
if (!strMa.includes('/p2p-circuit')) {
|
||||
log.err('invalid circuit address')
|
||||
return cb(new Error('invalid circuit address'))
|
||||
}
|
||||
|
||||
const addr = strMa.split('p2p-circuit') // extract relay address if any
|
||||
const relay = addr[0] === '/' ? null : multiaddr(addr[0])
|
||||
const peer = multiaddr(addr[1] || addr[0])
|
||||
|
||||
const dstConn = new Connection()
|
||||
setImmediate(
|
||||
this._dialPeer.bind(this),
|
||||
peer,
|
||||
relay,
|
||||
(err, conn) => {
|
||||
if (err) {
|
||||
log.err(err)
|
||||
return cb(err)
|
||||
}
|
||||
|
||||
dstConn.setInnerConn(conn)
|
||||
cb(null, dstConn)
|
||||
})
|
||||
|
||||
return dstConn
|
||||
}
|
||||
|
||||
/**
|
||||
* Does the peer support the HOP protocol
|
||||
*
|
||||
* @param {PeerInfo} peer
|
||||
* @param {Function} callback
|
||||
* @returns {void}
|
||||
*/
|
||||
canHop (peer, callback) {
|
||||
callback = once(callback || (() => { }))
|
||||
|
||||
this._dialRelayHelper(peer, (err, conn) => {
|
||||
if (err) {
|
||||
return callback(err)
|
||||
}
|
||||
|
||||
const sh = new StreamHandler(conn)
|
||||
waterfall([
|
||||
(cb) => sh.write(proto.CircuitRelay.encode({
|
||||
type: proto.CircuitRelay.Type.CAN_HOP
|
||||
}), cb),
|
||||
(cb) => sh.read(cb)
|
||||
], (err, msg) => {
|
||||
if (err) {
|
||||
return callback(err)
|
||||
}
|
||||
const response = proto.CircuitRelay.decode(msg)
|
||||
|
||||
if (response.code !== proto.CircuitRelay.Status.SUCCESS) {
|
||||
const err = new Error(`HOP not supported, skipping - ${this.utils.getB58String(peer)}`)
|
||||
log(err)
|
||||
return callback(err)
|
||||
}
|
||||
|
||||
log('HOP supported adding as relay - %s', this.utils.getB58String(peer))
|
||||
this.relayPeers.set(this.utils.getB58String(peer), peer)
|
||||
sh.close()
|
||||
callback()
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Dial the destination peer over a relay
|
||||
*
|
||||
* @param {multiaddr} dstMa
|
||||
* @param {Connection|PeerInfo} relay
|
||||
* @param {Function} cb
|
||||
* @return {Function|void}
|
||||
* @private
|
||||
*/
|
||||
_dialPeer (dstMa, relay, cb) {
|
||||
if (typeof relay === 'function') {
|
||||
cb = relay
|
||||
relay = null
|
||||
}
|
||||
|
||||
if (!cb) {
|
||||
cb = () => {}
|
||||
}
|
||||
|
||||
dstMa = multiaddr(dstMa)
|
||||
// if no relay provided, dial on all available relays until one succeeds
|
||||
if (!relay) {
|
||||
const relays = Array.from(this.relayPeers.values())
|
||||
const next = (nextRelay) => {
|
||||
if (!nextRelay) {
|
||||
const err = `no relay peers were found or all relays failed to dial`
|
||||
log.err(err)
|
||||
return cb(err)
|
||||
}
|
||||
|
||||
return this._negotiateRelay(
|
||||
nextRelay,
|
||||
dstMa,
|
||||
(err, conn) => {
|
||||
if (err) {
|
||||
log.err(err)
|
||||
return next(relays.shift())
|
||||
}
|
||||
cb(null, conn)
|
||||
})
|
||||
}
|
||||
next(relays.shift())
|
||||
} else {
|
||||
return this._negotiateRelay(
|
||||
relay,
|
||||
dstMa,
|
||||
(err, conn) => {
|
||||
if (err) {
|
||||
log.err('An error has occurred negotiating the relay connection', err)
|
||||
return cb(err)
|
||||
}
|
||||
|
||||
return cb(null, conn)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Negotiate the relay connection
|
||||
*
|
||||
* @param {Multiaddr|PeerInfo|Connection} relay - the Connection or PeerInfo of the relay
|
||||
* @param {multiaddr} dstMa - the multiaddr of the peer to relay the connection for
|
||||
* @param {Function} callback - a callback which gets the negotiated relay connection
|
||||
* @returns {void}
|
||||
* @private
|
||||
*
|
||||
* @memberOf Dialer
|
||||
*/
|
||||
_negotiateRelay (relay, dstMa, callback) {
|
||||
dstMa = multiaddr(dstMa)
|
||||
relay = this.utils.peerInfoFromMa(relay)
|
||||
const srcMas = this.swarm._peerInfo.multiaddrs.toArray()
|
||||
this._dialRelayHelper(relay, (err, conn) => {
|
||||
if (err) {
|
||||
log.err(err)
|
||||
return callback(err)
|
||||
}
|
||||
const sh = new StreamHandler(conn)
|
||||
waterfall([
|
||||
(cb) => {
|
||||
log('negotiating relay for peer %s', dstMa.getPeerId())
|
||||
let dstPeerId
|
||||
try {
|
||||
dstPeerId = PeerId.createFromB58String(dstMa.getPeerId()).id
|
||||
} catch (err) {
|
||||
return cb(err)
|
||||
}
|
||||
sh.write(
|
||||
proto.CircuitRelay.encode({
|
||||
type: proto.CircuitRelay.Type.HOP,
|
||||
srcPeer: {
|
||||
id: this.swarm._peerInfo.id.id,
|
||||
addrs: srcMas.map((addr) => addr.buffer)
|
||||
},
|
||||
dstPeer: {
|
||||
id: dstPeerId,
|
||||
addrs: [dstMa.buffer]
|
||||
}
|
||||
}), cb)
|
||||
},
|
||||
(cb) => sh.read(cb)
|
||||
], (err, msg) => {
|
||||
if (err) {
|
||||
return callback(err)
|
||||
}
|
||||
const message = proto.CircuitRelay.decode(msg)
|
||||
if (message.type !== proto.CircuitRelay.Type.STATUS) {
|
||||
return callback(new Error(`Got invalid message type - ` +
|
||||
`expected ${proto.CircuitRelay.Type.STATUS} got ${message.type}`))
|
||||
}
|
||||
|
||||
if (message.code !== proto.CircuitRelay.Status.SUCCESS) {
|
||||
return callback(new Error(`Got ${message.code} error code trying to dial over relay`))
|
||||
}
|
||||
|
||||
callback(null, new Connection(sh.rest()))
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Dial a relay peer by its PeerInfo
|
||||
*
|
||||
* @param {PeerInfo} peer - the PeerInfo of the relay peer
|
||||
* @param {Function} cb - a callback with the connection to the relay peer
|
||||
* @returns {void}
|
||||
* @private
|
||||
*/
|
||||
_dialRelay (peer, cb) {
|
||||
cb = once(cb || (() => { }))
|
||||
|
||||
this.swarm.dial(
|
||||
peer,
|
||||
multicodec.relay,
|
||||
once((err, conn) => {
|
||||
if (err) {
|
||||
log.err(err)
|
||||
return cb(err)
|
||||
}
|
||||
cb(null, conn)
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = Dialer
|
||||
283
src/circuit/circuit/hop.js
Normal file
283
src/circuit/circuit/hop.js
Normal file
@@ -0,0 +1,283 @@
|
||||
'use strict'
|
||||
|
||||
const pull = require('pull-stream/pull')
|
||||
const debug = require('debug')
|
||||
const PeerInfo = require('peer-info')
|
||||
const PeerId = require('peer-id')
|
||||
const EE = require('events').EventEmitter
|
||||
const once = require('once')
|
||||
const utilsFactory = require('./utils')
|
||||
const StreamHandler = require('./stream-handler')
|
||||
const proto = require('../protocol').CircuitRelay
|
||||
const multiaddr = require('multiaddr')
|
||||
const series = require('async/series')
|
||||
const waterfall = require('async/waterfall')
|
||||
const setImmediate = require('async/setImmediate')
|
||||
|
||||
const multicodec = require('./../multicodec')
|
||||
|
||||
const log = debug('libp2p:circuit:relay')
|
||||
log.err = debug('libp2p:circuit:error:relay')
|
||||
|
||||
class Hop extends EE {
|
||||
/**
|
||||
* Construct a Circuit object
|
||||
*
|
||||
* This class will handle incoming circuit connections and
|
||||
* either start a relay or hand the relayed connection to
|
||||
* the swarm
|
||||
*
|
||||
* @param {Swarm} swarm
|
||||
* @param {Object} options
|
||||
*/
|
||||
constructor (swarm, options) {
|
||||
super()
|
||||
this.swarm = swarm
|
||||
this.peerInfo = this.swarm._peerInfo
|
||||
this.utils = utilsFactory(swarm)
|
||||
this.config = options || { active: false, enabled: false }
|
||||
this.active = this.config.active
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle the relay message
|
||||
*
|
||||
* @param {CircuitRelay} message
|
||||
* @param {StreamHandler} sh
|
||||
* @returns {*}
|
||||
*/
|
||||
handle (message, sh) {
|
||||
if (!this.config.enabled) {
|
||||
this.utils.writeResponse(
|
||||
sh,
|
||||
proto.Status.HOP_CANT_SPEAK_RELAY)
|
||||
return sh.close()
|
||||
}
|
||||
|
||||
// check if message is `CAN_HOP`
|
||||
if (message.type === proto.Type.CAN_HOP) {
|
||||
this.utils.writeResponse(
|
||||
sh,
|
||||
proto.Status.SUCCESS)
|
||||
return sh.close()
|
||||
}
|
||||
|
||||
// This is a relay request - validate and create a circuit
|
||||
let srcPeerId = null
|
||||
let dstPeerId = null
|
||||
try {
|
||||
srcPeerId = PeerId.createFromBytes(message.srcPeer.id).toB58String()
|
||||
dstPeerId = PeerId.createFromBytes(message.dstPeer.id).toB58String()
|
||||
} catch (err) {
|
||||
log.err(err)
|
||||
|
||||
if (!srcPeerId) {
|
||||
this.utils.writeResponse(
|
||||
sh,
|
||||
proto.Status.HOP_SRC_MULTIADDR_INVALID)
|
||||
return sh.close()
|
||||
}
|
||||
|
||||
if (!dstPeerId) {
|
||||
this.utils.writeResponse(
|
||||
sh,
|
||||
proto.Status.HOP_DST_MULTIADDR_INVALID)
|
||||
return sh.close()
|
||||
}
|
||||
}
|
||||
|
||||
if (srcPeerId === dstPeerId) {
|
||||
this.utils.writeResponse(
|
||||
sh,
|
||||
proto.Status.HOP_CANT_RELAY_TO_SELF)
|
||||
return sh.close()
|
||||
}
|
||||
|
||||
if (!message.dstPeer.addrs.length) {
|
||||
// TODO: use encapsulate here
|
||||
const addr = multiaddr(`/p2p-circuit/ipfs/${dstPeerId}`).buffer
|
||||
message.dstPeer.addrs.push(addr)
|
||||
}
|
||||
|
||||
log('trying to establish a circuit: %s <-> %s', srcPeerId, dstPeerId)
|
||||
const noPeer = () => {
|
||||
// log.err(err)
|
||||
this.utils.writeResponse(
|
||||
sh,
|
||||
proto.Status.HOP_NO_CONN_TO_DST)
|
||||
return sh.close()
|
||||
}
|
||||
|
||||
const isConnected = (cb) => {
|
||||
let dstPeer
|
||||
try {
|
||||
dstPeer = this.swarm._peerBook.get(dstPeerId)
|
||||
if (!dstPeer.isConnected() && !this.active) {
|
||||
const err = new Error(`No Connection to peer ${dstPeerId}`)
|
||||
noPeer(err)
|
||||
return cb(err)
|
||||
}
|
||||
} catch (err) {
|
||||
if (!this.active) {
|
||||
noPeer(err)
|
||||
return cb(err)
|
||||
}
|
||||
}
|
||||
cb()
|
||||
}
|
||||
|
||||
series([
|
||||
(cb) => this.utils.validateAddrs(message, sh, proto.Type.HOP, cb),
|
||||
(cb) => isConnected(cb),
|
||||
(cb) => this._circuit(sh, message, cb)
|
||||
], (err) => {
|
||||
if (err) {
|
||||
log.err(err)
|
||||
sh.close()
|
||||
return setImmediate(() => this.emit('circuit:error', err))
|
||||
}
|
||||
setImmediate(() => this.emit('circuit:success'))
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Connect to STOP
|
||||
*
|
||||
* @param {PeerInfo} peer
|
||||
* @param {StreamHandler} srcSh
|
||||
* @param {function} callback
|
||||
* @returns {void}
|
||||
*/
|
||||
_connectToStop (peer, srcSh, callback) {
|
||||
this._dialPeer(peer, (err, dstConn) => {
|
||||
if (err) {
|
||||
this.utils.writeResponse(
|
||||
srcSh,
|
||||
proto.Status.HOP_CANT_DIAL_DST)
|
||||
log.err(err)
|
||||
return callback(err)
|
||||
}
|
||||
|
||||
return this.utils.writeResponse(
|
||||
srcSh,
|
||||
proto.Status.SUCCESS,
|
||||
(err) => {
|
||||
if (err) {
|
||||
log.err(err)
|
||||
return callback(err)
|
||||
}
|
||||
return callback(null, dstConn)
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Negotiate STOP
|
||||
*
|
||||
* @param {StreamHandler} dstSh
|
||||
* @param {StreamHandler} srcSh
|
||||
* @param {CircuitRelay} message
|
||||
* @param {function} callback
|
||||
* @returns {void}
|
||||
*/
|
||||
_negotiateStop (dstSh, srcSh, message, callback) {
|
||||
const stopMsg = Object.assign({}, message, {
|
||||
type: proto.Type.STOP // change the message type
|
||||
})
|
||||
dstSh.write(proto.encode(stopMsg),
|
||||
(err) => {
|
||||
if (err) {
|
||||
this.utils.writeResponse(
|
||||
srcSh,
|
||||
proto.Status.HOP_CANT_OPEN_DST_STREAM)
|
||||
log.err(err)
|
||||
return callback(err)
|
||||
}
|
||||
|
||||
// read response from STOP
|
||||
dstSh.read((err, msg) => {
|
||||
if (err) {
|
||||
log.err(err)
|
||||
return callback(err)
|
||||
}
|
||||
|
||||
const message = proto.decode(msg)
|
||||
if (message.code !== proto.Status.SUCCESS) {
|
||||
return callback(new Error(`Unable to create circuit!`))
|
||||
}
|
||||
|
||||
return callback(null, msg)
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Attempt to make a circuit from A <-> R <-> B where R is this relay
|
||||
*
|
||||
* @param {StreamHandler} srcSh - the source stream handler
|
||||
* @param {CircuitRelay} message - the message with the src and dst entries
|
||||
* @param {Function} callback - callback to signal success or failure
|
||||
* @returns {void}
|
||||
* @private
|
||||
*/
|
||||
_circuit (srcSh, message, callback) {
|
||||
let dstSh = null
|
||||
waterfall([
|
||||
(cb) => this._connectToStop(message.dstPeer, srcSh, cb),
|
||||
(_dstConn, cb) => {
|
||||
dstSh = new StreamHandler(_dstConn)
|
||||
this._negotiateStop(dstSh, srcSh, message, cb)
|
||||
}
|
||||
], (err) => {
|
||||
if (err) {
|
||||
// close/end the source stream if there was an error
|
||||
if (srcSh) {
|
||||
srcSh.close()
|
||||
}
|
||||
|
||||
if (dstSh) {
|
||||
dstSh.close()
|
||||
}
|
||||
return callback(err)
|
||||
}
|
||||
|
||||
const src = srcSh.rest()
|
||||
const dst = dstSh.rest()
|
||||
|
||||
const srcIdStr = PeerId.createFromBytes(message.srcPeer.id).toB58String()
|
||||
const dstIdStr = PeerId.createFromBytes(message.dstPeer.id).toB58String()
|
||||
|
||||
// circuit the src and dst streams
|
||||
pull(
|
||||
src,
|
||||
dst,
|
||||
src
|
||||
)
|
||||
log('circuit %s <-> %s established', srcIdStr, dstIdStr)
|
||||
callback()
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Dial the dest peer and create a circuit
|
||||
*
|
||||
* @param {Multiaddr} dstPeer
|
||||
* @param {Function} callback
|
||||
* @returns {void}
|
||||
* @private
|
||||
*/
|
||||
_dialPeer (dstPeer, callback) {
|
||||
const peerInfo = new PeerInfo(PeerId.createFromBytes(dstPeer.id))
|
||||
dstPeer.addrs.forEach((a) => peerInfo.multiaddrs.add(a))
|
||||
this.swarm.dial(peerInfo, multicodec.relay, once((err, conn) => {
|
||||
if (err) {
|
||||
log.err(err)
|
||||
return callback(err)
|
||||
}
|
||||
|
||||
callback(null, conn)
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = Hop
|
||||
56
src/circuit/circuit/stop.js
Normal file
56
src/circuit/circuit/stop.js
Normal file
@@ -0,0 +1,56 @@
|
||||
'use strict'
|
||||
|
||||
const setImmediate = require('async/setImmediate')
|
||||
|
||||
const EE = require('events').EventEmitter
|
||||
const Connection = require('interface-connection').Connection
|
||||
const utilsFactory = require('./utils')
|
||||
const PeerInfo = require('peer-info')
|
||||
const proto = require('../protocol').CircuitRelay
|
||||
const series = require('async/series')
|
||||
|
||||
const debug = require('debug')
|
||||
|
||||
const log = debug('libp2p:circuit:stop')
|
||||
log.err = debug('libp2p:circuit:error:stop')
|
||||
|
||||
class Stop extends EE {
|
||||
constructor (swarm) {
|
||||
super()
|
||||
this.swarm = swarm
|
||||
this.utils = utilsFactory(swarm)
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle the incoming STOP message
|
||||
*
|
||||
* @param {{}} msg - the parsed protobuf message
|
||||
* @param {StreamHandler} sh - the stream handler wrapped connection
|
||||
* @param {Function} callback - callback
|
||||
* @returns {undefined}
|
||||
*/
|
||||
handle (msg, sh, callback) {
|
||||
callback = callback || (() => {})
|
||||
|
||||
series([
|
||||
(cb) => this.utils.validateAddrs(msg, sh, proto.Type.STOP, cb),
|
||||
(cb) => this.utils.writeResponse(sh, proto.Status.Success, cb)
|
||||
], (err) => {
|
||||
if (err) {
|
||||
// we don't return the error here,
|
||||
// since multistream select don't expect one
|
||||
callback()
|
||||
return log(err)
|
||||
}
|
||||
|
||||
const peerInfo = new PeerInfo(this.utils.peerIdFromId(msg.srcPeer.id))
|
||||
msg.srcPeer.addrs.forEach((addr) => peerInfo.multiaddrs.add(addr))
|
||||
const newConn = new Connection(sh.rest())
|
||||
newConn.setPeerInfo(peerInfo)
|
||||
setImmediate(() => this.emit('connection', newConn))
|
||||
callback(newConn)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = Stop
|
||||
140
src/circuit/circuit/stream-handler.js
Normal file
140
src/circuit/circuit/stream-handler.js
Normal file
@@ -0,0 +1,140 @@
|
||||
'use strict'
|
||||
|
||||
const values = require('pull-stream/sources/values')
|
||||
const collect = require('pull-stream/sinks/collect')
|
||||
const empty = require('pull-stream/sources/empty')
|
||||
const pull = require('pull-stream/pull')
|
||||
const lp = require('pull-length-prefixed')
|
||||
const handshake = require('pull-handshake')
|
||||
|
||||
const debug = require('debug')
|
||||
const log = debug('libp2p:circuit:stream-handler')
|
||||
log.err = debug('libp2p:circuit:error:stream-handler')
|
||||
|
||||
class StreamHandler {
|
||||
/**
|
||||
* Create a stream handler for connection
|
||||
*
|
||||
* @param {Connection} conn - connection to read/write
|
||||
* @param {Function|undefined} cb - handshake callback called on error
|
||||
* @param {Number} timeout - handshake timeout
|
||||
* @param {Number} maxLength - max bytes length of message
|
||||
*/
|
||||
constructor (conn, cb, timeout, maxLength) {
|
||||
this.conn = conn
|
||||
this.stream = null
|
||||
this.shake = null
|
||||
this.timeout = cb || 1000 * 60
|
||||
this.maxLength = maxLength || 4096
|
||||
|
||||
if (typeof cb === 'function') {
|
||||
this.timeout = timeout || 1000 * 60
|
||||
}
|
||||
|
||||
this.stream = handshake({ timeout: this.timeout }, cb)
|
||||
this.shake = this.stream.handshake
|
||||
|
||||
pull(this.stream, conn, this.stream)
|
||||
}
|
||||
|
||||
isValid () {
|
||||
return this.conn && this.shake && this.stream
|
||||
}
|
||||
|
||||
/**
|
||||
* Read and decode message
|
||||
*
|
||||
* @param {Function} cb
|
||||
* @returns {void|Function}
|
||||
*/
|
||||
read (cb) {
|
||||
if (!this.isValid()) {
|
||||
return cb(new Error(`handler is not in a valid state`))
|
||||
}
|
||||
|
||||
lp.decodeFromReader(
|
||||
this.shake,
|
||||
{ maxLength: this.maxLength },
|
||||
(err, msg) => {
|
||||
if (err) {
|
||||
log.err(err)
|
||||
// this.shake.abort(err)
|
||||
return cb(err)
|
||||
}
|
||||
|
||||
return cb(null, msg)
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Encode and write array of buffers
|
||||
*
|
||||
* @param {Buffer[]} msg
|
||||
* @param {Function} [cb]
|
||||
* @returns {Function}
|
||||
*/
|
||||
write (msg, cb) {
|
||||
cb = cb || (() => {})
|
||||
|
||||
if (!this.isValid()) {
|
||||
return cb(new Error(`handler is not in a valid state`))
|
||||
}
|
||||
|
||||
pull(
|
||||
values([msg]),
|
||||
lp.encode(),
|
||||
collect((err, encoded) => {
|
||||
if (err) {
|
||||
log.err(err)
|
||||
this.shake.abort(err)
|
||||
return cb(err)
|
||||
}
|
||||
|
||||
encoded.forEach((e) => this.shake.write(e))
|
||||
cb()
|
||||
})
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the raw Connection
|
||||
*
|
||||
* @returns {null|Connection|*}
|
||||
*/
|
||||
getRawConn () {
|
||||
return this.conn
|
||||
}
|
||||
|
||||
/**
|
||||
* Return the handshake rest stream and invalidate handler
|
||||
*
|
||||
* @return {*|{source, sink}}
|
||||
*/
|
||||
rest () {
|
||||
const rest = this.shake.rest()
|
||||
|
||||
this.conn = null
|
||||
this.stream = null
|
||||
this.shake = null
|
||||
return rest
|
||||
}
|
||||
|
||||
/**
|
||||
* Close the stream
|
||||
*
|
||||
* @returns {undefined}
|
||||
*/
|
||||
close () {
|
||||
if (!this.isValid()) {
|
||||
return
|
||||
}
|
||||
|
||||
// close stream
|
||||
pull(
|
||||
empty(),
|
||||
this.rest()
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = StreamHandler
|
||||
118
src/circuit/circuit/utils.js
Normal file
118
src/circuit/circuit/utils.js
Normal file
@@ -0,0 +1,118 @@
|
||||
'use strict'
|
||||
|
||||
const multiaddr = require('multiaddr')
|
||||
const PeerInfo = require('peer-info')
|
||||
const PeerId = require('peer-id')
|
||||
const proto = require('../protocol')
|
||||
const { getPeerInfo } = require('../../get-peer-info')
|
||||
|
||||
module.exports = function (swarm) {
|
||||
/**
|
||||
* Get b58 string from multiaddr or peerinfo
|
||||
*
|
||||
* @param {Multiaddr|PeerInfo} peer
|
||||
* @return {*}
|
||||
*/
|
||||
function getB58String (peer) {
|
||||
let b58Id = null
|
||||
if (multiaddr.isMultiaddr(peer)) {
|
||||
const relayMa = multiaddr(peer)
|
||||
b58Id = relayMa.getPeerId()
|
||||
} else if (PeerInfo.isPeerInfo(peer)) {
|
||||
b58Id = peer.id.toB58String()
|
||||
}
|
||||
|
||||
return b58Id
|
||||
}
|
||||
|
||||
/**
|
||||
* Helper to make a peer info from a multiaddrs
|
||||
*
|
||||
* @param {Multiaddr|PeerInfo|PeerId} peer
|
||||
* @return {PeerInfo}
|
||||
* @private
|
||||
*/
|
||||
function peerInfoFromMa (peer) {
|
||||
return getPeerInfo(peer, swarm._peerBook)
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks if peer has an existing connection
|
||||
*
|
||||
* @param {String} peerId
|
||||
* @param {Swarm} swarm
|
||||
* @return {Boolean}
|
||||
*/
|
||||
function isPeerConnected (peerId) {
|
||||
return swarm.muxedConns[peerId] || swarm.conns[peerId]
|
||||
}
|
||||
|
||||
/**
|
||||
* Write a response
|
||||
*
|
||||
* @param {StreamHandler} streamHandler
|
||||
* @param {CircuitRelay.Status} status
|
||||
* @param {Function} cb
|
||||
* @returns {*}
|
||||
*/
|
||||
function writeResponse (streamHandler, status, cb) {
|
||||
cb = cb || (() => {})
|
||||
streamHandler.write(proto.CircuitRelay.encode({
|
||||
type: proto.CircuitRelay.Type.STATUS,
|
||||
code: status
|
||||
}))
|
||||
return cb()
|
||||
}
|
||||
|
||||
/**
|
||||
* Validate incomming HOP/STOP message
|
||||
*
|
||||
* @param {CircuitRelay} msg
|
||||
* @param {StreamHandler} streamHandler
|
||||
* @param {CircuitRelay.Type} type
|
||||
* @returns {*}
|
||||
* @param {Function} cb
|
||||
*/
|
||||
function validateAddrs (msg, streamHandler, type, cb) {
|
||||
try {
|
||||
msg.dstPeer.addrs.forEach((addr) => {
|
||||
return multiaddr(addr)
|
||||
})
|
||||
} catch (err) {
|
||||
writeResponse(streamHandler, type === proto.CircuitRelay.Type.HOP
|
||||
? proto.CircuitRelay.Status.HOP_DST_MULTIADDR_INVALID
|
||||
: proto.CircuitRelay.Status.STOP_DST_MULTIADDR_INVALID)
|
||||
return cb(err)
|
||||
}
|
||||
|
||||
try {
|
||||
msg.srcPeer.addrs.forEach((addr) => {
|
||||
return multiaddr(addr)
|
||||
})
|
||||
} catch (err) {
|
||||
writeResponse(streamHandler, type === proto.CircuitRelay.Type.HOP
|
||||
? proto.CircuitRelay.Status.HOP_SRC_MULTIADDR_INVALID
|
||||
: proto.CircuitRelay.Status.STOP_SRC_MULTIADDR_INVALID)
|
||||
return cb(err)
|
||||
}
|
||||
|
||||
return cb(null)
|
||||
}
|
||||
|
||||
function peerIdFromId (id) {
|
||||
if (typeof id === 'string') {
|
||||
return PeerId.createFromB58String(id)
|
||||
}
|
||||
|
||||
return PeerId.createFromBytes(id)
|
||||
}
|
||||
|
||||
return {
|
||||
getB58String,
|
||||
peerInfoFromMa,
|
||||
isPeerConnected,
|
||||
validateAddrs,
|
||||
writeResponse,
|
||||
peerIdFromId
|
||||
}
|
||||
}
|
||||
3
src/circuit/index.js
Normal file
3
src/circuit/index.js
Normal file
@@ -0,0 +1,3 @@
|
||||
'use strict'
|
||||
|
||||
module.exports = require('./circuit')
|
||||
149
src/circuit/listener.js
Normal file
149
src/circuit/listener.js
Normal file
@@ -0,0 +1,149 @@
|
||||
'use strict'
|
||||
|
||||
const setImmediate = require('async/setImmediate')
|
||||
|
||||
const multicodec = require('./multicodec')
|
||||
const EE = require('events').EventEmitter
|
||||
const multiaddr = require('multiaddr')
|
||||
const mafmt = require('mafmt')
|
||||
const Stop = require('./circuit/stop')
|
||||
const Hop = require('./circuit/hop')
|
||||
const proto = require('./protocol')
|
||||
const utilsFactory = require('./circuit/utils')
|
||||
|
||||
const StreamHandler = require('./circuit/stream-handler')
|
||||
|
||||
const debug = require('debug')
|
||||
|
||||
const log = debug('libp2p:circuit:listener')
|
||||
log.err = debug('libp2p:circuit:error:listener')
|
||||
|
||||
module.exports = (swarm, options, connHandler) => {
|
||||
const listener = new EE()
|
||||
const utils = utilsFactory(swarm)
|
||||
|
||||
listener.stopHandler = new Stop(swarm)
|
||||
listener.stopHandler.on('connection', (conn) => listener.emit('connection', conn))
|
||||
listener.hopHandler = new Hop(swarm, options.hop)
|
||||
|
||||
/**
|
||||
* Add swarm handler and listen for incoming connections
|
||||
*
|
||||
* @param {Multiaddr} ma
|
||||
* @param {Function} callback
|
||||
* @return {void}
|
||||
*/
|
||||
listener.listen = (ma, callback) => {
|
||||
callback = callback || (() => {})
|
||||
|
||||
swarm.handle(multicodec.relay, (_, conn) => {
|
||||
const sh = new StreamHandler(conn)
|
||||
|
||||
sh.read((err, msg) => {
|
||||
if (err) {
|
||||
log.err(err)
|
||||
return
|
||||
}
|
||||
|
||||
let request = null
|
||||
try {
|
||||
request = proto.CircuitRelay.decode(msg)
|
||||
} catch (err) {
|
||||
return utils.writeResponse(
|
||||
sh,
|
||||
proto.CircuitRelay.Status.MALFORMED_MESSAGE)
|
||||
}
|
||||
|
||||
switch (request.type) {
|
||||
case proto.CircuitRelay.Type.CAN_HOP:
|
||||
case proto.CircuitRelay.Type.HOP: {
|
||||
return listener.hopHandler.handle(request, sh)
|
||||
}
|
||||
|
||||
case proto.CircuitRelay.Type.STOP: {
|
||||
return listener.stopHandler.handle(request, sh, connHandler)
|
||||
}
|
||||
|
||||
default: {
|
||||
utils.writeResponse(
|
||||
sh,
|
||||
proto.CircuitRelay.Status.INVALID_MSG_TYPE)
|
||||
return sh.close()
|
||||
}
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
setImmediate(() => listener.emit('listen'))
|
||||
callback()
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove swarm listener
|
||||
*
|
||||
* @param {Function} cb
|
||||
* @return {void}
|
||||
*/
|
||||
listener.close = (cb) => {
|
||||
swarm.unhandle(multicodec.relay)
|
||||
setImmediate(() => listener.emit('close'))
|
||||
cb()
|
||||
}
|
||||
|
||||
/**
|
||||
* Get fixed up multiaddrs
|
||||
*
|
||||
* NOTE: This method will grab the peers multiaddrs and expand them such that:
|
||||
*
|
||||
* a) If it's an existing /p2p-circuit address for a specific relay i.e.
|
||||
* `/ip4/0.0.0.0/tcp/0/ipfs/QmRelay/p2p-circuit` this method will expand the
|
||||
* address to `/ip4/0.0.0.0/tcp/0/ipfs/QmRelay/p2p-circuit/ipfs/QmPeer` where
|
||||
* `QmPeer` is this peers id
|
||||
* b) If it's not a /p2p-circuit address, it will encapsulate the address as a /p2p-circuit
|
||||
* addr, such when dialing over a relay with this address, it will create the circuit using
|
||||
* the encapsulated transport address. This is useful when for example, a peer should only
|
||||
* be dialed over TCP rather than any other transport
|
||||
*
|
||||
* @param {Function} callback
|
||||
* @return {void}
|
||||
*/
|
||||
listener.getAddrs = (callback) => {
|
||||
let addrs = swarm._peerInfo.multiaddrs.toArray()
|
||||
|
||||
// get all the explicit relay addrs excluding self
|
||||
const p2pAddrs = addrs.filter((addr) => {
|
||||
return mafmt.Circuit.matches(addr) &&
|
||||
!addr.toString().includes(swarm._peerInfo.id.toB58String())
|
||||
})
|
||||
|
||||
// use the explicit relays instead of any relay
|
||||
if (p2pAddrs.length) {
|
||||
addrs = p2pAddrs
|
||||
}
|
||||
|
||||
const listenAddrs = []
|
||||
addrs.forEach((addr) => {
|
||||
const peerMa = `/p2p-circuit/ipfs/${swarm._peerInfo.id.toB58String()}`
|
||||
if (addr.toString() === peerMa) {
|
||||
listenAddrs.push(multiaddr(peerMa))
|
||||
return
|
||||
}
|
||||
|
||||
if (!mafmt.Circuit.matches(addr)) {
|
||||
if (addr.getPeerId()) {
|
||||
// by default we're reachable over any relay
|
||||
listenAddrs.push(multiaddr(`/p2p-circuit`).encapsulate(addr))
|
||||
} else {
|
||||
const ma = `${addr}/ipfs/${swarm._peerInfo.id.toB58String()}`
|
||||
listenAddrs.push(multiaddr(`/p2p-circuit`).encapsulate(ma))
|
||||
}
|
||||
} else {
|
||||
listenAddrs.push(addr.encapsulate(`/ipfs/${swarm._peerInfo.id.toB58String()}`))
|
||||
}
|
||||
})
|
||||
|
||||
callback(null, listenAddrs)
|
||||
}
|
||||
|
||||
return listener
|
||||
}
|
||||
5
src/circuit/multicodec.js
Normal file
5
src/circuit/multicodec.js
Normal file
@@ -0,0 +1,5 @@
|
||||
'use strict'
|
||||
|
||||
module.exports = {
|
||||
relay: '/libp2p/circuit/relay/0.1.0'
|
||||
}
|
||||
5
src/circuit/package.json
Normal file
5
src/circuit/package.json
Normal file
@@ -0,0 +1,5 @@
|
||||
{
|
||||
"name": "libp2p-circuit",
|
||||
"description": "JavaScript implementation of circuit/switch relaying",
|
||||
"main": "./index.js"
|
||||
}
|
||||
44
src/circuit/protocol/index.js
Normal file
44
src/circuit/protocol/index.js
Normal file
@@ -0,0 +1,44 @@
|
||||
'use strict'
|
||||
const protobuf = require('protons')
|
||||
module.exports = protobuf(`
|
||||
message CircuitRelay {
|
||||
|
||||
enum Status {
|
||||
SUCCESS = 100;
|
||||
HOP_SRC_ADDR_TOO_LONG = 220;
|
||||
HOP_DST_ADDR_TOO_LONG = 221;
|
||||
HOP_SRC_MULTIADDR_INVALID = 250;
|
||||
HOP_DST_MULTIADDR_INVALID = 251;
|
||||
HOP_NO_CONN_TO_DST = 260;
|
||||
HOP_CANT_DIAL_DST = 261;
|
||||
HOP_CANT_OPEN_DST_STREAM = 262;
|
||||
HOP_CANT_SPEAK_RELAY = 270;
|
||||
HOP_CANT_RELAY_TO_SELF = 280;
|
||||
STOP_SRC_ADDR_TOO_LONG = 320;
|
||||
STOP_DST_ADDR_TOO_LONG = 321;
|
||||
STOP_SRC_MULTIADDR_INVALID = 350;
|
||||
STOP_DST_MULTIADDR_INVALID = 351;
|
||||
STOP_RELAY_REFUSED = 390;
|
||||
MALFORMED_MESSAGE = 400;
|
||||
}
|
||||
|
||||
enum Type { // RPC identifier, either HOP, STOP or STATUS
|
||||
HOP = 1;
|
||||
STOP = 2;
|
||||
STATUS = 3;
|
||||
CAN_HOP = 4;
|
||||
}
|
||||
|
||||
message Peer {
|
||||
required bytes id = 1; // peer id
|
||||
repeated bytes addrs = 2; // peer's known addresses
|
||||
}
|
||||
|
||||
optional Type type = 1; // Type of the message
|
||||
|
||||
optional Peer srcPeer = 2; // srcPeer and dstPeer are used when Type is HOP or STATUS
|
||||
optional Peer dstPeer = 3;
|
||||
|
||||
optional Status code = 4; // Status code, used when Type is STATUS
|
||||
}
|
||||
`)
|
||||
Reference in New Issue
Block a user