Python class for calling methods of a REST based API. Supports basic authentication over HTTPS

APIWithBasicAuth.py

# Copyright (c) 2011 Cole Krumbholz
# 
# Permission is hereby granted, free of charge, to any person obtaining a 
# copy of this software and associated documentation files (the "Software"), 
# to deal in the Software without restriction, including without limitation 
# the rights to use, copy, modify, merge, publish, distribute, sublicense,
# and/or sell copies of the Software, and to permit persons to whom the 
# Software is furnished to do so, subject to the following conditions:
# 
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
# 
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS 
# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL 
# THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR 
# OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, 
# ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR 
# OTHER DEALINGS IN THE SOFTWARE.

# A good chunk of this library was adapted from the following article:
# http://thejosephturner.com/blog/2011/03/19/https-certificate-verification-in-python-with-urllib2/
# Thanks Joseph, you're the shit.

import urllib
import urllib2
import httplib
import socket
import ssl

def verified_HTTPS_connect(connection):
    # given an httplib.HTTPSConnection object,
    # creates a connection and verifies CA certs
    sock = socket.create_connection((connection.host, connection.port), 
                                    connection.timeout)
    
    # tunneling only supported in python 2.6.3 and later
    if hasattr(connection, '_tunnel_host') and \
       connection._tunnel_host:
        connection.sock = sock
        connection._tunnel()

    # wrap the socket using verification with the root
    #    certs in ca_cert_file
    
    cert_file_path = connection.ca_cert_file if \
        hasattr(connection, 'ca_cert_file') else None

    if not cert_file_path:
        cert_reqs = ssl.CERT_NONE
    else:
        cert_reqs = ssl.CERT_REQUIRED
    
    connection.sock = ssl.wrap_socket(sock,
                                      connection.key_file,
                                      connection.cert_file,
                                      cert_reqs=cert_reqs,
                                      ca_certs=cert_file_path)

# wraps https connections with ssl certificate verification
class VerifiedHTTPSHandler(urllib2.HTTPSHandler):
    def __init__(self, connection_class):
        self.specialized_conn_class = connection_class
        urllib2.HTTPSHandler.__init__(self)

    def https_open(self, req):
        return self.do_open(self.specialized_conn_class, req)

class RequestWithMethod(urllib2.Request):
    def __init__(self, *args, **kwargs):
        self._method = kwargs.pop('method', None)
        urllib2.Request.__init__(self, *args, **kwargs)

    def get_method(self):
        return self._method if self._method \
            else urllib2.Request.get_method(self)

class APIWithBasicAuth():
    
    def __init__(self, url, username=None, password=None, ca_cert_file=None):
        handlers = []
        
        VerifiedHTTPSConnection = type(
            'VerifiedHTTPSConnection',
            (httplib.HTTPSConnection, object),
            dict(
                ca_cert_file = ca_cert_file,
                connect = verified_HTTPS_connect,
            )
        )
        handlers.append(VerifiedHTTPSHandler(VerifiedHTTPSConnection))
            
        if username:
            auth_handler = urllib2.HTTPBasicAuthHandler(
                                urllib2.HTTPPasswordMgrWithDefaultRealm())
            auth_handler.add_password(realm=None,
                                      uri=(url,),
                                      user=username,
                                      passwd=password)
            handlers.append(auth_handler)

        self.url_opener = urllib2.build_opener(*handlers)

    def call(self, url, method=None, data=None, timeout=None):
        if data:
            data = urllib.urlencode(data)
        request = RequestWithMethod(url,
                                    method=method,
                                    data=data)
        try:
            handle = self.url_opener.open(request, timeout=timeout)
            response = handle.read()
            headers = handle.info().headers
            handle.close()
            code = 200
        except urllib2.HTTPError, e:
            response = e.read()
            headers = e.info().headers
            code = e.code
        return code, response, headers