Source Code for Module web.tests.testsecureconnection

  1  import unittest 
  2  import web 
  3  from web.wsgiserver import CherryPyWSGIServer 
  4  import threading 
  5  from clientjob.secure_connection import SecureConnection 
  6  from web.wsgiserver.ssl_pyopenssl import pyOpenSSLAdapter 
  7  import string 
  8  import random 
  9  import os 
 10  import sys 
 11  import time 
 12  import config 
 13  from ssl import SSLError 
 14   
 15 -class EchoView: 
 16      ''' 
 17      This is a view for the test server which prints the argument given to it 
 18      ''' 
 19 -    def GET(self, arg): 
 20          return arg 
 21   
 22 -class TestSecureConnection(unittest.TestCase): 
 23 -    def setUp(self): 
 24          self.ssl_certificate = os.path.join('ssl', 'localhost.pem') 
 25          self.ssl_private_key = os.path.join('ssl', 'localhost.key') 
 26          self.client_CA = os.path.join('ssl', 'ca-cert.pem') 
 27   
 28 -    def launchServer(self, port): 
 29          ''' 
 30          Launches a test server which always prints the args in the url requested 
 31          ''' 
 32          CherryPyWSGIServer.ssl_adapter = pyOpenSSLAdapter(self.ssl_certificate, 
 33              self.ssl_private_key, None, self.client_CA) 
 34   
 35          urls = ("/(.*)", "EchoView") 
 36          # the following is the only way to tell web.py which port to start the server.. 
 37          sys.argv[1:] = [str(port)] 
 38          app = web.application(urls, globals()) 
 39   
 40          app.run() 
 41   
 42 -    def testSecureConnection(self): 
 43          ''' 
 44          Request a random url from the test serverand checks the response, 
 45          using the secure connection with a valid certificate signed by the CA. 
 46          ''' 
 47          port = 8081 
 48          server = threading.Thread(target=self.launchServer, args=[port]) 
 49          server.start() 
 50          time.sleep(1) # let the server start in the thread 
 51          connection = SecureConnection("127.0.0.1", port) 
 52          random_string = ''.join(random.choice(string.letters) for i in xrange(40)) 
 53          connection.request("GET", "/" + random_string) 
 54          response = connection.getresponse() 
 55          self.assertEqual(response.status, 200) 
 56          self.assertEqual(response.read(), random_string) 
 57          server = None 
 58   
 59 -    def testInsecureClient(self): 
 60          ''' 
 61          Similar to testSecureConnection, but this time the client has an invalid 
 62          (self-signed) certificate and thus the server should reject the connection. 
 63          ''' 
 64          port = 8082 
 65          server = threading.Thread(target=self.launchServer, args=[port]) 
 66          server.start() 
 67          time.sleep(1) # let the server start in the thread 
 68   
 69          # use an invalid (self-signed) certificate in the client 
 70          connection = SecureConnection("127.0.0.1", port, 
 71              cert_file = os.path.join('ssl', 'self-signed-localhost.cert')) 
 72   
 73          random_string = ''.join(random.choice(string.letters) for i in xrange(40)) 
 74          #import ipdb; ipdb.set_trace() 
 75   
 76          # As it uses a self-signed certificate, it should raise an SSLError exception, 
 77          # otherwise the test should fail 
 78          try: 
 79              connection.request("GET", "/" + random_string) 
 80              raise Exception 
 81          except SSLError: 
 82              # Here we could check the excapt SSLError, problem is we could only verify 
 83              # the string it contains and it might vary from version to version, so we 
 84              # just verify that an SSLError exception is raised 
 85              server = None 
 86   
 87 -    def testInsecureServer(self): 
 88          ''' 
 89          Similar to testInsecureConnection, but now it's the server the one using an 
 90          invalid (self-signed) certificate, and thus the client should reject the connection. 
 91          ''' 
 92          port = 8083 
 93   
 94          # use an invalid (self-signed) certificate in the server 
 95          self.ssl_certificate = os.path.join('ssl', 'self-signed-localhost.cert') 
 96   
 97          server = threading.Thread(target=self.launchServer, args=[port]) 
 98          server.start() 
 99          time.sleep(1) # let the server start in the thread 
100          connection = SecureConnection("127.0.0.1", port) 
101          random_string = ''.join(random.choice(string.letters) for i in xrange(40)) 
102          try: 
103              connection.request("GET", "/" + random_string) 
104              raise Exception 
105          except SSLError: 
106              # See same case in testInsecureClient 
107              server = None 
108   
109 -def suite(): 
110     suite = unittest.TestSuite() 
111     suite.addTest(unittest.makeSuite(TestSecureConnection)) 
112     return suite 
113