nit: Format all code and remove unused libs

rawsocketpy/__init__.py

@@ -10,8 +10,10 @@
 
 from __future__ import absolute_import
 from __future__ import print_function
+
 try:
     from gevent import monkey
+
     monkey.patch_all()
     from .asyncserver import RawAsyncServer, RawAsyncServerCallback
 except ImportError:

rawsocketpy/asyncserver.py

@@ -13,6 +13,7 @@ class RawAsyncServer(RawServer):
 
     This will ensure you are not loosing data because the handler is too long.
     """
+
     pool = pool.Pool()
 
     def handle_handler(self, handler):

rawsocketpy/packet.py

@@ -5,7 +5,7 @@
 from .util import to_str
 
 
-class RawPacket():
+class RawPacket:
     """RawPacket is the resulting data container of the RawSocket class.
 
     It reads raw data and stores the MAC source, MAC destination, the Ethernet type and the data payload.
@@ -43,7 +43,17 @@ def __init__(self, data):
             self.success = False
 
     def __repr__(self):
-        return "".join([to_str(self.src), " == 0x", to_str(self.type, separator=""), " => ", to_str(self.dest), " - ", "OK" if self.success else "FAILED"])
+        return "".join(
+            [
+                to_str(self.src),
+                " == 0x",
+                to_str(self.type, separator=""),
+                " => ",
+                to_str(self.dest),
+                " - ",
+                "OK" if self.success else "FAILED",
+            ]
+        )
 
     def __str__(self):
-        return "".join([self.__repr__(), ":\n", self.data.decode('utf-8')])
+        return "".join([self.__repr__(), ":\n", self.data.decode("utf-8")])

rawsocketpy/server.py

@@ -2,9 +2,7 @@
 # -*- coding: utf-8 -*-
 
 from __future__ import absolute_import
-from .packet import RawPacket
 from .socket import RawSocket
-from .util import get_hw, to_bytes, protocol_to_ethertype
 
 
 class RawServer(object):

rawsocketpy/socket.py

@@ -3,18 +3,16 @@
 
 from __future__ import absolute_import
 import socket
-import select
-import struct
-import time
 from .packet import RawPacket
-from .util import get_hw, to_str, protocol_to_ethertype, to_bytes
+from .util import get_hw, protocol_to_ethertype, to_bytes
 
 
 class RawSocket(object):
     """RawSocket is using the socket library to send raw ethernet frames, using socket.RAW_SOCK
 
     It has a similar API to the socket library: send/recv/close/dup.
     """
+
     BROADCAST = b"\xff\xff\xff\xff\xff\xff"
     """:description: Default MAC address: ``"\\xff\\xff\\xff\\xff\\xff\\xff"``"""
 
@@ -47,9 +45,13 @@ def __init__(self, interface, protocol, sock=None, no_recv_protocol=False):
         self.close = self.sock.close
 
     def dup(self):
-        """Duplicates the RawSocket
-        """
-        return RawSocket(self.interface, self.non_processed_protocol, self.sock.dup(), self.no_recv_protocol)
+        """Duplicates the RawSocket"""
+        return RawSocket(
+            self.interface,
+            self.non_processed_protocol,
+            self.sock.dup(),
+            self.no_recv_protocol,
+        )
 
     @staticmethod
     def sock_create(interface, protocol, sock=None):

rawsocketpy/util.py

@@ -3,37 +3,56 @@
 
 from __future__ import absolute_import
 import socket
-import fcntl
 import struct
 import sys
 
-if sys.version_info >= (3, 0):
-    import binascii
+if sys.platform == "win32":
+    import psutil
 
     def get_hw(ifname):
-        """Returns a bytearray containing the MAC address of the interface.
+        addrs = psutil.net_if_addrs()
+        if ifname not in addrs:
+            raise ValueError(f"Interface '{ifname}' not found")
+
+        for snic in addrs[ifname]:
+            if snic.family == psutil.AF_LINK:
+                mac_str = snic.address.replace("-", ":")
+                mac_bytes = bytearray(int(b, 16) for b in mac_str.split(":"))
+                return mac_bytes
+
+        raise ValueError(f"No MAC address found for interface '{ifname}'")
 
-        :param ifname: Interface name such as ``wlp2s0``
-        :type ifname: str
-        :rtype: str
-        :rtype: bytearray
-        """
-        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-        info = fcntl.ioctl(s.fileno(), 0x8927,  struct.pack(
-            '256s',  bytearray(ifname[:15], 'utf-8')))
-        return info[18:24]
 else:
-    def get_hw(ifname):
-        """Returns a unicode string containing the MAC address of the interface.
+    import fcntl
 
-        :param ifname: Interface name such as ``wlp2s0``
-        :type ifname: str
-        :rtype: str
-        """
-        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-        info = fcntl.ioctl(s.fileno(), 0x8927,
-                           struct.pack('256s', ifname[:15]))
-        return info[18:24]
+    if sys.version_info >= (3, 0):
+
+        def get_hw(ifname):
+            """Returns a bytearray containing the MAC address of the interface.
+
+            :param ifname: Interface name such as ``wlp2s0``
+            :type ifname: str
+            :rtype: str
+            :rtype: bytearray
+            """
+            s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+            info = fcntl.ioctl(
+                s.fileno(), 0x8927, struct.pack("256s", bytearray(ifname[:15], "utf-8"))
+            )
+            return info[18:24]
+
+    else:
+
+        def get_hw(ifname):
+            """Returns a unicode string containing the MAC address of the interface.
+
+            :param ifname: Interface name such as ``wlp2s0``
+            :type ifname: str
+            :rtype: str
+            """
+            s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+            info = fcntl.ioctl(s.fileno(), 0x8927, struct.pack("256s", ifname[:15]))
+            return info[18:24]
 
 
 def to_str(data, separator=":"):
@@ -68,37 +87,42 @@ def protocol_to_ethertype(protocol):
 
 
 if sys.version_info >= (3, 0):
-  def to_bytes(*data):
-      """Flatten the arrays and Converts data to a bytearray
-
-      :param data: The data to be converted
-      :type data: [int, bytes, bytearray, str, [int], [bytes], [bytearray], [str]]
-      :rtype: bytearray
-
-      >>> to_bytes("123")
-      b'123'
-      >>> to_bytes(1, 2, 3)
-      b'\\x01\\x02\\x03'
-      >>> to_bytes("\\xff", "\\x01\\x02")
-      b'\\xff\\x01\\x02'
-      >>> to_bytes(1, 2, 3, [4,5,6])
-      b'\\x01\\x02\\x03\\x04\\x05\\x06'
-      >>> to_bytes(bytes([1,3,4]), bytearray([6,7,8]), "\\xff")
-      b'\\x01\\x03\\x04\\x06\\x07\\x08\\xff'
-      """
-      result = bytearray()
-      for d in data:
-          if type(d) in [tuple, list]:
-              baa = map(to_bytes, d)
-              for ba in baa:
-                  result += ba
-          if type(d) is int:
-              result += bytearray([d])
-          if type(d) is str:
-              result += bytearray(map(ord, d))
-          if type(d) in [bytes, bytearray]:
-              result += d
-      return result
+
+    def to_bytes(*data):
+        """Flatten the arrays and Converts data to a bytearray
+
+        :param data: The data to be converted
+        :type data: [int, bytes, bytearray, str, [int], [bytes], [bytearray], [str]]
+        :rtype: bytearray
+
+        >>> to_bytes("123")
+        b'123'
+        >>> to_bytes(1, 2, 3)
+        b'\\x01\\x02\\x03'
+        >>> to_bytes("\\xff", "\\x01\\x02")
+        b'\\xff\\x01\\x02'
+        >>> to_bytes(1, 2, 3, [4,5,6])
+        b'\\x01\\x02\\x03\\x04\\x05\\x06'
+        >>> to_bytes(bytes([1,3,4]), bytearray([6,7,8]), "\\xff")
+        b'\\x01\\x03\\x04\\x06\\x07\\x08\\xff'
+        """
+        result = bytearray()
+        for d in data:
+            if type(d) in [tuple, list]:
+                baa = map(to_bytes, d)
+                for ba in baa:
+                    result += ba
+            if type(d) is int:
+                result.extend(bytearray([d]))
+            elif type(d) is str:
+                result.extend(bytearray(map(ord, d)))
+            elif type(d) is bytearray:
+                result.extend(d)
+            elif type(d) is bytes:
+                result += d
+        return result
+
 else:
-  def to_bytes(*data):
-      return bytes("".join(map(str, data)))
+
+    def to_bytes(*data):
+        return bytes("".join(map(str, data)))

requirements-win.txt

@@ -0,0 +1,2 @@
+psutil==7.1.0
+gevent==25.9.1

requirements.txt

@@ -0,0 +1 @@
+gevent==25.9.1

setup.py

@@ -3,24 +3,26 @@
 with open("README.md", "r") as fh:
     long_description = fh.read()
 
-setuptools.setup(name='rawsocketpy',
-      version='0.4.0',
-      description='This library allows you to implement a custom layer 2 communication using raw sockets in Python 2 and Python 3, synchronous and asynchronous, with and without callbacks.',
-      long_description=long_description,
-      long_description_content_type="text/markdown",
-      url='https://github.com/AlexisTM/rawsocket_python',
-      author='AlexisTM',
-      author_email='alexis.paques@gmail.com',
-      license='MIT',
-      packages=setuptools.find_packages(),
-      classifiers=(
-          "Programming Language :: Python :: 2",
-          "Programming Language :: Python :: 3",
-          "Development Status :: 4 - Beta",
-          "License :: OSI Approved :: MIT License",
-          "Operating System :: POSIX :: Linux",
-          "Topic :: Internet",
-          "Topic :: System :: Networking",
-          "Topic :: System :: Networking :: Monitoring",
-      ),
-      zip_safe=False)
+setuptools.setup(
+    name="rawsocketpy",
+    version="0.4.0",
+    description="This library allows you to implement a custom layer 2 communication using raw sockets in Python 2 and Python 3, synchronous and asynchronous, with and without callbacks.",
+    long_description=long_description,
+    long_description_content_type="text/markdown",
+    url="https://github.com/AlexisTM/rawsocket_python",
+    author="AlexisTM",
+    author_email="alexis.paques@gmail.com",
+    license="MIT",
+    packages=setuptools.find_packages(),
+    classifiers=(
+        "Programming Language :: Python :: 2",
+        "Programming Language :: Python :: 3",
+        "Development Status :: 4 - Beta",
+        "License :: OSI Approved :: MIT License",
+        "Operating System :: POSIX :: Linux",
+        "Topic :: Internet",
+        "Topic :: System :: Networking",
+        "Topic :: System :: Networking :: Monitoring",
+    ),
+    zip_safe=False,
+)

test.py

@@ -1,13 +1,17 @@
-
-from rawsocketpy import RawRequestHandler, RawAsyncServerCallback
+#!/usr/bin/env python
+import argparse
 import time
+import sys
+from rawsocketpy import RawRequestHandler, RawAsyncServerCallback
+
 
 def callback(handler, server):
     print("Testing")
     handler.setup()
     handler.handle()
     handler.finish()
 
+
 class LongTaskTest(RawRequestHandler):
     def handle(self):
         time.sleep(1)
@@ -17,11 +21,39 @@ def finish(self):
         print("End")
 
     def setup(self):
-        print("Begin") 
+        print("Begin")
+
 
 def main():
-    rs = RawAsyncServerCallback("wlp2s0", 0xEEFA, LongTaskTest, callback)
+    parser = argparse.ArgumentParser(
+        description="Run a RawAsyncServerCallback on a given network interface."
+    )
+    default_iface = "wlp2s0" if sys.platform == "linux" else "Ethernet"
+
+    parser.add_argument(
+        "-i",
+        "--interface",
+        type=str,
+        default=default_iface,
+        help=f"Network interface to bind to (default: {default_iface})",
+    )
+
+    parser.add_argument(
+        "-p",
+        "--protocol",
+        type=lambda x: int(x, 0),  # allows hex like 0xEEFA or decimal
+        default=0xEEFA,
+        help="EtherType / protocol number (default: 0xEEFA)",
+    )
+    args = parser.parse_args()
+    rs = RawAsyncServerCallback(
+        args.interface,
+        args.protocol,
+        LongTaskTest,
+        callback,
+    )
     rs.spin()
 
-if __name__ == '__main__':
+
+if __name__ == "__main__":
     main()