oci/zeronet
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Decodes IGD response to utf-8; Makes function arguments immutable;

Browse source
This commit is contained in:
sirMackk 2017-01-11 16:43:39 +01:00
parent 2425f9f03e
commit 1d8fe61920
1 changed files with 11 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
src/util/UpnpPunch.py
View file

@ -4,6 +4,7 @@ import httplib
import logging
from urlparse import urlparse
from xml.dom.minidom import parseString
from xml.parsers.expat import ExpatError
from gevent import socket
@ -82,7 +83,7 @@ def _retrieve_igd_profile(url):
Retrieve the device's UPnP profile.
"""
try:
return urllib2.urlopen(url.geturl(), timeout=5).read()
return urllib2.urlopen(url.geturl(), timeout=5).read().decode('utf-8')
except socket.error:
raise IGDError('IGD profile query timed out')
@ -100,7 +101,11 @@ def _parse_igd_profile(profile_xml):
WANIPConnection or WANPPPConnection and return
the 'controlURL' and the service xml schema.
"""
dom = parseString(profile_xml)
try:
dom = parseString(profile_xml)
except ExpatError as e:
raise IGDError(
'Unable to parse IGD reply: {0} \n\n\n {1}'.format(profile_xml, e))
service_types = dom.getElementsByTagName('serviceType')
for service in service_types:
@ -268,7 +273,7 @@ def _send_requests(messages, location, upnp_schema, control_path):
raise UpnpError('Sending requests using UPnP failed.')
def _orchestrate_soap_request(ip, port, msg_fn, desc=None, protos=['TCP', 'UDP']):
def _orchestrate_soap_request(ip, port, msg_fn, desc=None, protos=("TCP", "UDP")):
logging.debug("Trying using local ip: %s" % ip)
idg_data = _collect_idg_data(ip)
@ -284,7 +289,7 @@ def _communicate_with_igd(port=15441,
desc="UpnpPunch",
retries=3,
fn=_create_open_message,
protos=["TCP", "UDP"]):
protos=("TCP", "UDP")):
"""
Manage sending a message generated by 'fn'.
"""
@ -310,7 +315,7 @@ def _communicate_with_igd(port=15441,
port, retries))
def ask_to_open_port(port=15441, desc="UpnpPunch", retries=3, protos=["TCP", "UDP"]):
def ask_to_open_port(port=15441, desc="UpnpPunch", retries=3, protos=("TCP", "UDP")):
logging.debug("Trying to open port %d." % port)
_communicate_with_igd(port=port,
desc=desc,
@ -319,7 +324,7 @@ def ask_to_open_port(port=15441, desc="UpnpPunch", retries=3, protos=["TCP", "UD
protos=protos)
def ask_to_close_port(port=15441, desc="UpnpPunch", retries=3, protos=["TCP", "UDP"]):
def ask_to_close_port(port=15441, desc="UpnpPunch", retries=3, protos=("TCP", "UDP")):
logging.debug("Trying to close port %d." % port)
# retries=1 because multiple successes cause 500 response and failure
_communicate_with_igd(port=port,