dgram: added setMulticastInterface()

Add wrapper for uv's uv_udp_set_multicast_interface which provides the
sender side mechanism to explicitly select an interface. The
equivalent receiver side mechanism is the optional 2nd argument of
addMembership().

PR-URL: #7855
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Colin Ihrig <cjihrig@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>

Author: lostnet

doc/api/dgram.md

@@ -336,6 +336,84 @@ added: v0.6.9
 Sets or clears the `SO_BROADCAST` socket option.  When set to `true`, UDP
 packets may be sent to a local interface's broadcast address.
 
+### socket.setMulticastInterface(multicastInterface)
+<!-- YAML
+added: REPLACEME
+-->
+
+* `multicastInterface` {String}
+
+*Note: All references to scope in this section are refering to
+[IPv6 Zone Indices][], which are defined by [RFC 4007][]. In string form, an IP
+with a scope index is written as `'IP%scope'` where scope is an interface name or
+interface number.*
+
+Sets the default outgoing multicast interface of the socket to a chosen
+interface or back to system interface selection. The `multicastInterface` must
+be a valid string representation of an IP from the socket's family.
+
+For IPv4 sockets, this should be the IP configured for the desired physical
+interface. All packets sent to multicast on the socket will be sent on the
+interface determined by the most recent successful use of this call.
+
+For IPv6 sockets, `multicastInterface` should include a scope to indicate the
+interface as in the examples that follow. In IPv6, individual `send` calls can
+also use explicit scope in addresses, so only packets sent to a multicast
+address without specifying an explicit scope are affected by the most recent
+successful use of this call.
+
+#### Examples: IPv6 Outgoing Multicast Interface
+
+On most systems, where scope format uses the interface name:
+
+```js
+const socket = dgram.createSocket('udp6');
+
+socket.bind(1234, () => {
+  socket.setMulticastInterface('::%eth1');
+});
+```
+
+On Windows, where scope format uses an interface number:
+
+```js
+const socket = dgram.createSocket('udp6');
+
+socket.bind(1234, () => {
+  socket.setMulticastInterface('::%2');
+});
+```
+
+#### Example: IPv4 Outgoing Multicast Interface
+All systems use an IP of the host on the desired physical interface:
+```js
+const socket = dgram.createSocket('udp4');
+
+socket.bind(1234, () => {
+  socket.setMulticastInterface('10.0.0.2');
+});
+```
+
+#### Call Results
+
+A call on a socket that is not ready to send or no longer open may throw a *Not
+running* [`Error`][].
+
+If `multicastInterface` can not be parsed into an IP then an *EINVAL*
+[`System Error`][] is thrown.
+
+On IPv4, if `multicastInterface` is a valid address but does not match any
+interface, or if the address does not match the family then
+a [`System Error`][] such as `EADDRNOTAVAIL` or `EPROTONOSUP` is thrown.
+
+On IPv6, most errors with specifying or omiting scope will result in the socket
+continuing to use (or returning to) the system's default interface selection.
+
+A socket's address family's ANY address (IPv4 `'0.0.0.0'` or IPv6 `'::'`) can be
+used to return control of the sockets default outgoing interface to the system
+for future multicast packets.
+
+
 ### socket.setMulticastLoopback(flag)
 <!-- YAML
 added: v0.3.8
@@ -490,4 +568,7 @@ and `udp6` sockets). The bound address and port can be retrieved using
 [`socket.address().address`]: #dgram_socket_address
 [`socket.address().port`]: #dgram_socket_address
 [`socket.bind()`]: #dgram_socket_bind_port_address_callback
+[`System Error`]: errors.html#errors_class_system_error
 [byte length]: buffer.html#buffer_class_method_buffer_bytelength_string_encoding
+[IPv6 Zone Indices]: https://en.wikipedia.org/wiki/IPv6_address#Link-local_addresses_and_zone_indices
+[RFC 4007]: https://tools.ietf.org/html/rfc4007

lib/dgram.js

@@ -493,6 +493,19 @@ Socket.prototype.setMulticastLoopback = function(arg) {
 };
 
 
+Socket.prototype.setMulticastInterface = function(interfaceAddress) {
+  this._healthCheck();
+
+  if (typeof interfaceAddress !== 'string') {
+    throw new TypeError('"interfaceAddress" argument must be a string');
+  }
+
+  const err = this._handle.setMulticastInterface(interfaceAddress);
+  if (err) {
+    throw errnoException(err, 'setMulticastInterface');
+  }
+};
+
 Socket.prototype.addMembership = function(multicastAddress,
                                           interfaceAddress) {
   this._healthCheck();

src/udp_wrap.cc

@@ -97,6 +97,7 @@ void UDPWrap::Initialize(Local<Object> target,
                       GetSockOrPeerName<UDPWrap, uv_udp_getsockname>);
   env->SetProtoMethod(t, "addMembership", AddMembership);
   env->SetProtoMethod(t, "dropMembership", DropMembership);
+  env->SetProtoMethod(t, "setMulticastInterface", SetMulticastInterface);
   env->SetProtoMethod(t, "setMulticastTTL", SetMulticastTTL);
   env->SetProtoMethod(t, "setMulticastLoopback", SetMulticastLoopback);
   env->SetProtoMethod(t, "setBroadcast", SetBroadcast);
@@ -208,6 +209,22 @@ X(SetMulticastLoopback, uv_udp_set_multicast_loop)
 
 #undef X
 
+void UDPWrap::SetMulticastInterface(const FunctionCallbackInfo<Value>& args) {
+  UDPWrap* wrap;
+  ASSIGN_OR_RETURN_UNWRAP(&wrap,
+                          args.Holder(),
+                          args.GetReturnValue().Set(UV_EBADF));
+
+  CHECK_EQ(args.Length(), 1);
+  CHECK(args[0]->IsString());
+
+  Utf8Value iface(args.GetIsolate(), args[0]);
+
+  const char* iface_cstr = *iface;
+
+  int err = uv_udp_set_multicast_interface(&wrap->handle_, iface_cstr);
+  args.GetReturnValue().Set(err);
+}
 
 void UDPWrap::SetMembership(const FunctionCallbackInfo<Value>& args,
                             uv_membership membership) {

src/udp_wrap.h

@@ -28,6 +28,8 @@ class UDPWrap: public HandleWrap {
   static void RecvStop(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void AddMembership(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void DropMembership(const v8::FunctionCallbackInfo<v8::Value>& args);
+  static void SetMulticastInterface(
+      const v8::FunctionCallbackInfo<v8::Value>& args);
   static void SetMulticastTTL(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void SetMulticastLoopback(
       const v8::FunctionCallbackInfo<v8::Value>& args);

test/internet/test-dgram-multicast-multi-process.js

@@ -8,6 +8,7 @@ const assert = require('assert');
 const dgram = require('dgram');
 const fork = require('child_process').fork;
 const LOCAL_BROADCAST_HOST = '224.0.0.114';
+const LOCAL_HOST_IFADDR = '0.0.0.0';
 const TIMEOUT = common.platformTimeout(5000);
 const messages = [
   Buffer.from('First message to send'),
@@ -136,6 +137,7 @@ if (process.argv[2] !== 'child') {
     sendSocket.setBroadcast(true);
     sendSocket.setMulticastTTL(1);
     sendSocket.setMulticastLoopback(true);
+    sendSocket.setMulticastInterface(LOCAL_HOST_IFADDR);
   });
 
   sendSocket.on('close', function() {
@@ -175,7 +177,7 @@ if (process.argv[2] === 'child') {
   });
 
   listenSocket.on('listening', function() {
-    listenSocket.addMembership(LOCAL_BROADCAST_HOST);
+    listenSocket.addMembership(LOCAL_BROADCAST_HOST, LOCAL_HOST_IFADDR);
 
     listenSocket.on('message', function(buf, rinfo) {
       console.error('[CHILD] %s received "%s" from %j', process.pid,

test/parallel/test-dgram-multicast-set-interface-lo.js

@@ -0,0 +1,293 @@
+'use strict';
+const common = require('../common');
+const assert = require('assert');
+const dgram = require('dgram');
+const util = require('util');
+
+if (common.inFreeBSDJail) {
+  common.skip('in a FreeBSD jail');
+  return;
+}
+
+// All SunOS systems must be able to pass this manual test before the
+//   following barrier can be removed:
+// $ socat UDP-RECVFROM:12356,ip-add-membership=224.0.0.115:127.0.0.1,fork \
+//   EXEC:hostname &
+// $ echo hi |socat STDIO \
+//   UDP4-DATAGRAM:224.0.0.115:12356,ip-multicast-if=127.0.0.1
+
+if (common.isSunOS) {
+  common.skip('SunOs is not correctly delivering to loopback multicast.');
+  return;
+}
+
+const networkInterfaces = require('os').networkInterfaces();
+const Buffer = require('buffer').Buffer;
+const fork = require('child_process').fork;
+const MULTICASTS = {
+  IPv4: ['224.0.0.115', '224.0.0.116', '224.0.0.117'],
+  IPv6: ['ff02::1:115', 'ff02::1:116', 'ff02::1:117']
+};
+const LOOPBACK = { IPv4: '127.0.0.1', IPv6: '::1' };
+const ANY = { IPv4: '0.0.0.0', IPv6: '::' };
+const FAM = 'IPv4';
+
+// Windows wont bind on multicasts so its filtering is by port.
+const PORTS = {};
+for (let i = 0; i < MULTICASTS[FAM].length; i++) {
+  PORTS[MULTICASTS[FAM][i]] = common.PORT + (common.isWindows ? i : 0);
+}
+
+const UDP = { IPv4: 'udp4', IPv6: 'udp6' };
+
+const TIMEOUT = common.platformTimeout(5000);
+const NOW = Date.now();
+const TMPL = (tail) => `${NOW} - ${tail}`;
+
+// Take the first non-internal interface as the other interface to isolate
+// from loopback. Ideally, this should check for whether or not this interface
+// and the loopback have the MULTICAST flag.
+const interfaceAddress = ((networkInterfaces) => {
+  for (const name in networkInterfaces) {
+    for (const localInterface of networkInterfaces[name]) {
+      if (!localInterface.internal && localInterface.family === FAM) {
+        let interfaceAddress = localInterface.address;
+        // On Windows, IPv6 would need: `%${localInterface.scopeid}`
+        if (FAM === 'IPv6')
+          interfaceAddress += `${interfaceAddress}%${name}`;
+        return interfaceAddress;
+      }
+    }
+  }
+})(networkInterfaces);
+
+assert.ok(interfaceAddress);
+
+const messages = [
+  { tail: 'First message to send', mcast: MULTICASTS[FAM][0], rcv: true },
+  { tail: 'Second message to send', mcast: MULTICASTS[FAM][0], rcv: true },
+  { tail: 'Third message to send', mcast: MULTICASTS[FAM][1], rcv: true,
+    newAddr: interfaceAddress },
+  { tail: 'Fourth message to send', mcast: MULTICASTS[FAM][2] },
+  { tail: 'Fifth message to send', mcast: MULTICASTS[FAM][1], rcv: true },
+  { tail: 'Sixth message to send', mcast: MULTICASTS[FAM][2], rcv: true,
+    newAddr: LOOPBACK[FAM] }
+];
+
+
+if (process.argv[2] !== 'child') {
+  const IFACES = [ANY[FAM], interfaceAddress, LOOPBACK[FAM]];
+  const workers = {};
+  const listeners = MULTICASTS[FAM].length * 2;
+  let listening = 0;
+  let dead = 0;
+  let i = 0;
+  let done = 0;
+  let timer = null;
+  // Exit the test if it doesn't succeed within the TIMEOUT.
+  timer = setTimeout(function() {
+    console.error('[PARENT] Responses were not received within %d ms.',
+                  TIMEOUT);
+    console.error('[PARENT] Skip');
+
+    killChildren(workers);
+    common.skip('Check filter policy');
+
+    process.exit(1);
+  }, TIMEOUT);
+
+  // Launch the child processes.
+  for (let i = 0; i < listeners; i++) {
+    const IFACE = IFACES[i % IFACES.length];
+    const MULTICAST = MULTICASTS[FAM][i % MULTICASTS[FAM].length];
+
+    const messagesNeeded = messages.filter((m) => m.rcv &&
+                                                  m.mcast === MULTICAST)
+                                   .map((m) => TMPL(m.tail));
+    const worker = fork(process.argv[1],
+                        ['child',
+                         IFACE,
+                         MULTICAST,
+                         messagesNeeded.length,
+                         NOW]);
+    workers[worker.pid] = worker;
+
+    worker.messagesReceived = [];
+    worker.messagesNeeded = messagesNeeded;
+
+    // Handle the death of workers.
+    worker.on('exit', function(code, signal) {
+      // Don't consider this a true death if the worker has finished
+      // successfully or if the exit code is 0.
+      if (worker.isDone || code === 0) {
+        return;
+      }
+
+      dead += 1;
+      console.error('[PARENT] Worker %d died. %d dead of %d',
+                    worker.pid,
+                    dead,
+                    listeners);
+
+      if (dead === listeners) {
+        console.error('[PARENT] All workers have died.');
+        console.error('[PARENT] Fail');
+
+        killChildren(workers);
+
+        process.exit(1);
+      }
+    });
+
+    worker.on('message', function(msg) {
+      if (msg.listening) {
+        listening += 1;
+
+        if (listening === listeners) {
+          // All child process are listening, so start sending.
+          sendSocket.sendNext();
+        }
+      } else if (msg.message) {
+        worker.messagesReceived.push(msg.message);
+
+        if (worker.messagesReceived.length === worker.messagesNeeded.length) {
+          done += 1;
+          worker.isDone = true;
+          console.error('[PARENT] %d received %d messages total.',
+                        worker.pid,
+                        worker.messagesReceived.length);
+        }
+
+        if (done === listeners) {
+          console.error('[PARENT] All workers have received the ' +
+                        'required number of ' +
+                        'messages. Will now compare.');
+
+          Object.keys(workers).forEach(function(pid) {
+            const worker = workers[pid];
+
+            let count = 0;
+
+            worker.messagesReceived.forEach(function(buf) {
+              for (let i = 0; i < worker.messagesNeeded.length; ++i) {
+                if (buf.toString() === worker.messagesNeeded[i]) {
+                  count++;
+                  break;
+                }
+              }
+            });
+
+            console.error('[PARENT] %d received %d matching messages.',
+                          worker.pid,
+                          count);
+
+            assert.strictEqual(count, worker.messagesNeeded.length,
+                               'A worker received ' +
+                               'an invalid multicast message');
+          });
+
+          clearTimeout(timer);
+          console.error('[PARENT] Success');
+          killChildren(workers);
+        }
+      }
+    });
+  }
+
+  const sendSocket = dgram.createSocket({
+    type: UDP[FAM],
+    reuseAddr: true
+  });
+
+  // Don't bind the address explicitly when sending and start with
+  // the OSes default multicast interface selection.
+  sendSocket.bind(common.PORT, ANY[FAM]);
+  sendSocket.on('listening', function() {
+    console.error(`outgoing iface ${interfaceAddress}`);
+  });
+
+  sendSocket.on('close', function() {
+    console.error('[PARENT] sendSocket closed');
+  });
+
+  sendSocket.sendNext = function() {
+    const msg = messages[i++];
+
+    if (!msg) {
+      sendSocket.close();
+      return;
+    }
+    console.error(TMPL(NOW, msg.tail));
+    const buf = Buffer.from(TMPL(msg.tail));
+    if (msg.newAddr) {
+      console.error(`changing outgoing multicast ${msg.newAddr}`);
+      sendSocket.setMulticastInterface(msg.newAddr);
+    }
+    sendSocket.send(
+      buf,
+      0,
+      buf.length,
+      PORTS[msg.mcast],
+      msg.mcast,
+      function(err) {
+        assert.ifError(err);
+        console.error('[PARENT] sent %s to %s:%s',
+                      util.inspect(buf.toString()),
+                      msg.mcast, PORTS[msg.mcast]);
+
+        process.nextTick(sendSocket.sendNext);
+      }
+    );
+  };
+
+  function killChildren(children) {
+    for (const i in children)
+      children[i].kill();
+  }
+}
+
+if (process.argv[2] === 'child') {
+  const IFACE = process.argv[3];
+  const MULTICAST = process.argv[4];
+  const NEEDEDMSGS = Number(process.argv[5]);
+  const SESSION = Number(process.argv[6]);
+  const receivedMessages = [];
+
+  console.error(`pid ${process.pid} iface ${IFACE} MULTICAST ${MULTICAST}`);
+  const listenSocket = dgram.createSocket({
+    type: UDP[FAM],
+    reuseAddr: true
+  });
+
+  listenSocket.on('message', function(buf, rinfo) {
+    // Examine udp messages only when they were sent by the parent.
+    if (!buf.toString().startsWith(SESSION)) return;
+
+    console.error('[CHILD] %s received %s from %j',
+                  process.pid,
+                  util.inspect(buf.toString()),
+                  rinfo);
+
+    receivedMessages.push(buf);
+
+    let closecb;
+
+    if (receivedMessages.length === NEEDEDMSGS) {
+      listenSocket.close();
+      closecb = () => process.exit();
+    }
+
+    process.send({ message: buf.toString() }, closecb);
+  });
+
+
+  listenSocket.on('listening', function() {
+    listenSocket.addMembership(MULTICAST, IFACE);
+    process.send({ listening: true });
+  });
+
+  if (common.isWindows)
+    listenSocket.bind(PORTS[MULTICAST], ANY[FAM]);
+  else
+    listenSocket.bind(common.PORT, MULTICAST);
+}

test/parallel/test-dgram-multicast-set-interface.js

@@ -0,0 +1,119 @@
+'use strict';
+const common = require('../common');
+const assert = require('assert');
+const dgram = require('dgram');
+
+{
+  const socket = dgram.createSocket('udp4');
+
+  socket.bind(0);
+  socket.on('listening', common.mustCall(() => {
+    // Explicitly request default system selection
+    socket.setMulticastInterface('0.0.0.0');
+
+    socket.close();
+  }));
+}
+
+{
+  const socket = dgram.createSocket('udp4');
+
+  socket.bind(0);
+  socket.on('listening', common.mustCall(() => {
+    socket.close(common.mustCall(() => {
+      assert.throws(() => { socket.setMulticastInterface('0.0.0.0'); },
+                    /Not running/);
+    }));
+  }));
+}
+
+{
+  const socket = dgram.createSocket('udp4');
+
+  socket.bind(0);
+  socket.on('listening', common.mustCall(() => {
+    // Try to set with an invalid interfaceAddress (wrong address class)
+    try {
+      socket.setMulticastInterface('::');
+      throw new Error('Not detected.');
+    } catch (e) {
+      console.error(`setMulticastInterface: wrong family error is: ${e}`);
+    }
+
+    socket.close();
+  }));
+}
+
+{
+  const socket = dgram.createSocket('udp4');
+
+  socket.bind(0);
+  socket.on('listening', common.mustCall(() => {
+    // Try to set with an invalid interfaceAddress (wrong Type)
+    assert.throws(() => {
+      socket.setMulticastInterface(1);
+    }, /TypeError/);
+
+    socket.close();
+  }));
+}
+
+{
+  const socket = dgram.createSocket('udp4');
+
+  socket.bind(0);
+  socket.on('listening', common.mustCall(() => {
+    // Try to set with an invalid interfaceAddress (non-unicast)
+    assert.throws(() => {
+      socket.setMulticastInterface('224.0.0.2');
+    }, /Error/);
+
+    socket.close();
+  }));
+}
+
+if (!common.hasIPv6) {
+  common.skip('Skipping udp6 tests, no IPv6 support.');
+  return;
+}
+
+{
+  const socket = dgram.createSocket('udp6');
+
+  socket.bind(0);
+  socket.on('listening', common.mustCall(() => {
+    // Try to set with an invalid interfaceAddress ('undefined')
+    assert.throws(() => {
+      socket.setMulticastInterface(String(undefined));
+    }, /EINVAL/);
+
+    socket.close();
+  }));
+}
+
+{
+  const socket = dgram.createSocket('udp6');
+
+  socket.bind(0);
+  socket.on('listening', common.mustCall(() => {
+    // Try to set with an invalid interfaceAddress ('')
+    assert.throws(() => {
+      socket.setMulticastInterface('');
+    }, /EINVAL/);
+
+    socket.close();
+  }));
+}
+
+{
+  const socket = dgram.createSocket('udp6');
+
+  socket.bind(0);
+  socket.on('listening', common.mustCall(() => {
+    // Using lo0 for OsX, on all other OSes, an invalid Scope gets
+    // turned into #0 (default selection) which is also acceptable.
+    socket.setMulticastInterface('::%lo0');
+
+    socket.close();
+  }));
+}