const { EventEmitter } = require('events')
const safetyCatch = require('safety-catch')
const NoiseSecretStream = require('@hyperswarm/secret-stream')
const b4a = require('b4a')
const relay = require('blind-relay')
const NoiseWrap = require('./noise-wrap')
const Announcer = require('./announcer')
const { FIREWALL, ERROR } = require('./constants')
const { unslabbedHash } = require('./crypto')
const SecurePayload = require('./secure-payload')
const Holepuncher = require('./holepuncher')
const {
clearRelayTimeout,
confirmDirectUpgrade,
destroyRelayConnection
} = require('./relay-connection')
const { isPrivate } = require('bogon')
const { ALREADY_LISTENING, NODE_DESTROYED, KEYPAIR_ALREADY_USED } = require('./errors')
const HANDSHAKE_CLEAR_WAIT = 10000
const HANDSHAKE_INITIAL_TIMEOUT = 10000
module.exports = class Server extends EventEmitter {
constructor(dht, opts = {}) {
super()
this.dht = dht
this.target = null
this.closed = false
this.firewall = opts.firewall || (() => false)
this.holepunch = opts.holepunch || (() => true)
this.relayThrough = opts.relayThrough || null
this.relayKeepAlive = opts.relayKeepAlive || 5000
this.pool = opts.pool || null
this.createHandshake = opts.createHandshake || defaultCreateHandshake
this.createSecretStream = opts.createSecretStream || defaultCreateSecretStream
this.suspended = false
this.handshakeClearWait = opts.handshakeClearWait || HANDSHAKE_CLEAR_WAIT
this._shareLocalAddress = opts.shareLocalAddress !== false
this._reusableSocket = !!opts.reusableSocket
this._neverPunch = opts.holepunch === false // useful for fully disabling punching
this._keyPair = null
this._announcer = null
this._connects = new Map()
this._holepunches = []
this._listening = null
this._closing = null
}
get listening() {
return this._listening !== null
}
get publicKey() {
return this._keyPair && this._keyPair.publicKey
}
get relayAddresses() {
return this._announcer ? this._announcer.relayAddresses : []
}
onconnection(encryptedSocket) {
this.emit('connection', encryptedSocket)
}
async suspend({ log = noop } = {}) {
log('Suspending hyperdht server')
if (this._listening !== null) await this._listening
log('Suspending hyperdht server (post listening)')
this.suspended = true
this._clearAll()
return this._announcer ? this._announcer.suspend({ log }) : Promise.resolve()
}
async resume() {
if (this._listening !== null) await this._listening
this.suspended = false
return this._announcer ? this._announcer.resume() : Promise.resolve()
}
address() {
if (!this._keyPair) return null
return {
publicKey: this._keyPair.publicKey,
host: this.dht.host,
port: this.dht.port
}
}
close() {
if (this._closing) return this._closing
this._closing = this._close()
return this._closing
}
_gc() {
this.dht.listening.delete(this)
if (this.target) this.dht._router.delete(this.target)
}
async _stopListening() {
try {
if (this._announcer) await this._announcer.stop()
} catch {
// ignore
}
this._announcer = null
this._listening = null
this._keyPair = null
}
async _close() {
if (this._listening === null) {
this.closed = true
this.emit('close')
return
}
try {
await this._listening
} catch (err) {
safetyCatch(err)
}
this._gc()
this._clearAll()
await this._stopListening()
this.closed = true
this.emit('close')
}
_clearAll() {
while (this._holepunches.length > 0) {
const h = this._holepunches.pop()
if (h && h.puncher) h.puncher.destroy()
if (h && h.clearing) clearTimeout(h.clearing)
if (h && h.prepunching) clearTimeout(h.prepunching)
if (h && h.rawStream) h.rawStream.destroy()
}
this._connects.clear()
}
async listen(keyPair = this.dht.defaultKeyPair, opts = {}) {
if (this._listening !== null) throw ALREADY_LISTENING()
if (this.dht.destroyed) throw NODE_DESTROYED()
this._listening = this._listen(keyPair, opts)
await this._listening
return this
}
async _listen(keyPair, opts) {
// From now on, the DHT object which created me is responsible for closing me
this.dht.listening.add(this)
try {
await this.dht.bind()
if (this._closing) return
for (const s of this.dht.listening) {
if (s._keyPair && b4a.equals(s._keyPair.publicKey, keyPair.publicKey)) {
throw KEYPAIR_ALREADY_USED()
}
}
this.target = unslabbedHash(keyPair.publicKey)
this._keyPair = keyPair
this._announcer = new Announcer(this.dht, keyPair, this.target, opts)
this.dht._router.set(this.target, {
relay: null,
record: this._announcer.record,
onpeerhandshake: this._onpeerhandshake.bind(this),
onpeerholepunch: this._onpeerholepunch.bind(this)
})
// warm it up for now
this._localAddresses().catch(safetyCatch)
await this._announcer.start()
} catch (err) {
await this._stopListening()
this._gc()
throw err
}
if (this._closing) return
if (this.suspended) await this._announcer.suspend()
if (this._closing) return
if (this.dht.destroyed) throw NODE_DESTROYED()
if (this.pool) this.pool._attachServer(this)
this.emit('listening')
}
refresh() {
if (this._announcer && !this.suspended) this._announcer.refresh()
}
notifyOnline() {
if (this._announcer) this._announcer.online.notify()
}
_localAddresses() {
return this.dht.validateLocalAddresses(Holepuncher.localAddresses(this.dht.io.serverSocket))
}
async _addHandshake(k, noise, clientAddress, { to: serverAddress, socket }, direct) {
let id = this._holepunches.indexOf(null)
if (id === -1) id = this._holepunches.push(null) - 1
const hs = {
round: 0,
reply: null,
puncher: null,
payload: null,
rawStream: null,
encryptedSocket: null,
prepunching: null,
firewalled: true,
clearing: null,
replyFromPuncher: false,
onsocket: null,
aborted: false,
// Relay state
relayTimeout: null,
relayToken: null,
relaySocket: null,
relayClient: null,
relayPaired: false,
validUpgrade: true
}
this._holepunches[id] = hs
const handshake = this.createHandshake(this._keyPair, null)
let remotePayload
try {
remotePayload = await handshake.recv(noise)
} catch (err) {
safetyCatch(err)
this._clearLater(hs, id, k)
return null
}
if (this._closing || this.suspended) {
this._clearLater(hs, id, k)
return null
}
try {
hs.firewalled = await this.firewall(handshake.remotePublicKey, remotePayload, clientAddress)
} catch (err) {
safetyCatch(err)
}
if (this._closing || this.suspended || hs.firewalled) {
this._clearLater(hs, id, k)
return null
}
const error =
remotePayload.version === 1
? remotePayload.udx
? ERROR.NONE
: ERROR.ABORTED
: ERROR.VERSION_MISMATCH
const addresses = []
const ourRemoteAddr = this.dht.remoteAddress()
const ourLocalAddrs = this._shareLocalAddress ? await this._localAddresses() : null
if (this._closing || this.suspended) {
this._clearLater(hs, id, k)
return null
}
if (ourRemoteAddr) addresses.push(ourRemoteAddr)
if (ourLocalAddrs) addresses.push(...ourLocalAddrs)
if (error === ERROR.NONE) {
hs.rawStream = this.dht.createRawStream({
framed: true,
firewall(socket, port, host) {
if (!(port > 0 && port < 65536)) return true
// Check if the traffic originated from the socket on which we're expecting relay traffic. If so,
// we haven't hole punched yet and the other side is just sending us traffic through the relay.
if (hs.relaySocket && isRelay(hs.relaySocket, socket, port, host)) {
hs.validUpgrade = false
return false
}
hs.onsocket(socket, port, host)
hs.validUpgrade = true
return false
}
})
hs.rawStream.on('error', autoDestroy)
// Handles the case where onsocket is never called, but the stream got setup
// This can happen on a relayed connection which never connects directly
// (onsocket is called there only when the direct connection is established)
const onrawstreamclose = () => {
if (this._closing) return
if (hs.puncher) {
hs.puncher.onabort = noop
hs.puncher.destroy()
}
this._clearLater(hs, id, k)
}
hs.rawStream.on('close', onrawstreamclose)
hs.onsocket = (socket, port, host) => {
if (hs.rawStream === null) return // Already hole punched
this._clearLater(hs, id, k)
if (hs.prepunching) {
clearTimeout(hs.prepunching)
hs.prepunching = null
}
if (this._reusableSocket && remotePayload.udx.reusableSocket) {
this.dht._socketPool.routes.add(handshake.remotePublicKey, hs.rawStream)
}
hs.rawStream.removeListener('error', autoDestroy)
hs.rawStream.removeListener('close', onrawstreamclose)
if (hs.rawStream.connected) {
const rawStream = hs.rawStream
const remoteChanging = rawStream.changeRemote(socket, remotePayload.udx.id, port, host)
confirmDirectUpgrade(hs, rawStream, remoteChanging)
} else {
hs.rawStream.connect(socket, remotePayload.udx.id, port, host)
hs.encryptedSocket = this.createSecretStream(false, hs.rawStream, {
handshake: h,
keepAlive: this.dht.connectionKeepAlive
})
this.onconnection(hs.encryptedSocket)
}
if (hs.puncher) {
hs.puncher.onabort = noop
hs.puncher.destroy()
}
hs.rawStream = null
}
function autoDestroy() {
if (hs.puncher) hs.puncher.destroy()
}
}
const relayAddresses = this.relayAddresses
const relayThrough = selectRelay(this.relayThrough)
if (relayThrough) hs.relayToken = relay.token()
try {
hs.reply = await handshake.send({
error,
firewall: ourRemoteAddr ? FIREWALL.OPEN : FIREWALL.UNKNOWN,
holepunch: ourRemoteAddr ? null : { id, relays: this._announcer.relays },
addresses4: addresses,
addresses6: null,
udx: {
reusableSocket: this._reusableSocket,
id: hs.rawStream ? hs.rawStream.id : 0,
seq: 0
},
secretStream: {},
relayThrough: relayThrough ? { publicKey: relayThrough, token: hs.relayToken } : null,
relayAddresses: relayAddresses.length ? relayAddresses : null
})
} catch (err) {
safetyCatch(err)
if (hs.rawStream) hs.rawStream.destroy()
this._clearLater(hs, id, k)
return null
}
if (this._closing || this.suspended) {
if (hs.rawStream) hs.rawStream.destroy()
this._clearLater(hs, id, k)
return null
}
const h = handshake.final()
if (error !== ERROR.NONE) {
if (hs.rawStream) hs.rawStream.destroy()
this._clearLater(hs, id, k)
return hs
}
if (remotePayload.firewall === FIREWALL.OPEN || direct) {
const sock = direct ? socket : this.dht.socket
this.dht.stats.punches.open++
hs.onsocket(sock, clientAddress.port, clientAddress.host)
return hs
}
if (relayThrough || remotePayload.relayThrough) {
this._relayConnection(hs, relayThrough, remotePayload, h)
}
const onabort = () => {
hs.aborted = true
if (hs.prepunching) clearTimeout(hs.prepunching)
hs.prepunching = null
if (hs.puncher) {
hs.puncher.onabort = noop
hs.puncher.destroy()
}
if (hs.rawStream.destroyed) {
this._clearLater(hs, id, k)
return
}
hs.rawStream.on('close', () => this._clearLater(hs, id, k))
if (hs.relayToken === null) hs.rawStream.destroy()
}
if (!direct && clientAddress.host === serverAddress.host) {
const clientAddresses = remotePayload.addresses4.filter((addr) => isPrivate(addr.host))
if (clientAddresses.length > 0 && this._shareLocalAddress) {
const myAddresses = await this._localAddresses()
const addr = Holepuncher.matchAddress(myAddresses, clientAddresses)
if (addr) {
hs.payload = new SecurePayload(h.holepunchSecret)
setupHolepuncher(this.dht, hs, remotePayload.firewall, onabort)
hs.prepunching = setTimeout(onabort, HANDSHAKE_INITIAL_TIMEOUT)
return hs
}
}
}
if (this._closing || this.suspended) {
this._clearLater(hs, id, k)
return null
}
if (ourRemoteAddr || this._neverPunch) {
hs.prepunching = setTimeout(onabort, HANDSHAKE_INITIAL_TIMEOUT)
return hs
}
hs.payload = new SecurePayload(h.holepunchSecret)
setupHolepuncher(this.dht, hs, remotePayload.firewall, onabort)
hs.replyFromPuncher = true
hs.prepunching = setTimeout(hs.puncher.destroy.bind(hs.puncher), HANDSHAKE_INITIAL_TIMEOUT)
return hs
}
_clearLater(hs, id, k) {
if (hs.clearing) return
hs.clearing = setTimeout(() => this._clear(hs, id, k), this.handshakeClearWait)
hs.clearing.unref()
}
_clear(hs, id, k) {
if (id >= this._holepunches.length || this._holepunches[id] !== hs) return
if (hs.clearing) clearTimeout(hs.clearing)
this._holepunches[id] = null
while (
this._holepunches.length > 0 &&
this._holepunches[this._holepunches.length - 1] === null
) {
this._holepunches.pop()
}
this._connects.delete(k)
}
async _onpeerhandshake({ noise, peerAddress }, req) {
const k = b4a.toString(noise, 'hex')
// The next couple of statements MUST run within the same tick to prevent
// a malicious peer from flooding us with handshakes.
let p = this._connects.get(k)
if (!p) {
p = this._addHandshake(k, noise, peerAddress || req.from, req, !peerAddress)
this._connects.set(k, p)
}
const h = await p
if (!h) return null
if (this._closing !== null || this.suspended) return null
// Routed handshake replies advertise the socket they are sent from as the
// server address. Same-LAN prepunch also creates a puncher, but it must
// keep advertising the normal server socket so the LAN shortcut can open.
return { socket: h.replyFromPuncher ? h.puncher.socket : null, noise: h.reply }
}
async _onpeerholepunch({ id, peerAddress, payload }, req) {
const h = id < this._holepunches.length ? this._holepunches[id] : null
if (!h) return null
if (!peerAddress || this._closing !== null || this.suspended) return null
if (!h.payload) return this._abort(h) // not opened
const remotePayload = h.payload.decrypt(payload)
if (!remotePayload) return null
const p = h.puncher
if (!p || !p.socket) return this._abort(h) // not opened
const isServerRelay = this._announcer.isRelay(req.from)
const { error, firewall, round, punching, addresses, remoteAddress, remoteToken } =
remotePayload
if (error !== ERROR.NONE) {
// We actually do not need to set the round here, but just do it for consistency.
if (round >= h.round) h.round = round
return this._abort(h)
}
const token = h.payload.token(peerAddress)
const echoed = isServerRelay && !!remoteToken && b4a.equals(token, remoteToken)
// Update our heuristics here
if (req.socket === p.socket) {
p.nat.add(req.to, req.from)
}
if (round >= h.round) {
h.round = round
p.updateRemote({ punching, firewall, addresses, verified: echoed ? peerAddress.host : null })
}
// Wait for the analyzer to reach a conclusion...
let stable = await p.analyze(false)
if (p.destroyed) return null
if (!p.remoteHolepunching && !stable) {
stable = await p.analyze(true)
if (p.destroyed) return null
if (!stable) return this._abort(h)
}
// Fast mode! If we are consistent and the remote has opened a session to us (remoteAddress)
// then fire a quick punch back. Note the await here just waits for the udp socket to flush.
if (
isConsistent(p.nat.firewall) &&
remoteAddress &&
hasSameAddr(p.nat.addresses, remoteAddress)
) {
await p.ping(peerAddress)
if (p.destroyed) return null
}
// Remote said they are punching (or willing to), so we will punch as well.
// Note that this returns when the punching has STARTED, so no guarantee
// we will have a connection after this promise etc.
if (p.remoteHolepunching) {
// TODO: still continue here if a local connection might work, but then do not holepunch...
if (!this.holepunch(p.remoteFirewall, p.nat.firewall, p.remoteAddresses, p.nat.addresses)) {
return p.destroyed ? null : this._abort(h)
}
if (h.prepunching) {
clearTimeout(h.prepunching)
h.prepunching = null
}
if (p.remoteFirewall >= FIREWALL.RANDOM || p.nat.firewall >= FIREWALL.RANDOM) {
if (
this.dht._randomPunches >= this.dht._randomPunchLimit ||
Date.now() - this.dht._lastRandomPunch < this.dht._randomPunchInterval
) {
this.dht.stats.punches.tryLater++
if (!h.relayToken) return this._abort(h, ERROR.TRY_LATER)
if (!h.tryLaterCounted) {
h.tryLaterCounted = true
this.dht.stats.punches.tryLaterRelayedHandshakesAtLeastOnce++
}
return {
socket: p.socket,
payload: h.payload.encrypt({
error: ERROR.TRY_LATER,
firewall: p.nat.firewall,
round: h.round,
connected: p.connected,
punching: p.punching,
addresses: p.nat.addresses,
remoteAddress: null,
token: isServerRelay ? token : null,
remoteToken: remotePayload.token
})
}
}
}
const punching = await p.punch()
if (p.destroyed) return null
if (!punching) return this._abort(h)
}
// Freeze that analysis as soon as we have a result we are giving to the other peer
if (p.nat.firewall !== FIREWALL.UNKNOWN) {
p.nat.freeze()
}
return {
socket: p.socket,
payload: h.payload.encrypt({
error: ERROR.NONE,
firewall: p.nat.firewall,
round: h.round,
connected: p.connected,
punching: p.punching,
addresses: p.nat.addresses,
remoteAddress: null,
token: isServerRelay ? token : null,
remoteToken: remotePayload.token
})
}
}
_abort(h, error = ERROR.ABORTED) {
if (!h.payload) {
if (h.puncher) h.puncher.destroy()
return null
}
const payload = h.payload.encrypt({
error,
firewall: FIREWALL.UNKNOWN,
round: h.round,
connected: false,
punching: false,
addresses: null,
remoteAddress: null,
token: null,
remoteToken: null
})
if (h.puncher) h.puncher.destroy()
return { socket: this.dht.socket, payload }
}
_relayConnection(hs, relayThrough, remotePayload, h) {
this.dht.stats.relaying.attempts++
let isInitiator
let publicKey
let token
if (relayThrough) {
isInitiator = true
publicKey = relayThrough
token = hs.relayToken
} else {
isInitiator = false
publicKey = remotePayload.relayThrough.publicKey
token = remotePayload.relayThrough.token
}
hs.relayToken = token
hs.relaySocket = this.dht.connect(publicKey)
hs.relaySocket.setKeepAlive(this.relayKeepAlive)
hs.relayClient = relay.Client.from(hs.relaySocket, { id: hs.relaySocket.publicKey })
hs.relayTimeout = setTimeout(onabort, 15000)
hs.relayClient
.pair(isInitiator, token, hs.rawStream)
.on('error', onabort)
.on('data', (remoteId) => {
if (hs.relayTimeout) clearRelayTimeout(hs)
if (hs.rawStream === null) {
onabort(null)
return
}
hs.relayPaired = true
this.dht.stats.relaying.successes++
if (hs.prepunching) clearTimeout(hs.prepunching)
hs.prepunching = null
const { remotePort, remoteHost, socket } = hs.relaySocket.rawStream
hs.rawStream
.on('close', () => destroyRelayConnection(hs))
.connect(socket, remoteId, remotePort, remoteHost)
hs.encryptedSocket = this.createSecretStream(false, hs.rawStream, {
handshake: h,
keepAlive: this.dht.connectionKeepAlive
})
this.onconnection(hs.encryptedSocket)
})
const dht = this.dht
function onabort() {
if (!hs.relayPaired) dht.stats.relaying.aborts++
destroyRelayConnection(hs)
if (hs.aborted && hs.rawStream) hs.rawStream.destroy()
}
}
}
function isConsistent(fw) {
return fw === FIREWALL.OPEN || fw === FIREWALL.CONSISTENT
}
function hasSameAddr(addrs, other) {
if (addrs === null) return false
for (const addr of addrs) {
if (addr.port === other.port && addr.host === other.host) return true
}
return false
}
function defaultCreateHandshake(keyPair, remotePublicKey) {
return new NoiseWrap(keyPair, remotePublicKey)
}
function defaultCreateSecretStream(isInitiator, rawStream, opts) {
return new NoiseSecretStream(isInitiator, rawStream, opts)
}
function setupHolepuncher(dht, hs, remoteFirewall, onabort) {
if (hs.puncher) return
hs.puncher = new Holepuncher(dht, dht.session(), false, remoteFirewall)
hs.puncher.onconnect = hs.onsocket
hs.puncher.onabort = onabort
}
function isRelay(relaySocket, socket, port, host) {
const stream = relaySocket.rawStream
if (!stream) return false
if (stream.socket !== socket) return false
return port === stream.remotePort && host === stream.remoteHost
}
function selectRelay(relayThrough) {
if (typeof relayThrough === 'function') relayThrough = relayThrough()
if (relayThrough === null) return null
if (Array.isArray(relayThrough)) {
return relayThrough[Math.floor(Math.random() * relayThrough.length)]
}
return relayThrough
}
function noop() {}