Merge pull request #25 from PhilippHomann/master

Add lease_time as parameter

.github/workflows/pypi-publish.yml

@@ -0,0 +1,31 @@
+# This workflows will upload a Python Package using Twine when a release is created
+# For more information see: https://help.github.com/en/actions/language-and-framework-guides/using-python-with-github-actions#publishing-to-package-registries
+
+name: pypi release
+
+on:
+  release:
+    types: [created]
+
+jobs:
+  deploy:
+
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v2
+    - name: Set up Python
+      uses: actions/setup-python@v2
+      with:
+        python-version: '3.6'
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install setuptools wheel twine
+    - name: Build and publish
+      env:
+        TWINE_USERNAME: ${{ secrets.PYPI_USERNAME }}
+        TWINE_PASSWORD: ${{ secrets.PYPI_PASSWORD }}
+      run: |
+        python setup.py sdist
+        twine upload dist/*

.github/workflows/python-package.yml

@@ -0,0 +1,53 @@
+name: CI
+
+on:
+  push:
+    branches: [ master ]
+  pull_request:
+    branches: [ master ]
+
+jobs:
+  code-quality:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: [3.8]
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v2
+        with:
+          python-version: ${{ matrix.python-version }}
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install mypy lxml flake8
+          pip install -e .
+      - name: Lint with flake8
+        run: |
+          # stop the build if there are Python syntax errors or undefined names
+          flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
+      - name: mypy
+        run: |
+          mypy aioupnp
+
+  tests:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: [3.6, 3.7, 3.8]
+    steps:
+    - uses: actions/checkout@v2
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v2
+      with:
+        python-version: ${{ matrix.python-version }}
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install coverage
+        pip install -e .
+    - name: Test with coverage
+      run: |
+        coverage run -m unittest discover -v tests
+        bash <(curl -s https://codecov.io/bash)

.travis.yml

@@ -1,85 +0,0 @@
-sudo: required
-dist: xenial
-language: python
-python: "3.8"
-
-jobs:
-  include:
-  - stage: code quality
-    name: "mypy"
-    before_install:
-    - pip install mypy lxml
-    - pip install -e .
-
-    script:
-    - mypy aioupnp --txt-report . --scripts-are-modules; cat index.txt; rm index.txt
-
-  - &tests
-      stage: test
-      name: "Unit Tests w/ Python 3.8"
-      python: "3.8"
-      before_install:
-      - pip install pylint coverage
-      - pip install -e .
-
-      script:
-      - HOME=/tmp coverage run -m unittest discover -v tests
-
-      after_success:
-      - bash <(curl -s https://codecov.io/bash)
-
-  - <<: *tests
-    name: "Unit Tests w/ Python 3.7"
-    python: "3.7"
-
-  - <<: *tests
-    name: "Unit Tests w/ Python 3.6"
-    python: "3.6"
-
-#
-#if: tag IS present
-#  - &build
-#    name: "Linux Release"
-#    python: "3.6"
-#    install:
-#      - pip install pyinstaller
-#      - pip install -e .
-#
-#    script:
-#      - pyinstaller -F -n aioupnp aioupnp/__main__.py
-#      - chmod +x dist/aioupnp
-#      - zip -j dist/aioupnp-${OS}.zip dist/aioupnp
-#
-#    env: OS=linux
-#
-#    addons:
-#      artifacts:
-#        working_dir: dist
-#        paths:
-#          - aioupnp-${OS}.zip
-#
-#  - <<: *build
-#    name: "Mac Release"
-#    os: osx
-#    osx_image: xcode9.4
-#    language: generic
-#    env: OS=mac
-#    install:
-#    - pip3 install pyinstaller
-#    - pip3 install -e .
-#
-#  - <<: *build
-#    name: "Windows Release"
-#    language: generic
-#    services:
-#    - docker
-#    install:
-#    - docker pull cdrx/pyinstaller-windows:python3-32bit
-#    script:
-#    - docker run -v "$(pwd):/src/aioupnp" cdrx/pyinstaller-windows:python3-32bit aioupnp/wine_build.sh
-#    - sudo zip -j dist/aioupnp-windows.zip dist/aioupnp.exe
-#    addons:
-#      artifacts:
-#        working_dir: dist
-#        paths:
-#          - aioupnp-windows.zip

README.md

@@ -1,4 +1,4 @@
-[![Build Status](https://travis-ci.org/lbryio/aioupnp.svg?branch=master)](https://travis-ci.org/lbryio/aioupnp)
+[![Build Status](https://github.com/lbryio/aioupnp/workflows/CI/badge.svg)
 [![codecov](https://codecov.io/gh/lbryio/aioupnp/branch/master/graph/badge.svg)](https://codecov.io/gh/lbryio/aioupnp)
 [![PyPI version](https://badge.fury.io/py/aioupnp.svg)](https://badge.fury.io/py/aioupnp)
 [![Python 3.6](https://img.shields.io/badge/python-3.6-blue.svg)](https://www.python.org/downloads/release/python-360/)
@@ -19,12 +19,12 @@ Verify python is version 3.6-8
 python --version
 ```
 
-Installation for normal usage
+#### Installation for normal usage
 ```
 pip install aioupnp
 ```
 
-Installation for development
+#### Installation for development
 ```
 git clone https://github.com/lbryio/aioupnp.git
 cd aioupnp
@@ -75,7 +75,7 @@ aioupnp [-h] [--debug_logging] [--interface=<interface>] [--gateway_address=<gat
     aioupnp delete_port_mapping --external_port=1234 --protocol=TCP
 
 #### M-Search headers
-UPnP uses multicast protocol - SSDP - to locate the gateway. Gateway discovery is automatic by default, but you may provide specific headers for the search to use to override automatic discovery.
+UPnP uses a multicast protocol (SSDP) to locate the gateway. Gateway discovery is automatic by default, but you may provide specific headers for the search to use to override automatic discovery.
 
 If m-search headers are provided as keyword arguments then all of the headers to be used must be provided, in the order they are to be used. For example:
 
@@ -109,16 +109,33 @@ By default, the network device will be automatically discovered. The interface m
 
 
 ## Troubleshooting
-
-#### Debug logging
-To enable verbose debug logging, add the `--debug_logging` argument before the command
+If `aioupnp` is failing with m-search timeouts this means the UPnP gateway (the router) isn't being found at all. To see if this error is expected try running m_search with debug logging, which will print out the packets sent and received:
 
     aioupnp --debug_logging m_search
 
-#### It really doesn't work
-If aioupnp doesn't work with a device, a debugging report can be collected with `aioupnp gather_debug_info`.
+If you only see packets being sent or the replies are only from devices that aren't your router (smart devices, speakers, etc), then there are three options:
+* your router does not support UPnP (this is unlikely)
+* UPnP is turned off in the web gui for your router (more likely)
+* `aioupnp` has a bug (very likely if you don't see your router manufacturer doing well in the supported devices table)
 
-This will attempt to discover the UPnP gateway, and then perform a functionality check where it will request the external address and existing port mappings before attempting to make and remove a port mapping. The final result is the zipped packet dump of these attempts, which allows writing tests replaying it.
+If you see replies from the router but it still fails, then it's likely a bug in aioupnp.
+
+If there are no replies and UPnP is certainly turned on, then a local firewall is the likely culprit.
+
+
+## Sending a bug report
+
+If it still doesn't work, you can send a bug report using an included script. This script will try finding the UPnP gateway using `aioupnp` as well as `miniupnpc` and then try add and remove a port mapping using each library. The script does this while capturing the packets sent/received, which makes figuring out what went wrong possible. The script will create a file with this packet capture (`aioupnp-bug-report.json`) and automatically send it.
+
+Note: the bug report script currently does not work on MacOS
+```
+git clone https://github.com/lbryio/aioupnp.git
+cd aioupnp
+python3 -m pip install -e .
+
+python3 -m pip install --user certifi aiohttp miniupnpc
+sudo -E python3 generate_bug_report.py
+```
 
 ## License
 

aioupnp/__init__.py

@@ -1,4 +1,4 @@
-__version__ = "0.0.17"
+__version__ = "0.0.18"
 __author__ = "Jack Robison"
 __maintainer__ = "Jack Robison"
 __license__ = "MIT"

aioupnp/__main__.py

@@ -121,4 +121,4 @@ def main(argv: typing.Optional[typing.List[typing.Optional[str]]] = None,
 
 
 if __name__ == "__main__":
-    sys.exit(main())
+    sys.exit(main())   # pragma: no cover

aioupnp/commands.py

@@ -5,6 +5,7 @@ import logging
 from aioupnp.protocols.scpd import scpd_post
 from aioupnp.device import Service
 from aioupnp.fault import UPnPError
+from aioupnp.util import is_valid_public_ipv4
 
 log = logging.getLogger(__name__)
 
@@ -134,7 +135,7 @@ class SOAPCommands:
 
     def is_registered(self, name: str) -> bool:
         if name not in self.SOAP_COMMANDS:
-            raise ValueError("unknown command")
+            raise ValueError("unknown command")  # pragma: no cover
         for service in self._registered.values():
             if name in service:
                 return True
@@ -142,11 +143,11 @@ class SOAPCommands:
 
     def get_service(self, name: str) -> Service:
         if name not in self.SOAP_COMMANDS:
-            raise ValueError("unknown command")
+            raise ValueError("unknown command")  # pragma: no cover
         for service, commands in self._registered.items():
             if name in commands:
                 return service
-        raise ValueError(name)
+        raise ValueError(name)  # pragma: no cover
 
     def _register_soap_wrapper(self, name: str) -> None:
         annotations: typing.Dict[str, typing.Any] = typing.get_type_hints(getattr(self, name))
@@ -173,7 +174,7 @@ class SOAPCommands:
                 self._request_debug_infos.append(SCPDRequestDebuggingInfo(name, kwargs, xml_bytes, result, None, time.time()))
             except Exception as err:
                 if isinstance(err, asyncio.CancelledError):
-                    raise
+                    raise  # pragma: no cover
                 self._request_debug_infos.append(SCPDRequestDebuggingInfo(name, kwargs, xml_bytes, None, err, time.time()))
                 raise UPnPError(f"Raised {str(type(err).__name__)}({str(err)}) parsing response for {name}")
             return result
@@ -200,7 +201,7 @@ class SOAPCommands:
         """Returns None"""
         name = "AddPortMapping"
         if not self.is_registered(name):
-            raise NotImplementedError()
+            raise NotImplementedError()  # pragma: no cover
         assert name in self._wrappers_kwargs
         await self._wrappers_kwargs[name](
             NewRemoteHost=NewRemoteHost, NewExternalPort=NewExternalPort, NewProtocol=NewProtocol,
@@ -216,7 +217,7 @@ class SOAPCommands:
         """
         name = "GetGenericPortMappingEntry"
         if not self.is_registered(name):
-            raise NotImplementedError()
+            raise NotImplementedError()  # pragma: no cover
         assert name in self._wrappers_kwargs
         result: GetGenericPortMappingEntryResponse = await self._wrappers_kwargs[name](
             NewPortMappingIndex=NewPortMappingIndex
@@ -228,7 +229,7 @@ class SOAPCommands:
         """Returns (NewInternalPort, NewInternalClient, NewEnabled, NewPortMappingDescription, NewLeaseDuration)"""
         name = "GetSpecificPortMappingEntry"
         if not self.is_registered(name):
-            raise NotImplementedError()
+            raise NotImplementedError()  # pragma: no cover
         assert name in self._wrappers_kwargs
         result: GetSpecificPortMappingEntryResponse = await self._wrappers_kwargs[name](
             NewRemoteHost=NewRemoteHost, NewExternalPort=NewExternalPort, NewProtocol=NewProtocol
@@ -239,7 +240,7 @@ class SOAPCommands:
         """Returns None"""
         name = "DeletePortMapping"
         if not self.is_registered(name):
-            raise NotImplementedError()
+            raise NotImplementedError()  # pragma: no cover
         assert name in self._wrappers_kwargs
         await self._wrappers_kwargs[name](
             NewRemoteHost=NewRemoteHost, NewExternalPort=NewExternalPort, NewProtocol=NewProtocol
@@ -250,18 +251,18 @@ class SOAPCommands:
         """Returns (NewExternalIPAddress)"""
         name = "GetExternalIPAddress"
         if not self.is_registered(name):
-            raise NotImplementedError()
+            raise NotImplementedError()  # pragma: no cover
         assert name in self._wrappers_no_args
-        result: str = await self._wrappers_no_args[name]()
-        if not result:
-            raise UPnPError("Got null external ip address")
-        return result
+        external_ip: str = await self._wrappers_no_args[name]()
+        if not is_valid_public_ipv4(external_ip):
+            raise UPnPError(f"Got invalid external ipv4 address: {external_ip}")
+        return external_ip
 
     # async def GetNATRSIPStatus(self) -> Tuple[bool, bool]:
     #     """Returns (NewRSIPAvailable, NewNATEnabled)"""
     #     name = "GetNATRSIPStatus"
     #     if not self.is_registered(name):
-    #         raise NotImplementedError()
+    #         raise NotImplementedError()  # pragma: no cover
     #     assert name in self._wrappers_no_args
     #     result: Tuple[bool, bool] = await self._wrappers_no_args[name]()
     #     return result[0], result[1]
@@ -270,7 +271,7 @@ class SOAPCommands:
     #     """Returns None"""
     #     name = "SetConnectionType"
     #     if not self.is_registered(name):
-    #         raise NotImplementedError()
+    #         raise NotImplementedError()  # pragma: no cover
     #     assert name in self._wrappers_kwargs
     #     await self._wrappers_kwargs[name](NewConnectionType=NewConnectionType)
     #     return None
@@ -279,7 +280,7 @@ class SOAPCommands:
     #     """Returns (NewConnectionType, NewPossibleConnectionTypes)"""
     #     name = "GetConnectionTypeInfo"
     #     if not self.is_registered(name):
-    #         raise NotImplementedError()
+    #         raise NotImplementedError()  # pragma: no cover
     #     assert name in self._wrappers_no_args
     #     result: Tuple[str, str] = await self._wrappers_no_args[name]()
     #     return result
@@ -288,7 +289,7 @@ class SOAPCommands:
     #     """Returns (NewConnectionStatus, NewLastConnectionError, NewUptime)"""
     #     name = "GetStatusInfo"
     #     if not self.is_registered(name):
-    #         raise NotImplementedError()
+    #         raise NotImplementedError()  # pragma: no cover
     #     assert name in self._wrappers_no_args
     #     result: Tuple[str, str, int] = await self._wrappers_no_args[name]()
     #     return result
@@ -297,7 +298,7 @@ class SOAPCommands:
     #     """Returns None"""
     #     name = "ForceTermination"
     #     if not self.is_registered(name):
-    #         raise NotImplementedError()
+    #         raise NotImplementedError()  # pragma: no cover
     #     assert name in self._wrappers_no_args
     #     await self._wrappers_no_args[name]()
     #     return None
@@ -306,7 +307,7 @@ class SOAPCommands:
     #     """Returns None"""
     #     name = "RequestConnection"
     #     if not self.is_registered(name):
-    #         raise NotImplementedError()
+    #         raise NotImplementedError()  # pragma: no cover
     #     assert name in self._wrappers_no_args
     #     await self._wrappers_no_args[name]()
     #     return None
@@ -316,7 +317,7 @@ class SOAPCommands:
     #      NewPhysicalLinkStatus)"""
     #     name = "GetCommonLinkProperties"
     #     if not self.is_registered(name):
-    #         raise NotImplementedError()
+    #         raise NotImplementedError()  # pragma: no cover
     #     assert name in self._wrappers_no_args
     #     result: Tuple[str, int, int, str] = await self._wrappers_no_args[name]()
     #     return result
@@ -325,7 +326,7 @@ class SOAPCommands:
     #     """Returns (NewTotalBytesSent)"""
     #     name = "GetTotalBytesSent"
     #     if not self.is_registered(name):
-    #         raise NotImplementedError()
+    #         raise NotImplementedError()  # pragma: no cover
     #     assert name in self._wrappers_no_args
     #     result: Tuple[int] = await self._wrappers_no_args[name]()
     #     return result[0]
@@ -334,7 +335,7 @@ class SOAPCommands:
     #     """Returns (NewTotalBytesReceived)"""
     #     name = "GetTotalBytesReceived"
     #     if not self.is_registered(name):
-    #         raise NotImplementedError()
+    #         raise NotImplementedError()  # pragma: no cover
     #     assert name in self._wrappers_no_args
     #     result: Tuple[int] = await self._wrappers_no_args[name]()
     #     return result[0]
@@ -343,7 +344,7 @@ class SOAPCommands:
     #     """Returns (NewTotalPacketsSent)"""
     #     name = "GetTotalPacketsSent"
     #     if not self.is_registered(name):
-    #         raise NotImplementedError()
+    #         raise NotImplementedError()  # pragma: no cover
     #     assert name in self._wrappers_no_args
     #     result: Tuple[int] = await self._wrappers_no_args[name]()
     #     return result[0]
@@ -352,7 +353,7 @@ class SOAPCommands:
     #     """Returns (NewTotalPacketsReceived)"""
     #     name = "GetTotalPacketsReceived"
     #     if not self.is_registered(name):
-    #         raise NotImplementedError()
+    #         raise NotImplementedError()  # pragma: no cover
     #     assert name in self._wrappers_no_args
     #     result: Tuple[int] = await self._wrappers_no_args[name]()
     #     return result[0]
@@ -362,7 +363,7 @@ class SOAPCommands:
     #     Layer1DownstreamMaxBitRate, Uptime)"""
     #     name = "X_GetICSStatistics"
     #     if not self.is_registered(name):
-    #         raise NotImplementedError()
+    #         raise NotImplementedError()  # pragma: no cover
     #     assert name in self._wrappers_no_args
     #     result: Tuple[int, int, int, int, str, str] = await self._wrappers_no_args[name]()
     #     return result
@@ -371,7 +372,7 @@ class SOAPCommands:
     #     """Returns (NewDefaultConnectionService)"""
     #     name = "GetDefaultConnectionService"
     #     if not self.is_registered(name):
-    #         raise NotImplementedError()
+    #         raise NotImplementedError()  # pragma: no cover
     #     assert name in self._wrappers_no_args
     #     result: Tuple[str] = await self._wrappers_no_args[name]()
     #     return result[0]
@@ -380,7 +381,7 @@ class SOAPCommands:
     #     """Returns (None)"""
     #     name = "SetDefaultConnectionService"
     #     if not self.is_registered(name):
-    #         raise NotImplementedError()
+    #         raise NotImplementedError()  # pragma: no cover
     #     assert name in self._wrappers_kwargs
     #     await self._wrappers_kwargs[name](NewDefaultConnectionService=NewDefaultConnectionService)
     #     return None
@@ -388,7 +389,7 @@ class SOAPCommands:
     # async def SetEnabledForInternet(self, NewEnabledForInternet: bool) -> None:
     #     name = "SetEnabledForInternet"
     #     if not self.is_registered(name):
-    #         raise NotImplementedError()
+    #         raise NotImplementedError()  # pragma: no cover
     #     assert name in self._wrappers_kwargs
     #     await self._wrappers_kwargs[name](NewEnabledForInternet=NewEnabledForInternet)
     #     return None
@@ -396,7 +397,7 @@ class SOAPCommands:
     # async def GetEnabledForInternet(self) -> bool:
     #     name = "GetEnabledForInternet"
     #     if not self.is_registered(name):
-    #         raise NotImplementedError()
+    #         raise NotImplementedError()  # pragma: no cover
     #     assert name in self._wrappers_no_args
     #     result: Tuple[bool] = await self._wrappers_no_args[name]()
     #     return result[0]
@@ -404,7 +405,7 @@ class SOAPCommands:
     # async def GetMaximumActiveConnections(self, NewActiveConnectionIndex: int) -> None:
     #     name = "GetMaximumActiveConnections"
     #     if not self.is_registered(name):
-    #         raise NotImplementedError()
+    #         raise NotImplementedError()  # pragma: no cover
     #     assert name in self._wrappers_kwargs
     #     await self._wrappers_kwargs[name](NewActiveConnectionIndex=NewActiveConnectionIndex)
     #     return None
@@ -413,7 +414,7 @@ class SOAPCommands:
     #     """Returns (NewActiveConnDeviceContainer, NewActiveConnectionServiceID"""
     #     name = "GetActiveConnections"
     #     if not self.is_registered(name):
-    #         raise NotImplementedError()
+    #         raise NotImplementedError()  # pragma: no cover
     #     assert name in self._wrappers_no_args
     #     result: Tuple[str, str] = await self._wrappers_no_args[name]()
     #     return result

aioupnp/constants.py

@@ -2,6 +2,7 @@ POST = "POST"
 ROOT = "root"
 SPEC_VERSION = "specVersion"
 XML_VERSION = "<?xml version=\"1.0\"?>"
+XML_VERSION_PREFIX = "<?xml version=\"1.0\""
 FAULT = "{http://schemas.xmlsoap.org/soap/envelope/}Fault"
 ENVELOPE = "{http://schemas.xmlsoap.org/soap/envelope/}Envelope"
 BODY = "{http://schemas.xmlsoap.org/soap/envelope/}Body"

aioupnp/interfaces.py

@@ -1,11 +1,10 @@
-import socket
 from collections import OrderedDict
 import typing
 import netifaces
 from aioupnp.fault import UPnPError
 
 
-def get_netifaces():
+def get_netifaces():  # pragma: no cover
     return netifaces
 
 
@@ -25,11 +24,11 @@ def _get_gateways() -> typing.Dict[typing.Union[str, int],
 
 def get_interfaces() -> typing.Dict[str, typing.Tuple[str, str]]:
     gateways = _get_gateways()
-    infos = gateways[socket.AF_INET]
+    infos = gateways[netifaces.AF_INET]
     assert isinstance(infos, list), TypeError(f"expected list from netifaces, got a dict")
     interface_infos: typing.List[typing.Tuple[str, str, bool]] = infos
     result: typing.Dict[str, typing.Tuple[str, str]] = OrderedDict(
-        (interface_name, (router_address, ifaddresses(interface_name)[socket.AF_INET][0]['addr']))
+        (interface_name, (router_address, ifaddresses(interface_name)[netifaces.AF_INET][0]['addr']))
         for router_address, interface_name, _ in interface_infos
     )
     for interface_name in _get_interfaces():
@@ -43,7 +42,7 @@ def get_interfaces() -> typing.Dict[str, typing.Tuple[str, str]]:
     _default = gateways['default']
     assert isinstance(_default, dict), TypeError(f"expected dict from netifaces, got a list")
     default: typing.Dict[int, typing.Tuple[str, str]] = _default
-    result['default'] = result[default[socket.AF_INET][1]]
+    result['default'] = result[default[netifaces.AF_INET][1]]
     return result
 
 

aioupnp/protocols/multicast.py

@@ -1,3 +1,4 @@
+import sys
 import struct
 import socket
 import typing
@@ -6,11 +7,17 @@ from asyncio.transports import DatagramTransport
 from unittest import mock
 
 
+SOCKET_TYPES = (socket.SocketType, mock.MagicMock)
+if sys.version_info >= (3, 8):  # pragma: no cover
+    from asyncio.trsock import TransportSocket
+    SOCKET_TYPES = (socket.SocketType, TransportSocket, mock.MagicMock)
+
+
 def _get_sock(transport: typing.Optional[DatagramTransport]) -> typing.Optional[socket.socket]:
     if transport is None or not hasattr(transport, "_extra"):
         return None
     sock: typing.Optional[socket.socket] = transport.get_extra_info('socket', None)
-    assert sock is None or isinstance(sock, (socket.SocketType, mock.MagicMock))
+    assert sock is None or isinstance(sock, SOCKET_TYPES)
     return sock
 
 

aioupnp/protocols/scpd.py

@@ -56,6 +56,7 @@ class SCPDHTTPClientProtocol(Protocol):
         self._response_msg = b""
         self._content_length = 0
         self._got_headers = False
+        self._has_content_length = True
         self._headers: typing.Dict[bytes, bytes] = {}
         self._body = b""
         self.transport: typing.Optional[asyncio.WriteTransport] = None
@@ -67,35 +68,57 @@ class SCPDHTTPClientProtocol(Protocol):
         return None
 
     def data_received(self, data: bytes) -> None:
+        if self.finished.done():  # possible to hit during tests
+            return
         self.response_buff += data
         for i, line in enumerate(self.response_buff.split(b'\r\n')):
             if not line:  # we hit the blank line between the headers and the body
                 if i == (len(self.response_buff.split(b'\r\n')) - 1):
                     return None  # the body is still yet to be written
                 if not self._got_headers:
-                    self._headers, self._response_code, self._response_msg = parse_headers(
-                        b'\r\n'.join(self.response_buff.split(b'\r\n')[:i])
-                    )
+                    try:
+                        self._headers, self._response_code, self._response_msg = parse_headers(
+                            b'\r\n'.join(self.response_buff.split(b'\r\n')[:i])
+                        )
+                    except ValueError as err:
+                        self.finished.set_exception(UPnPError(str(err)))
+                        return
                     content_length = get_dict_val_case_insensitive(
                         self._headers, b'Content-Length'
                     )
-                    if content_length is None:
-                        return None
-                    self._content_length = int(content_length or 0)
+                    if content_length is not None:
+                        self._content_length = int(content_length)
+                    else:
+                        self._has_content_length = False
                     self._got_headers = True
-                body = b'\r\n'.join(self.response_buff.split(b'\r\n')[i+1:])
-                if self._content_length == len(body):
-                    self.finished.set_result((self.response_buff, body, self._response_code, self._response_msg))
-                elif self._content_length > len(body):
-                    pass
-                else:
-                    self.finished.set_exception(
-                        UPnPError(
-                            "too many bytes written to response (%i vs %i expected)" % (
-                                len(body), self._content_length
+                if self._got_headers and self._has_content_length:
+                    body = b'\r\n'.join(self.response_buff.split(b'\r\n')[i+1:])
+                    if self._content_length == len(body):
+                        self.finished.set_result((self.response_buff, body, self._response_code, self._response_msg))
+                    elif self._content_length > len(body):
+                        pass
+                    else:
+                        self.finished.set_exception(
+                            UPnPError(
+                                "too many bytes written to response (%i vs %i expected)" % (
+                                    len(body), self._content_length
+                                )
                             )
                         )
-                    )
+                elif any(map(self.response_buff.endswith, (b"</root>\r\n", b"</scpd>\r\n"))):
+                    # Actiontec has a router that doesn't give a Content-Length for the gateway xml
+                    body = b'\r\n'.join(self.response_buff.split(b'\r\n')[i+1:])
+                    self.finished.set_result((self.response_buff, body, self._response_code, self._response_msg))
+                elif len(self.response_buff) >= 65535:
+                    self.finished.set_exception(
+                        UPnPError(
+                            "too many bytes written to response (%i) with unspecified content length" % len(self.response_buff)
+                            )
+                        )
+                    return
+                else:
+                    # needed for the actiontec case
+                    pass
                 return None
         return None
 

aioupnp/protocols/ssdp.py

@@ -171,8 +171,8 @@ async def multi_m_search(lan_address: str, gateway_address: str, timeout: int =
     protocol, gateway_address, lan_address = await listen_ssdp(
         lan_address, gateway_address, loop
     )
-    fut = asyncio.ensure_future(protocol.send_m_searches(
+    fut = protocol.send_m_searches(
         address=gateway_address, datagrams=list(packet_generator())
-    ), loop=loop)
+    )
     loop.call_later(timeout, lambda: None if not fut or fut.done() else fut.cancel())
     return protocol

aioupnp/serialization/scpd.py

@@ -1,13 +1,13 @@
 import re
 from typing import Dict, Any, List, Tuple
 from aioupnp.fault import UPnPError
-from aioupnp.constants import XML_VERSION
+from aioupnp.constants import XML_VERSION_PREFIX
 from aioupnp.serialization.xml import xml_to_dict
 from aioupnp.util import flatten_keys
 
 
 CONTENT_PATTERN = re.compile(
-    "(\<\?xml version=\"1\.0\"\?\>(\s*.)*|\>)"
+    "(\<\?xml version=\"1\.0\"[^>]*\?\>(\s*.)*|\>)"
 )
 
 XML_ROOT_SANITY_PATTERN = re.compile(
@@ -38,8 +38,8 @@ def serialize_scpd_get(path: str, address: str) -> bytes:
 
 
 def deserialize_scpd_get_response(content: bytes) -> Dict[str, Any]:
-    if XML_VERSION.encode() in content:
-        parsed: List[Tuple[bytes, bytes]] = CONTENT_PATTERN.findall(content.decode())
+    if XML_VERSION_PREFIX.encode() in content:
+        parsed: List[Tuple[str, str]] = CONTENT_PATTERN.findall(content.decode())
         xml_dict = xml_to_dict('' if not parsed else parsed[0][0])
         return parse_device_dict(xml_dict)
     return {}

aioupnp/serialization/soap.py

@@ -13,7 +13,7 @@ CONTENT_NO_XML_VERSION_PATTERN = re.compile(
 
 def serialize_soap_post(method: str, param_names: typing.List[str], service_id: bytes, gateway_address: bytes,
                         control_url: bytes, **kwargs: typing.Dict[str, str]) -> bytes:
-    args = "".join(f"<{n}>{kwargs.get(n)}</{n}>" for n in param_names)
+    args = "".join(f"<{param_name}>{kwargs.get(param_name, '')}</{param_name}>" for param_name in param_names)
     soap_body = (f'\r\n{XML_VERSION}\r\n<s:Envelope xmlns:s="http://schemas.xmlsoap.org/soap/envelope/" '
                  f's:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/"><s:Body>'
                  f'<u:{method} xmlns:u="{service_id.decode()}">{args}</u:{method}></s:Body></s:Envelope>')

aioupnp/serialization/ssdp.py

@@ -3,23 +3,13 @@ import logging
 import binascii
 import json
 from collections import OrderedDict
-from typing import List, Optional, Dict, Union, Tuple, Callable
+from typing import List, Optional, Dict, Union, Callable
 from aioupnp.fault import UPnPError
 from aioupnp.constants import line_separator
 
 log = logging.getLogger(__name__)
 
 _template = "(?i)^(%s):[ ]*(.*)$"
-vendor_pattern = re.compile("^([\w|\d]*)\.([\w|\d]*\.com):([ \"|\w|\d\:]*)$")
-
-
-def match_vendor(line: str) -> Optional[Tuple[str, str]]:
-    match: List[Tuple[str, str, str]] = vendor_pattern.findall(line)
-    if match:
-        vendor_key: str = match[-1][0].lstrip(" ").rstrip(" ")
-        vendor_value: str = match[-1][2].lstrip(" ").rstrip(" ")
-        return vendor_key, vendor_value
-    return None
 
 
 def compile_find(pattern: str) -> Callable[[str], Optional[str]]:
@@ -69,8 +59,6 @@ class SSDPDatagram:
         _OK: "m-search response"
     }
 
-    _vendor_field_pattern = vendor_pattern
-
     _required_fields: Dict[str, List[str]] = {
         _M_SEARCH: [
             'host',
@@ -130,10 +118,7 @@ class SSDPDatagram:
     def get_cli_igd_kwargs(self) -> str:
         fields = []
         for field in self._field_order:
-            v = getattr(self, field, None)
-            if v is None:
-                raise UPnPError("missing required field %s" % field)
-            fields.append("--%s=%s" % (self._case_mappings.get(field, field), v))
+            fields.append("--%s=%s" % (self._case_mappings.get(field, field), getattr(self, field, None)))
         return " ".join(fields)
 
     def __repr__(self) -> str:
@@ -174,11 +159,6 @@ class SSDPDatagram:
             raise UPnPError(
                 f"failed to decode datagram: {binascii.hexlify(datagram).decode()}"
             )
-        for attr_name in packet._required_fields[packet._packet_type]:
-            if getattr(packet, attr_name, None) is None:
-                raise UPnPError(
-                    "required field for {} is missing from m-search response: {}".format(packet._packet_type, attr_name)
-                )
         return packet
 
     @classmethod
@@ -206,10 +186,6 @@ class SSDPDatagram:
                             matched = True
                             matched_keys.append(name)
                             break
-            if not matched:
-                matched_vendor = match_vendor(line)
-                if matched_vendor and matched_vendor[0] not in result:
-                    result[matched_vendor[0]] = matched_vendor[1]
         return result
 
     @classmethod

aioupnp/upnp.py

@@ -37,7 +37,7 @@ class UPnP:
             return cls.delete_port_mapping.__annotations__, cls.delete_port_mapping.__doc__
         if command == "get_next_mapping":
             return cls.get_next_mapping.__annotations__, cls.get_next_mapping.__doc__
-        raise AttributeError(command)
+        raise AttributeError(command)  # pragma: no cover
 
     @staticmethod
     def get_lan_and_gateway(lan_address: str = '', gateway_address: str = '',
@@ -107,7 +107,7 @@ class UPnP:
         return await self.gateway.commands.GetExternalIPAddress()
 
     async def add_port_mapping(self, external_port: int, protocol: str, internal_port: int, lan_address: str,
-                               description: str) -> None:
+                               description: str, lease_time: int = 0) -> None:
         """
         Add a new port mapping
 
@@ -116,12 +116,13 @@ class UPnP:
         :param internal_port: (int) internal port
         :param lan_address: (str) internal lan address
         :param description: (str) mapping description
+        :param lease_time: (int) lease time in seconds
         :return: None
         """
         await self.gateway.commands.AddPortMapping(
             NewRemoteHost='', NewExternalPort=external_port, NewProtocol=protocol,
             NewInternalPort=internal_port, NewInternalClient=lan_address,
-            NewEnabled=1, NewPortMappingDescription=description, NewLeaseDuration='0'
+            NewEnabled=1, NewPortMappingDescription=description, NewLeaseDuration=str(lease_time)
         )
         return None
 
@@ -208,7 +209,7 @@ class UPnP:
         return None
 
     async def get_next_mapping(self, port: int, protocol: str, description: str,
-                               internal_port: Optional[int] = None) -> int:
+                               internal_port: Optional[int] = None, lease_time: int = 0) -> int:
         """
         Get a new port mapping. If the requested port is not available, increment until the next free port is mapped
 
@@ -216,6 +217,7 @@ class UPnP:
         :param protocol: (str) UDP | TCP
         :param description: (str) mapping description
         :param internal_port: (int) internal port
+        :param lease_time: (int) lease time in seconds
 
         :return: (int) mapped port
         """
@@ -235,7 +237,7 @@ class UPnP:
             if int_host == self.lan_address and int_port == requested_port and desc == description:
                 return port
             port += 1
-        await self.add_port_mapping(port, protocol, _internal_port, self.lan_address, description)
+        await self.add_port_mapping(port, protocol, _internal_port, self.lan_address, description, lease_time)
         return port
 
     async def gather_debug_info(self) -> str:  # pragma: no cover
@@ -260,15 +262,18 @@ class UPnP:
             await self.get_redirects()
         except UPnPError:
             pass
+        external_port = 0
+        made_mapping = False
         try:
             external_port = await self.get_next_mapping(1234, 'TCP', 'aioupnp testing')
+            made_mapping = True
         except UPnPError:
-            external_port = None
+            pass
         try:
             await self.get_redirects()
         except UPnPError:
             pass
-        if external_port:
+        if made_mapping:
             try:
                 await self.delete_port_mapping(external_port, 'TCP')
             except UPnPError:
@@ -407,12 +412,12 @@ def run_cli(method: str, igd_args: Dict[str, Union[bool, str, int]], lan_address
                 u = await UPnP.discover(
                     lan_address, gateway_address, timeout, igd_args, interface_name, loop=loop
                 )
-            except UPnPError as err:
+            except UPnPError as err:  # pragma: no cover
                 fut.set_exception(err)
                 return
             if method not in cli_commands:
-                fut.set_exception(UPnPError("\"%s\" is not a recognized command" % method))
-                return
+                fut.set_exception(UPnPError("\"%s\" is not a recognized command" % method))  # pragma: no cover
+                return  # pragma: no cover
             else:
                 fn = getattr(u, method)
 
@@ -421,8 +426,7 @@ def run_cli(method: str, igd_args: Dict[str, Union[bool, str, int]], lan_address
             fut.set_result(result)
         except UPnPError as err:
             fut.set_exception(err)
-
-        except Exception as err:
+        except Exception as err:  # pragma: no cover
             log.exception("uncaught error")
             fut.set_exception(UPnPError("uncaught error: %s" % str(err)))
 

aioupnp/util.py

@@ -1,3 +1,5 @@
+import sys
+import ipaddress
 import typing
 from collections import OrderedDict
 
@@ -47,3 +49,21 @@ def get_dict_val_case_insensitive(source: typing.Dict[typing.AnyStr, typing.AnyS
         matched_key: typing.AnyStr = match[0]
         return source[matched_key]
     raise KeyError("overlapping keys")
+
+
+# the ipaddress module does not show these subnets as reserved
+CARRIER_GRADE_NAT_SUBNET = ipaddress.ip_network('100.64.0.0/10')
+IPV4_TO_6_RELAY_SUBNET = ipaddress.ip_network('192.88.99.0/24')
+
+
+def is_valid_public_ipv4(address):
+    try:
+        parsed_ip = ipaddress.ip_address(address)
+        if any((parsed_ip.version != 4, parsed_ip.is_unspecified, parsed_ip.is_link_local, parsed_ip.is_loopback,
+                parsed_ip.is_multicast, parsed_ip.is_reserved, parsed_ip.is_private, parsed_ip.is_reserved)):
+            return False
+        else:
+            return not any((CARRIER_GRADE_NAT_SUBNET.overlaps(ipaddress.ip_network(f"{address}/32")),
+                            IPV4_TO_6_RELAY_SUBNET.overlaps(ipaddress.ip_network(f"{address}/32"))))
+    except (ipaddress.AddressValueError, ValueError):
+        return False

generate_bug_report.py

@@ -0,0 +1,511 @@
+import os
+import sys
+import socket
+import ctypes
+import contextlib
+import struct
+import binascii
+import enum
+import re
+import time
+import base64
+import ssl
+import asyncio
+import codecs
+import typing
+import json
+from ctypes import c_char, c_short
+from typing import Tuple
+import aioupnp
+from aioupnp.upnp import UPnP, UPnPError, get_gateway_and_lan_addresses
+from aioupnp.constants import SSDP_IP_ADDRESS
+import certifi
+import aiohttp
+import miniupnpc
+
+
+
+_IFF_PROMISC  = 0x0100
+_SIOCGIFFLAGS = 0x8913  # get the active flags
+_SIOCSIFFLAGS = 0x8914  # set the active flags
+_ETH_P_ALL    = 0x0003  # all protocols
+ETHER_HEADER_LEN = 6 + 6 + 2
+VLAN_HEADER_LEN = 2
+
+printable = re.compile(b"([a-z0-9!\"#$%&'()*+,.\/:;<=>?@\[\] ^_`{|}~-]*)")
+
+
+class PacketTypes(enum.Enum):  # if_packet.h
+    HOST = 0
+    BROADCAST = 1
+    MULTICAST = 2
+    OTHERHOST = 3
+    OUTGOING = 4
+    LOOPBACK = 5
+    FASTROUTE = 6
+
+
+class Layer2(enum.Enum):  # https://www.iana.org/assignments/ieee-802-numbers/ieee-802-numbers.xhtml
+    IPv4 = 0x0800
+    ARP =  0x0806
+    VLAN = 0x8100
+    MVRP = 0x88f5
+    MMRP = 0x88f6
+    IPv6 = 0x86dd
+    GRE  = 0xb7ea
+
+
+class Layer3(enum.Enum):  # https://www.iana.org/assignments/protocol-numbers/protocol-numbers.xhtml
+    ICMP = 1
+    IGMP = 2
+    TCP = 6
+    UDP = 17
+
+
+class _ifreq(ctypes.Structure):
+    _fields_ = [("ifr_ifrn", c_char * 16),
+                ("ifr_flags", c_short)]
+
+
+@contextlib.contextmanager
+def _promiscuous_posix_socket_context(interface: str):
+    import fcntl  # posix-only
+    sock = socket.socket(socket.PF_PACKET, socket.SOCK_RAW, socket.htons(_ETH_P_ALL))
+    ifr = _ifreq()
+    ifr.ifr_ifrn = interface.encode()[:16]
+    fcntl.ioctl(sock, _SIOCGIFFLAGS, ifr)  # get the flags
+    ifr.ifr_flags |= _IFF_PROMISC  # add the promiscuous flag
+    fcntl.ioctl(sock, _SIOCSIFFLAGS, ifr)  # update
+    sock.setblocking(False)
+    try:
+        yield sock
+    finally:
+        ifr.ifr_flags ^= _IFF_PROMISC  # mask it off (remove)
+        fcntl.ioctl(sock, _SIOCSIFFLAGS, ifr)  # update
+        print("closed posix promiscuous socket")
+
+
+@contextlib.contextmanager
+def _promiscuous_non_posix_socket_context():
+    # the public network interface
+    HOST = socket.gethostbyname(socket.gethostname())
+    # create a raw socket and bind it to the public interface
+    sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_IP)
+    # prevent socket from being left in TIME_WAIT state, enabling reuse
+    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    sock.bind((HOST, 0))
+    # Include IP headers
+    sock.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)
+    # receive all packages
+    sock.ioctl(socket.SIO_RCVALL, socket.RCVALL_ON)
+    sock.setblocking(False)
+    try:
+        yield sock
+    finally:
+        # disable promiscuous mode
+        sock.ioctl(socket.SIO_RCVALL, socket.RCVALL_OFF)
+        print("closed non-posix promiscuous socket")
+
+
+def promiscuous(interface: typing.Optional[str] = None) -> typing.ContextManager[socket.socket]:
+    if os.name == 'posix':
+        return _promiscuous_posix_socket_context(interface)
+    return _promiscuous_non_posix_socket_context()
+
+
+def ipv4_to_str(addr: bytes) -> str:
+    return ".".join((str(b) for b in addr))
+
+
+def pretty_mac(mac: bytes) -> str:
+    return ":".join((('0' if b < 16 else '') + hex(b)[2:] for b in mac))
+
+
+def split_byte(b: int, bit=4) -> Tuple[bytes, bytes]:
+    return chr(((b >> (8-bit)) % 256) << (8-bit) >> (8-bit)).encode(), chr(((b << bit) % 256) >> bit).encode()
+
+
+class EtherFrame:
+    __slots__ = [
+        'source_mac',
+        'target_mac',
+        'ether_type',
+        'vlan_id',
+        'tpid'
+    ]
+
+    def __init__(self, source_mac: bytes, target_mac: bytes, ether_type: int,
+                 vlan_id: typing.Optional[int] = None, tpid: typing.Optional[int] = None):
+        self.source_mac = source_mac
+        self.target_mac = target_mac
+        self.ether_type = ether_type
+        self.vlan_id = vlan_id
+        self.tpid = tpid
+
+    def encode(self) -> bytes:
+        if self.vlan_id is None:
+            return struct.pack("6s6sH", *(getattr(self, slot) for slot in self.__slots__[:-2]))
+        return struct.pack("6s6sHHH", *(getattr(self, slot) for slot in self.__slots__))
+
+    @classmethod
+    def decode(cls, packet: bytes) -> Tuple['EtherFrame', bytes]:
+        vlan_id = None
+        tpid = None
+        if struct.unpack(f'!H', packet[12:14])[0] == Layer2.VLAN.value:
+            target_mac, source_mac, tpid, vlan_id, ether_type, data = struct.unpack(f'!6s6sHHH{len(packet) - ETHER_HEADER_LEN - VLAN_HEADER_LEN}s', packet)
+        else:
+            target_mac, source_mac, ether_type, data = struct.unpack(f'!6s6sH{len(packet) - ETHER_HEADER_LEN}s', packet)
+        return cls(source_mac, target_mac, ether_type, vlan_id, tpid), data
+
+    def debug(self) -> str:
+        if self.vlan_id is None:
+            return f"EtherFrame(source={pretty_mac(self.source_mac)}, target={pretty_mac(self.target_mac)}, " \
+                f"ether_type={Layer2(self.ether_type).name})"
+        return f"EtherFrame(source={pretty_mac(self.source_mac)}, target={pretty_mac(self.target_mac)}, " \
+               f"ether_type={Layer2(self.ether_type).name}, vlan={self.vlan_id})"
+
+
+class IPv4Packet:
+    __slots__ = [
+        'ether_frame',
+        'version',
+        'header_length',
+        'dscp',
+        'ecn',
+        'total_length',
+        'identification',
+        'df',
+        'mf',
+        'flag',
+        'fragment_offset',
+        'ttl',
+        'protocol',
+        'header_checksum',
+        '_source_address',
+        '_destination_address',
+        'data',
+        'packet_type',
+        'interface'
+    ]
+
+    ETHER_TYPE = Layer2.IPv4
+
+    def __init__(self, ether_frame: EtherFrame, version: int, header_length: int, dscp: int, ecn: int,
+                 total_length: int, identification: int, mf: bool, df: bool, flag: bool, fragment_offset: int, ttl: int,
+                 protocol: int, header_checksum: int, source_address: bytes, destination_address: bytes, data: bytes,
+                 packet_type: int, interface: str):
+        self.ether_frame = ether_frame
+        self.version = version
+        self.header_length = header_length
+        self.dscp = dscp
+        self.ecn = ecn
+        self.total_length = total_length
+        self.identification = identification
+        self.mf = mf
+        self.df = df
+        self.flag = flag
+        self.fragment_offset = fragment_offset
+        self.ttl = ttl
+        self.protocol = Layer3(protocol)
+        self.header_checksum = header_checksum
+        self._source_address = source_address
+        self._destination_address = destination_address
+        self.data = data
+        self.packet_type = PacketTypes(packet_type)
+        self.interface = interface
+
+    @property
+    def source(self) -> str:
+        return ipv4_to_str(self._source_address)
+
+    @property
+    def destination(self) -> str:
+        return ipv4_to_str(self._destination_address)
+
+    @staticmethod
+    def checksum(header: bytes) -> int:
+        c = 0
+        for i in range(0, len(header), 2):
+            c += int.from_bytes(header[i:i + 2], 'big')
+            while c > 0xffff:
+                c %= 0xffff
+        while c > 0xffff:
+            c %= 0xffff
+        return c ^ 0xffff
+
+    def get_header(self) -> bytes:
+        version_and_hlen = (self.version << 4) + self.header_length
+        dscp_and_ecn = (self.dscp << 2) + self.ecn
+        flags = (4 if self.flag else 0) + (2 if self.df else 0) + (1 if self.mf else 0)
+        df_mf_and_fragment = (flags << 12) + self.fragment_offset
+        return struct.pack(
+            '!BBHHHBBH4s4s', version_and_hlen, dscp_and_ecn, self.total_length,
+            self.identification, df_mf_and_fragment, self.ttl, self.protocol.value,
+            self.header_checksum, self._source_address, self._destination_address
+        )
+
+    @classmethod
+    def decode(cls, ether_frame: EtherFrame, packet: bytes, packet_type: int, interface: str) -> 'IPv4Packet':
+        if cls.checksum(packet[:20]):
+            raise ValueError(f'\nipv4 checksum failed, frame: {ether_frame.debug()}\n'
+                             f'packet: {binascii.hexlify(packet).decode()}, checksum: {hex(cls.checksum(packet[:20]))}')
+        data_len = len(packet) - 20
+        version_and_hlen, dscp_and_ecn, tlen, ident, df_mf_and_fragment, ttl, proto, checksum, source, dest = \
+            struct.unpack(
+                f'!BBHHHBBH4s4s', packet[:20]
+            )
+        version, hlen = split_byte(version_and_hlen)
+        flags = df_mf_and_fragment >> 13
+        mask = (flags << 13) | df_mf_and_fragment
+        fragment = mask ^ df_mf_and_fragment
+        flag, df, mf = False, False, False
+        if flags % 2:
+            mf = True
+            flags -= 1
+        if flags % 2:
+            df = True
+            flags -= 2
+        if flags % 4:
+            flag = True
+            flags -= 4
+        dscp, ecn = split_byte(dscp_and_ecn, 6)
+        return cls(
+            ether_frame, ord(version), ord(hlen), ord(dscp), ord(ecn), tlen, ident, mf, df, flag, fragment, ttl,
+            proto, checksum, source, dest, packet[20:], packet_type, interface
+        )
+
+    def encode(self) -> bytes:
+        return self.ether_frame.encode() + self.get_header() + self.data
+
+    @property
+    def printable_data(self) -> str:
+        return b".".join(printable.findall(self.data)).decode()
+
+    def __repr__(self) -> str:
+        return f"IPv4(protocol={self.protocol.name}, " \
+               f"iface={self.interface}, " \
+               f"type={self.packet_type.name}, " \
+               f"source={ipv4_to_str(self._source_address)}, " \
+               f"destination={ipv4_to_str(self._destination_address)}, " \
+               f"data_len={len(self.data)})"
+
+
+def make_filter(l3_protocol=None, src=None, dst=None, invert=False):
+    def filter_packet(packet: IPv4Packet):
+        if l3_protocol and not Layer3(packet.protocol) == l3_protocol:
+            return False
+        if src and not packet.source == src:
+            return False
+        if dst and not packet.destination == dst:
+            return False
+        return True
+    if invert:
+        return lambda packet: not filter_packet(packet)
+    return filter_packet
+
+
+async def sniff_ipv4(filters=None, kill=None):
+    start = time.perf_counter()
+    loop = asyncio.get_event_loop()
+
+    async def sock_recv(sock, n):
+        """Receive data from the socket.
+
+        The return value is a bytes object representing the data received.
+        The maximum amount of data to be received at once is specified by
+        nbytes.
+        """
+        if loop._debug and sock.gettimeout() != 0:
+            raise ValueError("the socket must be non-blocking")
+        fut = loop.create_future()
+        _sock_recv(fut, None, sock, n)
+        return await fut
+
+    def _sock_recv(fut, registered_fd, sock, n):
+        # _sock_recv() can add itself as an I/O callback if the operation can't
+        # be done immediately. Don't use it directly, call sock_recv().
+        if registered_fd is not None:
+            # Remove the callback early.  It should be rare that the
+            # selector says the fd is ready but the call still returns
+            # EAGAIN, and I am willing to take a hit in that case in
+            # order to simplify the common case.
+            loop.remove_reader(registered_fd)
+        if fut.cancelled():
+            return
+        try:
+            data, flags = sock.recvfrom(n)
+        except (BlockingIOError, InterruptedError):
+            fd = sock.fileno()
+            loop.add_reader(fd, _sock_recv, fut, fd, sock, n)
+        except Exception as exc:
+            fut.set_exception(exc)
+        else:
+            fut.set_result((data, flags))
+
+    with promiscuous('lo') as sock:
+        while True:
+            if not kill:
+                data, flags = await sock_recv(sock, 9000)
+            else:
+                t = asyncio.create_task(sock_recv(sock, 9000))
+                await asyncio.wait([t, kill.wait()], return_when=asyncio.FIRST_COMPLETED)
+                if kill.is_set():
+                    break
+                data, flags = await t
+            # https://stackoverflow.com/questions/42821309/how-to-interpret-result-of-recvfrom-raw-socket/45215859#45215859
+            if data:
+                try:
+                    ether_frame, packet = EtherFrame.decode(data)
+                    if ether_frame.ether_type == Layer2.IPv4.value:
+                        interface, _, packet_type, _, _ = flags
+                        ipv4 = IPv4Packet.decode(ether_frame, packet, packet_type, interface)
+                        if not filters or any((f(ipv4) for f in filters)):
+                            yield time.perf_counter() - start, ipv4
+                except ValueError:
+                    pass
+
+
+async def main():
+    loop = asyncio.get_event_loop()
+    gateway, lan = get_gateway_and_lan_addresses('default')
+
+    done = asyncio.Event()
+
+    def discover_aioupnp():
+        async def _discover():
+            print("testing aioupnp")
+            try:
+                u = await UPnP.discover()
+                print("successfully detected router with aioupnp")
+                try:
+                    await u.get_external_ip()
+                    print("successfully detected external ip with aioupnp")
+                except UPnPError:
+                    print("failed to detect external ip with aioupnp")
+                try:
+                    await u.get_redirects()
+                    print("successfully detected redirects with aioupnp")
+                except UPnPError:
+                    print("failed to get redirects with aioupnp")
+                try:
+                    external_port = await u.get_next_mapping(1234, 'TCP', 'aioupnp testing')
+                    print("successfully set redirect with aioupnp")
+                except UPnPError:
+                    print("failed to set redirect with aioupnp")
+                    external_port = None
+                try:
+                    await u.get_redirects()
+                    print("successfully detected redirects with aioupnp")
+                except UPnPError:
+                    print("failed to get redirects with aioupnp")
+                if external_port:
+                    try:
+                        print("successfully removed redirect with aioupnp")
+                        await u.delete_port_mapping(external_port, 'TCP')
+                    except UPnPError:
+                        print("failed to delete redirect with aioupnp")
+                    try:
+                        await u.get_redirects()
+                        print("successfully detected redirects with aioupnp")
+                    except UPnPError:
+                        print("failed to get redirects with aioupnp")
+            except UPnPError:
+                print("failed to discover router with aioupnp")
+            finally:
+                print("done with aioupnp test")
+        asyncio.create_task(_discover())
+
+    def discover_miniupnpc():
+        def _miniupnpc_discover():
+            try:
+                u = miniupnpc.UPnP()
+            except:
+                print("failed to create upnp object with miniupnpc")
+                return
+            try:
+                u.discover()
+            except:
+                print("failed to detect router with miniupnpc")
+                return
+            try:
+                u.selectigd()
+                print("successfully detected router with miniupnpc")
+            except:
+                print("failed to detect router with miniupnpc")
+                return
+            try:
+                u.externalipaddress()
+                print("successfully detected external ip with miniupnpc")
+            except:
+                print("failed to detect external ip with miniupnpc")
+                return
+
+        async def _discover():
+            print("testing miniupnpc")
+            try:
+                await loop.run_in_executor(None, _miniupnpc_discover)
+            finally:
+                done.set()
+                print("done with miniupnpc test")
+
+        asyncio.create_task(_discover())
+
+    loop.call_later(0, discover_aioupnp)
+    loop.call_later(8, discover_miniupnpc)
+    start = time.perf_counter()
+    packets = []
+    try:
+        async for (ts, ipv4_packet) in sniff_ipv4([
+                make_filter(l3_protocol=Layer3.UDP, src=SSDP_IP_ADDRESS),
+                make_filter(l3_protocol=Layer3.UDP, dst=SSDP_IP_ADDRESS),
+                make_filter(l3_protocol=Layer3.UDP, src=lan, dst=gateway),
+                make_filter(l3_protocol=Layer3.UDP, src=gateway, dst=lan),
+                make_filter(l3_protocol=Layer3.TCP, src=lan, dst=gateway),
+                make_filter(l3_protocol=Layer3.TCP, src=gateway, dst=lan)], done):
+            packets.append(
+                (time.perf_counter() - start, ipv4_packet.packet_type.name,
+                 ipv4_packet.source, ipv4_packet.destination, base64.b64encode(ipv4_packet.data).decode())
+            )
+    except KeyboardInterrupt:
+        print("stopping")
+    finally:
+        with open("aioupnp-bug-report.json", "w") as cap_file:
+            cap_file.write(json.dumps(packets))
+        print(f"Wrote bug report: {os.path.abspath('aioupnp-bug-report.json')}")
+        print("Sending bug report")
+        ssl_ctx = ssl.create_default_context(
+            purpose=ssl.Purpose.CLIENT_AUTH, capath=certifi.where()
+        )
+        auth = aiohttp.BasicAuth(
+            base64.b64decode(codecs.encode('Ax5LZzR1o3q3Z3WjATASDwR5rKyHH0qOIRIbLmMXn2H=', 'rot_13')).decode(), ''
+        )
+        report_id = base64.b64encode(os.urandom(16)).decode()
+        async with aiohttp.ClientSession() as session:
+            for i, (ts, direction, source, destination, packet) in enumerate(packets):
+                post = {
+                    'userId': report_id,
+                    'event': 'aioupnp bug report',
+                    'context': {
+                        'library': {
+                            'name': 'aioupnp',
+                            'version': aioupnp.__version__
+                        }
+                    },
+                    'properties': {
+                        'sequence': i,
+                        'ts': ts,
+                        'direction': direction,
+                        'source': source,
+                        'destination': destination,
+                        'packet': packet
+                    },
+                }
+                async with session.request(method='POST', url='https://api.segment.io/v1/track',
+                                           headers={'Connection': 'Close'}, auth=auth, json=post, ssl=ssl_ctx):
+                    sys.stdout.write(f"\r{'.' * i}")
+        sys.stdout.write("\n")
+        print("Successfully sent bug report, thanks for your contribution!")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

tests/__init__.py

@@ -18,7 +18,7 @@ except ImportError:
 @contextlib.contextmanager
 def mock_tcp_and_udp(loop, udp_expected_addr=None, udp_replies=None, udp_delay_reply=0.0, sent_udp_packets=None,
                      tcp_replies=None, tcp_delay_reply=0.0, sent_tcp_packets=None, add_potato_datagrams=False,
-                     raise_oserror_on_bind=False, raise_connectionerror=False):
+                     raise_oserror_on_bind=False, raise_connectionerror=False, tcp_chunk_size=100):
     sent_udp_packets = sent_udp_packets if sent_udp_packets is not None else []
     udp_replies = udp_replies or {}
 
@@ -36,8 +36,8 @@ def mock_tcp_and_udp(loop, udp_expected_addr=None, udp_replies=None, udp_delay_r
                     reply = tcp_replies[data]
                     i = 0
                     while i < len(reply):
-                        loop.call_later(tcp_delay_reply, p.data_received, reply[i:i+100])
-                        i += 100
+                        loop.call_later(tcp_delay_reply, p.data_received, reply[i:i+tcp_chunk_size])
+                        i += tcp_chunk_size
                     return
                 else:
                     pass
@@ -71,6 +71,7 @@ def mock_tcp_and_udp(loop, udp_expected_addr=None, udp_replies=None, udp_delay_r
                     loop.call_later(udp_delay_reply, p.datagram_received, udp_replies[(data, addr)],
                                     (udp_expected_addr, 1900))
 
+
             return _sendto
 
         protocol = proto_lam()

tests/protocols/test_scpd.py

@@ -31,6 +31,28 @@ class TestSCPDGet(AsyncioTestCase):
                    b"\r\n" \
                    b"%s" % bad_xml
 
+
+    bad_response2 = b"HTTP/1.1 200 OK\r\n" \
+                   b"CONTENT-TYPE: text/xml\r\n" \
+                   b"DATE: Thu, 18 Oct 2018 01:20:23 GMT\r\n" \
+                   b"LAST-MODIFIED: Fri, 28 Sep 2018 18:35:48 GMT\r\n" \
+                   b"SERVER: Linux/3.14.28-Prod_17.2, UPnP/1.0, Portable SDK for UPnP devices/1.6.22\r\n" \
+                   b"X-User-Agent: redsonic\r\n" \
+                   b"CONNECTION: close\r\n" \
+                   b"\r\n" \
+                   b"%s" % (b'.' * 65300)
+
+    bad_response3 = b"HTTP/1.1 200 OK\r\n" \
+                   b"CONTENT-TYPE: text/xml\r\n" \
+                   b"DATE: Thu, 18 Oct 2018 01:20:23 GMT\r\n" \
+                   b"LAST-MODIFIED: Fri, 28 Sep 2018 18:35:48 GMT\r\n" \
+                   b"CONTENT-TYPE: text/xml\r\n" \
+                   b"SERVER: Linux/3.14.28-Prod_17.2, UPnP/1.0, Portable SDK for UPnP devices/1.6.22\r\n" \
+                   b"X-User-Agent: redsonic\r\n" \
+                   b"CONNECTION: close\r\n" \
+                   b"\r\n" \
+                   b"<?xml version=\"1.0\"?>\n<root xmlns=\"urn:schemas-upnp-org:device-1-0\">\n<specVersion>\n<major>1</major>\n<minor>0</minor>\n</specVersion>\n<device>\n<deviceType>urn:schemas-upnp-org:device:InternetGatewayDevice:1</deviceType>\n<friendlyName>CGA4131COM</friendlyName>\n<manufacturer>Cisco</manufacturer>\n<manufacturerURL>http://www.cisco.com/</manufacturerURL>\n<modelDescription>CGA4131COM</modelDescription>\n<modelName>CGA4131COM</modelName>\n<modelNumber>CGA4131COM</modelNumber>\n<modelURL>http://www.cisco.com</modelURL>\n<serialNumber></serialNumber>\n<UDN>uuid:11111111-2222-3333-4444-555555555556</UDN>\n<UPC>CGA4131COM</UPC>\n<serviceList>\n<service>\n<serviceType>urn:schemas-upnp-org:service:Layer3Forwarding:1</serviceType>\n<serviceId>urn:upnp-org:serviceId:L3Forwarding1</serviceId>\n<SCPDURL>/Layer3ForwardingSCPD.xml</SCPDURL>\n<controlURL>/upnp/control/Layer3Forwarding</controlURL>\n<eventSubURL>/upnp/event/Layer3Forwarding</eventSubURL>\n</service>\n</serviceList>\n<deviceList>\n<device>\n<deviceType>urn:schemas-upnp-org:device:WANDevice:1</deviceType>\n<friendlyName>WANDevice:1</friendlyName>\n<manufacturer>Cisco</manufacturer>\n<manufacturerURL>http://www.cisco.com/</manufacturerURL>\n<modelDescription>CGA4131COM</modelDescription>\n<modelName>CGA4131COM</modelName>\n<modelNumber>CGA4131COM</modelNumber>\n<modelURL>http://www.cisco.com</modelURL>\n<serialNumber></serialNumber>\n<UDN>uuid:ebf5a0a0-1dd1-11b2-a92f-603d266f9915</UDN>\n<UPC>CGA4131COM</UPC>\n<serviceList>\n<service>\n<serviceType>urn:schemas-upnp-org:service:WANCommonInterfaceConfig:1</serviceType>\n<serviceId>urn:upnp-org:serviceId:WANCommonIFC1</serviceId>\n<SCPDURL>/WANCommonInterfaceConfigSCPD.xml</SCPDURL>\n<controlURL>/upnp/control/WANCommonInterfaceConfig0</controlURL>\n<eventSubURL>/upnp/event/WANCommonInterfaceConfig0</eventSubURL>\n</service>\n</serviceList>\n<deviceList>\n    <device>\n        <deviceType>urn:schemas-upnp-org:device:WANConnectionDevice:1</deviceType>\n        <friendlyName>WANConnectionDevice:1</friendlyName>\n        <manufacturer>Cisco</manufacturer>\n        <manufacturerURL>http://www.cisco.com/</manufacturerURL>\n        <modelDescription>CGA4131COM</modelDescription>\n        <modelName>CGA4131COM</modelName>\n        <modelNumber>CGA4131COM</modelNumber>\n        <modelURL>http://www.cisco.com</modelURL>\n        <serialNumber></serialNumber>\n        <UDN>uuid:11111111-2222-3333-4444-555555555555</UDN>\n        <UPC>CGA4131COM</UPC>\n        <serviceList>\n       <service>\n           <serviceType>urn:schemas-upnp-org:service:WANIPConnection:1</serviceType>\n           <serviceId>urn:upnp-org:serviceId:WANIPConn1</serviceId>\n           <SCPDURL>/WANIPConnectionServiceSCPD.xml</SCPDURL>\n           <controlURL>/upnp/control/WANIPConnection0</controlURL>\n           <eventSubURL>/upnp/event/WANIPConnection0</eventSubURL>\n       </service>\n        </serviceList>\n    </device>\n</deviceList>\n</device>\n</deviceList>\n<presentationURL>http://10.1.10.1/</presentationURL></device>\n</root>\n"
+
     expected_parsed = {
         'specVersion': {'major': '1', 'minor': '0'},
         'device': {
@@ -140,6 +162,26 @@ class TestSCPDGet(AsyncioTestCase):
             self.assertIsInstance(err, UPnPError)
             self.assertTrue(str(err).startswith('too many bytes written'))
 
+    async def test_scpd_get_overrun_unspecified_content_length(self):
+        sent = []
+        replies = {self.get_request: self.bad_response2}
+        with mock_tcp_and_udp(self.loop, tcp_replies=replies, sent_tcp_packets=sent):
+            result, raw, err = await scpd_get(self.path, self.lan_address, self.port, self.loop)
+            self.assertDictEqual({}, result)
+            self.assertEqual(self.bad_response2.decode(), raw.decode())
+            self.assertIsInstance(err, UPnPError)
+            self.assertTrue(str(err).endswith('with unspecified content length'))
+
+    async def test_scpd_duplicate_header(self):
+        sent = []
+        replies = {self.get_request: self.bad_response3}
+        with mock_tcp_and_udp(self.loop, tcp_replies=replies, sent_tcp_packets=sent):
+            result, raw, err = await scpd_get(self.path, self.lan_address, self.port, self.loop)
+            self.assertDictEqual({}, result)
+            self.assertTrue(self.bad_response3.startswith(raw))
+            self.assertIsInstance(err, UPnPError)
+            self.assertEqual("duplicate headers", str(err))
+
 
 class TestSCPDPost(AsyncioTestCase):
     param_names: list = []
@@ -202,7 +244,7 @@ class TestSCPDPost(AsyncioTestCase):
             self.assertEqual(b'', raw)
             self.assertDictEqual({}, result)
 
-    async def test_scpd_post_connection_error(self):
+    async def test_scpd_connection_error(self):
         sent = []
         replies = {}
         with mock_tcp_and_udp(self.loop, tcp_replies=replies, sent_tcp_packets=sent, raise_connectionerror=True):
@@ -214,6 +256,14 @@ class TestSCPDPost(AsyncioTestCase):
             self.assertEqual(b'', raw)
             self.assertDictEqual({}, result)
 
+            result, raw, err = await scpd_get(
+                self.path, self.gateway_address, self.port
+            )
+            self.assertIsInstance(err, UPnPError)
+            self.assertEqual('ConnectionRefusedError()', str(err))
+            self.assertEqual(b'', raw)
+            self.assertDictEqual({}, result)
+
     async def test_scpd_post_bad_xml_response(self):
         sent = []
         replies = {self.post_bytes: self.bad_envelope_response}

tests/protocols/test_ssdp.py

@@ -1,3 +1,4 @@
+import json
 from collections import OrderedDict
 from aioupnp.fault import UPnPError
 from aioupnp.protocols.m_search_patterns import packet_generator
@@ -50,6 +51,16 @@ class TestSSDP(AsyncioTestCase):
             with self.assertRaises(UPnPError):
                 await m_search("10.0.0.2", "10.0.0.1", self.successful_args, timeout=1, loop=self.loop)
 
+    async def test_ssdp_pretty_print(self):
+        self.assertEqual(
+            json.dumps({
+                "HOST": "239.255.255.250:1900",
+                "MAN": "ssdp:discover",
+                "MX": 1,
+                "ST": "urn:schemas-upnp-org:device:WANDevice:1"
+            }, indent=2), str(self.query_packet)
+        )
+
     async def test_m_search_reply_unicast(self):
         replies = {
             (self.query_packet.encode().encode(), ("10.0.0.1", 1900)): self.reply_packet.encode().encode()

tests/serialization/test_ssdp.py

@@ -28,6 +28,15 @@ class TestSSDPDatagram(unittest.TestCase):
         with self.assertRaises(UPnPError):
             SSDPDatagram.decode(packet)
 
+        packet = \
+        b'M-SEARCH * HTTP/1.1\r\n' \
+        b'Host: 239.255.255.250:1900\r\n' \
+        b'MX: 5\r\n' \
+        b'\r\n'
+
+        with self.assertRaises(UPnPError):
+            SSDPDatagram.decode(packet)
+
     def test_fail_to_decode_blank(self):
         packet = b''
 

tests/test_case_insensitive.py

@@ -2,7 +2,7 @@ import unittest
 from aioupnp.device import CaseInsensitive
 
 
-class TestService(CaseInsensitive):
+class _TestService(CaseInsensitive):
     serviceType = None
     serviceId = None
     controlURL = None
@@ -12,14 +12,14 @@ class TestService(CaseInsensitive):
 
 class TestCaseInsensitive(unittest.TestCase):
     def test_initialize(self):
-        s = TestService(
+        s = _TestService(
             serviceType="test", serviceId="test id", controlURL="/test", eventSubURL="/test2", SCPDURL="/test3"
         )
         self.assertEqual('test', getattr(s, 'serviceType'))
         self.assertEqual('test', getattr(s, 'servicetype'))
         self.assertEqual('test', getattr(s, 'SERVICETYPE'))
 
-        s = TestService(
+        s = _TestService(
             servicetype="test", serviceid="test id", controlURL="/test", eventSubURL="/test2", SCPDURL="/test3"
         )
         self.assertEqual('test', getattr(s, 'serviceType'))
@@ -35,7 +35,7 @@ class TestCaseInsensitive(unittest.TestCase):
         }, s.as_dict())
 
     def test_set_attr(self):
-        s = TestService(
+        s = _TestService(
             serviceType="test", serviceId="test id", controlURL="/test", eventSubURL="/test2", SCPDURL="/test3"
         )
         self.assertEqual('test', getattr(s, 'serviceType'))

tests/test_cli.py

@@ -62,7 +62,7 @@ Get the external ip address from the gateway
 """
 
 expected_add_port_mapping_usage = """aioupnp [-h] [--debug_logging] add_port_mapping [--external_port=<int>] [--protocol=<str>]
-  [--internal_port=<int>] [--lan_address=<str>] [--description=<str>]
+  [--internal_port=<int>] [--lan_address=<str>] [--description=<str>] [--lease_time=<int>]
 
 Add a new port mapping
 
@@ -71,12 +71,13 @@ Add a new port mapping
 :param internal_port: (int) internal port
 :param lan_address: (str) internal lan address
 :param description: (str) mapping description
+:param lease_time: (int) lease time in seconds
 :return: None
 
 """
 
 expected_get_next_mapping_usage = """aioupnp [-h] [--debug_logging] get_next_mapping [--port=<int>] [--protocol=<str>]
-  [--description=<str>] [--internal_port=<typing.Union[int, NoneType]>]
+  [--description=<str>] [--internal_port=<typing.Union[int, NoneType]>] [--lease_time=<int>]
 
 Get a new port mapping. If the requested port is not available, increment until the next free port is mapped
 
@@ -84,6 +85,7 @@ Get a new port mapping. If the requested port is not available, increment until
 :param protocol: (str) UDP | TCP
 :param description: (str) mapping description
 :param internal_port: (int) internal port
+:param lease_time: (int) lease time in seconds
 
 :return: (int) mapped port
 

tests/test_gateway.py

@@ -1,8 +1,12 @@
+import os
+import json
+from collections import OrderedDict
 from aioupnp.fault import UPnPError
 from tests import AsyncioTestCase, mock_tcp_and_udp
-from collections import OrderedDict
 from aioupnp.gateway import Gateway, get_action_list
 from aioupnp.serialization.ssdp import SSDPDatagram
+from aioupnp.serialization.soap import serialize_soap_post
+from aioupnp.upnp import UPnP
 
 
 def gen_get_bytes(location: str, host: str) -> bytes:
@@ -296,3 +300,76 @@ class TestDiscoverNetgearNighthawkAC2350(TestDiscoverDLinkDIR890L):
                                                                            'RequestConnection', 'ForceTermination',
                                                                            'GetStatusInfo', 'GetNATRSIPStatus']},
                     'soap_requests': []}
+
+
+class TestActiontec(AsyncioTestCase):
+    name = "Actiontec GT784WN"
+    _location_key = 'Location'
+
+    @property
+    def data_path(self):
+        return os.path.join(os.path.dirname(os.path.abspath(__file__)), "replays", self.name)
+
+    def _get_location(self):
+        # return self.gateway_info['reply']['Location'].split(self.gateway_address)[-1]
+        return self.gateway_info['reply'][self._location_key].split(f"{self.gateway_address}:{self.gateway_info['soap_port']}")[-1]
+
+    def setUp(self) -> None:
+        with open(self.data_path, 'r') as f:
+            data = json.loads(f.read())
+            self.gateway_info = data['gateway']
+            self.client_address = data['client_address']
+        self.gateway_address = self.gateway_info['gateway_address']
+        self.udp_replies = {
+            (SSDPDatagram('M-SEARCH', self.gateway_info['m_search_args']).encode().encode(), ("239.255.255.250", 1900)): SSDPDatagram("OK", self.gateway_info['reply']).encode().encode()
+        }
+        self.tcp_replies = {
+            (
+                f"GET {path} HTTP/1.1\r\n"
+                f"Accept-Encoding: gzip\r\n"
+                f"Host: {self.gateway_info['gateway_address']}\r\n"
+                f"Connection: Close\r\n"
+                f"\r\n"
+            ).encode(): xml_bytes.encode()
+            for path, xml_bytes in self.gateway_info['service_descriptors'].items()
+        }
+        self.tcp_replies.update({
+            (
+                f"GET {self._get_location()} HTTP/1.1\r\n"
+                f"Accept-Encoding: gzip\r\n"
+                f"Host: {self.gateway_info['gateway_address']}\r\n"
+                f"Connection: Close\r\n"
+                f"\r\n"
+            ).encode(): self.gateway_info['gateway_xml'].encode()
+        })
+        self.registered_soap_commands = self.gateway_info['registered_soap_commands']
+        super().setUp()
+
+    async def setup_request_replay(self, u: UPnP):
+        for method, reqs in self.gateway_info['soap_requests'].items():
+            if not reqs:
+                continue
+            self.tcp_replies.update({
+                serialize_soap_post(
+                    method, list(args.keys()), self.registered_soap_commands[method].encode(),
+                    self.gateway_address.encode(), u.gateway.services[self.registered_soap_commands[method]].controlURL.encode()
+                ): response.encode() for args, response in reqs
+            })
+
+    async def replay(self, u: UPnP):
+        self.assertEqual('11.222.33.111', await u.get_external_ip())
+
+    async def test_replay(self):
+        with mock_tcp_and_udp(self.loop, udp_replies=self.udp_replies, tcp_replies=self.tcp_replies, udp_expected_addr=self.gateway_address, tcp_chunk_size=1450):
+            u = await UPnP.discover(lan_address=self.client_address, gateway_address=self.gateway_address, loop=self.loop)
+            await self.setup_request_replay(u)
+            await self.replay(u)
+
+
+class TestNewMediaNet(TestActiontec):
+    name = "NewMedia-NET GmbH Generic X86"
+
+    async def replay(self, u: UPnP):
+        self.assertEqual('11.222.33.111', await u.get_external_ip())
+        await u.get_redirects()
+        # print(await u.get_next_mapping(4567, 'UDP', 'aioupnp test mapping'))

tests/test_interfaces.py

@@ -1,4 +1,6 @@
 from unittest import mock
+from collections import OrderedDict
+from aioupnp import interfaces
 from aioupnp.fault import UPnPError
 from aioupnp.upnp import UPnP
 from tests import AsyncioTestCase
@@ -29,22 +31,26 @@ class mock_netifaces:
     @staticmethod
     def ifaddresses(interface):
         return {
-            "test0": {
-                17: [
-                    {
-                        "addr": "01:02:03:04:05:06",
-                        "broadcast": "ff:ff:ff:ff:ff:ff"
-                    }
-                ],
-                2: [
-                    {
-                        "addr": "192.168.1.2",
-                        "netmask": "255.255.255.0",
-                        "broadcast": "192.168.1.255"
-                    }
-                ],
-            },
-        }[interface]
+            17: [
+                {
+                    "addr": "01:02:03:04:05:06",
+                    "broadcast": "ff:ff:ff:ff:ff:ff"
+                }
+            ],
+            2: [
+                {
+                    "addr": "192.168.1.2",
+                    "netmask": "255.255.255.0",
+                    "broadcast": "192.168.1.255"
+                }
+            ],
+        }
+
+
+class mock_netifaces_extra_interface(mock_netifaces):
+    @staticmethod
+    def interfaces():
+        return ['lo', 'test0', 'test1']
 
 
 class TestParseInterfaces(AsyncioTestCase):
@@ -68,3 +74,16 @@ class TestParseInterfaces(AsyncioTestCase):
             else:
                 self.assertTrue(False)
         self.assertEqual(len(checked), 1)
+
+    def test_guess_gateway(self):
+        # handle edge case where netifaces gives more interfaces than it does gateways
+        with mock.patch('aioupnp.interfaces.get_netifaces') as patch:
+            patch.return_value = mock_netifaces_extra_interface
+            self.assertDictEqual(
+                OrderedDict(
+                    [
+                        ('test0', ('192.168.1.1', '192.168.1.2')),
+                        ('test1', ('192.168.1.1', '192.168.1.2')),
+                        ('default', ('192.168.1.1', '192.168.1.2'))
+                    ]), interfaces.get_interfaces()
+            )

tests/test_upnp.py

@@ -51,6 +51,42 @@ class TestGetExternalIPAddress(UPnPCommandTestCase):
             external_ip = await upnp.get_external_ip()
             self.assertEqual("11.222.3.44", external_ip)
 
+    async def test_handle_got_ipv6(self):
+        request = b'POST /soap.cgi?service=WANIPConn1 HTTP/1.1\r\nHost: 11.2.3.4\r\nUser-Agent: python3/aioupnp, UPnP/1.0, MiniUPnPc/1.9\r\nContent-Length: 285\r\nContent-Type: text/xml\r\nSOAPAction: "urn:schemas-upnp-org:service:WANIPConnection:1#GetExternalIPAddress"\r\nConnection: Close\r\nCache-Control: no-cache\r\nPragma: no-cache\r\n\r\n<?xml version="1.0"?>\r\n<s:Envelope xmlns:s="http://schemas.xmlsoap.org/soap/envelope/" s:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/"><s:Body><u:GetExternalIPAddress xmlns:u="urn:schemas-upnp-org:service:WANIPConnection:1"></u:GetExternalIPAddress></s:Body></s:Envelope>\r\n'
+        self.replies.update({request: b"HTTP/1.1 200 OK\r\nServer: WebServer\r\nDate: Wed, 22 May 2019 03:25:57 GMT\r\nConnection: close\r\nCONTENT-TYPE: text/xml; charset=\"utf-8\"\r\nCONTENT-LENGTH: 392 \r\nEXT:\r\n\r\n<?xml version=\"1.0\"?>\n<s:Envelope xmlns:s=\"http://schemas.xmlsoap.org/soap/envelope/\" s:encodingStyle=\"http://schemas.xmlsoap.org/soap/encoding/\">\n\t<s:Body>\n\t\t<u:GetExternalIPAddressResponse xmlns:u=\"urn:schemas-upnp-org:service:WANIPConnection:1\">\n<NewExternalIPAddress>2a02:2e02:30c:db00:4a8d:36ff:feb0:5202</NewExternalIPAddress>\n</u:GetExternalIPAddressResponse>\n\t</s:Body>\n</s:Envelope>\n"})
+        self.addCleanup(self.replies.pop, request)
+        with mock_tcp_and_udp(self.loop, tcp_replies=self.replies):
+            gateway = Gateway(self.reply, self.client_address, self.gateway_address, loop=self.loop)
+            await gateway.discover_commands()
+            upnp = UPnP(self.client_address, self.gateway_address, gateway)
+            with self.assertRaises(UPnPError) as err:
+                await upnp.get_external_ip()
+                self.assertTrue(str(err).startswith("Got invalid external ipv4 address"))
+
+    async def test_handle_got_invalid_ip(self):
+        request = b'POST /soap.cgi?service=WANIPConn1 HTTP/1.1\r\nHost: 11.2.3.4\r\nUser-Agent: python3/aioupnp, UPnP/1.0, MiniUPnPc/1.9\r\nContent-Length: 285\r\nContent-Type: text/xml\r\nSOAPAction: "urn:schemas-upnp-org:service:WANIPConnection:1#GetExternalIPAddress"\r\nConnection: Close\r\nCache-Control: no-cache\r\nPragma: no-cache\r\n\r\n<?xml version="1.0"?>\r\n<s:Envelope xmlns:s="http://schemas.xmlsoap.org/soap/envelope/" s:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/"><s:Body><u:GetExternalIPAddress xmlns:u="urn:schemas-upnp-org:service:WANIPConnection:1"></u:GetExternalIPAddress></s:Body></s:Envelope>\r\n'
+        self.replies.update({request: b"HTTP/1.1 200 OK\r\nServer: WebServer\r\nDate: Wed, 22 May 2019 03:25:57 GMT\r\nConnection: close\r\nCONTENT-TYPE: text/xml; charset=\"utf-8\"\r\nCONTENT-LENGTH: 360 \r\nEXT:\r\n\r\n<?xml version=\"1.0\"?>\n<s:Envelope xmlns:s=\"http://schemas.xmlsoap.org/soap/envelope/\" s:encodingStyle=\"http://schemas.xmlsoap.org/soap/encoding/\">\n\t<s:Body>\n\t\t<u:GetExternalIPAddressResponse xmlns:u=\"urn:schemas-upnp-org:service:WANIPConnection:1\">\n<NewExternalIPAddress>potato</NewExternalIPAddress>\n</u:GetExternalIPAddressResponse>\n\t</s:Body>\n</s:Envelope>\n"})
+        self.addCleanup(self.replies.pop, request)
+        with mock_tcp_and_udp(self.loop, tcp_replies=self.replies):
+            gateway = Gateway(self.reply, self.client_address, self.gateway_address, loop=self.loop)
+            await gateway.discover_commands()
+            upnp = UPnP(self.client_address, self.gateway_address, gateway)
+            with self.assertRaises(UPnPError) as err:
+                await upnp.get_external_ip()
+                self.assertTrue(str(err).startswith("Got invalid external ipv4 address"))
+
+    async def test_handle_got_carrier_nat_ip(self):
+        request = b'POST /soap.cgi?service=WANIPConn1 HTTP/1.1\r\nHost: 11.2.3.4\r\nUser-Agent: python3/aioupnp, UPnP/1.0, MiniUPnPc/1.9\r\nContent-Length: 285\r\nContent-Type: text/xml\r\nSOAPAction: "urn:schemas-upnp-org:service:WANIPConnection:1#GetExternalIPAddress"\r\nConnection: Close\r\nCache-Control: no-cache\r\nPragma: no-cache\r\n\r\n<?xml version="1.0"?>\r\n<s:Envelope xmlns:s="http://schemas.xmlsoap.org/soap/envelope/" s:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/"><s:Body><u:GetExternalIPAddress xmlns:u="urn:schemas-upnp-org:service:WANIPConnection:1"></u:GetExternalIPAddress></s:Body></s:Envelope>\r\n'
+        self.replies.update({request: b"HTTP/1.1 200 OK\r\nServer: WebServer\r\nDate: Wed, 22 May 2019 03:25:57 GMT\r\nConnection: close\r\nCONTENT-TYPE: text/xml; charset=\"utf-8\"\r\nCONTENT-LENGTH: 365 \r\nEXT:\r\n\r\n<?xml version=\"1.0\"?>\n<s:Envelope xmlns:s=\"http://schemas.xmlsoap.org/soap/envelope/\" s:encodingStyle=\"http://schemas.xmlsoap.org/soap/encoding/\">\n\t<s:Body>\n\t\t<u:GetExternalIPAddressResponse xmlns:u=\"urn:schemas-upnp-org:service:WANIPConnection:1\">\n<NewExternalIPAddress>192.88.99.2</NewExternalIPAddress>\n</u:GetExternalIPAddressResponse>\n\t</s:Body>\n</s:Envelope>\n"})
+        self.addCleanup(self.replies.pop, request)
+        with mock_tcp_and_udp(self.loop, tcp_replies=self.replies):
+            gateway = Gateway(self.reply, self.client_address, self.gateway_address, loop=self.loop)
+            await gateway.discover_commands()
+            upnp = UPnP(self.client_address, self.gateway_address, gateway)
+            with self.assertRaises(UPnPError) as err:
+                await upnp.get_external_ip()
+                self.assertTrue(str(err).startswith("Got invalid external ipv4 address"))
+
     async def test_null_external_ip(self):
         request = b'POST /soap.cgi?service=WANIPConn1 HTTP/1.1\r\nHost: 11.2.3.4\r\nUser-Agent: python3/aioupnp, UPnP/1.0, MiniUPnPc/1.9\r\nContent-Length: 285\r\nContent-Type: text/xml\r\nSOAPAction: "urn:schemas-upnp-org:service:WANIPConnection:1#GetExternalIPAddress"\r\nConnection: Close\r\nCache-Control: no-cache\r\nPragma: no-cache\r\n\r\n<?xml version="1.0"?>\r\n<s:Envelope xmlns:s="http://schemas.xmlsoap.org/soap/envelope/" s:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/"><s:Body><u:GetExternalIPAddress xmlns:u="urn:schemas-upnp-org:service:WANIPConnection:1"></u:GetExternalIPAddress></s:Body></s:Envelope>\r\n'
         self.replies.update({