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>

1 files changed, 229 insertions, 0 deletions

diff --git a/test/internet/test-dgram-multicast-ssmv6-multi-process.js b/test/internet/test-dgram-multicast-ssmv6-multi-process.js
new file mode 100644
index 0000000000..88bb563364
--- /dev/null
+++ b/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);
+}