diff --git a/aioupnp/util.py b/aioupnp/util.py index 540d33e..1adfa07 100644 --- a/aioupnp/util.py +++ b/aioupnp/util.py @@ -1,3 +1,4 @@ +import sys import ipaddress import typing from collections import OrderedDict @@ -62,7 +63,7 @@ def is_valid_public_ipv4(address): 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.supernet_of(ipaddress.ip_network(f"{address}/32")), - IPV4_TO_6_RELAY_SUBNET.supernet_of(ipaddress.ip_network(f"{address}/32")))) + 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 diff --git a/tests/test_upnp.py b/tests/test_upnp.py index cec00f6..4b9abf8 100644 --- a/tests/test_upnp.py +++ b/tests/test_upnp.py @@ -75,6 +75,18 @@ class TestGetExternalIPAddress(UPnPCommandTestCase): 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\r\n\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\n\n\t\n\t\t\n192.88.99.2\n\n\t\n\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\r\n\r\n' self.replies.update({