def test_rsa(self):
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives import serialization
private_key = rsa.generate_private_key(
public_exponent=65537,
key_size=2048,
backend=default_backend()
)
private_key_pem = private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.BestAvailableEncryption(passphrase)
)
public_key_pem = private_key.public_key().public_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PublicFormat.SubjectPublicKeyInfo
)
url = 'http://example.com/path?query#fragment'
auth = HTTPSignatureAuth(algorithm="rsa-sha256", key=private_key_pem, key_id="sekret", passphrase=passphrase)
self.session.get(url, auth=auth, headers=dict(pubkey=base64.b64encode(public_key_pem)))
python类generate_private_key()的实例源码
def generate_RSA(bits=4096):
new_key = rsa.generate_private_key(
public_exponent=65537,
key_size=bits,
backend=default_backend()
)
private_key = new_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption()
)
public_key = new_key.public_key().public_bytes(
encoding=serialization.Encoding.OpenSSH,
format=serialization.PublicFormat.OpenSSH
)
return private_key, public_key
def generate_RSA(bits=4096):
new_key = rsa.generate_private_key(
public_exponent=65537,
key_size=bits,
backend=default_backend()
)
private_key = new_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption()
)
public_key = new_key.public_key().public_bytes(
encoding=serialization.Encoding.OpenSSH,
format=serialization.PublicFormat.OpenSSH
)
return private_key, public_key
def create_decryptor(private_location, public_location):
try:
with open(private_location, "rb") as key_file:
private_key = serialization.load_pem_private_key(
key_file.read(),
password=None,
backend=default_backend()
)
except FileNotFoundError:
with open(private_location, "wb") as key_file:
private_key = rsa.generate_private_key(
public_exponent=65537,
key_size=2048,
backend=default_backend()
)
key_file.write(private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.TraditionalOpenSSL,
encryption_algorithm=serialization.NoEncryption()
))
with open(public_location, "wb") as public_file:
public_key = private_key.public_key()
pem = public_key.public_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PublicFormat.SubjectPublicKeyInfo
)
public_file.write(pem)
def decrypt(ciphertext):
return private_key.decrypt(
ciphertext,
padding.OAEP(
mgf=padding.MGF1(algorithm=hashes.SHA1()),
algorithm=hashes.SHA1(),
label=None
)
)
return decrypt
def test_cache_timer(self):
"""
Test if the cache max-age is retrieved from the HTTPS resource
"""
private_key = generate_private_key(
public_exponent=65537,
key_size=2048,
backend=default_backend()
)
public_numbers = private_key.public_key().public_numbers()
test_id = "thisisatestid"
server_address = create_webserver.start_server(public_numbers.n, public_numbers.e, test_id)
print(server_address)
_, cache_timer = self.keycache._get_issuer_publickey("http://localhost:{}/".format(server_address[1]),
key_id=test_id,
insecure=True)
self.assertEqual(cache_timer, 3600)
create_webserver.shutdown_server()
def generate_keys(self):
private_key = rsa.generate_private_key(
public_exponent=65537,
key_size=2048,
backend=default_backend()
)
private_pem = private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.BestAvailableEncryption(bytes(CloudLinkSettings.PRIVATE_KEY_PASSPHRASE))
)
public_pem = private_key.public_key().public_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PublicFormat.SubjectPublicKeyInfo
)
with open(CloudLinkSettings.PRIVATE_KEY_LOCATION, 'w') as key_file:
key_file.write(str(private_pem))
with open(CloudLinkSettings.PUBLIC_KEY_LOCATION, 'w') as key_file:
key_file.write(str(public_pem))
def create_rsa_key():
# generate private key
private_key = rsa.generate_private_key(
public_exponent=65537,
key_size=2048,
backend=default_backend()
)
# output PEM encoded version of private key
privateKey = private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption()
)
return privateKey.decode()
def generate_key():
# generate private/public key pair
key = rsa.generate_private_key(backend=default_backend(), public_exponent=65537, key_size=2048)
# get public key in OpenSSH format
public_key = key.public_key().public_bytes(
serialization.Encoding.OpenSSH, serialization.PublicFormat.OpenSSH
)
# get private key in PEM container format
pem = key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.TraditionalOpenSSL,
encryption_algorithm=serialization.NoEncryption()
)
return KeyPair(pem.decode('utf-8'), public_key.decode('utf-8'))
def __init__(self):
self._private_key = rsa.generate_private_key(
public_exponent=65537,
key_size=_KEY_SIZE,
backend=default_backend()
)
name = _name(u"Testing CA #" + random_text())
self._certificate = (
_cert_builder_common(name, name, self._private_key.public_key())
.add_extension(
x509.BasicConstraints(ca=True, path_length=9), critical=True,
)
.sign(
private_key=self._private_key,
algorithm=hashes.SHA256(),
backend=default_backend(),
)
)
self.cert_pem = Blob(self._certificate.public_bytes(Encoding.PEM))
def generate_ssh_key():
logging.info('NOTICE! Generating a new private/public key combination, be AWARE!')
key = rsa.generate_private_key(
backend=crypto_default_backend(),
public_exponent=65537,
key_size=2048
)
private_key = key.private_bytes(
crypto_serialization.Encoding.PEM,
crypto_serialization.PrivateFormat.PKCS8,
crypto_serialization.NoEncryption())
public_key = key.public_key().public_bytes(
crypto_serialization.Encoding.OpenSSH,
crypto_serialization.PublicFormat.OpenSSH
)
with open("{}/keys/private.key".format(SETTINGS['RESOURCESDIR']), 'w') as content_file:
content_file.write(private_key)
chmod("{}/keys/private.key".format(SETTINGS['RESOURCESDIR']), 0600)
with open("{}/keys/public.key".format(SETTINGS['RESOURCESDIR']), 'w') as content_file:
content_file.write(public_key)
return public_key
def generate_account(name='default'):
private_key = rsa.generate_private_key(
public_exponent=65537,
key_size=2048,
backend=default_backend()
)
private_serialized = private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption()
)
public_serialized = private_key.public_key().public_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PublicFormat.SubjectPublicKeyInfo
)
public_lines = public_serialized.splitlines()
public_shrunk = ''
for line in range(1, len(public_lines) - 1):
public_shrunk += public_lines[line].strip('\n')
return {
'name': name,
'private_key': private_serialized,
'public_key': public_shrunk
}
def gen_key_pair():
"""
This method generates the public and private key pair.
The public key format is OpenSSH and private key format is PEM container
:return:
"""
private_key = rsa.generate_private_key(backend=default_backend(),
public_exponent=65537,
key_size=2048)
public_key_str = private_key.public_key(). \
public_bytes(serialization.Encoding.OpenSSH,
serialization.PublicFormat.OpenSSH).decode('utf-8')
private_key_str = private_key. \
private_bytes(encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.TraditionalOpenSSL,
encryption_algorithm=serialization.NoEncryption()
).decode('utf-8')
return (private_key_str, public_key_str)
def make_ssh_key():
"""
Creates public and private ssh keys with 1024 bit RSA encryption
:return: private, public key
"""
key = rsa.generate_private_key(
backend=crypto_default_backend(),
public_exponent=65537,
key_size=1024
)
private_key = key.private_bytes(
crypto_serialization.Encoding.PEM,
crypto_serialization.PrivateFormat.PKCS8,
crypto_serialization.NoEncryption())
public_key = key.public_key().public_bytes(
crypto_serialization.Encoding.OpenSSH,
crypto_serialization.PublicFormat.OpenSSH
)
return private_key.decode('utf-8'), public_key.decode('utf-8')
def genkeys(self, key_size=KeySize.NORMAL, password=None):
"""
Generates a Private and Public Key set and returns them in a tuple
(private, public)
"""
self.private_key = rsa.generate_private_key(
# The public exponent of the new key. Usually one of the small
# Fermat primes 3, 5, 17, 257, 65537. If in doubt you should use
# 65537. See http://www.daemonology.net/blog/2009-06-11-\
# cryptographic-right-answers.html
public_exponent=65537,
key_size=key_size,
backend=default_backend()
)
# Generate our Public Key
self.public_key = self.private_key.public_key()
# Store our password; this will be used when we save our content
# via it's searialized value later on
self.password = password
# Returns a (RSAPrivateKey, RSAPublicKey)
return (self.private_key, self.public_key)
def generate_keypair(key_out_path):
private_key = rsa.generate_private_key(public_exponent=65537, key_size=4096, backend=default_backend())
pem = private_key.private_bytes(encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption())
with open(key_out_path, 'wb') as outf:
print >>outf, pem
def gen_key():
private_key = rsa.generate_private_key(
public_exponent=65537, key_size=2048, backend=default_backend()
)
return private_key
def generate_self_signed_cert(cert_file, key_file):
# type: (Any, Any) -> None
"""Given two file-like objects, generate an SSL key and certificate
Args:
cert_file: The certificate file you wish to write to
key_file: The key file you wish to write to
"""
one_day = timedelta(1, 0, 0)
private_key = rsa.generate_private_key(
public_exponent=65537, key_size=2048, backend=default_backend())
public_key = private_key.public_key()
builder = x509.CertificateBuilder()
builder = builder.subject_name(
x509.Name([
x509.NameAttribute(NameOID.COMMON_NAME, u'cryptography.io'),
]))
builder = builder.issuer_name(
x509.Name([
x509.NameAttribute(NameOID.COMMON_NAME, u'cryptography.io'),
]))
builder = builder.not_valid_before(datetime.today() - one_day)
builder = builder.not_valid_after(datetime.today() + timedelta(365 * 10))
builder = builder.serial_number(uuid4().int)
builder = builder.public_key(public_key)
builder = builder.add_extension(
x509.BasicConstraints(ca=False, path_length=None),
critical=True, )
certificate = builder.sign(
private_key=private_key,
algorithm=hashes.SHA256(),
backend=default_backend())
key_file.write(
private_key.private_bytes(
Encoding.PEM, PrivateFormat.TraditionalOpenSSL, NoEncryption()))
cert_file.write(certificate.public_bytes(Encoding.PEM))
def generate_ssh_keys(
cls, size=4096, passphrase=None, private_format=PrivateFormat.PKCS8,
public_format=PublicFormat.OpenSSH, private_encoding=Encoding.PEM,
public_encoding=Encoding.OpenSSH):
"""Generates a public and private rsa ssh key
Returns an SSHKeyResponse objects which has both the public and private
key as attributes
:param int size: RSA modulus length (must be a multiple of 256)
and >= 1024
:param str passphrase: The pass phrase to derive the encryption key
from
"""
encryption = (
BestAvailableEncryption(passphrase) if passphrase else
NoEncryption())
key = rsa.generate_private_key(
backend=default_backend(),
public_exponent=65537,
key_size=size)
return SSHKey(
public_key=key.public_key().public_bytes(
public_encoding, public_format),
private_key=key.private_bytes(
Encoding.PEM, private_format, encryption))
test_endpoint.py 文件源码
项目：Complete-Bunq-API-Python-Wrapper
作者: PJUllrich
项目源码
文件源码
阅读 27
收藏 0
点赞 0
评论 0
def create_random_privkey():
private_key = rsa.generate_private_key(
public_exponent=65537,
key_size=2048,
backend=default_backend()
)
return private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption()
).decode()
def create_new_key_pair(self, save_to_config=True):
"""Creates a new public/private key pair and saves them to the config file
:return: Prints out a success message
"""
private_key = rsa.generate_private_key(
public_exponent=65537,
key_size=2048,
backend=default_backend()
)
private_key_decoded = private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption()
).decode()
public_key_decoded = private_key.public_key().public_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PublicFormat.SubjectPublicKeyInfo
).decode()
print('New key pair was created')
if save_to_config:
self.config.set('key_private', private_key_decoded)
self.config.set('key_public', public_key_decoded)
else:
print('\tNew Private Key: %s' % private_key_decoded)
print('\tNew Public Key: %s' % public_key_decoded)
return private_key_decoded, public_key_decoded
def generate_key_pair(cls):
private_key = rsa.generate_private_key(
public_exponent=65537,
key_size=3072,
backend=default_backend()
)
return AsymmetricKeyPair(
private_key.private_bytes(encryption_algorithm=serialization.NoEncryption(),
encoding=serialization.Encoding.DER,
format=serialization.PrivateFormat.TraditionalOpenSSL),
private_key.public_key().public_bytes(encoding=serialization.Encoding.DER,
format=serialization.PublicFormat.PKCS1))
def generate_RSA(bits=4096):
# public_exponent set to 655537 is what pyCA recommends
new_key = rsa.generate_private_key(public_exponent=65537,
key_size=bits,
backend=default_backend())
# we strip trailing space for 1:1 compat with previous implementation
private_key = new_key.private_bytes(
encoding=Encoding.PEM,
format=PrivateFormat.PKCS8,
encryption_algorithm=NoEncryption())
public_key = new_key.public_key().public_bytes(encoding=Encoding.OpenSSH,
format=PublicFormat.OpenSSH)
return private_key, public_key
def generate_RSA(bits=4096):
# public_exponent set to 655537 is what pyCA recommends
new_key = rsa.generate_private_key(public_exponent=65537,
key_size=bits,
backend=default_backend())
# we strip trailing space for 1:1 compat with previous implementation
private_key = new_key.private_bytes(
encoding=Encoding.PEM,
format=PrivateFormat.PKCS8,
encryption_algorithm=NoEncryption())
public_key = new_key.public_key().public_bytes(encoding=Encoding.OpenSSH,
format=PublicFormat.OpenSSH)
return private_key, public_key
def test_verify_signature_ECC(self, mock_get_pub_key):
data = b'224626ae19824466f2a7f39ab7b80f7f'
# test every ECC curve
for curve in signature_utils.ECC_CURVES:
key_type_name = 'ECC_' + curve.name.upper()
try:
signature_utils.SignatureKeyType.lookup(key_type_name)
except exception.SignatureVerificationError:
import warnings
warnings.warn("ECC curve '%s' not supported" % curve.name)
continue
# Create a private key to use
private_key = ec.generate_private_key(curve,
default_backend())
mock_get_pub_key.return_value = private_key.public_key()
for hash_name, hash_alg in signature_utils.HASH_METHODS.items():
signer = private_key.signer(
ec.ECDSA(hash_alg)
)
signer.update(data)
signature = base64.b64encode(signer.finalize())
img_sig_cert_uuid = 'fea14bc2-d75f-4ba5-bccc-b5c924ad0693'
verifier = signature_utils.get_verifier(None,
img_sig_cert_uuid,
hash_name, signature,
key_type_name)
verifier.update(data)
verifier.verify()
def generate_private_key(key_size):
"""Generate rsa private key."""
private_key = rsa.generate_private_key(
public_exponent=65537,
key_size=key_size,
backend=default_backend()
)
return private_key
def set_key(self):
password = settings.ACCOUNT_KEY_PASSWORD.encode()
key = rsa.generate_private_key(public_exponent=65537, key_size=settings.BITS, backend=default_backend())
self.key = key.private_bytes(encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.TraditionalOpenSSL,
encryption_algorithm=serialization.BestAvailableEncryption(password))
def generate_asym_key(size):
assert size > 1023
private_key = rsa.generate_private_key(
public_exponent=65537,
key_size=size,
backend=default_backend()
)
return RSAPrivateKey(private_key)
def ca_file(tmpdir):
"""
Create a valid PEM file with CA certificates and return the path.
"""
key = rsa.generate_private_key(
public_exponent=65537,
key_size=2048,
backend=default_backend()
)
public_key = key.public_key()
builder = x509.CertificateBuilder()
builder = builder.subject_name(x509.Name([
x509.NameAttribute(NameOID.COMMON_NAME, u"pyopenssl.org"),
]))
builder = builder.issuer_name(x509.Name([
x509.NameAttribute(NameOID.COMMON_NAME, u"pyopenssl.org"),
]))
one_day = datetime.timedelta(1, 0, 0)
builder = builder.not_valid_before(datetime.datetime.today() - one_day)
builder = builder.not_valid_after(datetime.datetime.today() + one_day)
builder = builder.serial_number(int(uuid.uuid4()))
builder = builder.public_key(public_key)
builder = builder.add_extension(
x509.BasicConstraints(ca=True, path_length=None), critical=True,
)
certificate = builder.sign(
private_key=key, algorithm=hashes.SHA256(),
backend=default_backend()
)
ca_file = tmpdir.join("test.pem")
ca_file.write_binary(
certificate.public_bytes(
encoding=serialization.Encoding.PEM,
)
)
return str(ca_file).encode("ascii")
def generate_new_key(self, keysize=2048):
self._private_key = crypto_rsa.generate_private_key(
public_exponent=65537,
key_size=keysize,
backend=crypto_backend,
)
self._public_key = self._private_key.public_key()
def test_empty(self):
"""
Test when the keycache should be empty
"""
# Stand up an HTTP server
private_key = generate_private_key(
public_exponent=65537,
key_size=2048,
backend=default_backend()
)
public_numbers = private_key.public_key().public_numbers()
test_id = "thisisatestid"
server_address = create_webserver.start_server(public_numbers.n, public_numbers.e, test_id)
print(server_address)
# Now try to get the public key from the server
pubkey_from_keycache = self.keycache.getkeyinfo("http://localhost:{}/".format(server_address[1]),
test_id,
insecure=True)
# Now compare the 2 public keys
public_pem = private_key.public_key().public_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PublicFormat.SubjectPublicKeyInfo
)
pubkey_pem_from_keycache = pubkey_from_keycache.public_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PublicFormat.SubjectPublicKeyInfo
)
self.assertEqual(public_pem, pubkey_pem_from_keycache)
create_webserver.shutdown_server()
