internet/community.crypto
3
0
Fork 0
mirror of https://github.com/ansible-collections/community.crypto.git synced 2026-05-07 13:53:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

Reformat everything with black.

Browse Source 
I had to undo the u string prefix removals to not drop Python 2 compatibility.
That's why black isn't enabled in antsibull-nox.toml yet.
This commit is contained in:
Felix Fontein
2025-04-28 09:51:33 +02:00
parent 04a0d38e3b
commit aec1826c34
118 changed files with 11780 additions and 7565 deletions
Download Patch File Download Diff File Expand all files Collapse all files

208
plugins/module_utils/crypto/support.py
View File

@@ -32,6 +32,7 @@ from ansible_collections.community.crypto.plugins.module_utils.time import ( #
try:
from OpenSSL import crypto
HAS_PYOPENSSL = True
except (ImportError, AttributeError):
# Error handled in the calling module.
@@ -52,7 +53,14 @@ from .basic import OpenSSLBadPassphraseError, OpenSSLObjectError
# This list of preferred fingerprints is used when prefer_one=True is supplied to the
# fingerprinting methods.
PREFERRED_FINGERPRINTS = (
'sha256', 'sha3_256', 'sha512', 'sha3_512', 'sha384', 'sha3_384', 'sha1', 'md5'
"sha256",
"sha3_256",
"sha512",
"sha3_512",
"sha384",
"sha3_384",
"sha1",
"md5",
)
@@ -71,8 +79,16 @@ def get_fingerprint_of_bytes(source, prefer_one=False):
if prefer_one:
# Sort algorithms to have the ones in PREFERRED_FINGERPRINTS at the beginning
prefered_algorithms = [algorithm for algorithm in PREFERRED_FINGERPRINTS if algorithm in algorithms]
prefered_algorithms += sorted([algorithm for algorithm in algorithms if algorithm not in PREFERRED_FINGERPRINTS])
prefered_algorithms = [
algorithm for algorithm in PREFERRED_FINGERPRINTS if algorithm in algorithms
]
prefered_algorithms += sorted(
[
algorithm
for algorithm in algorithms
if algorithm not in PREFERRED_FINGERPRINTS
]
)
algorithms = prefered_algorithms
for algo in algorithms:
@@ -88,34 +104,47 @@ def get_fingerprint_of_bytes(source, prefer_one=False):
pubkey_digest = h.hexdigest()
except TypeError:
pubkey_digest = h.hexdigest(32)
fingerprint[algo] = ':'.join(pubkey_digest[i:i + 2] for i in range(0, len(pubkey_digest), 2))
fingerprint[algo] = ":".join(
pubkey_digest[i : i + 2] for i in range(0, len(pubkey_digest), 2)
)
if prefer_one:
break
return fingerprint
def get_fingerprint_of_privatekey(privatekey, backend='cryptography', prefer_one=False):
"""Generate the fingerprint of the public key. """
def get_fingerprint_of_privatekey(privatekey, backend="cryptography", prefer_one=False):
"""Generate the fingerprint of the public key."""
if backend == 'cryptography':
if backend == "cryptography":
publickey = privatekey.public_key().public_bytes(
serialization.Encoding.DER,
serialization.PublicFormat.SubjectPublicKeyInfo
serialization.Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo
)
return get_fingerprint_of_bytes(publickey, prefer_one=prefer_one)
def get_fingerprint(path, passphrase=None, content=None, backend='cryptography', prefer_one=False):
"""Generate the fingerprint of the public key. """
def get_fingerprint(
path, passphrase=None, content=None, backend="cryptography", prefer_one=False
):
"""Generate the fingerprint of the public key."""
privatekey = load_privatekey(path, passphrase=passphrase, content=content, check_passphrase=False, backend=backend)
privatekey = load_privatekey(
path,
passphrase=passphrase,
content=content,
check_passphrase=False,
backend=backend,
)
return get_fingerprint_of_privatekey(privatekey, backend=backend, prefer_one=prefer_one)
return get_fingerprint_of_privatekey(
privatekey, backend=backend, prefer_one=prefer_one
)
def load_privatekey(path, passphrase=None, check_passphrase=True, content=None, backend='cryptography'):
def load_privatekey(
path, passphrase=None, check_passphrase=True, content=None, backend="cryptography"
):
"""Load the specified OpenSSL private key.
The content can also be specified via content; in that case,
@@ -124,58 +153,72 @@ def load_privatekey(path, passphrase=None, check_passphrase=True, content=None,
try:
if content is None:
with open(path, 'rb') as b_priv_key_fh:
with open(path, "rb") as b_priv_key_fh:
priv_key_detail = b_priv_key_fh.read()
else:
priv_key_detail = content
except (IOError, OSError) as exc:
raise OpenSSLObjectError(exc)
if backend == 'pyopenssl':
if backend == "pyopenssl":
# First try: try to load with real passphrase (resp. empty string)
# Will work if this is the correct passphrase, or the key is not
# password-protected.
try:
result = crypto.load_privatekey(crypto.FILETYPE_PEM,
priv_key_detail,
to_bytes(passphrase or ''))
result = crypto.load_privatekey(
crypto.FILETYPE_PEM, priv_key_detail, to_bytes(passphrase or "")
)
except crypto.Error as e:
if len(e.args) > 0 and len(e.args[0]) > 0:
if e.args[0][0][2] in ('bad decrypt', 'bad password read'):
if e.args[0][0][2] in ("bad decrypt", "bad password read"):
# This happens in case we have the wrong passphrase.
if passphrase is not None:
raise OpenSSLBadPassphraseError('Wrong passphrase provided for private key!')
raise OpenSSLBadPassphraseError(
"Wrong passphrase provided for private key!"
)
else:
raise OpenSSLBadPassphraseError('No passphrase provided, but private key is password-protected!')
raise OpenSSLObjectError('Error while deserializing key: {0}'.format(e))
raise OpenSSLBadPassphraseError(
"No passphrase provided, but private key is password-protected!"
)
raise OpenSSLObjectError("Error while deserializing key: {0}".format(e))
if check_passphrase:
# Next we want to make sure that the key is actually protected by
# a passphrase (in case we did try the empty string before, make
# sure that the key is not protected by the empty string)
try:
crypto.load_privatekey(crypto.FILETYPE_PEM,
priv_key_detail,
to_bytes('y' if passphrase == 'x' else 'x'))
crypto.load_privatekey(
crypto.FILETYPE_PEM,
priv_key_detail,
to_bytes("y" if passphrase == "x" else "x"),
)
if passphrase is not None:
# Since we can load the key without an exception, the
# key is not password-protected
raise OpenSSLBadPassphraseError('Passphrase provided, but private key is not password-protected!')
raise OpenSSLBadPassphraseError(
"Passphrase provided, but private key is not password-protected!"
)
except crypto.Error as e:
if passphrase is None and len(e.args) > 0 and len(e.args[0]) > 0:
if e.args[0][0][2] in ('bad decrypt', 'bad password read'):
if e.args[0][0][2] in ("bad decrypt", "bad password read"):
# The key is obviously protected by the empty string.
# Do not do this at home (if it is possible at all)...
raise OpenSSLBadPassphraseError('No passphrase provided, but private key is password-protected!')
elif backend == 'cryptography':
raise OpenSSLBadPassphraseError(
"No passphrase provided, but private key is password-protected!"
)
elif backend == "cryptography":
try:
result = load_pem_private_key(priv_key_detail,
None if passphrase is None else to_bytes(passphrase),
cryptography_backend())
result = load_pem_private_key(
priv_key_detail,
None if passphrase is None else to_bytes(passphrase),
cryptography_backend(),
)
except TypeError:
raise OpenSSLBadPassphraseError('Wrong or empty passphrase provided for private key')
raise OpenSSLBadPassphraseError(
"Wrong or empty passphrase provided for private key"
)
except ValueError:
raise OpenSSLBadPassphraseError('Wrong passphrase provided for private key')
raise OpenSSLBadPassphraseError("Wrong passphrase provided for private key")
return result
@@ -183,60 +226,72 @@ def load_privatekey(path, passphrase=None, check_passphrase=True, content=None,
def load_publickey(path=None, content=None, backend=None):
if content is None:
if path is None:
raise OpenSSLObjectError('Must provide either path or content')
raise OpenSSLObjectError("Must provide either path or content")
try:
with open(path, 'rb') as b_priv_key_fh:
with open(path, "rb") as b_priv_key_fh:
content = b_priv_key_fh.read()
except (IOError, OSError) as exc:
raise OpenSSLObjectError(exc)
if backend == 'cryptography':
if backend == "cryptography":
try:
return serialization.load_pem_public_key(content, backend=cryptography_backend())
return serialization.load_pem_public_key(
content, backend=cryptography_backend()
)
except Exception as e:
raise OpenSSLObjectError('Error while deserializing key: {0}'.format(e))
raise OpenSSLObjectError("Error while deserializing key: {0}".format(e))
def load_certificate(path, content=None, backend='cryptography', der_support_enabled=False):
def load_certificate(
path, content=None, backend="cryptography", der_support_enabled=False
):
"""Load the specified certificate."""
try:
if content is None:
with open(path, 'rb') as cert_fh:
with open(path, "rb") as cert_fh:
cert_content = cert_fh.read()
else:
cert_content = content
except (IOError, OSError) as exc:
raise OpenSSLObjectError(exc)
if backend == 'pyopenssl':
if backend == "pyopenssl":
if der_support_enabled is False or identify_pem_format(cert_content):
return crypto.load_certificate(crypto.FILETYPE_PEM, cert_content)
elif der_support_enabled:
raise OpenSSLObjectError('Certificate in DER format is not supported by the pyopenssl backend.')
elif backend == 'cryptography':
raise OpenSSLObjectError(
"Certificate in DER format is not supported by the pyopenssl backend."
)
elif backend == "cryptography":
if der_support_enabled is False or identify_pem_format(cert_content):
try:
return x509.load_pem_x509_certificate(cert_content, cryptography_backend())
return x509.load_pem_x509_certificate(
cert_content, cryptography_backend()
)
except ValueError as exc:
raise OpenSSLObjectError(exc)
elif der_support_enabled:
try:
return x509.load_der_x509_certificate(cert_content, cryptography_backend())
return x509.load_der_x509_certificate(
cert_content, cryptography_backend()
)
except ValueError as exc:
raise OpenSSLObjectError('Cannot parse DER certificate: {0}'.format(exc))
raise OpenSSLObjectError(
"Cannot parse DER certificate: {0}".format(exc)
)
def load_certificate_request(path, content=None, backend='cryptography'):
def load_certificate_request(path, content=None, backend="cryptography"):
"""Load the specified certificate signing request."""
try:
if content is None:
with open(path, 'rb') as csr_fh:
with open(path, "rb") as csr_fh:
csr_content = csr_fh.read()
else:
csr_content = content
except (IOError, OSError) as exc:
raise OpenSSLObjectError(exc)
if backend == 'cryptography':
if backend == "cryptography":
try:
return x509.load_pem_x509_csr(csr_content, cryptography_backend())
except ValueError as exc:
@@ -245,23 +300,40 @@ def load_certificate_request(path, content=None, backend='cryptography'):
def parse_name_field(input_dict, name_field_name=None):
"""Take a dict with key: value or key: list_of_values mappings and return a list of tuples"""
error_str = '{key}' if name_field_name is None else '{key} in {name}'
error_str = "{key}" if name_field_name is None else "{key} in {name}"
result = []
for key, value in input_dict.items():
if isinstance(value, list):
for entry in value:
if not isinstance(entry, six.string_types):
raise TypeError(('Values %s must be strings' % error_str).format(key=key, name=name_field_name))
raise TypeError(
("Values %s must be strings" % error_str).format(
key=key, name=name_field_name
)
)
if not entry:
raise ValueError(('Values for %s must not be empty strings' % error_str).format(key=key))
raise ValueError(
("Values for %s must not be empty strings" % error_str).format(
key=key
)
)
result.append((key, entry))
elif isinstance(value, six.string_types):
if not value:
raise ValueError(('Value for %s must not be an empty string' % error_str).format(key=key))
raise ValueError(
("Value for %s must not be an empty string" % error_str).format(
key=key
)
)
result.append((key, value))
else:
raise TypeError(('Value for %s must be either a string or a list of strings' % error_str).format(key=key))
raise TypeError(
(
"Value for %s must be either a string or a list of strings"
% error_str
).format(key=key)
)
return result
@@ -272,28 +344,32 @@ def parse_ordered_name_field(input_list, name_field_name):
for index, entry in enumerate(input_list):
if len(entry) != 1:
raise ValueError(
'Entry #{index} in {name} must be a dictionary with exactly one key-value pair'.format(
name=name_field_name, index=index + 1))
"Entry #{index} in {name} must be a dictionary with exactly one key-value pair".format(
name=name_field_name, index=index + 1
)
)
try:
result.extend(parse_name_field(entry, name_field_name=name_field_name))
except (TypeError, ValueError) as exc:
raise ValueError(
'Error while processing entry #{index} in {name}: {error}'.format(
name=name_field_name, index=index + 1, error=exc))
"Error while processing entry #{index} in {name}: {error}".format(
name=name_field_name, index=index + 1, error=exc
)
)
return result
def select_message_digest(digest_string):
digest = None
if digest_string == 'sha256':
if digest_string == "sha256":
digest = hashes.SHA256()
elif digest_string == 'sha384':
elif digest_string == "sha384":
digest = hashes.SHA384()
elif digest_string == 'sha512':
elif digest_string == "sha512":
digest = hashes.SHA512()
elif digest_string == 'sha1':
elif digest_string == "sha1":
digest = hashes.SHA1()
elif digest_string == 'md5':
elif digest_string == "md5":
digest = hashes.MD5()
return digest
@@ -317,7 +393,7 @@ class OpenSSLObject(object):
def _check_perms(module):
file_args = module.load_file_common_arguments(module.params)
if module.check_file_absent_if_check_mode(file_args['path']):
if module.check_file_absent_if_check_mode(file_args["path"]):
return False
return not module.set_fs_attributes_if_different(file_args, False)