NODERAWSOCKETS: support UDP multicast group membership

src/lib/libsockfs_node.js

@@ -316,7 +316,7 @@ var NodeSockFSLibrary = {
     // never called bind() auto-binds to an ephemeral port here so getsockname
     // reports the assigned local address.
     startUdpRecv(sock) {
-      if (!sock.udp || sock.udpReceiving) return;
+      if (!sock.udp || sock.udpBound) return;
       if (sock.udpPublic) {
         if (sock.sport === undefined) {
           var a = sock.udp.bindSync({ address: sock.family === {{{ cDefs.AF_INET6 }}} ? '::' : '0.0.0.0', port: 0 });
@@ -333,10 +333,11 @@ var NodeSockFSLibrary = {
           }
         }
       }
-      sock.udpReceiving = true;
+      sock.udpBound = true;
       // node only honors these once the socket is bound, so (re)apply any
       // options that were set earlier.
       nodeSockHelpers.applyUdpOptions(sock);
+      nodeSockHelpers.applyMemberships(sock);
     },
     // Apply the buffered datagram options to a bound UDP socket, returning the
     // errno of the first rejected value (0 on success) so setsockopt can report
@@ -349,7 +350,7 @@ var NodeSockFSLibrary = {
     applyUdpOptions(sock) {
       var h = sock.udp;
       var o = sock.opts;
-      if (!h || !o || !sock.udpReceiving) return 0;
+      if (!h || !o || !sock.udpBound) return 0;
       var err = 0;
       // The public dgram setters throw on a bad value; the udp_wrap handle
       // setters instead return a negative libuv code. Capture both, and drop a
@@ -382,9 +383,40 @@ var NodeSockFSLibrary = {
       }
       return err;
     },
+    // Multicast joins are tracked so that one made before bind() can be
+    // applied once the socket is bound (libuv would otherwise auto-bind it to
+    // an ephemeral port). Returns an errno, or 0.
+    setMembership(sock, add, group, iface) {
+      var m = sock.memberships ??= new Map();
+      var key = group + '%' + (iface ?? '');
+      if (add && m.has(key)) return {{{ cDefs.EADDRINUSE }}};
+      if (!add && !m.has(key)) return {{{ cDefs.EADDRNOTAVAIL }}};
+      if (sock.udpBound) {
+        var err = nodeSockHelpers.udpMembership(sock, group, iface, add);
+        if (err) return err;
+      }
+      if (add) m.set(key, [group, iface]);
+      else m.delete(key);
+      return 0;
+    },
+    udpMembership(sock, group, iface, add = true) {
+      var h = sock.udp;
+      try {
+        var r = add ? h.addMembership(group, iface) : h.dropMembership(group, iface);
+        if (typeof r === 'number' && r < 0) return nodeSockHelpers.codeToErrno(r);
+      } catch (e) {
+        return nodeSockHelpers.nodeErrToErrno(e);
+      }
+      return 0;
+    },
+    applyMemberships(sock) {
+      for (var [group, iface] of sock.memberships?.values() ?? []) {
+        nodeSockHelpers.udpMembership(sock, group, iface);
+      }
+    },
     // The live OS buffer size from a bound UDP socket, or undefined.
     udpBufferSize(sock, recv) {
-      if (!sock.udp || !sock.udpReceiving) return undefined;
+      if (!sock.udp || !sock.udpBound) return undefined;
       try {
         if (sock.udpPublic) return recv ? sock.udp.getRecvBufferSize() : sock.udp.getSendBufferSize();
         return sock.udp.bufferSize(0, recv, {});
@@ -639,8 +671,10 @@ var NodeSockFSLibrary = {
           var a = udp.address();
           sock.saddr = a.address;
           sock.sport = a.port;
-          sock.udpReceiving = true; // a bound dgram socket already receives
+          var wasBound = sock.udpBound;
+          sock.udpBound = true; // a bound dgram socket already receives
           nodeSockHelpers.applyUdpOptions(sock);
+          if (!wasBound) nodeSockHelpers.applyMemberships(sock);
           return;
         }
         // Older node has no synchronous dgram connect, so just record the peer
@@ -911,6 +945,16 @@ var NodeSockFSLibrary = {
           case 34: // IP_MULTICAST_LOOP
             sock.opts.multicastLoop = !!val;
             return -nodeSockHelpers.applyUdpOptions(sock);
+          case 35: // IP_ADD_MEMBERSHIP (struct ip_mreq, or ip_mreqn)
+          case 36: { // IP_DROP_MEMBERSHIP
+            if (sock.type !== {{{ cDefs.SOCK_DGRAM }}}) return -{{{ cDefs.EPROTO }}};
+            if (optlen < 8) return -{{{ cDefs.EINVAL }}};
+            var iface = {{{ makeGetValue('optval', 4, 'i32') }}};
+            // node selects the interface by address only, so an ip_mreqn that
+            // names it by index alone cannot be honored.
+            if (!iface && optlen >= 12 && {{{ makeGetValue('optval', 8, 'i32') }}}) return -{{{ cDefs.ENODEV }}};
+            return -nodeSockHelpers.setMembership(sock, optname === 35, inetNtop4(val), iface ? inetNtop4(iface) : undefined);
+          }
         }
       } else if (level === {{{ cDefs.IPPROTO_IPV6 }}}) {
         switch (optname) {
@@ -927,6 +971,17 @@ var NodeSockFSLibrary = {
           case 19: // IPV6_MULTICAST_LOOP (same libuv setting as IP_MULTICAST_LOOP)
             sock.opts.multicastLoop = !!val;
             return -nodeSockHelpers.applyUdpOptions(sock);
+          case 20: // IPV6_JOIN_GROUP (struct ipv6_mreq)
+          case 21: { // IPV6_LEAVE_GROUP
+            if (sock.type !== {{{ cDefs.SOCK_DGRAM }}}) return -{{{ cDefs.EPROTO }}};
+            if (optlen < 20) return -{{{ cDefs.EINVAL }}};
+            // node selects the interface by name and has no index-to-name
+            // lookup, so only the default interface (index 0) is supported.
+            if ({{{ makeGetValue('optval', 16, 'i32') }}}) return -{{{ cDefs.ENODEV }}};
+            var src32 = {{{ getHeapOffset('optval', 'i32') }}};
+            var group = inetNtop6(HEAP32.subarray(src32, src32 + 4));
+            return -nodeSockHelpers.setMembership(sock, optname === 20, group, undefined);
+          }
         }
       } else if (level === {{{ cDefs.IPPROTO_TCP }}}) {
         switch (optname) {

test/sockets/test_udp_sockopts.c

@@ -10,7 +10,10 @@
  * default (TTL/HOPS 1, LOOP 1), and a set is observable by a subsequent get.
  * Then SO_REUSEADDR / SO_REUSEPORT, which take effect at bind: two sockets
  * with the option set may share a port, and a third without it is refused
- * with EADDRINUSE. Plain POSIX, also runs natively.
+ * with EADDRINUSE. Finally the set-only IP_ADD_MEMBERSHIP / IP_DROP_MEMBERSHIP
+ * and IPv6 equivalents reach the kernel: a duplicate join fails with
+ * EADDRINUSE, leaving an unjoined group with EADDRNOTAVAIL, and a join made
+ * before bind() is applied once bound. Plain POSIX, also runs natively.
  */
 
 #include <arpa/inet.h>
@@ -70,6 +73,58 @@ void check_reuse(int opt) {
   close(c);
 }
 
+int set4(int fd, int opt, const char* group) {
+  struct ip_mreq m = {0};
+  inet_pton(AF_INET, group, &m.imr_multiaddr);
+  inet_pton(AF_INET, "127.0.0.1", &m.imr_interface);
+  return setsockopt(fd, IPPROTO_IP, opt, &m, sizeof(m)) == 0 ? 0 : errno;
+}
+
+int set6(int fd, int opt, const char* group) {
+  struct ipv6_mreq m = {0};
+  inet_pton(AF_INET6, group, &m.ipv6mr_multiaddr);
+  return setsockopt(fd, IPPROTO_IPV6, opt, &m, sizeof(m)) == 0 ? 0 : errno;
+}
+
+void check_membership(void) {
+  int fd = socket(AF_INET, SOCK_DGRAM, 0);
+  assert(bind_any(fd, 0) == 0);
+  assert(set4(fd, IP_ADD_MEMBERSHIP, "227.1.1.100") == 0);
+  assert(set4(fd, IP_ADD_MEMBERSHIP, "227.1.1.101") == 0);
+  assert(set4(fd, IP_ADD_MEMBERSHIP, "227.1.1.100") == EADDRINUSE);
+  assert(set4(fd, IP_DROP_MEMBERSHIP, "227.1.1.100") == 0);
+  assert(set4(fd, IP_DROP_MEMBERSHIP, "227.1.1.100") == EADDRNOTAVAIL);
+
+  struct ip_mreq m;
+  socklen_t len = sizeof(m);
+  assert(getsockopt(fd, IPPROTO_IP, IP_ADD_MEMBERSHIP, &m, &len) == -1);
+  assert(errno == ENOPROTOOPT);
+  assert(setsockopt(fd, IPPROTO_IP, IP_ADD_MEMBERSHIP, &m, 4) == -1);
+  assert(errno == EINVAL);
+  close(fd);
+
+  fd = socket(AF_INET, SOCK_DGRAM, 0);
+  assert(set4(fd, IP_ADD_MEMBERSHIP, "227.1.1.100") == 0);
+  assert(set4(fd, IP_ADD_MEMBERSHIP, "227.1.1.100") == EADDRINUSE);
+  assert(bind_any(fd, 0) == 0);
+  assert(set4(fd, IP_ADD_MEMBERSHIP, "227.1.1.100") == EADDRINUSE);
+  assert(set4(fd, IP_DROP_MEMBERSHIP, "227.1.1.100") == 0);
+  close(fd);
+
+  fd = socket(AF_INET6, SOCK_DGRAM, 0);
+  struct sockaddr_in6 sa6 = {.sin6_family = AF_INET6};
+  assert(bind(fd, (struct sockaddr*)&sa6, sizeof(sa6)) == 0);
+  int err = set6(fd, IPV6_JOIN_GROUP, "ff02::1:3");
+  // ENODEV: no IPv6 multicast route on this host (e.g. a container).
+  if (err != ENODEV) {
+    assert(err == 0);
+    assert(set6(fd, IPV6_JOIN_GROUP, "ff02::1:3") == EADDRINUSE);
+    assert(set6(fd, IPV6_LEAVE_GROUP, "ff02::1:3") == 0);
+    assert(set6(fd, IPV6_LEAVE_GROUP, "ff02::1:3") == EADDRNOTAVAIL);
+  }
+  close(fd);
+}
+
 int main(void) {
   int v4 = socket(AF_INET, SOCK_DGRAM, 0);
   assert(v4 >= 0);
@@ -124,6 +179,7 @@ int main(void) {
 
   check_reuse(SO_REUSEADDR);
   check_reuse(SO_REUSEPORT);
+  check_membership();
   printf("done\n");
   return 0;
 }

test/test_sockets_node.py

@@ -323,7 +323,7 @@ def test_noderawsockets_udp_sockopts(self):
     # UDP socket options: IP_MULTICAST_TTL/LOOP and their IPv6 counterparts
     # round-trip through set/getsockopt, with POSIX defaults readable before
     # any set, and SO_REUSEADDR/SO_REUSEPORT are applied at bind so sockets can
-    # share a port. EXIT_RUNTIME so the plain synchronous main()
+    # share a port, and multicast group joins reach the kernel. EXIT_RUNTIME so the plain synchronous main()
     # tears down the proxy worker on return (otherwise noExitRuntime keeps the
     # worker, and thus node, alive under PROXY_TO_PTHREAD).
     self.do_runf('sockets/test_udp_sockopts.c', 'done\n', cflags=['-sNODERAWSOCKETS', '-sEXIT_RUNTIME'])