mirall-cli / node_modules / blind-relay / index.js

const EventEmitter = require('events')
const Protomux = require('protomux')
const { Readable } = require('streamx')
const sodium = require('sodium-universal')
const b4a = require('b4a')
const c = require('compact-encoding')
const bitfield = require('compact-encoding-bitfield')
const bits = require('bits-to-bytes')
const errors = require('./lib/errors')

exports.Server = class BlindRelayServer extends EventEmitter {
  constructor(opts = {}) {
    super()

    const { createStream } = opts

    this._createStream = createStream
    this._pairing = new Map()
    this._activePairingRefs = new Map()
    this._sessions = new Set()
    this.stats = createStats()
  }

  get sessions() {
    return this._sessions[Symbol.iterator]()
  }

  accept(stream, opts) {
    const session = new BlindRelaySession(this, stream, opts)

    this._sessions.add(session)
    this.stats.sessions.accepted++

    return session
  }

  async close() {
    const ending = []

    for (const session of this._sessions) {
      ending.push(session.end())
    }

    await Promise.all(ending)

    this._pairing.clear()
  }

  _trackActiveLink(keyString) {
    const refs = this._activePairingRefs.get(keyString) || 0

    if (refs === 0) this.stats.pairings.active++

    this._activePairingRefs.set(keyString, refs + 1)
  }

  _untrackActiveLink(keyString) {
    const refs = this._activePairingRefs.get(keyString)

    if (refs === undefined) return

    if (refs === 1) {
      this._activePairingRefs.delete(keyString)
      this.stats.pairings.active--
    } else {
      this._activePairingRefs.set(keyString, refs - 1)
    }
  }
}

class BlindRelaySession extends EventEmitter {
  constructor(server, stream, opts = {}) {
    super()

    const { id, handshake, handshakeEncoding } = opts

    this._server = server
    this._mux = Protomux.from(stream)

    this._channel = this._mux.createChannel({
      protocol: 'blind-relay',
      id,
      handshake: handshake ? handshakeEncoding || c.raw : null,
      onopen: this._onopen.bind(this),
      onclose: this._onclose.bind(this),
      ondestroy: this._ondestroy.bind(this)
    })

    this._pair = this._channel.addMessage({
      encoding: m.pair,
      onmessage: this._onpair.bind(this)
    })

    this._unpair = this._channel.addMessage({
      encoding: m.unpair,
      onmessage: this._onunpair.bind(this)
    })

    this._ending = null
    this._destroyed = false
    this._error = null
    this._pairing = new Set()
    this._links = new Map()

    this._onerror = (err) => {
      this._server.stats.streams.errors++
      this.emit('error', err)
    }

    this._channel.open(handshake)
  }

  get closed() {
    return this._channel.closed
  }

  get mux() {
    return this._mux
  }

  get stream() {
    return this._mux.stream
  }

  _onopen() {
    this._server.stats.sessions.opened++
    this.emit('open')
  }

  _onclose() {
    this._ending = Promise.resolve()

    const err = this._error || errors.CHANNEL_CLOSED()
    this._server.stats.pairings.cancelled += this._pairing.size

    for (const token of this._pairing) {
      if (this._server._pairing.delete(token.toString('hex'))) {
        this._server.stats.pairings.pending--
      }
    }

    for (const link of this._links.values()) {
      link.destroy(err)
    }

    this._pairing.clear()
    this._links.clear()

    this._server._sessions.delete(this)
    this._server.stats.sessions.closed++

    this.emit('close')
  }

  _ondestroy() {
    this._destroyed = true
    this.emit('destroy')
  }

  _onpair({ isInitiator, token, id: remoteId }) {
    const keyString = token.toString('hex')

    let pair = this._server._pairing.get(keyString)

    if (pair === undefined) {
      pair = new BlindRelayPair(token)
      this._server._pairing.set(keyString, pair)
      this._server.stats.pairings.pending++
    } else if (pair.links[+isInitiator]) return

    this._pairing.add(keyString)
    this._server.stats.pairings.requested++

    pair.links[+isInitiator] = new BlindRelayLink(this, isInitiator, remoteId)

    if (!pair.paired) return

    this._server._pairing.delete(keyString)
    this._server.stats.pairings.pending--
    this._server.stats.pairings.matched++

    // 1st pass: Create the raw streams needed for each end of the link.
    for (const link of pair.links) {
      link.createStream()
    }

    // 2nd pass: Connect the raw streams and set up handlers.
    for (const link of pair.links) {
      link.remote = pair.remote(link.isInitiator)
      link.activate(keyString)
    }

    // 3rd pass: Let either end of the link know the streams were set up.
    for (const { isInitiator, session, remoteId, stream } of pair.links) {
      session._pair.send({
        isInitiator,
        token,
        id: stream.id,
        seq: 0
      })

      session._endMaybe()

      session.emit('pair', isInitiator, token, stream, remoteId)
    }
  }

  _onunpair({ token }) {
    const keyString = token.toString('hex')

    const pair = this._server._pairing.get(keyString)

    if (pair) {
      for (const link of pair.links) {
        if (link) link.session._pairing.delete(keyString)
      }

      this._server._pairing.delete(keyString)
      this._server.stats.pairings.pending--
      this._server.stats.pairings.cancelled++
      return
    }

    const link = this._links.get(keyString)

    if (link) {
      link.destroyPair(errors.PAIRING_CANCELLED())
      this._server.stats.pairings.cancelled++
    }
  }

  cork() {
    this._channel.cork()
  }

  uncork() {
    this._channel.uncork()
  }

  async end() {
    if (this._ending) return this._ending

    this._ending = EventEmitter.once(this, 'close')
    this._endMaybe()

    return this._ending
  }

  _endMaybe() {
    if (this._ending && this._pairing.size === 0) {
      this._channel.close()
    }
  }

  destroy(err) {
    if (this._destroyed) return
    this._destroyed = true

    this._error = err || errors.CHANNEL_DESTROYED()
    this._channel.close()
  }
}

class BlindRelayPair {
  constructor(token) {
    this.token = token
    this.links = [null, null]
  }

  get paired() {
    return this.links[0] !== null && this.links[1] !== null
  }

  remote(isInitiator) {
    return this.links[isInitiator ? 0 : 1]
  }
}

class BlindRelayLink {
  constructor(session, isInitiator, remoteId) {
    this.session = session
    this.isInitiator = isInitiator
    this.remoteId = remoteId
    this.stream = null
    this.remote = null
    this.keyString = null
    this._onclose = null
  }

  createStream() {
    if (this.stream) return

    this.stream = this.session._server._createStream({
      firewall: this._onfirewall.bind(this)
    })
  }

  activate(keyString) {
    const stream = this.stream

    this.keyString = keyString
    this._onclose = () => {
      if (this.stream !== stream) return

      this.stream = null
      unlink(this)
      this.session._links.delete(keyString)
      this.session._server._untrackActiveLink(keyString)
      this.session._server.stats.streams.closed++
    }

    stream.on('error', this.session._onerror).on('close', this._onclose).relayTo(this.remote.stream)

    this.session._pairing.delete(keyString)
    this.session._links.set(keyString, this)
    this.session._server._trackActiveLink(keyString)
    this.session._server.stats.streams.opened++
  }

  destroy(err) {
    const stream = this.stream

    if (!stream) return

    this.stream = null

    stream.off('close', this._onclose)
    this._onclose = null

    stream.off('error', this.session._onerror)
    unlink(this)
    this.session._links.delete(this.keyString)
    this.session._server._untrackActiveLink(this.keyString)
    this.session._server.stats.streams.closed++
    stream.on('error', noop).destroy(err)
  }

  destroyPair(err) {
    const remote = this.remote

    this.destroy(err)
    if (remote) remote.destroy(err)
  }

  _onfirewall(socket, port, host) {
    this.stream.connect(socket, this.remoteId, port, host)

    return false
  }
}

exports.Client = class BlindRelayClient extends EventEmitter {
  static _clients = new WeakMap()

  static from(stream, opts) {
    let client = this._clients.get(stream)
    if (client) return client
    client = new this(stream, opts)
    this._clients.set(stream, client)
    return client
  }

  constructor(stream, opts = {}) {
    super()

    const { id, handshake, handshakeEncoding } = opts

    this._mux = Protomux.from(stream)

    this._channel = this._mux.createChannel({
      protocol: 'blind-relay',
      id,
      handshake: handshake ? handshakeEncoding || c.raw : null,
      onopen: this._onopen.bind(this),
      onclose: this._onclose.bind(this),
      ondestroy: this._ondestroy.bind(this)
    })

    this._pair = this._channel.addMessage({
      encoding: m.pair,
      onmessage: this._onpair.bind(this)
    })

    this._unpair = this._channel.addMessage({
      encoding: m.unpair
    })

    this._ending = false
    this._destroyed = false
    this._error = null
    this._requests = new Map()

    this._channel.open(handshake)
  }

  get closed() {
    return this._channel.closed
  }

  get mux() {
    return this._mux
  }

  get stream() {
    return this._mux.stream
  }

  get requests() {
    return this._requests.values()
  }

  _onopen() {
    this.emit('open')
  }

  _onclose() {
    this._ending = Promise.resolve()

    const err = this._error || errors.CHANNEL_CLOSED()

    for (const request of this._requests.values()) {
      request.destroy(err)
    }

    this._requests.clear()

    this.constructor._clients.delete(this.stream)

    this.emit('close')
  }

  _ondestroy() {
    this._destroyed = true
    this.emit('destroy')
  }

  _onpair({ isInitiator, token, id: remoteId }) {
    const request = this._requests.get(token.toString('hex'))

    if (request === undefined || request.isInitiator !== isInitiator) return

    request.push(remoteId)
    request.push(null)

    this.emit('pair', request.isInitiator, request.token, request.stream, remoteId)
  }

  pair(isInitiator, token, stream) {
    if (this._destroyed) throw errors.CHANNEL_DESTROYED()

    const keyString = token.toString('hex')

    if (this._requests.has(keyString)) throw errors.ALREADY_PAIRING()

    const request = new BlindRelayRequest(this, isInitiator, token, stream)

    this._requests.set(keyString, request)

    return request
  }

  unpair(token) {
    if (this._destroyed) throw errors.CHANNEL_DESTROYED()

    const request = this._requests.get(token.toString('hex'))

    if (request) request.destroy(errors.PAIRING_CANCELLED())

    this._unpair.send({ token })
  }

  cork() {
    this._channel.cork()
  }

  uncork() {
    this._channel.uncork()
  }

  async end() {
    if (this._ending) return this._ending

    this._ending = EventEmitter.once(this, 'close')
    this._endMaybe()

    return this._ending
  }

  _endMaybe() {
    if (this._ending && this._requests.size === 0) {
      this._channel.close()
    }
  }

  destroy(err) {
    if (this._destroyed) return
    this._destroyed = true

    this._error = err || errors.CHANNEL_DESTROYED()
    this._channel.close()
  }
}

class BlindRelayRequest extends Readable {
  constructor(client, isInitiator, token, stream) {
    super()

    this.client = client
    this.isInitiator = isInitiator
    this.token = token
    this.stream = stream
  }

  _open(cb) {
    if (this.client._destroyed) return cb(errors.CHANNEL_DESTROYED())

    this.client._pair.send({
      isInitiator: this.isInitiator,
      token: this.token,
      id: this.stream.id,
      seq: 0
    })

    cb(null)
  }

  _destroy(cb) {
    this.client._requests.delete(this.token.toString('hex'))

    cb(null)

    this.client._endMaybe()
  }
}

exports.token = function token(buf = b4a.allocUnsafe(32)) {
  sodium.randombytes_buf(buf)
  return buf
}

function noop() {}

function unlink(link) {
  const remote = link.remote

  link.remote = null

  if (remote && remote.remote === link) {
    remote.remote = null
  }
}

function createStats() {
  const stats = {
    sessions: {
      accepted: 0,
      opened: 0,
      closed: 0,
      get active() {
        return this.accepted - this.closed
      }
    },
    pairings: {
      requested: 0,
      matched: 0,
      cancelled: 0,
      pending: 0,
      active: 0
    },
    streams: {
      opened: 0,
      closed: 0,
      errors: 0,
      get active() {
        return this.opened - this.closed
      }
    }
  }

  return stats
}

const m = (exports.messages = {})

const flags = bitfield(7)

m.pair = {
  preencode(state, m) {
    flags.preencode(state)
    c.fixed32.preencode(state, m.token)
    c.uint.preencode(state, m.id)
    c.uint.preencode(state, m.seq)
  },
  encode(state, m) {
    flags.encode(state, bits.of(m.isInitiator))
    c.fixed32.encode(state, m.token)
    c.uint.encode(state, m.id)
    c.uint.encode(state, m.seq)
  },
  decode(state) {
    const [isInitiator] = bits.iterator(flags.decode(state))

    return {
      isInitiator,
      token: c.fixed32.decode(state),
      id: c.uint.decode(state),
      seq: c.uint.decode(state)
    }
  }
}

m.unpair = {
  preencode(state, m) {
    flags.preencode(state)
    c.fixed32.preencode(state, m.token)
  },
  encode(state, m) {
    flags.encode(state, bits.of())
    c.fixed32.encode(state, m.token)
  },
  decode(state) {
    flags.decode(state)

    return {
      token: c.fixed32.decode(state)
    }
  }
}