// session — the `--hyperident` mode. Auth is the hyperdht pubkey firewall (no
// SSH): the daemon runs on its own key, and per connection spawns its OWN
// mosh-server, sends the client the one-time key over a control frame, and
// bridges the mosh UDP over the hyperdht stream. Simpler than SSH-mode but the
// firewall allow-list IS the whole auth surface (see bin/hypermosh.mjs).
//
// Wire format: shared typed frames (frames.mjs) —
// type 0 (control): JSON — server sends {ok,key}|{ok:false,error}
// type 1 (data): raw — one frame == one UDP datagram (both directions)
import DHT from 'hyperdht'
import dgram from '#node:dgram'
import cp from '#node:child_process'
import process from '#node:process'
import b4a from 'b4a'
import { ctrlFrame, dataFrame, makeParser } from './frames.mjs'
import { mkLog } from './log.mjs'
export async function serveSession ({ keyPair, allow = true, node = null, moshTimeoutMs = 15000 } = {}) {
const log = mkLog('serve-hi')
const dht = node || new DHT()
const firewall = (rpk) => !(allow === true || allow.some((k) => b4a.equals(rpk, k)))
const server = dht.createServer({ firewall }, (socket) => {
socket.on('error', (e) => log('socket error', e.code || e.message))
let udp = null
let moshPort = null
let daemonPid = null
let replied = false
const reply = (obj) => { if (!replied) { replied = true; try { socket.write(ctrlFrame(obj)) } catch {} } }
const onCtrl = (start) => {
const args = ['new', '-i', '127.0.0.1']
if (start && Array.isArray(start.cmd) && start.cmd.length) args.push('--', ...start.cmd)
let proc
try {
proc = cp.spawn('mosh-server', args, { env: { ...process.env, LC_ALL: 'C.UTF-8', LANG: 'C.UTF-8' } })
} catch (e) {
reply({ ok: false, error: 'spawn mosh-server: ' + e.message })
return
}
// Async spawn failures (ENOENT) and early exits are NOT caught by the
// try/catch above — surface them to the client instead of hanging it.
proc.on('error', (e) => { log('mosh-server spawn error', e.code || e.message); reply({ ok: false, error: 'mosh-server not runnable: ' + (e.code || e.message) }) })
const failTimer = setTimeout(() => reply({ ok: false, error: 'mosh-server produced no MOSH CONNECT in ' + moshTimeoutMs + 'ms' }), moshTimeoutMs)
let out = ''
const scan = (d) => {
out += d
const m = out.match(/MOSH CONNECT (\d+) (\S+)/)
if (m && !moshPort) {
moshPort = Number(m[1])
udp = dgram.createSocket('udp4')
udp.on('message', (msg) => { try { socket.write(dataFrame(msg)) } catch {} })
udp.on('error', (e) => log('udp error', e.code || e.message))
clearTimeout(failTimer)
log('mosh-server up on', moshPort)
reply({ ok: true, key: m[2] })
}
const pm = out.match(/pid\s*=\s*(\d+)/)
if (pm) daemonPid = Number(pm[1])
}
proc.stdout.on('data', scan)
proc.stderr.on('data', scan)
proc.on('exit', (code) => { if (!moshPort) { clearTimeout(failTimer); reply({ ok: false, error: 'mosh-server exited (' + code + ') before starting a session' }) } })
// Do NOT socket.end() on proc 'exit': mosh-server DAEMONIZES — the spawned
// parent exits right after printing MOSH CONNECT while the real server runs
// detached. We track the detached pid for cleanup.
}
const onData = (dg) => { if (udp && moshPort) { try { udp.send(dg, moshPort, '127.0.0.1') } catch {} } }
socket.on('data', makeParser({ onCtrl, onData }))
socket.on('close', () => {
try { if (udp) udp.close() } catch {}
try { if (daemonPid) process.kill(daemonPid) } catch {} // reap the detached mosh-server
})
})
await server.listen(keyPair)
log('listening as', b4a.toString(keyPair.publicKey, 'hex').slice(0, 8))
return { dht, server, publicKey: keyPair.publicKey, close: async () => { await server.close(); if (!node) await dht.destroy() } }
}
const TRANSIENT = new Set(['HOLEPUNCH_ABORTED', 'ETIMEDOUT', 'PEER_NOT_FOUND', 'ECONNRESET', 'PEER_CONNECTION_FAILED'])
export function connectSession ({ remoteKey, keyPair = null, start = {}, onstarted, onerror = null, retries = 5, handshakeTimeoutMs = 20000 } = {}) {
const log = mkLog('client-hi')
const state = { dht: null, socket: null, udp: null, closed: false }
let attempt = 0
const fail = (err) => { if (onerror) onerror(err instanceof Error ? err : new Error(String(err))) }
const go = () => {
if (state.closed) return
attempt++
const dht = new DHT()
const socket = dht.connect(remoteKey, keyPair ? { keyPair } : undefined)
const udp = dgram.createSocket('udp4')
state.dht = dht; state.socket = socket; state.udp = udp
let established = false
let peer = null
let done = false
socket.setKeepAlive(5000)
const hsTimer = setTimeout(() => {
if (!done) { log('handshake timeout'); try { socket.destroy() } catch {}; fail(new Error('timed out waiting for session (server reachable but silent in ' + handshakeTimeoutMs + 'ms)')) }
}, handshakeTimeoutMs)
const onCtrl = (m) => {
done = true; clearTimeout(hsTimer)
if (m && m.ok) onstarted(m.key, udp.address().port)
else fail(new Error((m && m.error) || 'session failed'))
}
const onData = (dg) => { if (peer) { try { udp.send(dg, peer.port, peer.address) } catch {} } }
socket.on('data', makeParser({ onCtrl, onData }))
udp.on('error', (e) => log('local udp error', e.code || e.message))
udp.on('message', (msg, rinfo) => { peer = rinfo; try { socket.write(dataFrame(msg)) } catch {} })
socket.once('open', () => {
established = true
log('stream open (attempt ' + attempt + ')')
udp.bind(0, '127.0.0.1', () => { try { socket.write(ctrlFrame(start)) } catch {} })
socket.once('close', () => { clearTimeout(hsTimer); try { udp.close() } catch {}; dht.destroy() })
})
socket.on('error', (err) => {
if (established) { fail(err); return }
clearTimeout(hsTimer)
log('connect error', err.code || err.message)
try { socket.destroy() } catch {}
dht.destroy().finally(() => {
if (!state.closed && attempt <= retries && (TRANSIENT.has(err.code) || !err.code)) setTimeout(go, 400 * attempt)
else fail(err)
})
})
}
go()
return { close: async () => { state.closed = true; try { if (state.udp) state.udp.close() } catch {}; try { if (state.socket) state.socket.destroy() } catch {}; try { if (state.dht) await state.dht.destroy() } catch {} } }
}