dgram: add source-specific multicast support

This adds RFC 4607 support for IPv4 and IPv6.

Co-Authored-By: Nicolas Thumann <46975855+n-thumann@users.noreply.github.com>
Co-Authored-By: Rich Trott <rtrott@gmail.com>
PR-URL: https://github.com/nodejs/node/pull/15735
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: James M Snell <jasnell@gmail.com>

doc/api/dgram.md

@@ -123,6 +123,21 @@ if (cluster.isMaster) {
 }
 ```
 
+### socket.addSourceSpecificMembership(sourceAddress, groupAddress\[, multicastInterface\])
+<!-- YAML
+added: REPLACEME
+-->
+* `sourceAddress` {string}
+* `groupAddress` {string}
+* `multicastInterface` {string}
+
+Tells the kernel to join a source-specific multicast channel at the given
+`sourceAddress` and `groupAddress`, using the `multicastInterface` with the
+`IP_ADD_SOURCE_MEMBERSHIP` socket option. If the `multicastInterface` argument
+is not specified, the operating system will choose one interface and will add
+membership to it. To add membership to every available interface, call
+`socket.addSourceSpecificMembership()` multiple times, once per interface.
+
 ### socket.address()
 <!-- YAML
 added: v0.1.99
@@ -297,6 +312,24 @@ never have reason to call this.
 If `multicastInterface` is not specified, the operating system will attempt to
 drop membership on all valid interfaces.
 
+### socket.dropSourceSpecificMembership(sourceAddress, groupAddress\[, multicastInterface\])
+<!-- YAML
+added: REPLACEME
+-->
+
+* `sourceAddress` {string}
+* `groupAddress` {string}
+* `multicastInterface` {string}
+
+Instructs the kernel to leave a source-specific multicast channel at the given
+`sourceAddress` and `groupAddress` using the `IP_DROP_SOURCE_MEMBERSHIP`
+socket option. This method is automatically called by the kernel when the
+socket is closed or the process terminates, so most apps will never have
+reason to call this.
+
+If `multicastInterface` is not specified, the operating system will attempt to
+drop membership on all valid interfaces.
+
 ### socket.getRecvBufferSize()
 <!-- YAML
 added: v8.7.0

lib/dgram.js

@@ -832,6 +832,55 @@ Socket.prototype.dropMembership = function(multicastAddress,
   }
 };
 
+Socket.prototype.addSourceSpecificMembership = function(sourceAddress,
+                                                        groupAddress,
+                                                        interfaceAddress) {
+  healthCheck(this);
+
+  if (typeof sourceAddress !== 'string') {
+    throw new errors.TypeError('ERR_INVALID_ARG_TYPE', 'sourceAddress',
+                               'string');
+  }
+
+  if (typeof groupAddress !== 'string') {
+    throw new errors.TypeError('ERR_INVALID_ARG_TYPE', 'groupAddress',
+                               'string');
+  }
+
+  const err =
+    this[kStateSymbol].handle.addSourceSpecificMembership(sourceAddress,
+                                                          groupAddress,
+                                                          interfaceAddress);
+  if (err) {
+    throw errnoException(err, 'addSourceSpecificMembership');
+  }
+};
+
+
+Socket.prototype.dropSourceSpecificMembership = function(sourceAddress,
+                                                         groupAddress,
+                                                         interfaceAddress) {
+  healthCheck(this);
+
+  if (typeof sourceAddress !== 'string') {
+    throw new errors.TypeError('ERR_INVALID_ARG_TYPE', 'sourceAddress',
+                               'string');
+  }
+
+  if (typeof groupAddress !== 'string') {
+    throw new errors.TypeError('ERR_INVALID_ARG_TYPE', 'groupAddress',
+                               'string');
+  }
+
+  const err =
+    this[kStateSymbol].handle.dropSourceSpecificMembership(sourceAddress,
+                                                           groupAddress,
+                                                           interfaceAddress);
+  if (err) {
+    throw errnoException(err, 'dropSourceSpecificMembership');
+  }
+};
+
 
 function healthCheck(socket) {
   if (!socket[kStateSymbol].handle) {

src/udp_wrap.cc

@@ -128,6 +128,10 @@ void UDPWrap::Initialize(Local<Object> target,
                       GetSockOrPeerName<UDPWrap, uv_udp_getsockname>);
   env->SetProtoMethod(t, "addMembership", AddMembership);
   env->SetProtoMethod(t, "dropMembership", DropMembership);
+  env->SetProtoMethod(t, "addSourceSpecificMembership",
+                      AddSourceSpecificMembership);
+  env->SetProtoMethod(t, "dropSourceSpecificMembership",
+                      DropSourceSpecificMembership);
   env->SetProtoMethod(t, "setMulticastInterface", SetMulticastInterface);
   env->SetProtoMethod(t, "setMulticastTTL", SetMulticastTTL);
   env->SetProtoMethod(t, "setMulticastLoopback", SetMulticastLoopback);
@@ -397,6 +401,44 @@ void UDPWrap::DropMembership(const FunctionCallbackInfo<Value>& args) {
   SetMembership(args, UV_LEAVE_GROUP);
 }
 
+void UDPWrap::SetSourceMembership(const FunctionCallbackInfo<Value>& args,
+                                  uv_membership membership) {
+  UDPWrap* wrap;
+  ASSIGN_OR_RETURN_UNWRAP(&wrap,
+                          args.Holder(),
+                          args.GetReturnValue().Set(UV_EBADF));
+
+  CHECK_EQ(args.Length(), 3);
+
+  node::Utf8Value source_address(args.GetIsolate(), args[0]);
+  node::Utf8Value group_address(args.GetIsolate(), args[1]);
+  node::Utf8Value iface(args.GetIsolate(), args[2]);
+
+  if (*iface == nullptr) return;
+  const char* iface_cstr = *iface;
+  if (args[2]->IsUndefined() || args[2]->IsNull()) {
+    iface_cstr = nullptr;
+  }
+
+  int err = uv_udp_set_source_membership(&wrap->handle_,
+                                         *group_address,
+                                         iface_cstr,
+                                         *source_address,
+                                         membership);
+  args.GetReturnValue().Set(err);
+}
+
+void UDPWrap::AddSourceSpecificMembership(
+  const FunctionCallbackInfo<Value>& args) {
+  SetSourceMembership(args, UV_JOIN_GROUP);
+}
+
+
+void UDPWrap::DropSourceSpecificMembership(
+  const FunctionCallbackInfo<Value>& args) {
+  SetSourceMembership(args, UV_LEAVE_GROUP);
+}
+
 
 void UDPWrap::DoSend(const FunctionCallbackInfo<Value>& args, int family) {
   Environment* env = Environment::GetCurrent(args);

src/udp_wrap.h

@@ -55,6 +55,10 @@ 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 AddSourceSpecificMembership(
+      const v8::FunctionCallbackInfo<v8::Value>& args);
+  static void DropSourceSpecificMembership(
+      const v8::FunctionCallbackInfo<v8::Value>& args);
   static void SetMulticastInterface(
       const v8::FunctionCallbackInfo<v8::Value>& args);
   static void SetMulticastTTL(const v8::FunctionCallbackInfo<v8::Value>& args);
@@ -88,6 +92,9 @@ class UDPWrap: public HandleWrap {
                      int family);
   static void SetMembership(const v8::FunctionCallbackInfo<v8::Value>& args,
                             uv_membership membership);
+  static void SetSourceMembership(
+      const v8::FunctionCallbackInfo<v8::Value>& args,
+      uv_membership membership);
 
   static void OnAlloc(uv_handle_t* handle,
                       size_t suggested_size,

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

@@ -0,0 +1,229 @@
+'use strict';
+const common = require('../common');
+// Skip test in FreeBSD jails.
+if (common.inFreeBSDJail)
+  common.skip('In a FreeBSD jail');
+
+const assert = require('assert');
+const dgram = require('dgram');
+const fork = require('child_process').fork;
+const networkInterfaces = require('os').networkInterfaces();
+const GROUP_ADDRESS = '232.1.1.1';
+const TIMEOUT = common.platformTimeout(5000);
+const messages = [
+  Buffer.from('First message to send'),
+  Buffer.from('Second message to send'),
+  Buffer.from('Third message to send'),
+  Buffer.from('Fourth message to send')
+];
+const workers = {};
+const listeners = 3;
+let listening, sendSocket, done, timer, dead;
+
+let sourceAddress = null;
+
+// Take the first non-internal interface as the IPv4 address for binding.
+// Ideally, this should favor internal/private interfaces.
+get_sourceAddress: for (const name in networkInterfaces) {
+  const interfaces = networkInterfaces[name];
+  for (let i = 0; i < interfaces.length; i++) {
+    const localInterface = interfaces[i];
+    if (!localInterface.internal && localInterface.family === 'IPv4') {
+      sourceAddress = localInterface.address;
+      break get_sourceAddress;
+    }
+  }
+}
+assert.ok(sourceAddress);
+
+function launchChildProcess() {
+  const worker = fork(__filename, ['child']);
+  workers[worker.pid] = worker;
+
+  worker.messagesReceived = [];
+
+  // Handle the death of workers.
+  worker.on('exit', function(code) {
+    // Don't consider this the 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');
+      assert.fail();
+    }
+  });
+
+  worker.on('message', function(msg) {
+    if (msg.listening) {
+      listening += 1;
+
+      if (listening === listeners) {
+        // All child process are listening, so start sending.
+        sendSocket.sendNext();
+      }
+      return;
+    }
+    if (msg.message) {
+      worker.messagesReceived.push(msg.message);
+
+      if (worker.messagesReceived.length === messages.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 < messages.length; ++i) {
+              if (buf.toString() === messages[i].toString()) {
+                count++;
+                break;
+              }
+            }
+          });
+
+          console.error('[PARENT] %d received %d matching messages.',
+                        worker.pid, count);
+
+          assert.strictEqual(count, messages.length);
+        });
+
+        clearTimeout(timer);
+        console.error('[PARENT] Success');
+        killChildren(workers);
+      }
+    }
+  });
+}
+
+function killChildren(children) {
+  Object.keys(children).forEach(function(key) {
+    const child = children[key];
+    child.kill();
+  });
+}
+
+if (process.argv[2] !== 'child') {
+  listening = 0;
+  dead = 0;
+  let i = 0;
+  done = 0;
+
+  // Exit the test if it doesn't succeed within TIMEOUT.
+  timer = setTimeout(function() {
+    console.error('[PARENT] Responses were not received within %d ms.',
+                  TIMEOUT);
+    console.error('[PARENT] Fail');
+
+    killChildren(workers);
+
+    assert.fail();
+  }, TIMEOUT);
+
+  // Launch child processes.
+  for (let x = 0; x < listeners; x++) {
+    launchChildProcess(x);
+  }
+
+  sendSocket = dgram.createSocket('udp4');
+
+  // The socket is actually created async now.
+  sendSocket.on('listening', function() {
+    sendSocket.setTTL(1);
+    sendSocket.setBroadcast(true);
+    sendSocket.setMulticastTTL(1);
+    sendSocket.setMulticastLoopback(true);
+    sendSocket.addSourceSpecificMembership(sourceAddress, GROUP_ADDRESS);
+  });
+
+  sendSocket.on('close', function() {
+    console.error('[PARENT] sendSocket closed');
+  });
+
+  sendSocket.sendNext = function() {
+    const buf = messages[i++];
+
+    if (!buf) {
+      try { sendSocket.close(); } catch {}
+      return;
+    }
+
+    sendSocket.send(
+      buf,
+      0,
+      buf.length,
+      common.PORT,
+      GROUP_ADDRESS,
+      function(err) {
+        assert.ifError(err);
+        console.error('[PARENT] sent "%s" to %s:%s',
+                      buf.toString(),
+                      GROUP_ADDRESS, common.PORT);
+        process.nextTick(sendSocket.sendNext);
+      }
+    );
+  };
+}
+
+if (process.argv[2] === 'child') {
+  const receivedMessages = [];
+  const listenSocket = dgram.createSocket({
+    type: 'udp4',
+    reuseAddr: true
+  });
+
+  listenSocket.on('listening', function() {
+    listenSocket.setMulticastLoopback(true);
+    listenSocket.addSourceSpecificMembership(sourceAddress, GROUP_ADDRESS);
+
+    listenSocket.on('message', function(buf, rinfo) {
+      console.error('[CHILD] %s received "%s" from %j', process.pid,
+                    buf.toString(), rinfo);
+
+      receivedMessages.push(buf);
+
+      process.send({ message: buf.toString() });
+
+      if (receivedMessages.length === messages.length) {
+        // .dropSourceSpecificMembership() not strictly needed,
+        // it is here as a sanity check.
+        listenSocket.dropSourceSpecificMembership(sourceAddress, GROUP_ADDRESS);
+        process.nextTick(function() {
+          listenSocket.close();
+        });
+      }
+    });
+
+    listenSocket.on('close', function() {
+      // HACK: Wait to exit the process to ensure that the parent
+      // process has had time to receive all messages via process.send()
+      // This may be indicative of some other issue.
+      setTimeout(function() {
+        process.exit();
+      }, common.platformTimeout(1000));
+    });
+    process.send({ listening: true });
+  });
+
+  listenSocket.bind(common.PORT);
+}

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

@@ -0,0 +1,229 @@
+'use strict';
+const common = require('../common');
+// Skip test in FreeBSD jails.
+if (common.inFreeBSDJail)
+  common.skip('In a FreeBSD jail');
+
+const assert = require('assert');
+const dgram = require('dgram');
+const fork = require('child_process').fork;
+const networkInterfaces = require('os').networkInterfaces();
+const GROUP_ADDRESS = 'ff3e::1234';
+const TIMEOUT = common.platformTimeout(5000);
+const messages = [
+  Buffer.from('First message to send'),
+  Buffer.from('Second message to send'),
+  Buffer.from('Third message to send'),
+  Buffer.from('Fourth message to send')
+];
+const workers = {};
+const listeners = 3;
+let listening, sendSocket, done, timer, dead;
+
+let sourceAddress = null;
+
+// Take the first non-internal interface as the IPv6 address for binding.
+// Ideally, this should check favor internal/private interfaces.
+get_sourceAddress: for (const name in networkInterfaces) {
+  const interfaces = networkInterfaces[name];
+  for (let i = 0; i < interfaces.length; i++) {
+    const localInterface = interfaces[i];
+    if (!localInterface.internal && localInterface.family === 'IPv6') {
+      sourceAddress = localInterface.address;
+      break get_sourceAddress;
+    }
+  }
+}
+assert.ok(sourceAddress);
+
+function launchChildProcess() {
+  const worker = fork(__filename, ['child']);
+  workers[worker.pid] = worker;
+
+  worker.messagesReceived = [];
+
+  // Handle the death of workers.
+  worker.on('exit', function(code) {
+    // Don't consider this the 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');
+      assert.fail();
+    }
+  });
+
+  worker.on('message', function(msg) {
+    if (msg.listening) {
+      listening += 1;
+
+      if (listening === listeners) {
+        // All child process are listening, so start sending.
+        sendSocket.sendNext();
+      }
+      return;
+    }
+    if (msg.message) {
+      worker.messagesReceived.push(msg.message);
+
+      if (worker.messagesReceived.length === messages.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 < messages.length; ++i) {
+              if (buf.toString() === messages[i].toString()) {
+                count++;
+                break;
+              }
+            }
+          });
+
+          console.error('[PARENT] %d received %d matching messages.',
+                        worker.pid, count);
+
+          assert.strictEqual(count, messages.length);
+        });
+
+        clearTimeout(timer);
+        console.error('[PARENT] Success');
+        killChildren(workers);
+      }
+    }
+  });
+}
+
+function killChildren(children) {
+  Object.keys(children).forEach(function(key) {
+    const child = children[key];
+    child.kill();
+  });
+}
+
+if (process.argv[2] !== 'child') {
+  listening = 0;
+  dead = 0;
+  let i = 0;
+  done = 0;
+
+  // Exit the test if it doesn't succeed within TIMEOUT.
+  timer = setTimeout(function() {
+    console.error('[PARENT] Responses were not received within %d ms.',
+                  TIMEOUT);
+    console.error('[PARENT] Fail');
+
+    killChildren(workers);
+
+    assert.fail();
+  }, TIMEOUT);
+
+  // Launch child processes.
+  for (let x = 0; x < listeners; x++) {
+    launchChildProcess(x);
+  }
+
+  sendSocket = dgram.createSocket('udp6');
+
+  // The socket is actually created async now.
+  sendSocket.on('listening', function() {
+    sendSocket.setTTL(1);
+    sendSocket.setBroadcast(true);
+    sendSocket.setMulticastTTL(1);
+    sendSocket.setMulticastLoopback(true);
+    sendSocket.addSourceSpecificMembership(sourceAddress, GROUP_ADDRESS);
+  });
+
+  sendSocket.on('close', function() {
+    console.error('[PARENT] sendSocket closed');
+  });
+
+  sendSocket.sendNext = function() {
+    const buf = messages[i++];
+
+    if (!buf) {
+      try { sendSocket.close(); } catch {}
+      return;
+    }
+
+    sendSocket.send(
+      buf,
+      0,
+      buf.length,
+      common.PORT,
+      GROUP_ADDRESS,
+      function(err) {
+        assert.ifError(err);
+        console.error('[PARENT] sent "%s" to %s:%s',
+                      buf.toString(),
+                      GROUP_ADDRESS, common.PORT);
+        process.nextTick(sendSocket.sendNext);
+      }
+    );
+  };
+}
+
+if (process.argv[2] === 'child') {
+  const receivedMessages = [];
+  const listenSocket = dgram.createSocket({
+    type: 'udp6',
+    reuseAddr: true
+  });
+
+  listenSocket.on('listening', function() {
+    listenSocket.setMulticastLoopback(true);
+    listenSocket.addSourceSpecificMembership(sourceAddress, GROUP_ADDRESS);
+
+    listenSocket.on('message', function(buf, rinfo) {
+      console.error('[CHILD] %s received "%s" from %j', process.pid,
+                    buf.toString(), rinfo);
+
+      receivedMessages.push(buf);
+
+      process.send({ message: buf.toString() });
+
+      if (receivedMessages.length === messages.length) {
+        // .dropSourceSpecificMembership() not strictly needed,
+        // it is here as a sanity check.
+        listenSocket.dropSourceSpecificMembership(sourceAddress, GROUP_ADDRESS);
+        process.nextTick(function() {
+          listenSocket.close();
+        });
+      }
+    });
+
+    listenSocket.on('close', function() {
+      // HACK: Wait to exit the process to ensure that the parent
+      // process has had time to receive all messages via process.send()
+      // This may be indicative of some other issue.
+      setTimeout(function() {
+        process.exit();
+      }, common.platformTimeout(1000));
+    });
+    process.send({ listening: true });
+  });
+
+  listenSocket.bind(common.PORT);
+}