I'm writing in Python software that runs on my Windows (and Linux) PC, while the PC is connected via LAN (with or without a router) to a second device.

The second device sends UDP multicast packets to a known multicast group address and port. The software in the computer is configured to be part of the same multicast group. This is working OK as long as both my computer and the server network configuration are on the same sub-net.

Now, most of the times I will don't know the IP of the device in advance and I'll be connecting my computer directly to the server point-to-point. (Imagine the software that comes with IP security cameras that allows you to discover or know the IP of the camera when connecting directly to them with out knowing it in advance and without being in the same sub-net). E.g my computer has IP 169.x.x.x/24 and the server has IP 10.1.1.100 but I do not know the server IP in advance.

For reasons out of my control, the device cannot be configured to be a DHCP server so it cannot assign IP to my computer and cannot use DNS.

How can I receive the UDP multicast packets without raw capture?

This is my current code for the socket configuration that is working when both the computer and the server have the same sub-net. Ex 10.1.1.100/16 and 10.1.1.60/16 but needs to work also as mentioned above.

class MulticastSocket(object):
    """Sends UDP packets to multicast addressses."""

    def __init__(self, bind_ip=None):
        self._create(bind_ip)

    def _create(self, bind_ip):
        self.bind_addr = bind_ip

        # chosen arbitrary from IANA's Scoped Multicast Range
        # https://www.iana.org/assignments/multicast-addresses/multicast-addresses.xhtml#unicast-prefix-based
        self.multicast_group = '239.6.2.86'
        self.multicast_port = 6286

        """Creates a multicast UDP socket"""
        self._sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
        self._sock.settimeout(6)
        ttl = 2
        self._sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, ttl)
        
        self._sock.setsockopt(socket.SOL_IP, socket.IP_MULTICAST_IF, socket.inet_aton(self.bind_addr))
        membership_request = struct.pack('4s4s', socket.inet_aton(self.multicast_group),
                                         socket.inet_aton(self.bind_addr))
        self._sock.setsockopt(
            socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, membership_request)

        self._sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

        
        if sys.platform == 'linux':
            bind_addr = self.multicast_group
        else:
            bind_addr = self.bind_addr if self.bind_addr else ''
        self._sock.bind((bind_addr, self.multicast_port))

    def get(self):
        """use to get the socket object to work with the select as select only accept sockets objects"""
        return self._sock

    def send(self, msg):
        return self._sock.sendto(msg, (self.multicast_group, self.multicast_port))

    def recv(self, len):
        try:
            response = self._sock.recvfrom(len)
            return response
        except socket.timeout:
            return "", ""

    def close(self):
        self._sock.close()
0

There are 0 answers