<s> """<STR_LIT>""" <EOL> from OpenSSL . SSL import VERIFY_PEER , VERIFY_FAIL_IF_NO_PEER_CERT <EOL> from zope . interface import implementer <EOL> from twisted . web . iweb import IPolicyForHTTPS <EOL> from twisted . internet . ssl import optionsForClientTLS , Certificate <EOL> from twisted . web . client import Agent <EOL> from treq . client import HTTPClient <EOL> from pyrsistent import PClass , field <EOL> from . _ca import UserCredential <EOL> @ implementer ( IPolicyForHTTPS ) <EOL> class ControlServicePolicy ( PClass ) : <EOL> """<STR_LIT>""" <EOL> ca_certificate = field ( mandatory = True ) <EOL> client_credential = field ( mandatory = True ) <EOL> def creatorForNetloc ( self , hostname , port ) : <EOL> return optionsForClientTLS ( <EOL> u"<STR_LIT>" , <EOL> trustRoot = self . ca_certificate , <EOL> clientCertificate = self . client_credential . private_certificate ( ) ) <EOL> class _ControlServiceContextFactory ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , ca_certificate , control_credential , prefix ) : <EOL> """<STR_LIT>""" <EOL> self . prefix = prefix <EOL> self . control_credential = control_credential <EOL> self . ca_certificate = ca_certificate <EOL> def getContext ( self ) : <EOL> default_options = self . control_credential . _default_options ( <EOL> self . ca_certificate ) <EOL> def verify ( conn , cert , errno , depth , preverify_ok ) : <EOL> if depth > <NUM_LIT:0> : <EOL> return preverify_ok <EOL> if not preverify_ok : <EOL> return preverify_ok <EOL> return cert . get_subject ( ) . commonName . startswith ( self . prefix ) <EOL> context = default_options . getContext ( ) <EOL> context . set_verify ( VERIFY_PEER | VERIFY_FAIL_IF_NO_PEER_CERT , <EOL> verify ) <EOL> return context <EOL> def amp_server_context_factory ( ca_certificate , control_credential ) : <EOL> """<STR_LIT>""" <EOL> return _ControlServiceContextFactory ( <EOL> ca_certificate , control_credential , b"<STR_LIT>" ) <EOL> def rest_api_context_factory ( ca_certificate , control_credential ) : <EOL> """<STR_LIT>""" <EOL> return _ControlServiceContextFactory ( <EOL> ca_certificate , control_credential , b"<STR_LIT>" ) <EOL> def treq_with_authentication ( reactor , ca_path , user_cert_path , user_key_path ) : <EOL> """<STR_LIT>""" <EOL> ca = Certificate . loadPEM ( ca_path . getContent ( ) ) <EOL> user_credential = UserCredential . from_files ( user_cert_path , user_key_path ) <EOL> policy = ControlServicePolicy ( <EOL> ca_certificate = ca , client_credential = user_credential . credential ) <EOL> return HTTPClient ( Agent ( reactor , contextFactory = policy ) ) </s>
