import sys

if sys.version_info >= (3,):

import http.client as httplib

import urllib.parse as urlparse

import urllib.request as urllib2

else:

import httplib

import urllib2

import urlparse

try:

import ssl

try:

from ssl import match_hostname, CertificateError

except ImportError:

from tuf.compatibility.ssl_match_hostname import match_hostname, CertificateError

except ImportError:

ssl = None

import socket

if not hasattr(socket, 'create_connection'): # for Python 2.5

# monkey-patch socket module

from tuf.compatibility.socket_create_connection import create_connection

socket.create_connection = create_connection

_GLOBAL_DEFAULT_TIMEOUT = object()

def create_connection(address, timeout=_GLOBAL_DEFAULT_TIMEOUT,

source_address=None):

"""Connect to *address* and return the socket object.

host, port = address

err = None

for res in getaddrinfo(host, port, 0, SOCK_STREAM):

af, socktype, proto, canonname, sa = res

sock = None

try:

sock = socket(af, socktype, proto)

if timeout is not _GLOBAL_DEFAULT_TIMEOUT:

sock.settimeout(timeout)

if source_address:

sock.bind(source_address)

sock.connect(sa)

return sock

except error as _:

err = _

if sock is not None:

sock.close()

if err is not None:

raise err

else:

raise error("getaddrinfo returns an empty list")

import re

class CertificateError(ValueError):

pass

def _dnsname_to_pat(dn):

pats = []

for frag in dn.split(r'.'):

if frag == '*':

# When '*' is a fragment by itself, it matches a non-empty dotless

# fragment.

pats.append('[^.]+')

else:

# Otherwise, '*' matches any dotless fragment.

frag = re.escape(frag)

pats.append(frag.replace(r'\*', '[^.]*'))

return re.compile(r'\A' + r'\.'.join(pats) + r'\Z', re.IGNORECASE)

def match_hostname(cert, hostname):

"""Verify that *cert* (in decoded format as returned by

if not cert:

raise ValueError("empty or no certificate")

dnsnames = []

san = cert.get('subjectAltName', ())

for key, value in san:

if key == 'DNS':

if _dnsname_to_pat(value).match(hostname):

return

dnsnames.append(value)

if not dnsnames:

# The subject is only checked when there is no dNSName entry

# in subjectAltName

for sub in cert.get('subject', ()):

for key, value in sub:

# XXX according to RFC 2818, the most specific Common Name

# must be used.

if key == 'commonName':

if _dnsname_to_pat(value).match(hostname):

return

dnsnames.append(value)

if len(dnsnames) > 1:

raise CertificateError("hostname %r "

"doesn't match either of %s"

% (hostname, ', '.join(map(repr, dnsnames))))

elif len(dnsnames) == 1:

raise CertificateError("hostname %r "

"doesn't match %r"

% (hostname, dnsnames[0]))

else:

raise CertificateError("no appropriate commonName or "

"subjectAltName fields were found")

ca_certs = None

import os.path

import socket

import tuf

from tuf.compatibility import httplib, ssl, urllib2, urlparse

if ssl:

from tuf.compatibility import match_hostname

else:

raise tuf.Error( "No SSL support!" ) # TODO: degrade gracefully

class VerifiedHTTPSConnection( httplib.HTTPSConnection ):

"""

def connect(self):

self.connection_kwargs = {}

#TODO: refactor compatibility logic into tuf.compatibility?

# for > py2.5

if hasattr(self, 'timeout'):

self.connection_kwargs.update(timeout = self.timeout)

# for >= py2.7

if hasattr(self, 'source_address'):

self.connection_kwargs.update(source_address = self.source_address)

sock = socket.create_connection((self.host, self.port), **self.connection_kwargs)

# for >= py2.7

if getattr(self, '_tunnel_host', None):

self.sock = sock

self._tunnel()

# set location of certificate authorities

assert os.path.isfile( tuf.conf.ca_certs )

cert_path = tuf.conf.ca_certs

self.sock = ssl.wrap_socket(sock,

self.key_file,

self.cert_file,

cert_reqs=ssl.CERT_REQUIRED,

ca_certs=cert_path)

match_hostname(self.sock.getpeercert(), self.host)

class VerifiedHTTPSHandler( urllib2.HTTPSHandler ):

"""

def __init__(self, connection_class = VerifiedHTTPSConnection):

self.specialized_conn_class = connection_class

urllib2.HTTPSHandler.__init__(self)

def https_open(self, req):

return self.do_open(self.specialized_conn_class, req)

def _get_request(url):

"""

return urllib2.Request(url, headers={'Accept-encoding': 'identity'})

def _get_opener( scheme = None ):

"""

if scheme == "https":

assert os.path.isfile( tuf.conf.ca_certs )

# If we are going over https, use an opener which will provide SSL

# certificate verification.

https_handler = VerifiedHTTPSHandler()

opener = urllib2.build_opener( https_handler )

# strip out HTTPHandler to prevent MITM spoof

for handler in opener.handlers:

if isinstance( handler, urllib2.HTTPHandler ):

opener.handlers.remove( handler )

else:

# Otherwise, use the default opener.

opener = urllib2.build_opener()

return opener

parsed_url = urlparse.urlparse( url )

opener = _get_opener( scheme = parsed_url.scheme )

request = _get_request( url )

return opener.open( request )

raise

- Uses some Python features (e.g. string formatting) not available in earlier versions (e.g. < 2.6).

- Uses some Python features (e.g. `urllib, urllib2, urlparse`) not available in later versions (e.g. >= 3).