Repository: MoshiBin/ssdpy

Issue:
Windows support
This issue tracks the work needed for supporting ssdpy on Windows.

- `socket.if_nametoindex` only exists in Windows since Python 3.8 - check how to bind to interface in Windows prior to that.
- TBD

Discussion:
codecov[bot]: # [Codecov](https://codecov.io/gh/MoshiBin/ssdpy/pull/58?src=pr&el=h1) Report
> Merging [#58](https://codecov.io/gh/MoshiBin/ssdpy/pull/58?src=pr&el=desc) (d53d537) into [master](https://codecov.io/gh/MoshiBin/ssdpy/commit/9f67a5f9fba13a7918418cf2e1c53f61dac6ee6e?el=desc) (9f67a5f) will **not change** coverage.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/MoshiBin/ssdpy/pull/58/graphs/tree.svg?width=650&height=150&src=pr&token=GNLwBMjmd4)](https://codecov.io/gh/MoshiBin/ssdpy/pull/58?src=pr&el=tree)

```diff
@@           Coverage Diff           @@
##           master      #58   +/-   ##
=======================================
  Coverage   94.11%   94.11%           
=======================================
  Files           6        6           
  Lines         119      119           
  Branches       19       19           
=======================================
  Hits          112      112           
  Misses          4        4           
  Partials        3        3           
```



------

[Continue to review full report at Codecov](https://codecov.io/gh/MoshiBin/ssdpy/pull/58?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/MoshiBin/ssdpy/pull/58?src=pr&el=footer). Last update [9f67a5f...d53d537](https://codecov.io/gh/MoshiBin/ssdpy/pull/58?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

MoshiBin: Not sure what's wrong with codecov, it shows missing coverage for the Linux Python 2.7 implementation of `if_nametoindex`. Testing manually on a Linux machine (and combining coverage reports manually) I can confirm that these lines *do* get coverage.

Patch:
diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index 86c6fbf..a3cefc8 100644
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@ -44,7 +44,7 @@ jobs:
     runs-on: ${{ matrix.os }}
     strategy:
       matrix:
-        os: [ubuntu-latest]
+        os: [ubuntu-latest, windows-latest]
         python-version: [2.7, 3.5, 3.6, 3.7, 3.8, 3.9]
 
     # Steps represent a sequence of tasks that will be executed as part of the job
@@ -69,7 +69,12 @@ jobs:
 
       - name: pytest
         run: sudo -E $pythonLocation/bin/coverage run --source ssdpy,tests -m pytest tests/
-        
+        if: runner.os == 'Linux'
+
+      - name: pytest (Windows)
+        run: coverage run --source ssdpy,tests -m pytest tests/
+        if: runner.os == 'Windows'
+
       - name: Upload to codecov
         uses: codecov/codecov-action@v1
         with:
diff --git a/CHANGES.md b/CHANGES.md
index 315287c..19a192f 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -1,5 +1,8 @@
+# Changelog
+
 ### Latest changes (not in a release)
 
+- Add support for Win32
 
 ### 0.4.1
 (2020-12-07)
diff --git a/README.md b/README.md
index 820ec78..e513012 100644
--- a/README.md
+++ b/README.md
@@ -2,6 +2,8 @@
 
 SSDPy is a lightweight implementation of [SSDP](https://en.wikipedia.org/wiki/Simple_Service_Discovery_Protocol) (Simple Service Discovery Protocol). It is designed for ease of use and high compatibility with the protocol in real-life use. It supports both the IETF and UPnP versions of the protocol. You can read the [full documentation here](https://ssdpy.readthedocs.io/en/latest/).
 
+This library works on both Windows and Linux.
+
 ## Example usage
 
 Send an SSDP discover packet (M-SEARCH):
diff --git a/ssdpy/client.py b/ssdpy/client.py
index 18e295f..5d60019 100644
--- a/ssdpy/client.py
+++ b/ssdpy/client.py
@@ -2,29 +2,18 @@
 from __future__ import absolute_import, division, print_function, unicode_literals
 
 import socket
+import sys
 from .constants import ipv4_multicast_ip, ipv6_multicast_ip
 from .http_helper import parse_headers
 from .protocol import create_msearch_payload
-from .compat import if_nametoindex, SO_BINDTODEVICE
+from .compat import if_nametoindex, SO_BINDTODEVICE, IPPROTO_IPV6
 
 
 class SSDPClient(object):
-    def __init__(
-        self,
-        proto="ipv4",
-        port=1900,
-        ttl=2,
-        iface=None,
-        timeout=5,
-        address=None,
-        *args,
-        **kwargs
-    ):
+    def __init__(self, proto="ipv4", port=1900, ttl=2, iface=None, timeout=5, address=None, *args, **kwargs):
         allowed_protos = ("ipv4", "ipv6")
         if proto not in allowed_protos:
-            raise ValueError(
-                "Invalid proto - expected one of {}".format(allowed_protos)
-            )
+            raise ValueError("Invalid proto - expected one of {}".format(allowed_protos))
         self.port = port
         if proto == "ipv4":
             af_type = socket.AF_INET
@@ -36,8 +25,12 @@ class SSDPClient(object):
             self._address = (self.broadcast_ip, port, 0, 0)
         self.sock = socket.socket(af_type, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
         self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-        self.sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, ttl)
-        self.sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_LOOP, 1)
+        if sys.platform == "win32" and proto == "ipv6":
+            self.sock.setsockopt(IPPROTO_IPV6, socket.IPV6_MULTICAST_HOPS, ttl)
+            self.sock.setsockopt(IPPROTO_IPV6, socket.IPV6_MULTICAST_LOOP, 1)
+        else:
+            self.sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, ttl)
+            self.sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_LOOP, 1)
         self.sock.settimeout(timeout)
         if address is not None:
             self.sock.bind((address, 0))
@@ -46,9 +39,7 @@ class SSDPClient(object):
             if proto == "ipv6":
                 # Specifically set multicast on interface
                 iface_index = if_nametoindex(iface)
-                self.sock.setsockopt(
-                    socket.IPPROTO_IPV6, socket.IPV6_MULTICAST_IF, iface_index
-                )
+                self.sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_MULTICAST_IF, iface_index)
 
     def send(self, data):
         self.sock.sendto(data, self._address)
diff --git a/ssdpy/compat.py b/ssdpy/compat.py
index 957f0ae..8650ac6 100644
--- a/ssdpy/compat.py
+++ b/ssdpy/compat.py
@@ -1,34 +1,86 @@
 # -*- coding: utf-8 -*-
 from __future__ import absolute_import, division, print_function, unicode_literals
 import socket
+import ctypes
+import errno
 import sys
 
 
 PY2 = sys.version_info[0] == 2
+WINDOWS = sys.platform == "win32"
+LINUX = sys.platform.startswith("linux")
 
 string_types = basestring if PY2 else str  # noqa
 
 
 # Python 2 doesn't have socket.if_nametoindex so we need to implement it manually
-if PY2:
-    import ctypes
-    import ctypes.util
+if LINUX:
+    if PY2:
+        import ctypes.util
 
-    libc = ctypes.CDLL(ctypes.util.find_library('c'))
+        libc = ctypes.CDLL(ctypes.util.find_library("c"))
+
+        def if_nametoindex(name):
+            """
+            Return the logical index number of the given interface name.
+            This is backported for Python 2.7 on Linux.
+            """
+            if not isinstance(name, string_types):
+                raise TypeError("Expected string type, got '{}'".format(type(name)))
+
+            rc = libc.if_nametoindex(name)
+            if rc == 0:
+                raise OSError("no interface with this name '{}'".format(name))
+            return rc
+
+    else:
+        if_nametoindex = socket.if_nametoindex
+else:
 
     def if_nametoindex(name):
+        raise NotImplementedError("if_nametoindex is not available on this platform")
+
+
+SO_BINDTODEVICE = getattr(socket, "SO_BINDTODEVICE", 25)
+IPPROTO_IPV6 = getattr(socket, "IPPROTO_IPV6", 41)
+
+
+if PY2 and WINDOWS:
+    def inet_pton(socket_af, ip_address):
         """
-        Return the logical index number of the given interface name.
+        Python 2.7 on Windows doesn't have the inet_pton function, which SSDPy uses for IPv6
+        servers. This function backports inet_pton using raw winsock functions.
+        This should ONLY be used on Python 2.7 on Windows.
         """
-        if not isinstance(name, string_types):
-            raise TypeError("Expected string type, got '{}'".format(type(name)))
+        if socket_af != socket.AF_INET6 or sys.platform != "win32":
+            raise ValueError("inet_pton workaround function only works for IPv6 on Windows")
 
-        rc = libc.if_nametoindex(name)
-        if rc == 0:
-            raise OSError("no interface with this name '{}'".format(name))
-        return rc
-else:
-    if_nametoindex = socket.if_nametoindex
+        # typedef struct in6_addr {
+        #     union {
+        #         u_char  Byte[16];
+        #         u_short Word[8];
+        #     } u;
+        # } IN6_ADDR, *PIN6_ADDR, FAR *LPIN6_ADDR;
+        in6_addr = ctypes.c_ubyte * 16
 
+        # INT WSAAPI InetPtonW(
+        #     INT    Family,
+        #     PCWSTR pszAddrString,
+        #     PVOID  pAddrBuf
+        # );
+        InetPtonW = ctypes.windll.ws2_32.InetPtonW
+        WSAGetLastError = ctypes.windll.ws2_32.WSAGetLastError
 
-SO_BINDTODEVICE = getattr(socket, "SO_BINDTODEVICE", 25)
+        socket_af = ctypes.c_int(socket.AF_INET6)
+        ip_address = ctypes.c_wchar_p(ip_address)
+        address_buffer = in6_addr()
+        return_code = InetPtonW(socket_af, ip_address, address_buffer)
+        if return_code != 1:
+            error_code = WSAGetLastError()
+            raise OSError("inet_pton failed: %s", errno.errorcode.get(error_code))
+        else:
+            return ctypes.string_at(address_buffer, 16)
+
+
+else:
+    inet_pton = socket.inet_pton
diff --git a/ssdpy/server.py b/ssdpy/server.py
index 4965804..431d8ce 100644
--- a/ssdpy/server.py
+++ b/ssdpy/server.py
@@ -4,10 +4,11 @@ from __future__ import absolute_import, division, print_function, unicode_litera
 import logging
 import socket
 import struct
+import sys
 from .constants import ipv6_multicast_ip, ipv4_multicast_ip
 from .protocol import create_notify_payload
 from .http_helper import parse_headers
-from .compat import if_nametoindex, SO_BINDTODEVICE
+from .compat import if_nametoindex, SO_BINDTODEVICE, inet_pton, IPPROTO_IPV6
 
 
 logger = logging.getLogger("ssdpy.server")
@@ -96,6 +97,8 @@ class SSDPServer(object):
 
         # Bind to specific interface
         if iface is not None:
+            if sys.platform == "win32":
+                raise ValueError("Unable to bind to device under Windows, use a bind address instead")
             self.sock.setsockopt(socket.SOL_SOCKET, SO_BINDTODEVICE, iface)
 
         # Subscribe to multicast address
@@ -105,25 +108,21 @@ class SSDPServer(object):
                 mreq += socket.inet_aton(address)
             else:
                 mreq += struct.pack(b"@I", socket.INADDR_ANY)
-            self.sock.setsockopt(
-                socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq,
-            )
+            self.sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
             # Allow multicasts on loopback devices (necessary for testing)
             self.sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_LOOP, 1)
         elif proto == "ipv6":
             # In IPv6 we use the interface index, not the address when subscribing to the group
-            mreq = socket.inet_pton(socket.AF_INET6, self._broadcast_ip)
+            mreq = inet_pton(socket.AF_INET6, self._broadcast_ip)
             if iface is not None:
                 iface_index = if_nametoindex(iface)
                 # Send outgoing packets from the same interface
-                self.sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_MULTICAST_IF, iface_index)
+                self.sock.setsockopt(IPPROTO_IPV6, socket.IPV6_MULTICAST_IF, iface_index)
                 mreq += struct.pack(b"@I", iface_index)
             else:
-                mreq += socket.inet_pton(socket.AF_INET6, "::")
-            self.sock.setsockopt(
-                socket.IPPROTO_IPV6, socket.IPV6_JOIN_GROUP, mreq,
-            )
-            self.sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_MULTICAST_LOOP, 1)
+                mreq += inet_pton(socket.AF_INET6, "::")
+            self.sock.setsockopt(IPPROTO_IPV6, socket.IPV6_JOIN_GROUP, mreq)
+            self.sock.setsockopt(IPPROTO_IPV6, socket.IPV6_MULTICAST_LOOP, 1)
         self.sock.bind((bind_address, port))
 
     def on_recv(self, data, address):


Test patch:
diff --git a/tests/test_client.py b/tests/test_client.py
index 8f66b5d..6b1f6a1 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 from __future__ import absolute_import, division, print_function, unicode_literals
 import os
+import sys
 import errno
 import pytest
 from ssdpy import SSDPClient
@@ -20,10 +21,8 @@ def test_client_rejects_bad_proto():
         SSDPClient(proto="invalid")
 
 
-@pytest.mark.skipif(
-    os.environ.get("CI") == "true",
-    reason="Not all development environments have a predictable loopback device name",
-)
+@pytest.mark.skipif(os.environ.get("CI") == "true", reason="No loopback in CI")
+@pytest.mark.skipif(sys.platform == "win32", reason="No bind to interface on Windows")
 def test_client_binds_iface():
     SSDPClient(iface=b"lo")
 
@@ -32,6 +31,7 @@ def test_client_binds_iface():
     os.environ.get("CI") == "true",
     reason="IPv6 testing is broken in GitHub Actions, see https://github.com/actions/virtual-environments/issues/668",
 )
+@pytest.mark.skipif(sys.platform == "win32", reason="No bind to interface on Windows")
 def test_client_bind_iface_ipv6():
     try:
         SSDPClient(proto="ipv6", iface=b"lo")
diff --git a/tests/test_compat.py b/tests/test_compat.py
index 214aec8..66009c7 100644
--- a/tests/test_compat.py
+++ b/tests/test_compat.py
@@ -1,26 +1,51 @@
 # -*- coding: utf-8 -*-
 from __future__ import absolute_import, division, print_function, unicode_literals
 import pytest
+import sys
+import socket
 from .compat import PY2
-from ssdpy.compat import if_nametoindex
+from ssdpy.compat import if_nametoindex, inet_pton, WINDOWS
 
 
+@pytest.mark.skipif(sys.platform == "win32", reason="No bind to interface on Windows")
 def test_if_nametoindex_none():
     if PY2:
         with pytest.raises(TypeError):
             if_nametoindex(None)
 
 
+@pytest.mark.skipif(sys.platform == "win32", reason="No bind to interface on Windows")
 def test_if_nametoindex_int():
     if PY2:
         with pytest.raises(TypeError):
             if_nametoindex(0)
 
 
+@pytest.mark.skipif(sys.platform == "win32", reason="No bind to interface on Windows")
 def test_if_nametoindex_nodevice():
     with pytest.raises(OSError):
         if_nametoindex("does-not-exist")
 
 
+@pytest.mark.skipif(sys.platform == "win32", reason="No bind to interface on Windows")
 def test_if_nametoindex():
     assert type(if_nametoindex(b"lo")) is int
+
+
+@pytest.mark.skipif(not (PY2 and WINDOWS), reason="Only test compat inet_pton")
+def test_inet_pton():
+    assert (
+        inet_pton(socket.AF_INET6, "dead::beef") == b"\xde\xad\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xbe\xef"
+    )
+
+
+@pytest.mark.skipif(not (PY2 and WINDOWS), reason="Only test compat inet_pton")
+def test_inet_pton_bad_addr():
+    with pytest.raises(OSError):
+        inet_pton(socket.AF_INET6, "bad address")
+
+
+@pytest.mark.skipif(not (PY2 and WINDOWS), reason="Only test compat inet_pton")
+def test_inet_pton_wrong_proto():
+    with pytest.raises(ValueError):
+        inet_pton(socket.AF_INET, "127.0.0.1")
diff --git a/tests/test_server.py b/tests/test_server.py
index c16b507..caa9af3 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -5,7 +5,9 @@ import time
 import pytest
 import errno
 import os
+import sys
 from ssdpy import SSDPServer
+from ssdpy.compat import LINUX
 
 
 def test_server_ipv4():
@@ -41,6 +43,7 @@ def test_server_invalid_proto():
     os.environ.get("CI") == "true",
     reason="Not all development environments have a predictable loopback device name",
 )
+@pytest.mark.skipif(sys.platform == "win32", reason="No bind to interface on Windows")
 def test_server_binds_iface():
     SSDPServer("test-server", iface=b"lo")
 
@@ -61,6 +64,7 @@ def test_server_bind_address_ipv6():
     os.environ.get("CI") == "true",
     reason="IPv6 testing is broken in GitHub Actions, see https://github.com/actions/virtual-environments/issues/668",
 )
+@pytest.mark.skipif(sys.platform == "win32", reason="No bind to interface on Windows")
 def test_server_bind_address_and_iface_ipv6():
     try:
         SSDPServer("test-server", address="::1", proto="ipv6", iface=b"lo")
@@ -76,3 +80,9 @@ def test_server_extra_fields():
 def test_server_extra_fields_non_ascii():
     with pytest.raises(ValueError):
         SSDPServer("test-server", extra_fields={"invalid-field™": "foo"})
+
+
+@pytest.mark.skipif(LINUX, reason="This limitation only occurs on Windows")
+def test_server_bind_interface_in_windows():
+    with pytest.raises(ValueError):
+        SSDPServer("test-server", iface=b"Ethernet")
diff --git a/tests/test_system.py b/tests/test_system.py
new file mode 100644
index 0000000..3eef259
--- /dev/null
+++ b/tests/test_system.py
@@ -0,0 +1,21 @@
+# -*- coding: utf-8 -*-
+from __future__ import absolute_import, division, print_function, unicode_literals
+import threading
+import pytest
+from ssdpy import SSDPServer, SSDPClient
+
+
+@pytest.fixture(scope="session", autouse=True)
+def ssdpy_server():
+    server = SSDPServer("test-server")
+    thread = threading.Thread(target=server.serve_forever)
+    thread.daemon = True
+    thread.start()
+
+
+def test_system(ssdpy_server):
+    """System test: Start a server on localhost, then try to discover the server"""
+    client = SSDPClient()
+    results = client.m_search()
+    assert results
+    assert list(filter(lambda response: response.get("usn") == "test-server", results))