path: root/packages/meshbay-client/src/cast-relay.js

/**
 * Local HTTP relay for LAN casting.
 *
 * Re-serves decrypted fMP4 segments from the renderer over HTTP so that a
 * Chromecast, Smart TV or any player on the same Wi-Fi can stream the video.
 *
 * The renderer sends raw byte chunks from the WebRTC DataChannel — these are
 * arbitrary-sized slices of the fMP4 stream, NOT aligned to MP4 box boundaries.
 * MSE handles this internally, but external players (VLC, Chromecast) need
 * properly framed fMP4 fragments. The BoxAccumulator reassembles the byte
 * stream and emits complete moof+mdat pairs.
 *
 * Subtitles ride alongside the video as a side-loaded WebVTT file at
 * `/subs.vtt`. A receiver fetches that one with XHR rather than handing it to
 * a media element, so unlike the stream it needs CORS headers to be readable
 * at all.
 *
 * A whole file — a photo, or a music track with its cover at `/cover` — is
 * served at `/file`, one at a time: the page decrypts it and hands it over in
 * pieces (begin, writes, end), and the receiver loads it as a picture or as
 * music. Ranges are honoured there, which is what a receiver seeks with. The
 * relay starts for a file when it is not already running.
 *
 * Mitigations:
 *   · Bind to the LAN interface, never 0.0.0.0
 *   · Fixed port range (19550-19553), opened only during active cast
 *   · Unguessable token in the URL (32 hex chars)
 *   · CORS is granted on both served paths, and both stay behind the token
 *   · Cache-Control: no-store on every response
 *   · Server destroyed when playback stops — zero residual surface
 */

'use strict';

const crypto = require('node:crypto');
const http = require('node:http');
const os = require('node:os');
const util = require('node:util');

/**
 * Progress logging, off unless asked for:
 *
 *     NODE_DEBUG=cast-relay npm start
 *
 * Most of what this class has to say repeats without bound — a line every
 * fiftieth fragment for the length of a film, one per dropped fragment
 * whenever a client falls behind — and it lands in the terminal the app was
 * started from. Node's own `debuglog` is used rather than a flag of our own:
 * it costs nothing when disabled, since the message is never even formatted.
 *
 * Nothing that stops a cast is hidden behind it. Every hard failure here
 * throws — no free port, no such file — and the renderer already reports the
 * rejection. The one fault that neither throws nor reaches anybody else is a
 * loss of fMP4 framing, and that one stays on `console.warn`.
 */
const debug = util.debuglog('cast-relay');

const RING_CAP = 64;
// A fragment is a whole segment, megabytes at a film's bitrate: 64 of them
// can be hundreds of megabytes, so the backlog is bounded in bytes too.
const RING_MAX_BYTES = 32 * 1024 * 1024;
// What the receiver is ever told a photo is. The page re-encodes every photo
// to JPEG, so this list is a guard, not a menu.
const FILE_TYPES = new Set([
  'image/jpeg', 'image/png', 'image/webp',
  'audio/mpeg', 'audio/mp4', 'audio/aac', 'audio/flac', 'audio/ogg', 'audio/opus',
  'audio/wav', 'audio/webm',
]);
// A track is a few megabytes, a long lossless one a few hundred; this only
// stops a page from filling the machine's memory.
const FILE_MAX_BYTES = 1024 * 1024 * 1024;
// How far a receiver may fall behind before fragments are dropped for it.
// The page runs ahead of the television by its whole read-ahead (up to 48 MB,
// video-player.js) plus the receiver's start-up, and a fragment is a few
// seconds of picture: the 8 MB this used to be dropped one every few seconds
// from a receiver reading at playback speed, and each froze the picture for
// its length. Node holds what waits in the response's own buffer.
const CLIENT_LEAD_MAX = 512 * 1024 * 1024;
// A receiver that connects before the first moof waits for the whole header.
const HEADER_WAIT_MS = 10000;
const FTYP = 0x66747970;
const MOOF = 0x6d6f6f66;
const PORT_BASE = 19550;
const PORT_COUNT = 4;

// A receiver reads a side-loaded subtitle with XHR, from its own origin, so
// the three headers it sends have to be allowed by name — `Range` included,
// which it sends even for a document it will read whole.
const CORS_HEADERS = Object.freeze({
  'Access-Control-Allow-Origin': '*',
  'Access-Control-Allow-Methods': 'GET, OPTIONS',
  'Access-Control-Allow-Headers': 'Content-Type, Accept-Encoding, Range',
  'Access-Control-Expose-Headers': 'Content-Length, Content-Range',
});

function lanAddress() {
  const ifaces = os.networkInterfaces();
  for (const name of Object.keys(ifaces)) {
    for (const iface of ifaces[name]) {
      if (!iface.internal && iface.family === 'IPv4') {
        return iface.address;
      }
    }
  }
  return '127.0.0.1';
}

class BoxAccumulator {
  constructor() {
    this._buf = Buffer.alloc(0);
    this._synced = false;
    this._preambleSeen = false;
    // Called once, with every byte before the first moof: the stream's
    // header (ftyp, moov), however many pushes it came in.
    this.onPreamble = null;
  }

  push(data) {
    this._buf = Buffer.concat([this._buf, data]);
    const fragments = [];

    if (!this._synced) {
      const idx = this._findMoof();
      if (idx === -1) return fragments;
      debug(`[cast-relay] box sync: found first moof at byte offset ${idx}`);
      if (!this._preambleSeen) {
        this._preambleSeen = true;
        if (this.onPreamble) this.onPreamble(Buffer.from(this._buf.subarray(0, idx)));
      }
      this._buf = this._buf.subarray(idx);
      this._synced = true;
    }

    while (this._buf.length >= 8) {
      const size = this._buf.readUInt32BE(0);
      const type = this._buf.readUInt32BE(4);

      if (size < 8) {
        // Not `debug`: the byte stream stopped being framed fMP4. It recovers
        // by rescanning, so nothing throws and nobody else ever hears about
        // it — this line is the only trace that the picture on the television
        // is missing a piece.
        console.warn(`[cast-relay] box sync lost: invalid size ${size}, rescanning`);
        this._synced = false;
        const idx = this._findMoof();
        if (idx === -1) return fragments;
        this._buf = this._buf.subarray(idx);
        this._synced = true;
        continue;
      }

      if (type === MOOF) {
        if (this._buf.length < size + 8) break;
        const mdatSize = this._buf.readUInt32BE(size);
        const pairSize = size + mdatSize;
        if (this._buf.length < pairSize) break;

        fragments.push(Buffer.from(this._buf.subarray(0, pairSize)));
        this._buf = this._buf.subarray(pairSize);
      } else {
        if (this._buf.length < size) break;
        this._buf = this._buf.subarray(size);
      }
    }

    return fragments;
  }

  reset() {
    this._buf = Buffer.alloc(0);
    this._synced = false;
    this._preambleSeen = false;
  }

  _findMoof() {
    for (let i = 0; i <= this._buf.length - 8; i++) {
      if (this._buf.readUInt32BE(i + 4) === MOOF) {
        const size = this._buf.readUInt32BE(i);
        if (size >= 8 && size < 1_000_000) {
          return i;
        }
      }
    }
    return -1;
  }
}

/**
 * The header is what comes before the first moof, and only that.
 *
 * The page hands over the first chunk it decrypted, which is a slice of the
 * stream and not a box: measured, it was by turns the 28-byte ftyp alone, the
 * header exactly, or the header and the start of the first fragment,
 * depending on when the node read ffmpeg's output. Served as the header, the
 * first left the receiver without a moov and the third broke the box
 * structure, so the same film cast one try in three. The pushed stream
 * carries the whole header; this trims the page's chunk for the case where
 * nothing came before the first moof.
 */
function headerOnly(init) {
  let at = 0;
  while (at + 8 <= init.length) {
    if (init.readUInt32BE(at + 4) === MOOF) return init.subarray(0, at);
    const size = init.readUInt32BE(at);
    if (size < 8) break;
    at += size;
  }
  return init;
}

class CastRelay {
  constructor() {
    this._server = null;
    this._port = null;
    this._token = null;
    this._lanIP = null;
    this._initSegment = null;
    this._headerReady = false;
    this._waiting = new Set();
    this._ring = [];
    this._ringBytes = 0;
    this._clients = new Set();
    this._accum = new BoxAccumulator();
    this._fragCount = 0;
    this._chunkCount = 0;
    this._bytesSent = 0;
    this._dropped = 0;
    this._subtitle = null;
    this._subtitleVersion = 0;
    this._file = null;
    this._fileVersion = 0;
    this._incoming = null;
  }

  get active() { return this._server !== null; }

  get url() {
    if (!this._server) return null;
    return `http://${this._lanIP}:${this._port}/stream.mp4?t=${this._token}`;
  }

  /**
   * Where the current subtitle can be fetched, or null when there is none.
   *
   * The version is part of the URL because a receiver caches a side-loaded
   * track by its address: changing the cues behind a fixed URL would leave the
   * old ones on screen.
   */
  get subtitleUrl() {
    if (!this._server || !this._subtitle) return null;
    return `http://${this._lanIP}:${this._port}/subs.vtt`
      + `?t=${this._token}&v=${this._subtitleVersion}`;
  }

  /**
   * Where the current file can be fetched, or null when there is none.
   * Versioned for the same reason as the subtitle: a receiver handed the same
   * address twice shows what it already has.
   */
  get fileUrl() {
    if (!this._server || !this._file) return null;
    return `http://${this._lanIP}:${this._port}/file`
      + `?t=${this._token}&v=${this._fileVersion}`;
  }

  get fileType() { return this._file ? this._file.type : null; }

  /** What the receiver is told the file is: title, artist, album, cover. */
  get fileMeta() {
    if (!this._file) return null;
    const { title, artist, album } = this._file.meta;
    const cover = this._file.cover
      ? `http://${this._lanIP}:${this._port}/cover?t=${this._token}&v=${this._fileVersion}`
      : null;
    return { title, artist, album, cover };
  }

  /**
   * A file arrives in three steps, so that a large one never has to cross in
   * a single message: `beginFile` names it, `writeFile` appends, `endFile`
   * puts it up and answers its address. The relay starts if nothing has.
   */
  async beginFile({ type, size, cover, title, artist, album }) {
    if (!FILE_TYPES.has(type)) throw new Error(`Not something a receiver shows: ${type}`);
    if (!(size > 0 && size <= FILE_MAX_BYTES)) throw new Error(`Bad file size: ${size}`);
    if (cover && !FILE_TYPES.has(cover.type)) throw new Error(`Bad cover type: ${cover.type}`);
    if (!this._server) await this.start({});
    const text = (v) => (typeof v === 'string' ? v.slice(0, 500) : '');
    this._incoming = {
      type, size, parts: [], received: 0,
      cover: cover && cover.type.startsWith('image/')
        ? { bytes: Buffer.from(cover.bytes), type: cover.type } : null,
      meta: { title: text(title), artist: text(artist), album: text(album) },
    };
    return true;
  }

  writeFile(bytes) {
    const f = this._incoming;
    if (!f) throw new Error('No file is being sent');
    const buf = Buffer.from(bytes);
    if (f.received + buf.length > f.size) throw new Error('More bytes than announced');
    f.parts.push(buf);
    f.received += buf.length;
    return true;
  }

  endFile() {
    const f = this._incoming;
    this._incoming = null;
    if (!f) throw new Error('No file is being sent');
    if (f.received !== f.size) throw new Error(`File cut short: ${f.received} of ${f.size} bytes`);
    this._file = { bytes: Buffer.concat(f.parts), type: f.type, cover: f.cover, meta: f.meta };
    this._fileVersion++;
    debug(`[cast-relay] file set: ${f.size} bytes, ${f.type}, v${this._fileVersion}`);
    return { url: this.fileUrl };
  }

  get subtitle() {
    if (!this._subtitle) return null;
    return {
      url: this.subtitleUrl,
      language: this._subtitle.language,
      label: this._subtitle.label,
    };
  }

  /**
   * Carry a WebVTT document for the receiver to side-load.
   *
   * The cues must already be expressed on the *stream's* timeline, which
   * starts at zero at the seek point — not on the film's. The renderer shifts
   * them before they get here; the relay only serves bytes.
   */
  setSubtitle(sub) {
    if (!sub || !sub.vtt) {
      this._subtitle = null;
      this._subtitleVersion++;
      debug('[cast-relay] subtitle cleared');
      return null;
    }
    this._subtitle = {
      vtt: Buffer.from(sub.vtt, 'utf8'),
      language: sub.language || '',
      label: sub.label || '',
    };
    this._subtitleVersion++;
    debug(`[cast-relay] subtitle set: ${this._subtitle.vtt.length} bytes`
      + `, lang "${this._subtitle.language}", v${this._subtitleVersion}`);
    return this.subtitle;
  }

  async start({ codec, initSegment, subtitle }) {
    // Everything the new stream needs is set before the first await: the
    // page's pushes arrive while this waits for its port, and they belong to
    // the new stream's accumulator, header first.
    const closing = this._teardown();

    this._token = crypto.randomBytes(16).toString('hex');
    this._lanIP = lanAddress();
    this._initSegment = initSegment ? headerOnly(Buffer.from(initSegment)) : null;
    // Nothing to wait for without a first chunk: no header is coming.
    this._headerReady = !initSegment;
    this._accum.onPreamble = (bytes) => this._preamble(bytes);
    this.setSubtitle(subtitle);
    const token = this._token;
    await closing;

    const server = http.createServer((req, res) => this._handle(req, res));

    let bound = false;
    for (let i = 0; i < PORT_COUNT && !bound; i++) {
      const port = PORT_BASE + i;
      try {
        await new Promise((resolve, reject) => {
          const onError = (err) => {
            server.removeListener('error', onError);
            reject(err);
          };
          server.on('error', onError);
          server.listen(port, this._lanIP, () => {
            server.removeListener('error', onError);
            resolve();
          });
        });
        bound = port;
      } catch (err) {
        if (err.code !== 'EADDRINUSE') throw err;
        debug(`[cast-relay] port ${port} busy, trying next`);
      }
    }
    if (!bound) throw new Error('All cast relay ports are in use');
    if (this._token !== token) {
      server.close();
      throw new Error('Cast relay stopped while starting');
    }

    this._server = server;
    this._port = bound;
    debug(`[cast-relay] started on ${this.url}`);
    debug(`[cast-relay] init segment: ${this._initSegment ? this._initSegment.length + ' bytes' : 'none'}`);
    return {
      url: this.url,
      port: this._port,
      token: this._token,
      subtitle: this.subtitle,
    };
  }

  /**
   * Every byte the page pushed before the first moof. Kept as the header when
   * it starts with an ftyp; the page's own first chunk, trimmed, otherwise.
   */
  _preamble(bytes) {
    if (bytes.length >= 8 && bytes.readUInt32BE(4) === FTYP) this._initSegment = bytes;
    debug(`[cast-relay] header complete: ${this._initSegment ? this._initSegment.length : 0} bytes`);
    this._headerReady = true;
    for (const res of this._waiting) this._admit(res);
    this._waiting.clear();
  }

  pushSegment(data) {
    const buf = Buffer.isBuffer(data) ? data : Buffer.from(data);
    this._chunkCount++;

    const fragments = this._accum.push(buf);

    for (const frag of fragments) {
      this._fragCount++;
      if (this._fragCount <= 3 || this._fragCount % 50 === 0) {
        debug(`[cast-relay] fragment #${this._fragCount}: ${frag.length} bytes (from ${this._chunkCount} chunks), ${this._clients.size} client(s)`);
      }

      this._ring.push(frag);
      this._ringBytes += frag.length;
      while (this._ring.length > RING_CAP
             || (this._ringBytes > RING_MAX_BYTES && this._ring.length > 1)) {
        this._ringBytes -= this._ring.shift().length;
      }

      for (const res of this._clients) {
        if (res.writableLength > CLIENT_LEAD_MAX) {
          // The picture on the television freezes for this fragment's length.
          this._dropped++;
          if (this._dropped <= 3 || this._dropped % 20 === 0) {
            debug(`[cast-relay] dropped a fragment of ${frag.length} bytes for a`
              + ` receiver ${(res.writableLength / 1048576).toFixed(0)} MB behind,`
              + ` ${this._dropped} dropped so far`);
          }
          continue;
        }
        res.write(frag);
        this._bytesSent += frag.length;
      }
    }
  }

  finish() {
    debug('[cast-relay] finishing stream');
    for (const res of this._waiting) this._admit(res);
    this._waiting.clear();
    for (const res of this._clients) {
      try { res.end(); } catch { /* already closed */ }
    }
  }

  async stop() {
    await this._teardown();
  }

  /**
   * Everything is reset before anything is awaited, and the promise only says
   * when the old port is closed. The page calls stop() without waiting and
   * start() next: a stop that cleared its state after the close cleared the
   * new stream's instead (its token, port and header) and the receiver was
   * handed `:null/stream.mp4?t=null`. The close used to wait, too, for the
   * receiver to let go of the old stream, which it does only once loaded with
   * the new one; its connections are cut instead.
   */
  _teardown() {
    debug(`[cast-relay] stopping: ${this._chunkCount} chunks, ${this._fragCount} fragments, ${(this._bytesSent / 1048576).toFixed(1)} MB sent, ${this._dropped} dropped`);
    for (const res of [...this._clients, ...this._waiting]) {
      try { res.end(); } catch { /* already closed */ }
    }
    this._clients = new Set();
    this._waiting = new Set();

    const srv = this._server;
    this._server = null;
    this._port = null;
    this._token = null;
    this._initSegment = null;
    this._headerReady = false;
    this._subtitle = null;
    this._file = null;
    this._incoming = null;
    this._ring = [];
    this._ringBytes = 0;
    this._accum = new BoxAccumulator();
    this._fragCount = 0;
    this._chunkCount = 0;
    this._bytesSent = 0;
    this._dropped = 0;

    if (!srv) return Promise.resolve();
    const closed = new Promise((resolve) => srv.close(resolve));
    srv.closeAllConnections();
    return closed;
  }

  _handle(req, res) {
    if (req.method !== 'GET' && req.method !== 'OPTIONS') {
      res.writeHead(405);
      res.end();
      return;
    }

    let url;
    try {
      url = new URL(req.url, `http://${req.headers.host}`);
    } catch {
      res.writeHead(400);
      res.end();
      return;
    }

    // The preflight is answered before the token is examined: a browser sends
    // it without credentials and rejecting it here would read, on the receiver,
    // as a network failure rather than as a refusal.
    if (req.method === 'OPTIONS') {
      res.writeHead(204, CORS_HEADERS);
      res.end();
      return;
    }

    if (url.searchParams.get('t') !== this._token) {
      res.writeHead(403);
      res.end();
      return;
    }

    if (url.pathname === '/subs.vtt') {
      this._handleSubtitle(res);
      return;
    }

    if (url.pathname === '/file') {
      this._handleFile(req, res);
      return;
    }

    if (url.pathname === '/cover') {
      this._handleCover(res);
      return;
    }

    if (url.pathname !== '/stream.mp4') {
      res.writeHead(404);
      res.end();
      return;
    }

    res.writeHead(200, {
      // A receiver that has been given a side-loaded track reads the media
      // through the same CORS-checked path as the track, so the headers go on
      // both or neither. They widen nothing: the URL is already unguessable,
      // and a page that has it could embed it in a media element regardless.
      ...CORS_HEADERS,
      'Content-Type': 'video/mp4',
      'Cache-Control': 'no-store',
      'Accept-Ranges': 'none',
      'Connection': 'keep-alive',
    });

    res.flushHeaders();
    req.on('close', () => {
      this._clients.delete(res);
      this._waiting.delete(res);
      debug(`[cast-relay] client disconnected, ${this._clients.size} remaining`);
    });

    if (this._headerReady) {
      this._admit(res);
      return;
    }
    // Before the first moof the header may be a piece of itself: the
    // receiver waits for all of it, or for the page's chunk at worst.
    this._waiting.add(res);
    const token = this._token;
    setTimeout(() => {
      if (this._token !== token || !this._waiting.delete(res)) return;
      debug('[cast-relay] no moof yet, serving the first chunk as the header');
      this._admit(res);
    }, HEADER_WAIT_MS).unref();
  }

  /** The header, the backlog, then live fragments from here on. */
  _admit(res) {
    let sent = 0;
    if (this._initSegment) {
      res.write(this._initSegment);
      sent += this._initSegment.length;
    }

    for (const frag of this._ring) {
      res.write(frag);
      sent += frag.length;
    }

    debug(`[cast-relay] client connected from ${res.socket && res.socket.remoteAddress}: sent init + ${this._ring.length} fragments (${(sent / 1024).toFixed(0)} KB)`);

    this._clients.add(res);
  }

  _handleSubtitle(res) {
    if (!this._subtitle) {
      res.writeHead(404, CORS_HEADERS);
      res.end();
      return;
    }
    res.writeHead(200, {
      ...CORS_HEADERS,
      'Content-Type': 'text/vtt; charset=utf-8',
      'Content-Length': this._subtitle.vtt.length,
      'Cache-Control': 'no-store',
    });
    res.end(this._subtitle.vtt);
    debug(`[cast-relay] subtitle served: ${this._subtitle.vtt.length} bytes`);
  }

  /**
   * The whole file, or the one range asked for: a receiver seeks in a track
   * by asking for the bytes from there on.
   */
  _handleFile(req, res) {
    const f = this._file;
    if (!f) {
      res.writeHead(404, CORS_HEADERS);
      res.end();
      return;
    }
    const total = f.bytes.length;
    const base = {
      ...CORS_HEADERS,
      'Content-Type': f.type,
      'Cache-Control': 'no-store',
      'Accept-Ranges': 'bytes',
    };
    const m = /^bytes=(\d*)-(\d*)$/.exec(req.headers.range || '');
    if (m && (m[1] !== '' || m[2] !== '')) {
      let from = m[1] === '' ? total - Number(m[2]) : Number(m[1]);
      let to = m[1] === '' || m[2] === '' ? total - 1 : Math.min(Number(m[2]), total - 1);
      from = Math.max(0, from);
      if (from > to || from >= total) {
        res.writeHead(416, { ...base, 'Content-Range': `bytes */${total}` });
        res.end();
        return;
      }
      res.writeHead(206, { ...base, 'Content-Range': `bytes ${from}-${to}/${total}`,
                           'Content-Length': to - from + 1 });
      res.end(f.bytes.subarray(from, to + 1));
      return;
    }
    res.writeHead(200, { ...base, 'Content-Length': total });
    res.end(f.bytes);
    debug(`[cast-relay] file served: ${total} bytes`);
  }

  _handleCover(res) {
    const c = this._file && this._file.cover;
    if (!c) {
      res.writeHead(404, CORS_HEADERS);
      res.end();
      return;
    }
    res.writeHead(200, {
      ...CORS_HEADERS,
      'Content-Type': c.type,
      'Content-Length': c.bytes.length,
      'Cache-Control': 'no-store',
    });
    res.end(c.bytes);
  }
}

module.exports = CastRelay;
module.exports.headerOnly = headerOnly;