feat: Add Windows compatibility

Author: AlexisTM

rawsocketpy/util.py

@@ -3,38 +3,56 @@
 
 from __future__ import absolute_import
 import socket
-import fcntl
 import struct
 import sys
 
-if sys.version_info >= (3, 0):
+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")
 
-        :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]
+        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}'")
 
 else:
+    import fcntl
 
-    def get_hw(ifname):
-        """Returns a unicode string containing the MAC address of the interface.
+    if sys.version_info >= (3, 0):
 
-        :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 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=":"):

requirements-win.txt

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

requirements.txt

@@ -1 +1 @@
-gevent==25.8.2
+gevent==25.9.1

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()