<s> """<STR_LIT>""" <EOL> import BaseHTTPServer <EOL> import shutil <EOL> import signal <EOL> import socket <EOL> import tempfile <EOL> import threading <EOL> import time <EOL> import unittest <EOL> import certutils <EOL> import sslproxy <EOL> class Client ( object ) : <EOL> def __init__ ( self , ca_cert_path , verify_cb , port , host_name = '<STR_LIT>' , <EOL> host = '<STR_LIT:localhost>' ) : <EOL> self . host_name = host_name <EOL> self . verify_cb = verify_cb <EOL> self . ca_cert_path = ca_cert_path <EOL> self . port = port <EOL> self . host_name = host_name <EOL> self . host = host <EOL> self . connection = None <EOL> def run_request ( self ) : <EOL> context = certutils . get_ssl_context ( ) <EOL> context . set_verify ( certutils . VERIFY_PEER , self . verify_cb ) <EOL> context . use_certificate_file ( self . ca_cert_path ) <EOL> context . load_verify_locations ( self . ca_cert_path ) <EOL> s = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> self . connection = certutils . get_ssl_connection ( context , s ) <EOL> self . connection . connect ( ( self . host , self . port ) ) <EOL> self . connection . set_tlsext_host_name ( self . host_name ) <EOL> try : <EOL> self . connection . send ( '<STR_LIT>' ) <EOL> finally : <EOL> self . connection . shutdown ( ) <EOL> self . connection . close ( ) <EOL> class Handler ( BaseHTTPServer . BaseHTTPRequestHandler ) : <EOL> protocol_version = '<STR_LIT>' <EOL> def handle_one_request ( self ) : <EOL> """<STR_LIT>""" <EOL> self . raw_requestline = self . rfile . readline ( <NUM_LIT> ) <EOL> class WrappedErrorHandler ( Handler ) : <EOL> """<STR_LIT>""" <EOL> def setup ( self ) : <EOL> Handler . setup ( self ) <EOL> try : <EOL> sslproxy . _SetUpUsingDummyCert ( self ) <EOL> except certutils . Error : <EOL> self . server . error_function = certutils . Error <EOL> def finish ( self ) : <EOL> Handler . finish ( self ) <EOL> self . connection . shutdown ( ) <EOL> self . connection . close ( ) <EOL> class DummyArchive ( object ) : <EOL> def __init__ ( self ) : <EOL> pass <EOL> class DummyFetch ( object ) : <EOL> def __init__ ( self ) : <EOL> self . http_archive = DummyArchive ( ) <EOL> class Server ( BaseHTTPServer . HTTPServer ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , ca_cert_path , use_error_handler = False , port = <NUM_LIT:0> , <EOL> host = '<STR_LIT:localhost>' ) : <EOL> self . ca_cert_path = ca_cert_path <EOL> with open ( ca_cert_path , '<STR_LIT:r>' ) as ca_file : <EOL> self . ca_cert_str = ca_file . read ( ) <EOL> self . http_archive_fetch = DummyFetch ( ) <EOL> if use_error_handler : <EOL> self . HANDLER = WrappedErrorHandler <EOL> else : <EOL> self . HANDLER = sslproxy . wrap_handler ( Handler ) <EOL> try : <EOL> BaseHTTPServer . HTTPServer . __init__ ( self , ( host , port ) , self . HANDLER ) <EOL> except Exception , e : <EOL> raise RuntimeError ( '<STR_LIT>' <EOL> % ( port , e ) ) <EOL> def __enter__ ( self ) : <EOL> thread = threading . Thread ( target = self . serve_forever ) <EOL> thread . daemon = True <EOL> thread . start ( ) <EOL> return self <EOL> def cleanup ( self ) : <EOL> try : <EOL> self . shutdown ( ) <EOL> except KeyboardInterrupt : <EOL> pass <EOL> def __exit__ ( self , type_ , value_ , traceback_ ) : <EOL> self . cleanup ( ) <EOL> def get_certificate ( self , host ) : <EOL> return certutils . generate_cert ( self . ca_cert_str , '<STR_LIT>' , host ) <EOL> class TestClient ( unittest . TestCase ) : <EOL> _temp_dir = None <EOL> def setUp ( self ) : <EOL> self . _temp_dir = tempfile . mkdtemp ( prefix = '<STR_LIT>' , dir = '<STR_LIT>' ) <EOL> self . ca_cert_path = self . _temp_dir + '<STR_LIT>' <EOL> self . cert_path = self . _temp_dir + '<STR_LIT>' <EOL> self . wrong_ca_cert_path = self . _temp_dir + '<STR_LIT>' <EOL> self . wrong_cert_path = self . _temp_dir + '<STR_LIT>' <EOL> certutils . write_dummy_ca_cert ( * certutils . generate_dummy_ca_cert ( ) , <EOL> cert_path = self . ca_cert_path ) <EOL> certutils . write_dummy_ca_cert ( * certutils . generate_dummy_ca_cert ( ) , <EOL> cert_path = self . ca_cert_path ) <EOL> def tearDown ( self ) : <EOL> if self . _temp_dir : <EOL> shutil . rmtree ( self . _temp_dir ) <EOL> def verify_cb ( self , conn , cert , errnum , depth , ok ) : <EOL> """<STR_LIT>""" <EOL> self . assertFalse ( cert . has_expired ( ) ) <EOL> self . assertGreater ( time . strftime ( '<STR_LIT>' , time . gmtime ( ) ) , <EOL> cert . get_notBefore ( ) ) <EOL> return ok <EOL> def test_no_host ( self ) : <EOL> with Server ( self . ca_cert_path ) as server : <EOL> c = Client ( self . cert_path , self . verify_cb , server . server_port , '<STR_LIT>' ) <EOL> self . assertRaises ( certutils . Error , c . run_request ) <EOL> def test_client_connection ( self ) : <EOL> with Server ( self . ca_cert_path ) as server : <EOL> c = Client ( self . cert_path , self . verify_cb , server . server_port , '<STR_LIT>' ) <EOL> c . run_request ( ) <EOL> c = Client ( self . cert_path , self . verify_cb , server . server_port , <EOL> '<STR_LIT>' ) <EOL> c . run_request ( ) <EOL> def test_wrong_cert ( self ) : <EOL> with Server ( self . ca_cert_path , True ) as server : <EOL> c = Client ( self . wrong_cert_path , self . verify_cb , server . server_port , <EOL> '<STR_LIT>' ) <EOL> self . assertRaises ( certutils . Error , c . run_request ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> signal . signal ( signal . SIGINT , signal . SIG_DFL ) <EOL> unittest . main ( ) </s>
