def generate_adhoc_ssl_context():
"""Generates an adhoc SSL context for the development server."""
crypto = _get_openssl_crypto_module()
import tempfile
import atexit
cert, pkey = generate_adhoc_ssl_pair()
cert_handle, cert_file = tempfile.mkstemp()
pkey_handle, pkey_file = tempfile.mkstemp()
atexit.register(os.remove, pkey_file)
atexit.register(os.remove, cert_file)
os.write(cert_handle, crypto.dump_certificate(crypto.FILETYPE_PEM, cert))
os.write(pkey_handle, crypto.dump_privatekey(crypto.FILETYPE_PEM, pkey))
os.close(cert_handle)
os.close(pkey_handle)
ctx = load_ssl_context(cert_file, pkey_file)
return ctx
python类dump_privatekey()的实例源码
def generate_adhoc_ssl_context():
"""Generates an adhoc SSL context for the development server."""
crypto = _get_openssl_crypto_module()
import tempfile
import atexit
cert, pkey = generate_adhoc_ssl_pair()
cert_handle, cert_file = tempfile.mkstemp()
pkey_handle, pkey_file = tempfile.mkstemp()
atexit.register(os.remove, pkey_file)
atexit.register(os.remove, cert_file)
os.write(cert_handle, crypto.dump_certificate(crypto.FILETYPE_PEM, cert))
os.write(pkey_handle, crypto.dump_privatekey(crypto.FILETYPE_PEM, pkey))
os.close(cert_handle)
os.close(pkey_handle)
ctx = load_ssl_context(cert_file, pkey_file)
return ctx
def generate(self, passphrase: str = None, common_name=None, days=DEFAULT_CERT_VALIDITY, is_server=False):
k = crypto.PKey()
k.generate_key(crypto.TYPE_RSA, self.key_length)
cert = crypto.X509()
# cert.get_subject().CN = common_name
cert.get_subject().commonName = common_name
cert.set_serial_number(random.randint(990000, 999999999999999999999999999))
cert.gmtime_adj_notBefore(-600)
cert.gmtime_adj_notAfter(int(datetime.timedelta(days=days).total_seconds()))
cert.set_issuer(self.ca_cert.get_subject())
cert.set_pubkey(k)
cert = self._add_extensions(cert, is_server)
cert.sign(self.ca_key, self.digest)
self.certificate = crypto.dump_certificate(crypto.FILETYPE_PEM, cert)
if passphrase:
self.private_key = crypto.dump_privatekey(crypto.FILETYPE_PEM, k, cipher="DES-EDE3-CBC", passphrase=passphrase.encode())
else:
self.private_key = crypto.dump_privatekey(crypto.FILETYPE_PEM, k)
return self
def test_load_pkcs12(self):
"""
A PKCS12 string generated using the openssl command line can be loaded
with L{load_pkcs12} and its components extracted and examined.
"""
passwd = 'whatever'
pem = client_key_pem + client_cert_pem
p12_str = _runopenssl(
pem, "pkcs12", '-export', '-clcerts', '-passout', 'pass:' + passwd)
p12 = load_pkcs12(p12_str, passwd)
# verify
self.assertTrue(isinstance(p12, PKCS12))
cert_pem = dump_certificate(FILETYPE_PEM, p12.get_certificate())
self.assertEqual(cert_pem, client_cert_pem)
key_pem = dump_privatekey(FILETYPE_PEM, p12.get_privatekey())
self.assertEqual(key_pem, client_key_pem)
self.assertEqual(None, p12.get_ca_certificates())
def test_dump_privatekey(self):
"""
L{dump_privatekey} writes a PEM, DER, and text.
"""
key = load_privatekey(FILETYPE_PEM, cleartextPrivateKeyPEM)
dumped_pem = dump_privatekey(FILETYPE_PEM, key)
self.assertEqual(dumped_pem, cleartextPrivateKeyPEM)
dumped_der = dump_privatekey(FILETYPE_ASN1, key)
# XXX This OpenSSL call writes "writing RSA key" to standard out. Sad.
good_der = _runopenssl(dumped_pem, "rsa", "-outform", "DER")
self.assertEqual(dumped_der, good_der)
key2 = load_privatekey(FILETYPE_ASN1, dumped_der)
dumped_pem2 = dump_privatekey(FILETYPE_PEM, key2)
self.assertEqual(dumped_pem2, cleartextPrivateKeyPEM)
dumped_text = dump_privatekey(FILETYPE_TEXT, key)
good_text = _runopenssl(dumped_pem, "rsa", "-noout", "-text")
self.assertEqual(dumped_text, good_text)
def test_dump_privatekey_passphraseCallback(self):
"""
L{dump_privatekey} writes an encrypted PEM when given a callback which
returns the correct passphrase.
"""
passphrase = "foo"
called = []
def cb(writing):
called.append(writing)
return passphrase
key = load_privatekey(FILETYPE_PEM, cleartextPrivateKeyPEM)
pem = dump_privatekey(FILETYPE_PEM, key, "blowfish", cb)
self.assertTrue(isinstance(pem, str))
self.assertEqual(called, [True])
loadedKey = load_privatekey(FILETYPE_PEM, pem, passphrase)
self.assertTrue(isinstance(loadedKey, PKeyType))
self.assertEqual(loadedKey.type(), key.type())
self.assertEqual(loadedKey.bits(), key.bits())
def test_set_passwd_cb(self):
"""
L{Context.set_passwd_cb} accepts a callable which will be invoked when
a private key is loaded from an encrypted PEM.
"""
key = PKey()
key.generate_key(TYPE_RSA, 128)
pemFile = self.mktemp()
fObj = file(pemFile, 'w')
passphrase = "foobar"
fObj.write(dump_privatekey(FILETYPE_PEM, key, "blowfish", passphrase))
fObj.close()
calledWith = []
def passphraseCallback(maxlen, verify, extra):
calledWith.append((maxlen, verify, extra))
return passphrase
context = Context(TLSv1_METHOD)
context.set_passwd_cb(passphraseCallback)
context.use_privatekey_file(pemFile)
self.assertTrue(len(calledWith), 1)
self.assertTrue(isinstance(calledWith[0][0], int))
self.assertTrue(isinstance(calledWith[0][1], int))
self.assertEqual(calledWith[0][2], None)
def generate_adhoc_ssl_context():
"""Generates an adhoc SSL context for the development server."""
crypto = _get_openssl_crypto_module()
import tempfile
import atexit
cert, pkey = generate_adhoc_ssl_pair()
cert_handle, cert_file = tempfile.mkstemp()
pkey_handle, pkey_file = tempfile.mkstemp()
atexit.register(os.remove, pkey_file)
atexit.register(os.remove, cert_file)
os.write(cert_handle, crypto.dump_certificate(crypto.FILETYPE_PEM, cert))
os.write(pkey_handle, crypto.dump_privatekey(crypto.FILETYPE_PEM, pkey))
os.close(cert_handle)
os.close(pkey_handle)
ctx = load_ssl_context(cert_file, pkey_file)
return ctx
def generate_adhoc_ssl_context():
"""Generates an adhoc SSL context for the development server."""
crypto = _get_openssl_crypto_module()
import tempfile
import atexit
cert, pkey = generate_adhoc_ssl_pair()
cert_handle, cert_file = tempfile.mkstemp()
pkey_handle, pkey_file = tempfile.mkstemp()
atexit.register(os.remove, pkey_file)
atexit.register(os.remove, cert_file)
os.write(cert_handle, crypto.dump_certificate(crypto.FILETYPE_PEM, cert))
os.write(pkey_handle, crypto.dump_privatekey(crypto.FILETYPE_PEM, pkey))
os.close(cert_handle)
os.close(pkey_handle)
ctx = load_ssl_context(cert_file, pkey_file)
return ctx
def generate_adhoc_ssl_context():
"""Generates an adhoc SSL context for the development server."""
crypto = _get_openssl_crypto_module()
import tempfile
import atexit
cert, pkey = generate_adhoc_ssl_pair()
cert_handle, cert_file = tempfile.mkstemp()
pkey_handle, pkey_file = tempfile.mkstemp()
atexit.register(os.remove, pkey_file)
atexit.register(os.remove, cert_file)
os.write(cert_handle, crypto.dump_certificate(crypto.FILETYPE_PEM, cert))
os.write(pkey_handle, crypto.dump_privatekey(crypto.FILETYPE_PEM, pkey))
os.close(cert_handle)
os.close(pkey_handle)
ctx = load_ssl_context(cert_file, pkey_file)
return ctx
def make_identity_pem_string(given_key,given_cert,given_ca_cert):
'''Given objects for client key, client cert, and CA cert,
extract their contents in PEM format and combine them
into a single text string. Return that string or None.'''
common.logging_info("Generating client identity PEM.")
try:
key_pem = crypto.dump_privatekey(crypto.FILETYPE_PEM,given_key)
except crypto.Error:
common.logging_error("Could not get PEM contents of private key.")
return None
try:
cert_pem = crypto.dump_certificate(crypto.FILETYPE_PEM,given_cert)
except crypto.Error:
common.logging_error("Could not get PEM contents of client certificate.")
return None
try:
ca_pem = crypto.dump_certificate(crypto.FILETYPE_PEM,given_ca_cert)
except crypto.Error:
common.logging_error("Could not get PEM contents of CA certificate.")
return None
combined_pem = '%(key_pem)s\n%(cert_pem)s\n%(ca_pem)s' % {'key_pem':key_pem,'cert_pem':cert_pem,'ca_pem':ca_pem}
return combined_pem
def generate_adhoc_ssl_context():
"""Generates an adhoc SSL context for the development server."""
crypto = _get_openssl_crypto_module()
import tempfile
import atexit
cert, pkey = generate_adhoc_ssl_pair()
cert_handle, cert_file = tempfile.mkstemp()
pkey_handle, pkey_file = tempfile.mkstemp()
atexit.register(os.remove, pkey_file)
atexit.register(os.remove, cert_file)
os.write(cert_handle, crypto.dump_certificate(crypto.FILETYPE_PEM, cert))
os.write(pkey_handle, crypto.dump_privatekey(crypto.FILETYPE_PEM, pkey))
os.close(cert_handle)
os.close(pkey_handle)
ctx = load_ssl_context(cert_file, pkey_file)
return ctx
def generate_adhoc_ssl_context():
"""Generates an adhoc SSL context for the development server."""
crypto = _get_openssl_crypto_module()
import tempfile
import atexit
cert, pkey = generate_adhoc_ssl_pair()
cert_handle, cert_file = tempfile.mkstemp()
pkey_handle, pkey_file = tempfile.mkstemp()
atexit.register(os.remove, pkey_file)
atexit.register(os.remove, cert_file)
os.write(cert_handle, crypto.dump_certificate(crypto.FILETYPE_PEM, cert))
os.write(pkey_handle, crypto.dump_privatekey(crypto.FILETYPE_PEM, pkey))
os.close(cert_handle)
os.close(pkey_handle)
ctx = load_ssl_context(cert_file, pkey_file)
return ctx
def generate_adhoc_ssl_context():
"""Generates an adhoc SSL context for the development server."""
crypto = _get_openssl_crypto_module()
import tempfile
import atexit
cert, pkey = generate_adhoc_ssl_pair()
cert_handle, cert_file = tempfile.mkstemp()
pkey_handle, pkey_file = tempfile.mkstemp()
atexit.register(os.remove, pkey_file)
atexit.register(os.remove, cert_file)
os.write(cert_handle, crypto.dump_certificate(crypto.FILETYPE_PEM, cert))
os.write(pkey_handle, crypto.dump_privatekey(crypto.FILETYPE_PEM, pkey))
os.close(cert_handle)
os.close(pkey_handle)
ctx = load_ssl_context(cert_file, pkey_file)
return ctx
def generate_adhoc_ssl_context():
"""Generates an adhoc SSL context for the development server."""
crypto = _get_openssl_crypto_module()
import tempfile
import atexit
cert, pkey = generate_adhoc_ssl_pair()
cert_handle, cert_file = tempfile.mkstemp()
pkey_handle, pkey_file = tempfile.mkstemp()
atexit.register(os.remove, pkey_file)
atexit.register(os.remove, cert_file)
os.write(cert_handle, crypto.dump_certificate(crypto.FILETYPE_PEM, cert))
os.write(pkey_handle, crypto.dump_privatekey(crypto.FILETYPE_PEM, pkey))
os.close(cert_handle)
os.close(pkey_handle)
ctx = load_ssl_context(cert_file, pkey_file)
return ctx
def generate_adhoc_ssl_context():
"""Generates an adhoc SSL context for the development server."""
crypto = _get_openssl_crypto_module()
import tempfile
import atexit
cert, pkey = generate_adhoc_ssl_pair()
cert_handle, cert_file = tempfile.mkstemp()
pkey_handle, pkey_file = tempfile.mkstemp()
atexit.register(os.remove, pkey_file)
atexit.register(os.remove, cert_file)
os.write(cert_handle, crypto.dump_certificate(crypto.FILETYPE_PEM, cert))
os.write(pkey_handle, crypto.dump_privatekey(crypto.FILETYPE_PEM, pkey))
os.close(cert_handle)
os.close(pkey_handle)
ctx = load_ssl_context(cert_file, pkey_file)
return ctx
def create_v1_id_card(self, node_admin_tuple, conn_profile):
from zipfile import ZipFile
from io import BytesIO
user_name = node_admin_tuple.user
user = self.directory.getUser(userName=user_name)
in_memory = BytesIO()
zf = ZipFile(in_memory, mode="w")
metadata = self.create_v1_metadata(user_name=node_admin_tuple.nodeName)
zf.writestr("metadata.json", json.dumps(metadata, separators=(',', ':')))
zf.writestr("connection.json", json.dumps(conn_profile, separators=(',', ':')))
cert = self.directory.findCertForNodeAdminTuple(node_admin_tuple)
zf.writestr(os.path.join("credentials","certificate"), crypto.dump_certificate(crypto.FILETYPE_PEM, cert))
zf.writestr(os.path.join("credentials","privateKey"), crypto.dump_privatekey(crypto.FILETYPE_PEM, user.pKey))
zf.close()
in_memory.seek(0)
return in_memory.read()
def dump(self, output):
'Will dump the directory to the provided store'
import cPickle
data = {'users' : {}, 'organizations' : {}, 'nats' : {}}
dump_cert = lambda cert: crypto.dump_certificate(crypto.FILETYPE_PEM, cert)
for userName, user in self.users.iteritems():
# for k, v in user.tags.iteritems():
# try:
# cPickle.dumps(v)
# except:
# raise Exception("Failed on key {0}".format(k))
data['users'][userName] = (user.ecdsaSigningKey.to_pem(), crypto.dump_privatekey(crypto.FILETYPE_PEM, user.rsaSigningKey), user.tags)
for orgName, org in self.organizations.iteritems():
networks = [n.name for n in org.networks]
data['organizations'][orgName] = (
org.ecdsaSigningKey.to_pem(), crypto.dump_privatekey(crypto.FILETYPE_PEM, org.rsaSigningKey),
dump_cert(org.getSelfSignedCert()), networks)
for nat, cert in self.ordererAdminTuples.iteritems():
data['nats'][nat] = dump_cert(cert)
cPickle.dump(data, output)
def test_dump_privatekey_passphraseCallback(self):
"""
:py:obj:`dump_privatekey` writes an encrypted PEM when given a callback which
returns the correct passphrase.
"""
passphrase = b("foo")
called = []
def cb(writing):
called.append(writing)
return passphrase
key = load_privatekey(FILETYPE_PEM, cleartextPrivateKeyPEM)
pem = dump_privatekey(FILETYPE_PEM, key, GOOD_CIPHER, cb)
self.assertTrue(isinstance(pem, binary_type))
self.assertEqual(called, [True])
loadedKey = load_privatekey(FILETYPE_PEM, pem, passphrase)
self.assertTrue(isinstance(loadedKey, PKeyType))
self.assertEqual(loadedKey.type(), key.type())
self.assertEqual(loadedKey.bits(), key.bits())
def pkcs12_key_as_pem(private_key_text, private_key_password):
"""Convert the contents of a PKCS12 key to PEM using OpenSSL.
Args:
private_key_text: String. Private key.
private_key_password: String. Password for PKCS12.
Returns:
String. PEM contents of ``private_key_text``.
"""
decoded_body = base64.b64decode(private_key_text)
private_key_password = _to_bytes(private_key_password)
pkcs12 = crypto.load_pkcs12(decoded_body, private_key_password)
return crypto.dump_privatekey(crypto.FILETYPE_PEM,
pkcs12.get_privatekey())
