From c47792ded1e99793d56f7563d5fdcbcd94b16a0b Mon Sep 17 00:00:00 2001 From: Steffen Christgau Date: Tue, 24 Dec 2024 00:14:09 +0100 Subject: [PATCH] refactor(src): streamline SunOS address enum, fix PEP8 issues --- src/wsdd.py | 45 ++++++++++++++++++++------------------------- 1 file changed, 20 insertions(+), 25 deletions(-) diff --git a/src/wsdd.py b/src/wsdd.py index a63f354..30a207f 100755 --- a/src/wsdd.py +++ b/src/wsdd.py @@ -297,8 +297,7 @@ def init_v4(self) -> None: self.uc_send_socket.bind((self.address.address_str, WSD_UDP_PORT)) if platform.system() == 'SunOS': - self.mc_send_socket.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_IF, - self.address.raw) + self.mc_send_socket.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_IF, self.address.raw) else: self.mc_send_socket.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_IF, mreq) # OpenBSD requires the optlen to be sizeof(char) for LOOP and TTL options @@ -1790,23 +1789,25 @@ def cleanup(self) -> None: self.socket.close() super().cleanup() + class DladmAddressMonitor(NetworkAddressMonitor): libsocket: ctypes.CDLL class sockaddr(ctypes.Structure): - _fields_ = [ ("family", ctypes.c_ushort), ("dum", ctypes.c_ushort), - ("data", ctypes.c_ubyte * 14)] + _fields_ = [("family", ctypes.c_ushort), + ("dummy", ctypes.c_ushort), + ("data", ctypes.c_ubyte * 14)] class ifaddrs(ctypes.Structure): pass ifaddrs._fields_ = [("next", ctypes.POINTER(ifaddrs)), - ("name", ctypes.c_char_p), - ("flags", ctypes.c_ulonglong), - ("addr", ctypes.POINTER(sockaddr)), - ("netmask", ctypes.POINTER(sockaddr)), - ("dstaddr", ctypes.POINTER(sockaddr)), - ("data", ctypes.c_void_p)] + ("name", ctypes.c_char_p), + ("flags", ctypes.c_ulonglong), + ("addr", ctypes.POINTER(sockaddr)), + ("netmask", ctypes.POINTER(sockaddr)), + ("dstaddr", ctypes.POINTER(sockaddr)), + ("data", ctypes.c_void_p)] def __init__(self, aio_loop: asyncio.AbstractEventLoop) -> None: super().__init__(aio_loop) @@ -1815,30 +1816,24 @@ def do_enumerate(self) -> None: super().do_enumerate() libsocket = ctypes.CDLL(ctypes.util.find_library('socket'), use_errno=True) nifas = self.ifaddrs() - if libsocket.getifaddrs(ctypes.byref(nifas))==0: + if libsocket.getifaddrs(ctypes.byref(nifas)) == 0: ifa = nifas ifa_idx = 0 while ifa.next: if ifa.name: - print("{}%{}".format(socket.inet_ntop(ifa.addr[0].family, bytes(ifa.addr[0].data[:4])), ifa.name.decode())) - addr_family = ifa.addr[0].family + logger.debug("{}%{}".format( + socket.inet_ntop(ifa.addr[0].family, bytes(ifa.addr[0].data[:4])), + ifa.name.decode())) addr = socket.inet_ntop(ifa.addr[0].family, bytes(ifa.addr[0].data[:4])) - if_name = ifa.name.decode() - intf = NetworkInterface(if_name, 0, ifa_idx) + intf = NetworkInterface(ifa.name.decode(), 0, ifa_idx) self.add_interface(intf) - address = NetworkAddress(addr_family, addr, intf) - self.handle_new_address(address) - ifa_idx += 1; + self.handle_new_address(NetworkAddress(ifa.addr[0].family, addr, intf)) + ifa_idx += 1 ifa = ifa.next[0] - else: - libsocket.freeifaddrs(nifas) - def cleanup(self) -> None: - self.aio_loop.remove_reader(self.socket.fileno()) - self.socket.close() - self.libsocket.freeifaddrs(self.ifaddrs()) - super().cleanup() + libsocket.freeifaddrs(nifas) + def sigterm_handler() -> None: logger.info('received termination/interrupt signal, tearing down')