def test_set_and_add_client_ca(self):
"""
A call to `Context.set_client_ca_list` followed by a call to
`Context.add_client_ca` results in using the CA names from the
first call and the CA name from the second call.
"""
cacert = load_certificate(FILETYPE_PEM, root_cert_pem)
secert = load_certificate(FILETYPE_PEM, server_cert_pem)
clcert = load_certificate(FILETYPE_PEM, server_cert_pem)
cadesc = cacert.get_subject()
sedesc = secert.get_subject()
cldesc = clcert.get_subject()
def mixed_set_add_ca(ctx):
ctx.set_client_ca_list([cadesc, sedesc])
ctx.add_client_ca(clcert)
return [cadesc, sedesc, cldesc]
self._check_client_ca_list(mixed_set_add_ca)
python类load_certificate()的实例源码
def test_set_after_add_client_ca(self):
"""
A call to `Context.set_client_ca_list` after a call to
`Context.add_client_ca` replaces the CA name specified by the
former call with the names specified by the latter call.
"""
cacert = load_certificate(FILETYPE_PEM, root_cert_pem)
secert = load_certificate(FILETYPE_PEM, server_cert_pem)
clcert = load_certificate(FILETYPE_PEM, server_cert_pem)
cadesc = cacert.get_subject()
sedesc = secert.get_subject()
def set_replaces_add_ca(ctx):
ctx.add_client_ca(clcert)
ctx.set_client_ca_list([cadesc])
ctx.add_client_ca(secert)
return [cadesc, sedesc]
self._check_client_ca_list(set_replaces_add_ca)
def from_string(key_pem, is_x509_cert):
"""Construct a Verified instance from a string.
Args:
key_pem: string, public key in PEM format.
is_x509_cert: bool, True if key_pem is an X509 cert, otherwise it
is expected to be an RSA key in PEM format.
Returns:
Verifier instance.
Raises:
OpenSSL.crypto.Error: if the key_pem can't be parsed.
"""
key_pem = _helpers._to_bytes(key_pem)
if is_x509_cert:
pubkey = crypto.load_certificate(crypto.FILETYPE_PEM, key_pem)
else:
pubkey = crypto.load_privatekey(crypto.FILETYPE_PEM, key_pem)
return OpenSSLVerifier(pubkey)
def from_string(key_pem, is_x509_cert):
"""Construct a Verified instance from a string.
Args:
key_pem: string, public key in PEM format.
is_x509_cert: bool, True if key_pem is an X509 cert, otherwise it is
expected to be an RSA key in PEM format.
Returns:
Verifier instance.
Raises:
OpenSSL.crypto.Error if the key_pem can't be parsed.
"""
if is_x509_cert:
pubkey = crypto.load_certificate(crypto.FILETYPE_PEM, key_pem)
else:
pubkey = crypto.load_privatekey(crypto.FILETYPE_PEM, key_pem)
return OpenSSLVerifier(pubkey)
def gen_pkcs12(self, cert_pem=None, key_pem=None, ca_pem=None, friendly_name=None):
"""
Generate a PKCS12 object with components from PEM. Verify that the set
functions return None.
"""
p12 = PKCS12()
if cert_pem:
ret = p12.set_certificate(load_certificate(FILETYPE_PEM, cert_pem))
self.assertEqual(ret, None)
if key_pem:
ret = p12.set_privatekey(load_privatekey(FILETYPE_PEM, key_pem))
self.assertEqual(ret, None)
if ca_pem:
ret = p12.set_ca_certificates((load_certificate(FILETYPE_PEM, ca_pem),))
self.assertEqual(ret, None)
if friendly_name:
ret = p12.set_friendlyname(friendly_name)
self.assertEqual(ret, None)
return p12
def test_replace(self):
"""
L{PKCS12.set_certificate} replaces the certificate in a PKCS12 cluster.
L{PKCS12.set_privatekey} replaces the private key.
L{PKCS12.set_ca_certificates} replaces the CA certificates.
"""
p12 = self.gen_pkcs12(client_cert_pem, client_key_pem, root_cert_pem)
p12.set_certificate(load_certificate(FILETYPE_PEM, server_cert_pem))
p12.set_privatekey(load_privatekey(FILETYPE_PEM, server_key_pem))
root_cert = load_certificate(FILETYPE_PEM, root_cert_pem)
client_cert = load_certificate(FILETYPE_PEM, client_cert_pem)
p12.set_ca_certificates([root_cert]) # not a tuple
self.assertEqual(1, len(p12.get_ca_certificates()))
self.assertEqual(root_cert, p12.get_ca_certificates()[0])
p12.set_ca_certificates([client_cert, root_cert])
self.assertEqual(2, len(p12.get_ca_certificates()))
self.assertEqual(client_cert, p12.get_ca_certificates()[0])
self.assertEqual(root_cert, p12.get_ca_certificates()[1])
def _server(self, sock):
"""
Create a new server-side SSL L{Connection} object wrapped around
C{sock}.
"""
# Create the server side Connection. This is mostly setup boilerplate
# - use TLSv1, use a particular certificate, etc.
server_ctx = Context(TLSv1_METHOD)
server_ctx.set_options(OP_NO_SSLv2 | OP_NO_SSLv3 | OP_SINGLE_DH_USE )
server_ctx.set_verify(VERIFY_PEER|VERIFY_FAIL_IF_NO_PEER_CERT|VERIFY_CLIENT_ONCE, verify_cb)
server_store = server_ctx.get_cert_store()
server_ctx.use_privatekey(load_privatekey(FILETYPE_PEM, server_key_pem))
server_ctx.use_certificate(load_certificate(FILETYPE_PEM, server_cert_pem))
server_ctx.check_privatekey()
server_store.add_cert(load_certificate(FILETYPE_PEM, root_cert_pem))
# Here the Connection is actually created. If None is passed as the 2nd
# parameter, it indicates a memory BIO should be created.
server_conn = Connection(server_ctx, sock)
server_conn.set_accept_state()
return server_conn
def _client(self, sock):
"""
Create a new client-side SSL L{Connection} object wrapped around
C{sock}.
"""
# Now create the client side Connection. Similar boilerplate to the
# above.
client_ctx = Context(TLSv1_METHOD)
client_ctx.set_options(OP_NO_SSLv2 | OP_NO_SSLv3 | OP_SINGLE_DH_USE )
client_ctx.set_verify(VERIFY_PEER|VERIFY_FAIL_IF_NO_PEER_CERT|VERIFY_CLIENT_ONCE, verify_cb)
client_store = client_ctx.get_cert_store()
client_ctx.use_privatekey(load_privatekey(FILETYPE_PEM, client_key_pem))
client_ctx.use_certificate(load_certificate(FILETYPE_PEM, client_cert_pem))
client_ctx.check_privatekey()
client_store.add_cert(load_certificate(FILETYPE_PEM, root_cert_pem))
client_conn = Connection(client_ctx, sock)
client_conn.set_connect_state()
return client_conn
def test_set_multiple_ca_list(self):
"""
If passed a list containing multiple X509Name objects,
L{Context.set_client_ca_list} configures the context to send those CA
names to the client and, on both the server and client sides,
L{Connection.get_client_ca_list} returns a list containing those
X509Names after the connection is set up.
"""
secert = load_certificate(FILETYPE_PEM, server_cert_pem)
clcert = load_certificate(FILETYPE_PEM, server_cert_pem)
sedesc = secert.get_subject()
cldesc = clcert.get_subject()
def multiple_ca(ctx):
L = [sedesc, cldesc]
ctx.set_client_ca_list(L)
return L
self._check_client_ca_list(multiple_ca)
def test_reset_ca_list(self):
"""
If called multiple times, only the X509Names passed to the final call
of L{Context.set_client_ca_list} are used to configure the CA names
sent to the client.
"""
cacert = load_certificate(FILETYPE_PEM, root_cert_pem)
secert = load_certificate(FILETYPE_PEM, server_cert_pem)
clcert = load_certificate(FILETYPE_PEM, server_cert_pem)
cadesc = cacert.get_subject()
sedesc = secert.get_subject()
cldesc = clcert.get_subject()
def changed_ca(ctx):
ctx.set_client_ca_list([sedesc, cldesc])
ctx.set_client_ca_list([cadesc])
return [cadesc]
self._check_client_ca_list(changed_ca)
def test_mutated_ca_list(self):
"""
If the list passed to L{Context.set_client_ca_list} is mutated
afterwards, this does not affect the list of CA names sent to the
client.
"""
cacert = load_certificate(FILETYPE_PEM, root_cert_pem)
secert = load_certificate(FILETYPE_PEM, server_cert_pem)
cadesc = cacert.get_subject()
sedesc = secert.get_subject()
def mutated_ca(ctx):
L = [cadesc]
ctx.set_client_ca_list([cadesc])
L.append(sedesc)
return [cadesc]
self._check_client_ca_list(mutated_ca)
def test_set_and_add_client_ca(self):
"""
A call to L{Context.set_client_ca_list} followed by a call to
L{Context.add_client_ca} results in using the CA names from the first
call and the CA name from the second call.
"""
cacert = load_certificate(FILETYPE_PEM, root_cert_pem)
secert = load_certificate(FILETYPE_PEM, server_cert_pem)
clcert = load_certificate(FILETYPE_PEM, server_cert_pem)
cadesc = cacert.get_subject()
sedesc = secert.get_subject()
cldesc = clcert.get_subject()
def mixed_set_add_ca(ctx):
ctx.set_client_ca_list([cadesc, sedesc])
ctx.add_client_ca(clcert)
return [cadesc, sedesc, cldesc]
self._check_client_ca_list(mixed_set_add_ca)
def test_set_after_add_client_ca(self):
"""
A call to L{Context.set_client_ca_list} after a call to
L{Context.add_client_ca} replaces the CA name specified by the former
call with the names specified by the latter cal.
"""
cacert = load_certificate(FILETYPE_PEM, root_cert_pem)
secert = load_certificate(FILETYPE_PEM, server_cert_pem)
clcert = load_certificate(FILETYPE_PEM, server_cert_pem)
cadesc = cacert.get_subject()
sedesc = secert.get_subject()
cldesc = clcert.get_subject()
def set_replaces_add_ca(ctx):
ctx.add_client_ca(clcert)
ctx.set_client_ca_list([cadesc])
ctx.add_client_ca(secert)
return [cadesc, sedesc]
self._check_client_ca_list(set_replaces_add_ca)
def _secure_viewer(self):
cert_file='install/viewer-certs/ca-cert.pem'
cert_file=''
try:
with open(cert_file, "r") as caFile:
ca=caFile.read()
from OpenSSL import crypto
cert = crypto.load_certificate(crypto.FILETYPE_PEM, open(cert_file).read())
return {'defaultMode':'Secure',
'certificate':ca,
'domain':cert.get_issuer().organizationName}
except Exception as e:
print(e)
return {'defaultMode':'Insecure',
'certificate':'',
'domain':''}
def from_string(key_pem, is_x509_cert):
"""Construct a Verified instance from a string.
Args:
key_pem: string, public key in PEM format.
is_x509_cert: bool, True if key_pem is an X509 cert, otherwise it
is expected to be an RSA key in PEM format.
Returns:
Verifier instance.
Raises:
OpenSSL.crypto.Error: if the key_pem can't be parsed.
"""
key_pem = _to_bytes(key_pem)
if is_x509_cert:
pubkey = crypto.load_certificate(crypto.FILETYPE_PEM, key_pem)
else:
pubkey = crypto.load_privatekey(crypto.FILETYPE_PEM, key_pem)
return OpenSSLVerifier(pubkey)
def from_string(key_pem, is_x509_cert):
"""Construct a Verified instance from a string.
Args:
key_pem: string, public key in PEM format.
is_x509_cert: bool, True if key_pem is an X509 cert, otherwise it is
expected to be an RSA key in PEM format.
Returns:
Verifier instance.
Raises:
OpenSSL.crypto.Error if the key_pem can't be parsed.
"""
if is_x509_cert:
pubkey = crypto.load_certificate(crypto.FILETYPE_PEM, key_pem)
else:
pubkey = crypto.load_privatekey(crypto.FILETYPE_PEM, key_pem)
return OpenSSLVerifier(pubkey)
def read_ca_cert():
'''Loads CA certificate from the filesystem and returns it as an object.
Returns None if something went wrong.'''
common.logging_info("Loading CA certificate.")
# Check for missing CA certificate:
if not os.path.exists(config_ca.CA_CERT_FILE_PATH):
common.logging_error("No CA certificate file found at %s." % config_ca.CA_CERT_FILE_PATH)
return None
# Read and load CA certificate:
try:
file_object = open(config_ca.CA_CERT_FILE_PATH,'r')
file_contents = file_object.read()
file_object.close()
except IOError:
return None
# Return CA cert:
try:
return crypto.load_certificate(crypto.FILETYPE_PEM,file_contents)
except crypto.Error:
common.logging_error("Could not read CA certificate from %s." % config_ca.CA_CERT_FILE_PATH)
return None
def initFromPath(self, path):
'Will initialize the directory from the path supplied'
import cPickle
data = None
with open(path,'r') as f:
data = cPickle.load(f)
assert data != None, "Expected some data, did not load any."
priv_key_from_pem = lambda x: crypto.load_privatekey(crypto.FILETYPE_PEM, x)
for userName, keyTuple in data['users'].iteritems():
self.users[userName] = User(userName, directory=self,
ecdsaSigningKey=ecdsa.SigningKey.from_pem(keyTuple[0]),
rsaSigningKey=priv_key_from_pem(keyTuple[1]))
self.users[userName].tags = keyTuple[2]
for orgName, tuple in data['organizations'].iteritems():
org = Organization(orgName, ecdsaSigningKey=ecdsa.SigningKey.from_pem(tuple[0]),
rsaSigningKey=priv_key_from_pem(tuple[0]))
org.signedCert = crypto.load_certificate(crypto.FILETYPE_PEM, tuple[2])
org.networks = [Network[name] for name in tuple[3]]
self.organizations[orgName] = org
for nat, cert_as_pem in data['nats'].iteritems():
self.ordererAdminTuples[nat] = crypto.load_certificate(crypto.FILETYPE_PEM, cert_as_pem)
def __new__(typ, value):
if isinstance(value, basestring):
path = process.config_file(value)
if path is None:
log.warn("Certificate file '%s' is not readable" % value)
return None
try:
f = open(path, 'rt')
except:
log.warn("Certificate file '%s' could not be open" % value)
return None
try:
try:
return crypto.load_certificate(crypto.FILETYPE_PEM, f.read())
except crypto.Error, e:
log.warn("Certificate file '%s' could not be loaded: %s" % (value, str(e)))
return None
finally:
f.close()
else:
raise TypeError, 'value should be a string'
