Robert Brewer avatar Robert Brewer committed 7334a65

SSL support for _cpwsgiserver. One test fails immaterially. Run "test.py --ssl" to test.

Comments (0)

Files changed (11)

cherrypy/_cpserver.py

     max_request_header_size = 500 * 1024
     max_request_body_size = 100 * 1024 * 1024
     instance = None
+    ssl_certificate = None
+    ssl_private_key = None
     
     def __init__(self):
         self.httpservers = {}

cherrypy/_cpwsgi.py

                    timeout = server.socket_timeout,
                    )
         s.protocol = server.protocol_version
+        s.ssl_certificate = server.ssl_certificate
+        s.ssl_private_key = server.ssl_private_key
 

cherrypy/_cpwsgiserver.py

 from urllib import unquote
 from urlparse import urlparse
 
+try:
+    from OpenSSL import SSL
+except ImportError:
+    SSL = None
+
 import errno
 socket_errors_to_ignore = []
 # Not all of these names will be defined for every platform.
         wfile.flush()
 
 
+def _ssl_wrap_method(method):
+    def ssl_method_wrapper(self, *args, **kwargs):
+##        print (id(self), method, args, kwargs)
+        while True:
+            try:
+                return method(self, *args, **kwargs)
+            except (SSL.WantReadError, SSL.WantWriteError):
+                # Sleep and try again
+                time.sleep(self.ssl_retry)
+            except SSL.SysCallError, e:
+                errno = e.args[0]
+                if errno not in socket_errors_to_ignore:
+                    raise socket.error(errno)
+                return ""
+            except SSL.Error, e:
+                if e.args == (-1, 'Unexpected EOF'):
+                    return ""
+                elif e.args[0][0][2] == 'ssl handshake failure':
+                    return ""
+                else:
+                    raise
+##        raise socket.timeout()
+    return ssl_method_wrapper
+
+class SSL_fileobject(socket._fileobject):
+    """Faux file object attached to a socket object."""
+    
+    ssl_timeout = 3
+    ssl_retry = .01
+    
+    close = _ssl_wrap_method(socket._fileobject.close)
+    flush = _ssl_wrap_method(socket._fileobject.flush)
+    write = _ssl_wrap_method(socket._fileobject.write)
+    writelines = _ssl_wrap_method(socket._fileobject.writelines)
+    read = _ssl_wrap_method(socket._fileobject.read)
+    readline = _ssl_wrap_method(socket._fileobject.readline)
+    readlines = _ssl_wrap_method(socket._fileobject.readlines)
+
+
 class HTTPConnection(object):
     
-    bufsize = -1
+    rbufsize = -1
+    wbufsize = -1
     RequestHandlerClass = HTTPRequest
     environ = {"wsgi.version": (1, 0),
                "wsgi.url_scheme": "http",
                "wsgi.errors": sys.stderr,
                }
     
-    def __init__(self, socket, addr, server):
-        self.socket = socket
+    def __init__(self, sock, addr, server):
+        self.socket = sock
         self.addr = addr
         self.server = server
         
-        self.rfile = self.socket.makefile("r", self.bufsize)
-        self.wfile = self.socket.makefile("w", self.bufsize)
-        
         # Copy the class environ into self.
         self.environ = self.environ.copy()
+        
+        if type(sock) is socket.socket:
+            self.rfile = self.socket.makefile("r", self.rbufsize)
+            self.wfile = self.socket.makefile("w", self.wbufsize)
+        else:
+            # Assume it's an HTTPS socket wrapper
+            self.environ["wsgi.url_scheme"] = "https"
+            self.rfile = SSL_fileobject(sock, "r", self.rbufsize)
+            self.wfile = SSL_fileobject(sock, "w", self.wbufsize)
+        
         self.environ.update({"wsgi.input": self.rfile,
                              "SERVER_NAME": self.server.server_name,
                              })
                 # This order of operations should guarantee correct pipelining.
                 req.parse_request()
                 if not req.ready:
-                    break
+                    return
                 req.respond()
                 if req.close_connection:
-                    break
+                    return
         except socket.error, e:
             errno = e.args[0]
             if errno not in socket_errors_to_ignore:
                 if req:
                     req.simple_response("500 Internal Server Error",
                                         format_exc())
+            return
         except (KeyboardInterrupt, SystemExit):
             raise
         except:
     _interrupt = None
     ConnectionClass = HTTPConnection
     
+    # Paths to certificate and private key files
+    ssl_certificate = None
+    ssl_private_key = None
+    
     def __init__(self, bind_addr, wsgi_app, numthreads=10, server_name=None,
                  max=-1, request_queue_size=5, timeout=10):
         self.requests = Queue.Queue(max)
             """Create (or recreate) the actual socket object."""
             self.socket = socket.socket(family, type, proto)
             self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+            if self.ssl_certificate and self.ssl_private_key:
+                # See http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/442473
+                ctx = SSL.Context(SSL.SSLv23_METHOD)
+                ctx.use_privatekey_file(self.ssl_private_key)
+                ctx.use_certificate_file(self.ssl_certificate)
+                self.socket = SSL.Connection(ctx, self.socket)
             self.socket.bind(self.bind_addr)
         
         # Select the appropriate socket

cherrypy/test/modpy.py

         from cherrypy.test import webtest
         webtest.WebCase.PORT = self.port
         webtest.WebCase.harness = self
+        webtest.WebCase.scheme = "http"
         print
         print "Running tests:", self.server
         

cherrypy/test/test.pem

+-----BEGIN RSA PRIVATE KEY-----
+MIICXAIBAAKBgQDBKo554mzIMY+AByUNpaUOP9bJnQ7ZLQe9XgHwoLJR4VzpyZZZ
+R9L4WtImEew05FY3Izerfm3MN3+MC0tJ6yQU9sOiU3vBW6RrLIMlfKsnRwBRZ0Kn
+da+O6xldVSosu8Ev3z9VZ94iC/ZgKzrH7Mjj/U8/MQO7RBS/LAqee8bFNQIDAQAB
+AoGAWOCF0ZrWxn3XMucWq2LNwPKqlvVGwbIwX3cDmX22zmnM4Fy6arXbYh4XlyCj
+9+ofqRrxIFz5k/7tFriTmZ0xag5+Jdx+Kwg0/twiP7XCNKipFogwe1Hznw8OFAoT
+enKBdj2+/n2o0Bvo/tDB59m9L/538d46JGQUmJlzMyqYikECQQDyoq+8CtMNvE18
+8VgHcR/KtApxWAjj4HpaHYL637ATjThetUZkW92mgDgowyplthusxdNqhHWyv7E8
+tWNdYErZAkEAy85ShTR0M5aWmrE7o0r0SpWInAkNBH9aXQRRARFYsdBtNfRu6I0i
+0lvU9wiu3eF57FMEC86yViZ5UBnQfTu7vQJAVesj/Zt7pwaCDfdMa740OsxMUlyR
+MVhhGx4OLpYdPJ8qUecxGQKq13XZ7R1HGyNEY4bd2X80Smq08UFuATfC6QJAH8UB
+yBHtKz2GLIcELOg6PIYizW/7v3+6rlVF60yw7sb2vzpjL40QqIn4IKoR2DSVtOkb
+8FtAIX3N21aq0VrGYQJBAIPiaEc2AZ8Bq2GC4F3wOz/BxJ/izvnkiotR12QK4fh5
+yjZMhTjWCas5zwHR5PDjlD88AWGDMsZ1PicD4348xJQ=
+-----END RSA PRIVATE KEY-----
+-----BEGIN CERTIFICATE-----
+MIIDxTCCAy6gAwIBAgIJAI18BD7eQxlGMA0GCSqGSIb3DQEBBAUAMIGeMQswCQYD
+VQQGEwJVUzETMBEGA1UECBMKQ2FsaWZvcm5pYTESMBAGA1UEBxMJU2FuIERpZWdv
+MRkwFwYDVQQKExBDaGVycnlQeSBQcm9qZWN0MREwDwYDVQQLEwhkZXYtdGVzdDEW
+MBQGA1UEAxMNQ2hlcnJ5UHkgVGVhbTEgMB4GCSqGSIb3DQEJARYRcmVtaUBjaGVy
+cnlweS5vcmcwHhcNMDYwOTA5MTkyMDIwWhcNMzQwMTI0MTkyMDIwWjCBnjELMAkG
+A1UEBhMCVVMxEzARBgNVBAgTCkNhbGlmb3JuaWExEjAQBgNVBAcTCVNhbiBEaWVn
+bzEZMBcGA1UEChMQQ2hlcnJ5UHkgUHJvamVjdDERMA8GA1UECxMIZGV2LXRlc3Qx
+FjAUBgNVBAMTDUNoZXJyeVB5IFRlYW0xIDAeBgkqhkiG9w0BCQEWEXJlbWlAY2hl
+cnJ5cHkub3JnMIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQDBKo554mzIMY+A
+ByUNpaUOP9bJnQ7ZLQe9XgHwoLJR4VzpyZZZR9L4WtImEew05FY3Izerfm3MN3+M
+C0tJ6yQU9sOiU3vBW6RrLIMlfKsnRwBRZ0Knda+O6xldVSosu8Ev3z9VZ94iC/Zg
+KzrH7Mjj/U8/MQO7RBS/LAqee8bFNQIDAQABo4IBBzCCAQMwHQYDVR0OBBYEFDIQ
+2feb71tVZCWpU0qJ/Tw+wdtoMIHTBgNVHSMEgcswgciAFDIQ2feb71tVZCWpU0qJ
+/Tw+wdtooYGkpIGhMIGeMQswCQYDVQQGEwJVUzETMBEGA1UECBMKQ2FsaWZvcm5p
+YTESMBAGA1UEBxMJU2FuIERpZWdvMRkwFwYDVQQKExBDaGVycnlQeSBQcm9qZWN0
+MREwDwYDVQQLEwhkZXYtdGVzdDEWMBQGA1UEAxMNQ2hlcnJ5UHkgVGVhbTEgMB4G
+CSqGSIb3DQEJARYRcmVtaUBjaGVycnlweS5vcmeCCQCNfAQ+3kMZRjAMBgNVHRME
+BTADAQH/MA0GCSqGSIb3DQEBBAUAA4GBAL7AAQz7IePV48ZTAFHKr88ntPALsL5S
+8vHCZPNMevNkLTj3DYUw2BcnENxMjm1kou2F2BkvheBPNZKIhc6z4hAml3ed1xa2
+D7w6e6OTcstdK/+KrPDDHeOP1dhMWNs2JE1bNlfF1LiXzYKSXpe88eCKjCXsCT/T
+NluCaWQys3MS
+-----END CERTIFICATE-----

cherrypy/test/test.py

 # or vice-versa, unless you *really* know what you're doing.
 
 
+import getopt
+import httplib
+import os, os.path
+localDir = os.path.dirname(__file__)
+serverpem = os.path.join(os.getcwd(), localDir, 'test.pem')
 import sys
-import os, os.path
-import getopt
 
 
 class TestHarness(object):
     """A test harness for the CherryPy framework and CherryPy applications."""
     
-    def __init__(self, tests=None, server=None, protocol="HTTP/1.1", port=8000):
+    def __init__(self, tests=None, server=None, protocol="HTTP/1.1",
+                 port=8000, scheme="http"):
         """Constructor to populate the TestHarness instance.
         
         tests should be a list of module names (strings).
         """
+        self.tests = tests or []
+        self.server = server
         self.protocol = protocol
         self.port = port
-        self.server = server
-        self.tests = tests or []
+        self.scheme = scheme
     
     def run(self, conf=None):
         """Run the test harness."""
         v = sys.version.split()[0]
         print "Python version used to run this test script:", v
         print "CherryPy version", cherrypy.__version__
-        print "HTTP server version", self.protocol
+        if self.scheme == "https":
+            ssl = "(ssl)"
+        else:
+            ssl = ""
+        print "HTTP server version", self.protocol, ssl
         print
         
         if isinstance(conf, basestring):
         baseconf.update(conf or {})
         
         baseconf['server.protocol_version'] = self.protocol
+        if self.scheme == "https":
+            baseconf['server.ssl_certificate'] = serverpem
+            baseconf['server.ssl_private_key'] = serverpem
         self._run(baseconf)
     
     def _run(self, conf):
         from cherrypy.test import helper, webtest
         webtest.WebCase.PORT = self.port
         webtest.WebCase.harness = self
+        webtest.WebCase.scheme = self.scheme
+        if self.scheme == "https":
+            webtest.WebCase.HTTP_CONN = httplib.HTTPSConnection
         print
         print "Running tests:", self.server
         helper.run_test_suite(self.tests, self.server, conf)
                          'modpygw': "modpygw",
                          }
     default_server = "wsgi"
+    scheme = "http"
+    protocol = "HTTP/1.1"
     port = 8080
+    cover = False
+    profile = False
+    validate = False
+    server = None
     basedir = None
     
     def __init__(self, available_tests, args=sys.argv[1:]):
             set of args if you like.
         """
         self.available_tests = available_tests
-        self.cover = False
-        self.profile = False
-        self.validate = False
-        self.server = None
-        self.protocol = "HTTP/1.1"
         
-        longopts = ['cover', 'profile', 'validate', '1.0', 'help', 'basedir=', 'port=',
-                    'server=']
+        longopts = ['cover', 'profile', 'validate', '1.0', 'ssl', 'help',
+                    'basedir=', 'port=', 'server=']
         longopts.extend(self.available_tests)
         try:
             opts, args = getopt.getopt(args, "", longopts)
                 self.validate = True
             elif o == "--1.0":
                 self.protocol = "HTTP/1.0"
+            elif o == "--ssl":
+                self.scheme = "https"
             elif o == "--basedir":
                 self.basedir = a
             elif o == "--port":
         basedir = self.basedir
         if basedir is None:
             # Assume we want to cover everything in "../../cherrypy/"
-            localDir = os.path.dirname(__file__)
             basedir = os.path.normpath(os.path.join(os.getcwd(), localDir, '../'))
         else:
             if not os.path.isabs(basedir):
                                            self.protocol, self.port)
             h.use_wsgi = True
         else:
-            h = TestHarness(self.tests, self.server, self.protocol, self.port)
+            h = TestHarness(self.tests, self.server, self.protocol,
+                            self.port, self.scheme)
         
         h.run(conf)
         
 def prefer_parent_path():
     # Place this __file__'s grandparent (../../) at the start of sys.path,
     # so that all cherrypy/* imports are from this __file__'s package.
-    localDir = os.path.dirname(__file__)
     curpath = os.path.normpath(os.path.join(os.getcwd(), localDir))
     grandparent = os.path.normpath(os.path.join(curpath, '../../'))
     if grandparent not in sys.path:

cherrypy/test/test_conn.py

         self.PROTOCOL = "HTTP/1.1"
         
         # Set our HTTP_CONN to an instance so it persists between requests.
-        self.HTTP_CONN = httplib.HTTPConnection(self.HOST, self.PORT)
+        if self.scheme == "https":
+            self.HTTP_CONN = httplib.HTTPSConnection(self.HOST, self.PORT)
+        else:
+            self.HTTP_CONN = httplib.HTTPConnection(self.HOST, self.PORT)
         # Don't automatically re-connect
         self.HTTP_CONN.auto_open = False
         self.HTTP_CONN.connect()
             self.assertRaises(httplib.NotConnected, self.getPage, "/")
         
         # Test client-side close.
-        self.HTTP_CONN = httplib.HTTPConnection(self.HOST, self.PORT)
+        if self.scheme == "https":
+            self.HTTP_CONN = httplib.HTTPSConnection(self.HOST, self.PORT)
+        else:
+            self.HTTP_CONN = httplib.HTTPConnection(self.HOST, self.PORT)
         self.HTTP_CONN.auto_open = False
         self.HTTP_CONN.connect()
         self.getPage("/page2", headers=[("Connection", "close")])
             self.PROTOCOL = "HTTP/1.1"
             
             # Make an initial request
-            conn = httplib.HTTPConnection(self.HOST, self.PORT)
+            if self.scheme == "https":
+                conn = httplib.HTTPSConnection(self.HOST, self.PORT)
+            else:
+                conn = httplib.HTTPConnection(self.HOST, self.PORT)
             conn.auto_open = False
             conn.connect()
             conn.putrequest("GET", "/", skip_host=True)
                               " as it should have: %s" % sys.exc_info()[1])
             else:
                 self.fail("Writing to timed out socket didn't fail"
-                          " as it should have.")
+                          " as it should have: %s" %
+                          response.read())
             
             conn.close()
             
             # Make another request on a new socket, which should work
-            conn = httplib.HTTPConnection(self.HOST, self.PORT)
+            if self.scheme == "https":
+                conn = httplib.HTTPSConnection(self.HOST, self.PORT)
+            else:
+                conn = httplib.HTTPConnection(self.HOST, self.PORT)
             conn.auto_open = False
             conn.connect()
             conn.putrequest("GET", "/", skip_host=True)
         self.PROTOCOL = "HTTP/1.1"
         
         # Test pipelining. httplib doesn't support this directly.
-        conn = httplib.HTTPConnection(self.HOST, self.PORT)
+        if self.scheme == "https":
+            conn = httplib.HTTPSConnection(self.HOST, self.PORT)
+        else:
+            conn = httplib.HTTPConnection(self.HOST, self.PORT)
         conn.auto_open = False
         conn.connect()
         
         
         self.PROTOCOL = "HTTP/1.1"
         
-        conn = httplib.HTTPConnection(self.HOST, self.PORT)
+        if self.scheme == "https":
+            conn = httplib.HTTPSConnection(self.HOST, self.PORT)
+        else:
+            conn = httplib.HTTPConnection(self.HOST, self.PORT)
         conn.auto_open = False
         conn.connect()
         
         self.PROTOCOL = "HTTP/1.1"
         
         # Set our HTTP_CONN to an instance so it persists between requests.
-        conn = httplib.HTTPConnection(self.HOST, self.PORT)
+        if self.scheme == "https":
+            conn = httplib.HTTPSConnection(self.HOST, self.PORT)
+        else:
+            conn = httplib.HTTPConnection(self.HOST, self.PORT)
         
         # Try a normal chunked request
         body = ("8\r\nxx\r\nxxxx\r\n5\r\nyyyyy\r\n0\r\n"
     
     def test_HTTP10(self):
         self.PROTOCOL = "HTTP/1.0"
-        self.HTTP_CONN = httplib.HTTPConnection
+        if self.scheme == "https":
+            self.HTTP_CONN = httplib.HTTPSConnection
+        else:
+            self.HTTP_CONN = httplib.HTTPConnection
         
         # Test a normal HTTP/1.0 request.
         self.getPage("/page2")
 ##        self.assertNoHeader("Connection")
         
         # Test a keep-alive HTTP/1.0 request.
-        self.HTTP_CONN = httplib.HTTPConnection(self.HOST, self.PORT)
+        if self.scheme == "https":
+            self.HTTP_CONN = httplib.HTTPSConnection(self.HOST, self.PORT)
+        else:
+            self.HTTP_CONN = httplib.HTTPConnection(self.HOST, self.PORT)
         self.HTTP_CONN.auto_open = False
         self.HTTP_CONN.connect()
         

cherrypy/test/test_core.py

         # Make sure GET params are preserved.
         self.getPage("/redirect?id=3")
         self.assertStatus(('302 Found', '303 See Other'))
-        self.assertInBody("<a href='http://127.0.0.1:%s%s/redirect/?id=3'>"
-                          "http://127.0.0.1:%s%s/redirect/?id=3</a>" %
-                          (self.PORT, self.prefix(), self.PORT, self.prefix()))
+        self.assertInBody("<a href='%s://127.0.0.1:%s%s/redirect/?id=3'>"
+                          "%s://127.0.0.1:%s%s/redirect/?id=3</a>" %
+                          (self.scheme, self.PORT, self.prefix(),
+                           self.scheme, self.PORT, self.prefix()))
         
         if self.prefix():
             # Corner case: the "trailing slash" redirect could be tricky if
             # we're using a virtual root and the URI is "/vroot" (no slash).
             self.getPage("")
             self.assertStatus(('302 Found', '303 See Other'))
-            self.assertInBody("<a href='http://127.0.0.1:%s%s/'>"
-                              "http://127.0.0.1:%s%s/</a>" %
-                              (self.PORT, self.prefix(), self.PORT, self.prefix()))
+            self.assertInBody("<a href='%s://127.0.0.1:%s%s/'>"
+                              "%s://127.0.0.1:%s%s/</a>" %
+                              (self.scheme, self.PORT, self.prefix(),
+                               self.scheme, self.PORT, self.prefix()))
         
         self.getPage("/redirect/by_code?code=300")
         self.assertMatchesBody(r"<a href='(.*)somewhere else'>\1somewhere else</a>")
         # Make sure str(HTTPRedirect()) works.
         self.getPage("/redirect/stringify", protocol="HTTP/1.0")
         self.assertStatus(200)
-        self.assertBody("(['http://127.0.0.1:%s/'], 302)" % self.PORT)
+        self.assertBody("(['%s://127.0.0.1:%s/'], 302)" % (self.scheme, self.PORT))
         if cherrypy.server.protocol_version == "HTTP/1.1":
             self.getPage("/redirect/stringify", protocol="HTTP/1.1")
             self.assertStatus(200)
-            self.assertBody("(['http://127.0.0.1:%s/'], 303)" % self.PORT)
+            self.assertBody("(['%s://127.0.0.1:%s/'], 303)" % (self.scheme, self.PORT))
     
     def testFlatten(self):
         for url in ["/flatten/as_string", "/flatten/as_list",

cherrypy/test/test_objectmapping.py

             # Test omitted trailing slash (should be redirected by default).
             self.getPage("/dir1/dir2")
             self.assertStatus((302, 303))
-            self.assertHeader('Location', 'http://%s:%s%s/dir1/dir2/'
-                              % (self.HOST, self.PORT, prefix))
+            self.assertHeader('Location', '%s://%s:%s%s/dir1/dir2/'
+                              % (self.scheme, self.HOST, self.PORT, prefix))
             
             # Test extra trailing slash (should be redirected if configured).
             self.getPage("/dir1/myMethod/")
             self.assertStatus((302, 303))
-            self.assertHeader('Location', 'http://%s:%s%s/dir1/myMethod'
-                              % (self.HOST, self.PORT, prefix))
+            self.assertHeader('Location', '%s://%s:%s%s/dir1/myMethod'
+                              % (self.scheme, self.HOST, self.PORT, prefix))
             
             # Test that default method must be exposed in order to match.
             self.getPage("/dir1/dir2/dir3/dir4/index")
             
             self.getPage("/redirect")
             self.assertStatus('302 Found')
-            self.assertHeader('Location', 'http://%s:%s%s/dir1/'
-                              % (self.HOST, self.PORT, prefix))
+            self.assertHeader('Location', '%s://%s:%s%s/dir1/'
+                              % (self.scheme, self.HOST, self.PORT, prefix))
             
             # Test that we can use URL's which aren't all valid Python identifiers
             # This should also test the %XX-unquoting of URL's.
             self.getPage("/dir1/dir2/script_name")
             self.assertBody(url)
             self.getPage("/dir1/dir2/tree_url")
-            self.assertBody("http://%s:%s%s/extra" %
-                            (self.HOST, self.PORT, prefix))
+            self.assertBody("%s://%s:%s%s/extra" %
+                            (self.scheme, self.HOST, self.PORT, prefix))
             
             # Test that configs don't overwrite each other from diferent apps
             self.getPage("/confvalue")

cherrypy/test/test_proxy.py

     def testProxy(self):
         self.getPage("/")
         self.assertHeader('Location',
-                          "http://www.mydomain.com%s/dummy" % self.prefix())
+                          "%s://www.mydomain.com%s/dummy" % (self.scheme, self.prefix()))
         
         # Test X-Forwarded-Host (Apache 1.3.33+ and Apache 2)
         self.getPage("/", headers=[('X-Forwarded-Host', 'http://www.yetanother.com')])
         self.assertHeader('Location', "http://www.yetanother.com/dummy")
         self.getPage("/", headers=[('X-Forwarded-Host', 'www.yetanother.com')])
-        self.assertHeader('Location', "http://www.yetanother.com/dummy")
+        self.assertHeader('Location', "%s://www.yetanother.com/dummy" % self.scheme)
         
         # Test X-Forwarded-For (Apache2)
         self.getPage("/remoteip",
         
         # Test X-Host (lighttpd; see https://trac.lighttpd.net/trac/ticket/418)
         self.getPage("/xhost", headers=[('X-Host', 'www.yetanother.com')])
-        self.assertHeader('Location', "http://www.yetanother.com/blah")
+        self.assertHeader('Location', "%s://www.yetanother.com/blah" % self.scheme)
         
         # Test X-Forwarded-Proto (lighttpd)
         self.getPage("/base", headers=[('X-Forwarded-Proto', 'https')])
         # Test tree.url()
         for sn in script_names:
             self.getPage(sn + "/newurl")
-            self.assertBody("Browse to <a href='http://www.mydomain.com"
+            self.assertBody("Browse to <a href='%s://www.mydomain.com" % self.scheme
                             + sn + "/this/new/page'>this page</a>.")
             self.getPage(sn + "/newurl", headers=[('X-Forwarded-Host',
                                                    'http://www.yetanother.com')])

cherrypy/test/test_xmlrpc.py

     cherrypy.config.update({'environment': 'test_suite'})
 
 
+class HTTPSTransport(xmlrpclib.SafeTransport):
+    """Subclass of SafeTransport to fix sock.recv errors (by using file)."""
+    
+    def request(self, host, handler, request_body, verbose=0):
+        # issue XML-RPC request
+        h = self.make_connection(host)
+        if verbose:
+            h.set_debuglevel(1)
+        
+        self.send_request(h, handler, request_body)
+        self.send_host(h, host)
+        self.send_user_agent(h)
+        self.send_content(h, request_body)
+        
+        errcode, errmsg, headers = h.getreply()
+        if errcode != 200:
+            raise xmlrpclib.ProtocolError(host + handler, errcode, errmsg,
+                                          headers)
+        
+        self.verbose = verbose
+        return self.parse_response(h.getfile())
+
+
 from cherrypy.test import helper
 
 class XmlRpcTest(helper.CPWebCase):
     def testXmlRpc(self):
         
         # load the appropriate xmlrpc proxy
-        url = 'http://localhost:%s/xmlrpc/' % (self.PORT)
-        proxy = xmlrpclib.ServerProxy(url)
+        if getattr(self.harness, "scheme", "http") == "https":
+            url = 'https://localhost:%s/xmlrpc/' % self.PORT
+            proxy = xmlrpclib.ServerProxy(url, transport=HTTPSTransport())
+        else:
+            url = 'http://localhost:%s/xmlrpc/' % self.PORT
+            proxy = xmlrpclib.ServerProxy(url)
         
         # begin the tests ...
         self.assertEqual(proxy.return_single_item_list(), [42])
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.