Add option to enable loop back on external interfaces

multicast_expert/rx_socket.py

@@ -18,7 +18,7 @@ class McastRxSocket:
     Class to wrap a socket that receives from one or more multicast groups.
     """
 
-    def __init__(self, addr_family: int, mcast_ips: List[str], port: int, iface_ip: Optional[str] = None, iface_ips: Optional[List[str]] = None, source_ips: Optional[List[str]] = None, timeout: Optional[float] = None, blocking: Optional[bool] = None):
+    def __init__(self, addr_family: int, mcast_ips: List[str], port: int, iface_ip: Optional[str] = None, iface_ips: Optional[List[str]] = None, source_ips: Optional[List[str]] = None, timeout: Optional[float] = None, blocking: Optional[bool] = None, enable_external_loopback: bool = False):
         """
         Create a socket which receives UDP datagrams over multicast.  The socket must be opened
         (e.g. using a with statement) before it can be used.
@@ -43,6 +43,7 @@ def __init__(self, addr_family: int, mcast_ips: List[str], port: int, iface_ip:
             this value is processed.
         :param blocking: Legacy alias for timeout.  If set to True, timeout is set to None (block forever).  If set to
             False, timeout is set to 0 (nonblocking).
+        :param enable_external_loopback: Enable receiving multicast packets sent over external interfaces.
         """
         self.addr_family = addr_family
         self.mcast_ips = mcast_ips
@@ -100,6 +101,8 @@ def __init__(self, addr_family: int, mcast_ips: List[str], port: int, iface_ip:
         self.is_source_specific = not (source_ips is None or len(source_ips) == 0)
         if self.is_source_specific and self.addr_family == socket.AF_INET6:
             raise MulticastExpertError("Source-specific multicast currently cannot be used with IPv6!")
+
+        self.enable_external_loopback = enable_external_loopback
 
     def __enter__(self) -> McastRxSocket:
         if self.is_opened:
@@ -126,7 +129,7 @@ def __enter__(self) -> McastRxSocket:
                 # On Windows, by default, sent packets are looped back to local sockets on the same interface, even for interfaces
                 # that are not loopback.  Change this by disabling IP_MULTICAST_LOOP unless the loopback interface is used.
                 # Note: multicast_expert submitted a PR to clarify this in the Windows docs, and it was accepted!
-                loop_enabled = (iface_ip == LOCALHOST_IPV4 or iface_ip == LOCALHOST_IPV6)
+                loop_enabled = self.enable_external_loopback or iface_ip == LOCALHOST_IPV4 or iface_ip == LOCALHOST_IPV6
                 if self.addr_family == socket.AF_INET:
                     new_socket.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_LOOP, loop_enabled)
                 else:

multicast_expert/tx_socket.py

@@ -16,7 +16,7 @@ class McastTxSocket:
     Class to wrap a socket that sends to one or more multicast groups.
     """
 
-    def __init__(self, addr_family: int, mcast_ips: List[str], iface_ip: Optional[str] = None, ttl: int = 1):
+    def __init__(self, addr_family: int, mcast_ips: List[str], iface_ip: Optional[str] = None, ttl: int = 1, enable_external_loopback: bool = False):
         """
         Create a socket which transmits UDP datagrams over multicast.  The socket must be opened
         (e.g. using a with statement) before it can be used.
@@ -30,6 +30,7 @@ def __init__(self, addr_family: int, mcast_ips: List[str], iface_ip: Optional[st
         :param ttl: Time-to-live parameter to set on the packets sent by this socket, AKA hop limit for ipv6.
             This controls the number of routers that the packets may pass through until they are dropped.  The
             default value of 1 prevents the multicast packets from passing through any routers.
+        :param enable_external_loopback: Enable loopback of multicast packets sent on external interfaces.
         """
 
         self.addr_family = addr_family
@@ -56,6 +57,8 @@ def __init__(self, addr_family: int, mcast_ips: List[str], iface_ip: Optional[st
         for mcast_ip in self.mcast_ips:
             validate_mcast_ip(mcast_ip, self.addr_family)
 
+        self.enable_external_loopback = enable_external_loopback
+
     def __enter__(self) -> McastTxSocket:
         if self.is_opened:
             raise MulticastExpertError("Attempt to open an McastTxSocket that is already open!")
@@ -84,7 +87,7 @@ def __enter__(self) -> McastTxSocket:
 
             # On Mac, we do want to keep loopback enabled but only on the loopback interface.
             # On Linux, always disable it.
-            enable_loopback = is_mac and (self.iface_ip == LOCALHOST_IPV4 or self.iface_ip == LOCALHOST_IPV6)
+            enable_loopback = self.enable_external_loopback or (is_mac and (self.iface_ip == LOCALHOST_IPV4 or self.iface_ip == LOCALHOST_IPV6))
 
             if self.addr_family == socket.AF_INET:
                 self.socket.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_LOOP, enable_loopback)

tests/test_multicast_expert.py

@@ -340,4 +340,43 @@ def test_v6_loopback_multiple() -> None:
                     print("\nRx: " + repr(packet_alt))
                     assert packet_alt is not None
                     assert packet_alt[0] == test_string_alternate
-                    assert packet_alt[1][0:2] == (multicast_expert.LOCALHOST_IPV6, mcast_tx_sock_alt.getsockname()[1])
+                    assert packet_alt[1][0:2] == (multicast_expert.LOCALHOST_IPV6, mcast_tx_sock_alt.getsockname()[1])
+
+
+def test_external_loopback_v4() -> None:
+    """
+    Check that packets sent over external interface can be received when `enable_external_loopback` is set.
+    """
+    with multicast_expert.McastTxSocket(socket.AF_INET,
+                                        mcast_ips=[mcast_address_v4],
+                                        enable_external_loopback=True) as tx_socket:
+        assert tx_socket.iface_ip is not None and tx_socket.iface_ip != multicast_expert.LOCALHOST_IPV4
+
+        with multicast_expert.McastRxSocket(socket.AF_INET,
+                                            mcast_ips=[mcast_address_v4],
+                                            iface_ips=[tx_socket.iface_ip],
+                                            port=port,
+                                            timeout=1,
+                                            enable_external_loopback=True) as rx_socket:
+            tx_socket.sendto(test_string, (mcast_address_v4, port))
+            data = rx_socket.recv()
+            assert data == test_string
+
+
+def test_external_loopback_v6(nonloopback_iface_ipv6: str) -> None:
+    """
+    Check that packets sent over external interface can be received when `enable_external_loopback` is set.
+    """
+    with multicast_expert.McastTxSocket(socket.AF_INET6,
+                                        mcast_ips=[mcast_address_v6],
+                                        iface_ip=nonloopback_iface_ipv6,
+                                        enable_external_loopback=True) as tx_socket:
+        with multicast_expert.McastRxSocket(socket.AF_INET6,
+                                            mcast_ips=[mcast_address_v6],
+                                            iface_ips=[nonloopback_iface_ipv6],
+                                            port=port,
+                                            timeout=1,
+                                            enable_external_loopback=True) as rx_socket:
+            tx_socket.sendto(test_string, (mcast_address_v6, port))
+            data = rx_socket.recv()
+            assert data == test_string