LBRYCommunity/aioupnp
1
0
Fork 0
Code Issues 7 Pull requests 1 Projects Releases 1 Packages Wiki Activity

fix https://github.com/lbryio/aioupnp/issues/14

Browse source
This commit is contained in:
Jack Robison 2019-08-15 14:35:57 -04:00
parent 6e8eb9a87f
commit a0d1ca73e0
No known key found for this signature in database
GPG key ID: DF25C68FE0239BB2
2 changed files with 52 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
aioupnp/commands.py
View file

@ -4,6 +4,7 @@ import typing
import logging
from aioupnp.protocols.scpd import scpd_post
from aioupnp.device import Service
from aioupnp.fault import UPnPError
log = logging.getLogger(__name__)
@ -45,17 +46,28 @@ class SCPDRequestDebuggingInfo(typing.NamedTuple):
ts: float
def recast_return(return_annotation, result: typing.Dict[str, typing.Union[int, str]],
def recast_return(return_annotation, result: typing.Union[str, int, bool, typing.Dict[str, typing.Union[int, str]]],
result_keys: typing.List[str]) -> typing.Optional[
typing.Union[str, int, bool, GetSpecificPortMappingEntryResponse, GetGenericPortMappingEntryResponse]]:
if len(result_keys) == 1:
if isinstance(result, (str, int, bool)):
single_result = result
else:
if result_keys[0] in result:
single_result = result[result_keys[0]]
else: # check for the field having incorrect capitalization
flattened = {k.lower(): v for k, v in result.items()}
if result_keys[0].lower() in flattened:
single_result = flattened[result_keys[0].lower()]
else:
raise UPnPError(f"expected response key {result_keys[0]}, got {list(result.keys())}")
if return_annotation is bool:
return soap_bool(single_result)
if return_annotation is str:
return soap_optional_str(single_result)
return int(result[result_keys[0]]) if result_keys[0] in result else None
return None if single_result is None else int(single_result)
elif return_annotation in [GetGenericPortMappingEntryResponse, GetSpecificPortMappingEntryResponse]:
assert isinstance(result, dict)
arg_types: typing.Dict[str, typing.Type[typing.Any]] = return_annotation._field_types
assert len(arg_types) == len(result_keys)
recast_results: typing.Dict[str, typing.Optional[typing.Union[str, int, bool]]] = {}

37
tests/test_upnp.py
View file

@ -51,6 +51,43 @@ class TestGetExternalIPAddress(UPnPCommandTestCase):
self.assertEqual("11.222.3.44", external_ip)
class TestMalformedGetExternalIPAddressResponse(UPnPCommandTestCase):
client_address = '11.2.3.222'
get_ip_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'
async def test_response_key_mismatch(self):
self.replies.update({self.get_ip_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: 333 \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"
b"<derp>11.222.3.44</derp>\n</u:GetExternalIPAddressResponse>\n\t</s:Body>\n</s:Envelope>\n"})
self.addCleanup(self.replies.pop, self.get_ip_request)
with mock_tcp_and_udp(self.loop, tcp_replies=self.replies):
gateway = Gateway(self.reply, self.m_search_args, self.client_address, self.gateway_address)
await gateway.discover_commands(self.loop)
upnp = UPnP(self.client_address, self.gateway_address, gateway)
with self.assertRaises(UPnPError):
await upnp.get_external_ip()
async def test_response_key_case_sensitivity(self):
self.replies.update({self.get_ip_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"
b"<newexternalipaddress>11.222.3.44</newexternalipaddress>\n</u:GetExternalIPAddressResponse>\n\t</s:Body>\n</s:Envelope>\n"})
self.addCleanup(self.replies.pop, self.get_ip_request)
with mock_tcp_and_udp(self.loop, tcp_replies=self.replies):
gateway = Gateway(self.reply, self.m_search_args, self.client_address, self.gateway_address)
await gateway.discover_commands(self.loop)
upnp = UPnP(self.client_address, self.gateway_address, gateway)
external_ip = await upnp.get_external_ip()
self.assertEqual("11.222.3.44", external_ip)
async def test_non_encapsulated_single_field_response(self):
self.replies.update({self.get_ip_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: 320 \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"
b"11.222.3.44\n</u:GetExternalIPAddressResponse>\n\t</s:Body>\n</s:Envelope>\n"})
self.addCleanup(self.replies.pop, self.get_ip_request)
with mock_tcp_and_udp(self.loop, tcp_replies=self.replies):
gateway = Gateway(self.reply, self.m_search_args, self.client_address, self.gateway_address)
await gateway.discover_commands(self.loop)
upnp = UPnP(self.client_address, self.gateway_address, gateway)
external_ip = await upnp.get_external_ip()
self.assertEqual("11.222.3.44", external_ip)
class TestGetGenericPortMappingEntry(UPnPCommandTestCase):
def setUp(self) -> None:
query = 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: 341\r\nContent-Type: text/xml\r\nSOAPAction: "urn:schemas-upnp-org:service:WANIPConnection:1#GetGenericPortMappingEntry"\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:GetGenericPortMappingEntry xmlns:u="urn:schemas-upnp-org:service:WANIPConnection:1"><NewPortMappingIndex>0</NewPortMappingIndex></u:GetGenericPortMappingEntry></s:Body></s:Envelope>\r\n'