| 1 | import { createServer } from 'node:net' |
| 2 | import { start, stop, TypedEventEmitter } from '@libp2p/interface' |
| 3 | import { getThinWaistAddresses, isPrivate } from '@libp2p/utils' |
| 4 | import { multiaddr } from '@multiformats/multiaddr' |
| 5 | import { encode, decode } from 'it-length-prefixed' |
| 6 | import { pushable } from 'it-pushable' |
| 7 | import multicastDNS from 'multicast-dns' |
| 8 | import { duplex } from 'stream-to-it' |
| 9 | import { fromString as uint8ArrayFromString } from 'uint8arrays/from-string' |
| 10 | import { toString as uint8ArrayToString } from 'uint8arrays/to-string' |
| 11 | import { MessageEvent } from './message-event.ts' |
| 12 | import type { InspectorMessage } from '../index.js' |
| 13 | import type { ChannelMessages, Messages, MessagesComponents, MessagesInit } from './index.js' |
| 14 | import type { Logger } from '@libp2p/interface' |
| 15 | import type { Multiaddr } from '@multiformats/multiaddr' |
| 16 | import type { Answer, TxtAnswer } from 'dns-packet' |
| 17 | import type { Pushable } from 'it-pushable' |
| 18 | import type { MulticastDNS } from 'multicast-dns' |
| 19 | import type { Server, Socket, ListenOptions, AddressInfo } from 'node:net' |
| 20 | |
| 21 | interface MDNSPortInit extends MessagesInit { |
| 22 | messages: TCPPortMessages |
| 23 | } |
| 24 | |
| 25 | class MDNSPortAdvertisement { |
| 26 | private readonly components: MessagesComponents |
| 27 | private readonly log: Logger |
| 28 | private readonly serviceTag: string |
| 29 | private mdns?: MulticastDNS |
| 30 | private messages: TCPPortMessages |
| 31 | |
| 32 | constructor (components: MessagesComponents, init: MDNSPortInit) { |
| 33 | this.components = components |
| 34 | this.log = components.logger.forComponent('libp2p:inspector-metrics:messages') |
| 35 | this.serviceTag = init.serviceTag ?? '_libp2p_inspector_metrics._tcp.local' |
| 36 | this.messages = init.messages |
| 37 | |
| 38 | this._onMdnsQuery = this._onMdnsQuery.bind(this) |
| 39 | this._onMdnsWarning = this._onMdnsWarning.bind(this) |
| 40 | this._onMdnsError = this._onMdnsError.bind(this) |
| 41 | } |
| 42 | |
| 43 | async start (): Promise<void> { |
| 44 | this.mdns = multicastDNS() |
| 45 | this.mdns.on('query', this._onMdnsQuery) |
| 46 | this.mdns.on('warning', this._onMdnsWarning) |
| 47 | this.mdns.on('error', this._onMdnsError) |
| 48 | } |
| 49 | |
| 50 | async stop (): Promise<void> { |
| 51 | this.mdns?.destroy() |
| 52 | this.mdns = undefined |
| 53 | } |
| 54 | |
| 55 | _onMdnsQuery (event: multicastDNS.QueryPacket): void { |
| 56 | const address = this.messages.server?.address() |
| 57 | |
| 58 | if (this.mdns == null || address == null) { |
| 59 | return |
| 60 | } |
| 61 | |
| 62 | if (event.questions[0]?.name !== this.serviceTag) { |
| 63 | return |
| 64 | } |
| 65 | |
| 66 | const answers: Answer[] = [{ |
| 67 | name: this.serviceTag, |
| 68 | type: 'PTR', |
| 69 | class: 'IN', |
| 70 | ttl: 120, |
| 71 | data: `${this.components.peerId}.${this.serviceTag}` |
| 72 | }, ...toMultiaddrs(address) |
| 73 | // mDNS requires link-local addresses only |
| 74 | // https://github.com/libp2p/specs/blob/master/discovery/mdns.md#issues |
| 75 | .filter(isLinkLocal) |
| 76 | .map((ma): TxtAnswer => { |
| 77 | return { |
| 78 | name: `${this.components.peerId}.${this.serviceTag}`, |
| 79 | type: 'TXT', |
| 80 | class: 'IN', |
| 81 | ttl: 120, |
| 82 | data: ma.toString() |
| 83 | } |
| 84 | }) |
| 85 | ] |
| 86 | |
| 87 | this.mdns.respond(answers) |
| 88 | } |
| 89 | |
| 90 | _onMdnsWarning (err: Error): void { |
| 91 | this.log.error('mdns warning', err) |
| 92 | } |
| 93 | |
| 94 | _onMdnsError (err: Error): void { |
| 95 | this.log.error('mdns error', err) |
| 96 | } |
| 97 | } |
| 98 | |
| 99 | function toMultiaddrs (addr: string | AddressInfo): Multiaddr[] { |
| 100 | if (typeof addr === 'string') { |
| 101 | return [multiaddr(`/unix/${encodeURIComponent(addr)}`)] |
| 102 | } |
| 103 | |
| 104 | const { family, address, port } = addr |
| 105 | |
| 106 | return getThinWaistAddresses(multiaddr(`/ip${family === 'IPv6' ? 6 : 4}/${address}/tcp/${port}`)) |
| 107 | } |
| 108 | |
| 109 | function isLinkLocal (ma: Multiaddr): boolean { |
| 110 | // match private ip4/ip6 & loopback addresses |
| 111 | if (isPrivate(ma)) { |
| 112 | return true |
| 113 | } |
| 114 | |
| 115 | return false |
| 116 | } |
| 117 | |
| 118 | interface Client { |
| 119 | pushable: Pushable<Uint8Array> |
| 120 | socket: Socket |
| 121 | } |
| 122 | |
| 123 | class TCPPortMessages extends TypedEventEmitter<ChannelMessages> { |
| 124 | private readonly log: Logger |
| 125 | public server?: Server |
| 126 | private clients: Client[] |
| 127 | private readonly listenOptions: ListenOptions | string | number |
| 128 | private advertisement: MDNSPortAdvertisement |
| 129 | |
| 130 | constructor (components: MessagesComponents, init: MessagesInit) { |
| 131 | super() |
| 132 | |
| 133 | this.log = components.logger.forComponent('libp2p:inspector-metrics:messages') |
| 134 | this.clients = [] |
| 135 | this.listenOptions = init.listenOptions ?? 0 |
| 136 | |
| 137 | this.advertisement = new MDNSPortAdvertisement(components, { |
| 138 | ...init, |
| 139 | messages: this |
| 140 | }) |
| 141 | |
| 142 | this.onSocket = this.onSocket.bind(this) |
| 143 | } |
| 144 | |
| 145 | async start (): Promise<void> { |
| 146 | this.server = createServer(this.onSocket) |
| 147 | |
| 148 | await new Promise<void>((resolve, reject) => { |
| 149 | this.server?.listen(this.listenOptions, () => { |
| 150 | resolve() |
| 151 | }) |
| 152 | this.server?.on('error', (err) => { |
| 153 | reject(err) |
| 154 | }) |
| 155 | }) |
| 156 | |
| 157 | await start(this.advertisement) |
| 158 | } |
| 159 | |
| 160 | async stop (): Promise<void> { |
| 161 | await stop(this.advertisement) |
| 162 | this.server?.close() |
| 163 | this.clients.forEach(({ socket }) => { |
| 164 | socket.destroy() |
| 165 | }) |
| 166 | } |
| 167 | |
| 168 | onSocket (socket: Socket): void { |
| 169 | const client = { |
| 170 | socket, |
| 171 | pushable: pushable() |
| 172 | } |
| 173 | |
| 174 | this.clients.push(client) |
| 175 | |
| 176 | const duplexSocket = duplex(socket) |
| 177 | |
| 178 | Promise.all([ |
| 179 | duplexSocket.sink(encode(client.pushable)), |
| 180 | (async () => { |
| 181 | for await (const buf of decode(duplexSocket.source)) { |
| 182 | try { |
| 183 | const data = JSON.parse(uint8ArrayToString(buf.subarray())) |
| 184 | |
| 185 | this.dispatchEvent(new MessageEvent(this, data)) |
| 186 | } catch (err) { |
| 187 | // eslint-disable-next-line no-console |
| 188 | console.error('could not parse message', uint8ArrayToString(buf.subarray()), err) |
| 189 | } |
| 190 | } |
| 191 | })() |
| 192 | ]) |
| 193 | .catch(err => { |
| 194 | // eslint-disable-next-line no-console |
| 195 | console.error('socket error during messages decode', err) |
| 196 | }) |
| 197 | } |
| 198 | |
| 199 | postMessage (message: InspectorMessage): void { |
| 200 | const buf = uint8ArrayFromString(JSON.stringify(message)) |
| 201 | this.clients.forEach(({ pushable }) => { |
| 202 | pushable.push(buf) |
| 203 | }) |
| 204 | } |
| 205 | } |
| 206 | |
| 207 | export function messages (components: MessagesComponents, init: MessagesInit = {}): Messages { |
| 208 | return new TCPPortMessages(components, init) |
| 209 | } |