From 0b4a0af8343c7a459b2b7cf15ce496c2edfcf80a Mon Sep 17 00:00:00 2001
From: Adam Goodman <akgood@duosecurity.com>
Date: Tue, 19 Jun 2012 12:06:32 -0400
Subject: add SSL certificate validation

---
 https_wrapper.py | 146 +++++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 146 insertions(+)
 create mode 100644 https_wrapper.py

(limited to 'https_wrapper.py')

diff --git a/https_wrapper.py b/https_wrapper.py
new file mode 100644
index 0000000..c3a9ce3
--- /dev/null
+++ b/https_wrapper.py
@@ -0,0 +1,146 @@
+#!/usr/bin/env python
+#
+# Adapted from:
+# https://googleappengine.googlecode.com/svn-history/r136/trunk/python/google/appengine/tools/https_wrapper.py
+#
+# Copyright 2007 Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Extensions to allow HTTPS requests with SSL certificate validation."""
+
+
+import httplib
+import re
+import socket
+import urllib2
+import ssl
+
+
+class InvalidCertificateException(httplib.HTTPException):
+  """Raised when a certificate is provided with an invalid hostname."""
+
+  def __init__(self, host, cert, reason):
+    """Constructor.
+
+    Args:
+      host: The hostname the connection was made to.
+      cert: The SSL certificate (as a dictionary) the host returned.
+    """
+    httplib.HTTPException.__init__(self)
+    self.host = host
+    self.cert = cert
+    self.reason = reason
+
+  def __str__(self):
+    return ('Host %s returned an invalid certificate (%s): %s\n'
+            'To learn more, see '
+            'http://code.google.com/appengine/kb/general.html#rpcssl' %
+            (self.host, self.reason, self.cert))
+
+class CertValidatingHTTPSConnection(httplib.HTTPConnection):
+  """An HTTPConnection that connects over SSL and validates certificates."""
+
+  default_port = httplib.HTTPS_PORT
+
+  def __init__(self, host, port=None, key_file=None, cert_file=None,
+               ca_certs=None, strict=None, **kwargs):
+    """Constructor.
+
+    Args:
+      host: The hostname. Can be in 'host:port' form.
+      port: The port. Defaults to 443.
+      key_file: A file containing the client's private key
+      cert_file: A file containing the client's certificates
+      ca_certs: A file contianing a set of concatenated certificate authority
+          certs for validating the server against.
+      strict: When true, causes BadStatusLine to be raised if the status line
+          can't be parsed as a valid HTTP/1.0 or 1.1 status line.
+    """
+    httplib.HTTPConnection.__init__(self, host, port, strict, **kwargs)
+    self.key_file = key_file
+    self.cert_file = cert_file
+    self.ca_certs = ca_certs
+    if self.ca_certs:
+      self.cert_reqs = ssl.CERT_REQUIRED
+    else:
+      self.cert_reqs = ssl.CERT_NONE
+
+  def _GetValidHostsForCert(self, cert):
+    """Returns a list of valid host globs for an SSL certificate.
+
+    Args:
+      cert: A dictionary representing an SSL certificate.
+    Returns:
+      list: A list of valid host globs.
+    """
+    if 'subjectAltName' in cert:
+      return [x[1] for x in cert['subjectAltName'] if x[0].lower() == 'dns']
+    else:
+      return [x[0][1] for x in cert['subject']
+              if x[0][0].lower() == 'commonname']
+
+  def _ValidateCertificateHostname(self, cert, hostname):
+    """Validates that a given hostname is valid for an SSL certificate.
+
+    Args:
+      cert: A dictionary representing an SSL certificate.
+      hostname: The hostname to test.
+    Returns:
+      bool: Whether or not the hostname is valid for this certificate.
+    """
+    hosts = self._GetValidHostsForCert(cert)
+    for host in hosts:
+      host_re = host.replace('.', '\.').replace('*', '[^.]*')
+      if re.search('^%s$' % (host_re,), hostname, re.I):
+        return True
+    return False
+
+  def connect(self):
+    "Connect to a host on a given (SSL) port."
+    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    sock.connect((self.host, self.port))
+    self.sock = ssl.wrap_socket(sock, keyfile=self.key_file,
+                                certfile=self.cert_file,
+                                cert_reqs=self.cert_reqs,
+                                ca_certs=self.ca_certs)
+    if self.cert_reqs & ssl.CERT_REQUIRED:
+      cert = self.sock.getpeercert()
+      hostname = self.host.split(':', 0)[0]
+      if not self._ValidateCertificateHostname(cert, hostname):
+        raise InvalidCertificateException(hostname, cert, 'hostname mismatch')
+
+
+class CertValidatingHTTPSHandler(urllib2.HTTPSHandler):
+  """An HTTPHandler that validates SSL certificates."""
+
+  def __init__(self, **kwargs):
+    """Constructor. Any keyword args are passed to the httplib handler."""
+    urllib2.HTTPSHandler.__init__(self)
+    self._connection_args = kwargs
+
+  def https_open(self, req):
+    def http_class_wrapper(host, **kwargs):
+      full_kwargs = dict(self._connection_args)
+      full_kwargs.update(kwargs)
+      return CertValidatingHTTPSConnection(host, **full_kwargs)
+    try:
+      return self.do_open(http_class_wrapper, req)
+    except urllib2.URLError, e:
+      if type(e.reason) == ssl.SSLError and e.reason.args[0] == 1:
+        raise InvalidCertificateException(req.host, '',
+                                          e.reason.args[1])
+      raise
+
+  https_request = urllib2.HTTPSHandler.do_request_
-- 
cgit v1.2.3