Using Python and UPnP to Forward a Port

Published: 2015-02-22

Tagged: python networking guide

I'm currently working on a sweet piece of code for the ZeroNet project, which is really cool and you should check it out and contribute if you have the time and love things like p2p, and bitcoin. The code uses UPnP to tell the router to forward a specific port and it only uses modules from Python's awesome standard library. A user can do this manually, but why should a human do a machine's job?

First I'll give a top level overview of what technologies are involved and how they interact together. Later, how to get it to work using vanilla Python, but if you want, skip right to it

What's involved

The main component is UPnP or Universal Plug and Play. In short, UPnP is a network protocol that allows network-enabled devices to discover other devices that use UPnP and communicate with them. UPnP focuses on Addressing, Discovery, Description, Control, Events, and Presentation. Each part does its share of work in facilitating communication between devices, but we'll focus on Discovery, Description, and Control. A more in-depth look can be found here.

Discovery, as the name implies, is about UPnP devices discovering other UPnP-enabled devices on the network. This step utilizes the HTTPU protocol, which in the simplest terms, is pretty much your regular HTTP except it rides on top of UDP instead of TCP. To discover UPnP-enabled devices, we send out a HTTPU message to the 239.255.255.250 broadcast IP on port 1900, which are specified in the UPnP spec. This message includes a few interesting headers, but one stands out in particular: ST. ST stands for "search target". Only devices that match the search target will reply to your message. If you put ssdp:all in this header, all UPnP devices on the network should reply to your message. Since we're interested only in routers, our ST header will contain urn:schemas-upnp-org:device:InternetGatewayDevice:1. The responses to this message will include the IP addresses of each device matching the ST, as well as a device profile. The device profile information is located in the "location" header in the response.

Now we have the IP of any routers on the network as well as the device profile information. The device profile information is simply a URL pointing at an XML file served up by that device. It contains a list of services that this device offers. These services range from returning information about the device's current status to accepting and setting internal parameters. The service we're looking for is WANIPConnection or WANPPPConnection. Both of these services provide an API to query or set internal router variables. Some of the variables we can read are ConnectionStatus or Uptime. Some of the actions available to us are GetExternalIPAddress or AddPortMapping. The last one is what we're going to use to, as the name implies, add a port mapping.

To sum up the Description part: we need to get the device's main profile. The location of that profile is contained in the "location" header of the reply we get back from the Discovery phase. We send a regular GET request to that location, download the XML file and parse it, looking for one of two services - WANIPConnection or WANPPPConnection (contained in a "serviceType" tag). If we found either one, then one of its sibling elements is called "controlURL" and contains the URL of a SOAP API endpoint. POST requests submitted to that endpoint will change a router's internal state.

This is where we get to the Control part of UPnP. Control encompasses both services variables as well as ways to change them through API calls. Both of the documents I linked to above contain a chapter on this stage called Actions.

We will make SOAP calls to the URL we obtained from the "controlURL" element. Normally, you'd want to work with SOAP using some library to abstract away a lot of the nitty gritty details, but punching open a port requires a single SOAP request. I used tcpdump to intercept that kind of request and simply copied it over since a SOAP request is XML riding inside of an HTTP request - pure ASCII text.

Python has a wonderful built-in library for handling XML, including things like attributes or namespaces, but if we're interested in making a single request, I think it's easier to keep that request a string and interpolate it with variables.

Once we have prepared the XML payload and made a POST request to the controlURL, we will get a response. If the SOAP request is mangled, we'll get back a HTTP response with code 500, and the body will contain a SOAP response, which we can parse with the built-in XML module. If the request was correct, we will get a HTTP response with status code 200 and a small SOAP response saying everything is OK.

One thing that caught me off guard here was the fact that the elements of the SOAP request have to be in specific order. Change the sequence of SOAP function parameters and you'll get back a 500 error with the 402 UPnP error.

Laying down some code

First, we'll make a simple M-SEARCH request to the broadcast address 239.255.255.250 on port 1900 to ping any UPnP enabled devices with the IGD profile:

import socket

SSDP_ADDR = "239.255.255.250"
SSDP_PORT = 1900
SSDP_MX = 2
SSDP_ST = "urn:schemas-upnp-org:device:InternetGatewayDevice:1"

ssdpRequest = "M-SEARCH * HTTP/1.1\r\n" + \
                "HOST: %s:%d\r\n" % (SSDP_ADDR, SSDP_PORT) + \
                "MAN: \"ssdp:discover\"\r\n" + \
                "MX: %d\r\n" % (SSDP_MX, ) + \
                "ST: %s\r\n" % (SSDP_ST, ) + "\r\n"

sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.sendto(ssdpRequest, (SSDP_ADDR, SSDP_PORT))
resp = sock.recv(1000)

Thanks to Trinh Hoang Nhu, I didn't have to poke around for the right search target. Now would be a good time to refresh your knowledge of sockets if you were rusty like me. The response to that request, stored in the resp variable is supposed to contain a location header with the device profile.

We can easily parse the response using regular expressions and use urlparse.urlparse to store the url in an object for easier use later on. We'll also use a simple filter to isolate the contents of the location header into its own variable:

import re
from urlparse import urlparse

parsed = re.findall(r'(?P<name>.*?): (?P<value>.*?)\r\n', resp)

# get the location header
location = filter(lambda x: x[0].lower() == "location", parsed)

# use the urlparse function to create an easy to use object to hold a URL
router_path = urlparse(location[0][1])

If location holds anything then we're in luck. We have to obtain the file pointed at by the location header and parse it, searching for the right serviceType element. Here we can use Python's built-in xml.dom.minidom or xml.etree.ElementTree. I picked xml.dom.minidom because it supports javascript-like syntax for exploring the DOM. More info about these two modules here.

import urllib2
from xml.dom.minidom import parseString
# get the profile xml file and read it into a variable
directory = urllib2.urlopen(location[0][1]).read()

# create a DOM object that represents the `directory` document
dom = parseString(directory)

# find all 'serviceType' elements
service_types = dom.getElementsByTagName('serviceType')

# iterate over service_types until we get either WANIPConnection
# (this should also check for WANPPPConnection, which, if I remember correctly
# exposed a similar SOAP interface on ADSL routers.
for service in service_types:
    # I'm using the fact that a 'serviceType' element contains a single text node, who's data can
    # be accessed by the 'data' attribute.
    # When I find the right element, I take a step up into its parent and search for 'controlURL'
    if service.childNodes[0].data.find('WANIPConnection') > 0:
        path = service.parentNode.getElementsByTagName('controlURL')[0].childNodes[0].data)

If path isn't empty, it's all downhill from here. At this point we have the correct address of where to submit our SOAP requests - IP, port, path. Now all we need is to actually create the SOAP request and POST it to that address and listen for a reply. In the simple case of sending out only one request, I'd use a string representation of the request and interpolate the values, but since we have our learning hats on, I'll use the Document class to build a SOAP request node by node. This also demonstrates how to create namespaced elements using xml.dom.minidom.

from xml.dom.minidom import Document

doc = Document()

# create the envelope element and set its attributes
envelope = doc.createElementNS('', 's:Envelope')
envelope.setAttribute('xmlns:s', 'http://schemas.xmlsoap.org/soap/envelope/')
envelope.setAttribute('s:encodingStyle', 'http://schemas.xmlsoap.org/soap/encoding/')

# create the body element
body = doc.createElementNS('', 's:Body')

# create the function element and set its attribute
fn = doc.createElementNS('', 'u:AddPortMapping')
fn.setAttribute('xmlns:u', 'urn:schemas-upnp-org:service:WANIPConnection:1')

# setup the argument element names and values
# using a list of tuples to preserve order
arguments = [
    ('NewExternalPort', '35000'),           # specify port on router
    ('NewProtocol', 'TCP'),                 # specify protocol
    ('NewInternalPort', '35000'),           # specify port on internal host
    ('NewInternalClient', '192.168.1.90'), # specify IP of internal host
    ('NewEnabled', '1'),                    # turn mapping ON
    ('NewPortMappingDescription', 'Test desc'), # add a description
    ('NewLeaseDuration', '0')]              # how long should it be opened?

# NewEnabled should be 1 by default, but better supply it.
# NewPortMappingDescription Can be anything you want, even an empty string.
# NewLeaseDuration can be any integer BUT some UPnP devices don't support it,
# so set it to 0 for better compatibility.

# container for created nodes
argument_list = []

# iterate over arguments, create nodes, create text nodes,
# append text nodes to nodes, and finally add the ready product
# to argument_list
for k, v in arguments:
    tmp_node = doc.createElement(k)
    tmp_text_node = doc.createTextNode(v)
    tmp_node.appendChild(tmp_text_node)
    argument_list.append(tmp_node)

# append the prepared argument nodes to the function element
for arg in argument_list:
    fn.appendChild(arg)

# append function element to the body element
body.appendChild(fn)

# append body element to envelope element
envelope.appendChild(body)

# append envelope element to document, making it the root element
doc.appendChild(envelope)

# our tree is ready, conver it to a string
pure_xml = doc.toxml()

We have our payload ready as a string with no newlines. We can now create a HTTP POST request and add pure_xml as it's body. For this, we'll use the HTTPConnection class from the <code>httplib</code> module. HTTPConnection allows us to specify the host and port of the receiving device, as well as the method, path, body, as well as any extra headers. We'll make use of all of these!

import httplib

# use the object returned by urlparse.urlparse to get the hostname and port
conn = httplib.HTTPConnection(router_path.hostname, router_path.port)

# use the path of WANIPConnection (or WANPPPConnection) to target that service,
# insert the xml payload,
# add two headers to make tell the server what we're sending exactly.
conn.request('POST',
    path,
    pure_xml,
    {'SOAPAction': '"urn:schemas-upnp-org:service:WANIPConnection:1#AddPortMapping"',
     'Content-Type': 'text/xml'}
)

# wait for a response
resp = conn.getresponse()

# print the response status
print resp.status

# print the response body
print resp.read()

If all went well, the status should be a code 200 and the response body should contain a minimal XML formatted SOAP message along the line of "OK". If things didn't go well, the response will most likely be a code 500 and the response body will contain an XML formatted SOAP message that provides a UPnP specified error code and a short (too short) description of what went wrong. Combined with the UPnP spec and documentation, it should be fairly easy to find out what went wrong.

It's easy to build on this and check out what other services are available on your router. Why even stop at routers? What about your Roku box? How about networked printer? How about the router at your local coffeeshop? There's a ton of devices implementing UPnP all waiting to be explored.

For a look at the whole code, check it out in my fork of ZeroNet. I've tested it on 3 routers so far and it appears to work correctly so I think I'll be able to submit a pull request to ZeroNet soon :).