mirror of
https://github.com/ansible-collections/community.crypto.git
synced 2026-05-07 13:53:06 +00:00
openssh_keypair - Add diff support and general cleanup (#260)
* Initial commit * Matching tests to overwritten permissions behavior with cryptography * Ensuring key validation only occurs when state=present and accomodating CentOS6 restrictions * Making ssh-keygen behavior explicit by version in tests * Ensuring cyrptography not excluded in new conditions * Adding changelog fragment * Fixing sanity checks * Improving readability * Applying review suggestions * addressing restore_on_failure conflict
This commit is contained in:
Ajpantuso
GitHub
@@ -20,16 +20,13 @@ from __future__ import absolute_import, division, print_function
|
||||
__metaclass__ = type
|
||||
|
||||
import abc
|
||||
import errno
|
||||
import os
|
||||
import stat
|
||||
from distutils.version import LooseVersion
|
||||
|
||||
from ansible.module_utils import six
|
||||
from ansible.module_utils.basic import missing_required_lib
|
||||
from ansible.module_utils.common.text.converters import to_native, to_text, to_bytes
|
||||
|
||||
from ansible_collections.community.crypto.plugins.module_utils.openssh.utils import parse_openssh_version
|
||||
from ansible_collections.community.crypto.plugins.module_utils.openssh.cryptography import (
|
||||
HAS_OPENSSH_SUPPORT,
|
||||
HAS_OPENSSH_PRIVATE_FORMAT,
|
||||
@@ -39,322 +36,334 @@ from ansible_collections.community.crypto.plugins.module_utils.openssh.cryptogra
|
||||
OpenSSHError,
|
||||
OpensshKeypair,
|
||||
)
|
||||
from ansible_collections.community.crypto.plugins.module_utils.openssh.backends.common import (
|
||||
KeygenCommand,
|
||||
OpensshModule,
|
||||
PrivateKey,
|
||||
PublicKey,
|
||||
)
|
||||
from ansible_collections.community.crypto.plugins.module_utils.openssh.utils import (
|
||||
any_in,
|
||||
file_mode,
|
||||
secure_write,
|
||||
)
|
||||
|
||||
|
||||
@six.add_metaclass(abc.ABCMeta)
|
||||
class KeypairBackend(object):
|
||||
class KeypairBackend(OpensshModule):
|
||||
|
||||
def __init__(self, module):
|
||||
self.module = module
|
||||
super(KeypairBackend, self).__init__(module)
|
||||
|
||||
self.path = module.params['path']
|
||||
self.force = module.params['force']
|
||||
self.size = module.params['size']
|
||||
self.type = module.params['type']
|
||||
self.comment = module.params['comment']
|
||||
self.passphrase = module.params['passphrase']
|
||||
self.regenerate = module.params['regenerate']
|
||||
self.comment = self.module.params['comment']
|
||||
self.private_key_path = self.module.params['path']
|
||||
self.public_key_path = self.private_key_path + '.pub'
|
||||
self.regenerate = self.module.params['regenerate'] if not self.module.params['force'] else 'always'
|
||||
self.state = self.module.params['state']
|
||||
self.type = self.module.params['type']
|
||||
|
||||
self.changed = False
|
||||
self.fingerprint = ''
|
||||
self.public_key = {}
|
||||
self.size = self._get_size(self.module.params['size'])
|
||||
self._validate_path()
|
||||
|
||||
if self.regenerate == 'always':
|
||||
self.force = True
|
||||
self.original_private_key = None
|
||||
self.original_public_key = None
|
||||
self.private_key = None
|
||||
self.public_key = None
|
||||
|
||||
def _get_size(self, size):
|
||||
if self.type in ('rsa', 'rsa1'):
|
||||
self.size = 4096 if self.size is None else self.size
|
||||
if self.size < 1024:
|
||||
module.fail_json(msg=('For RSA keys, the minimum size is 1024 bits and the default is 4096 bits. '
|
||||
'Attempting to use bit lengths under 1024 will cause the module to fail.'))
|
||||
result = 4096 if size is None else size
|
||||
if result < 1024:
|
||||
return self.module.fail_json(
|
||||
msg="For RSA keys, the minimum size is 1024 bits and the default is 4096 bits. " +
|
||||
"Attempting to use bit lengths under 1024 will cause the module to fail."
|
||||
)
|
||||
elif self.type == 'dsa':
|
||||
self.size = 1024 if self.size is None else self.size
|
||||
if self.size != 1024:
|
||||
module.fail_json(msg=('DSA keys must be exactly 1024 bits as specified by FIPS 186-2.'))
|
||||
result = 1024 if size is None else size
|
||||
if result != 1024:
|
||||
return self.module.fail_json(msg="DSA keys must be exactly 1024 bits as specified by FIPS 186-2.")
|
||||
elif self.type == 'ecdsa':
|
||||
self.size = 256 if self.size is None else self.size
|
||||
if self.size not in (256, 384, 521):
|
||||
module.fail_json(msg=('For ECDSA keys, size determines the key length by selecting from '
|
||||
'one of three elliptic curve sizes: 256, 384 or 521 bits. '
|
||||
'Attempting to use bit lengths other than these three values for '
|
||||
'ECDSA keys will cause this module to fail. '))
|
||||
result = 256 if size is None else size
|
||||
if result not in (256, 384, 521):
|
||||
return self.module.fail_json(
|
||||
msg="For ECDSA keys, size determines the key length by selecting from one of " +
|
||||
"three elliptic curve sizes: 256, 384 or 521 bits. " +
|
||||
"Attempting to use bit lengths other than these three values for ECDSA keys will " +
|
||||
"cause this module to fail."
|
||||
)
|
||||
elif self.type == 'ed25519':
|
||||
# User input is ignored for `key size` when `key type` is ed25519
|
||||
self.size = 256
|
||||
result = 256
|
||||
else:
|
||||
module.fail_json(msg="%s is not a valid value for key type" % self.type)
|
||||
return self.module.fail_json(msg="%s is not a valid value for key type" % self.type)
|
||||
|
||||
def generate(self):
|
||||
if self.force or not self.is_private_key_valid(perms_required=False):
|
||||
try:
|
||||
if self.exists() and not os.access(self.path, os.W_OK):
|
||||
os.chmod(self.path, stat.S_IWUSR + stat.S_IRUSR)
|
||||
self._generate_keypair()
|
||||
self.changed = True
|
||||
except (IOError, OSError) as e:
|
||||
self.remove()
|
||||
self.module.fail_json(msg="%s" % to_native(e))
|
||||
return result
|
||||
|
||||
self.fingerprint = self._get_current_key_properties()[2]
|
||||
self.public_key = self._get_public_key()
|
||||
elif not self.is_public_key_valid(perms_required=False):
|
||||
pubkey = self._get_public_key()
|
||||
try:
|
||||
with open(self.path + ".pub", "w") as pubkey_f:
|
||||
pubkey_f.write(pubkey + '\n')
|
||||
os.chmod(self.path + ".pub", stat.S_IWUSR + stat.S_IRUSR + stat.S_IRGRP + stat.S_IROTH)
|
||||
except (IOError, OSError):
|
||||
self.module.fail_json(
|
||||
msg='The public key is missing or does not match the private key. '
|
||||
'Unable to regenerate the public key.')
|
||||
self.changed = True
|
||||
self.public_key = pubkey
|
||||
def _validate_path(self):
|
||||
self._check_if_base_dir(self.private_key_path)
|
||||
|
||||
if self.comment:
|
||||
try:
|
||||
if self.exists() and not os.access(self.path, os.W_OK):
|
||||
os.chmod(self.path, stat.S_IWUSR + stat.S_IRUSR)
|
||||
except (IOError, OSError):
|
||||
self.module.fail_json(msg='Unable to update the comment for the public key.')
|
||||
self._update_comment()
|
||||
if os.path.isdir(self.private_key_path):
|
||||
self.module.fail_json(msg='%s is a directory. Please specify a path to a file.' % self.private_key_path)
|
||||
|
||||
private_key_perms_changed = self._permissions_changed()
|
||||
public_key_perms_changed = self._permissions_changed(public_key=True)
|
||||
if private_key_perms_changed or public_key_perms_changed:
|
||||
self.changed = True
|
||||
def _execute(self):
|
||||
self.original_private_key = self._load_private_key()
|
||||
self.original_public_key = self._load_public_key()
|
||||
|
||||
def is_private_key_valid(self, perms_required=True):
|
||||
if not self.exists():
|
||||
return False
|
||||
if self.state == 'present':
|
||||
self._validate_key_load()
|
||||
|
||||
if self._check_pass_protected_or_broken_key():
|
||||
if self.regenerate in ('full_idempotence', 'always'):
|
||||
return False
|
||||
self.module.fail_json(msg='Unable to read the key. The key is protected with a passphrase or broken.'
|
||||
' Will not proceed. To force regeneration, call the module with `generate`'
|
||||
' set to `full_idempotence` or `always`, or with `force=yes`.')
|
||||
if self._should_generate():
|
||||
self._generate()
|
||||
elif not self._public_key_valid():
|
||||
self._restore_public_key()
|
||||
|
||||
if not self._private_key_loadable():
|
||||
if os.path.isdir(self.path):
|
||||
self.module.fail_json(msg='%s is a directory. Please specify a path to a file.' % self.path)
|
||||
self.private_key = self._load_private_key()
|
||||
self.public_key = self._load_public_key()
|
||||
|
||||
if self.regenerate in ('full_idempotence', 'always'):
|
||||
return False
|
||||
self.module.fail_json(msg='Unable to read the key. The key is protected with a passphrase or broken.'
|
||||
' Will not proceed. To force regeneration, call the module with `generate`'
|
||||
' set to `full_idempotence` or `always`, or with `force=yes`.')
|
||||
|
||||
keysize, keytype, self.fingerprint = self._get_current_key_properties()
|
||||
|
||||
if self.regenerate == 'never':
|
||||
return True
|
||||
|
||||
if not (self.type == keytype and self.size == keysize):
|
||||
if self.regenerate in ('partial_idempotence', 'full_idempotence', 'always'):
|
||||
return False
|
||||
self.module.fail_json(
|
||||
msg='Key has wrong type and/or size.'
|
||||
' Will not proceed. To force regeneration, call the module with `generate`'
|
||||
' set to `partial_idempotence`, `full_idempotence` or `always`, or with `force=yes`.'
|
||||
)
|
||||
|
||||
# Perms required short-circuits evaluation to prevent the side-effects of running _permissions_changed
|
||||
# when check_mode is not enabled
|
||||
return not (perms_required and self._permissions_changed())
|
||||
|
||||
def is_public_key_valid(self, perms_required=True):
|
||||
|
||||
def _get_pubkey_content():
|
||||
if self.exists(public_key=True):
|
||||
with open(self.path + ".pub", "r") as pubkey_f:
|
||||
present_pubkey = pubkey_f.read().strip(' \n')
|
||||
return present_pubkey
|
||||
else:
|
||||
return ''
|
||||
|
||||
def _parse_pubkey(pubkey_content):
|
||||
if pubkey_content:
|
||||
parts = pubkey_content.split(' ', 2)
|
||||
if len(parts) < 2:
|
||||
return ()
|
||||
return parts[0], parts[1], '' if len(parts) <= 2 else parts[2]
|
||||
return ()
|
||||
|
||||
def _pubkey_valid(pubkey):
|
||||
if pubkey_parts and _parse_pubkey(pubkey):
|
||||
return pubkey_parts[:2] == _parse_pubkey(pubkey)[:2]
|
||||
return False
|
||||
|
||||
def _comment_valid():
|
||||
if pubkey_parts:
|
||||
return pubkey_parts[2] == self.comment
|
||||
return False
|
||||
|
||||
pubkey_parts = _parse_pubkey(_get_pubkey_content())
|
||||
|
||||
pubkey = self._get_public_key()
|
||||
if _pubkey_valid(pubkey):
|
||||
self.public_key = pubkey
|
||||
for path in (self.private_key_path, self.public_key_path):
|
||||
self._update_permissions(path)
|
||||
else:
|
||||
return False
|
||||
if self._should_remove():
|
||||
self._remove()
|
||||
|
||||
if self.comment and not _comment_valid():
|
||||
return False
|
||||
|
||||
# Perms required short-circuits evaluation to prevent the side-effects of running _permissions_changes
|
||||
# when check_mode is not enabled
|
||||
return not (perms_required and self._permissions_changed(public_key=True))
|
||||
|
||||
def _permissions_changed(self, public_key=False):
|
||||
file_args = self.module.load_file_common_arguments(self.module.params)
|
||||
if public_key:
|
||||
file_args['path'] = file_args['path'] + '.pub'
|
||||
if self.module.check_file_absent_if_check_mode(file_args['path']):
|
||||
return True
|
||||
return self.module.set_fs_attributes_if_different(file_args, False)
|
||||
|
||||
@property
|
||||
def result(self):
|
||||
return {
|
||||
'changed': self.changed,
|
||||
'size': self.size,
|
||||
'type': self.type,
|
||||
'filename': self.path,
|
||||
'fingerprint': self.fingerprint if self.fingerprint else '',
|
||||
'public_key': self.public_key,
|
||||
'comment': self.comment if self.comment else '',
|
||||
}
|
||||
|
||||
def remove(self):
|
||||
"""Remove the resource from the filesystem."""
|
||||
|
||||
try:
|
||||
os.remove(self.path)
|
||||
self.changed = True
|
||||
except (IOError, OSError) as exc:
|
||||
if exc.errno != errno.ENOENT:
|
||||
self.module.fail_json(msg=to_native(exc))
|
||||
else:
|
||||
def _load_private_key(self):
|
||||
result = None
|
||||
if self._private_key_exists():
|
||||
try:
|
||||
result = self._get_private_key()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if self.exists(public_key=True):
|
||||
return result
|
||||
|
||||
def _private_key_exists(self):
|
||||
return os.path.exists(self.private_key_path)
|
||||
|
||||
@abc.abstractmethod
|
||||
def _get_private_key(self):
|
||||
pass
|
||||
|
||||
def _load_public_key(self):
|
||||
result = None
|
||||
if self._public_key_exists():
|
||||
try:
|
||||
os.remove(self.path + ".pub")
|
||||
self.changed = True
|
||||
except (IOError, OSError) as exc:
|
||||
if exc.errno != errno.ENOENT:
|
||||
self.module.fail_json(msg=to_native(exc))
|
||||
else:
|
||||
pass
|
||||
result = PublicKey.load(self.public_key_path)
|
||||
except (IOError, OSError):
|
||||
pass
|
||||
return result
|
||||
|
||||
def exists(self, public_key=False):
|
||||
return os.path.exists(self.path if not public_key else self.path + ".pub")
|
||||
def _public_key_exists(self):
|
||||
return os.path.exists(self.public_key_path)
|
||||
|
||||
def _validate_key_load(self):
|
||||
if (self._private_key_exists()
|
||||
and self.regenerate in ('never', 'fail', 'partial_idempotence')
|
||||
and (self.original_private_key is None or not self._private_key_readable())):
|
||||
self.module.fail_json(
|
||||
msg="Unable to read the key. The key is protected with a passphrase or broken. " +
|
||||
"Will not proceed. To force regeneration, call the module with `generate` " +
|
||||
"set to `full_idempotence` or `always`, or with `force=yes`."
|
||||
)
|
||||
|
||||
@abc.abstractmethod
|
||||
def _generate_keypair(self):
|
||||
def _private_key_readable(self):
|
||||
pass
|
||||
|
||||
def _should_generate(self):
|
||||
if self.regenerate == 'never':
|
||||
return self.original_private_key is None
|
||||
elif self.regenerate == 'fail':
|
||||
if not self._private_key_valid():
|
||||
self.module.fail_json(
|
||||
msg="Key has wrong type and/or size. Will not proceed. " +
|
||||
"To force regeneration, call the module with `generate` set to " +
|
||||
"`partial_idempotence`, `full_idempotence` or `always`, or with `force=yes`."
|
||||
)
|
||||
return self.original_private_key is None
|
||||
elif self.regenerate in ('partial_idempotence', 'full_idempotence'):
|
||||
return not self._private_key_valid()
|
||||
else:
|
||||
return True
|
||||
|
||||
def _private_key_valid(self):
|
||||
if self.original_private_key is None:
|
||||
return False
|
||||
|
||||
return all([
|
||||
self.size == self.original_private_key.size,
|
||||
self.type == self.original_private_key.type,
|
||||
])
|
||||
|
||||
@OpensshModule.trigger_change
|
||||
@OpensshModule.skip_if_check_mode
|
||||
def _generate(self):
|
||||
temp_private_key, temp_public_key = self._generate_temp_keypair()
|
||||
|
||||
try:
|
||||
self._safe_secure_move([(temp_private_key, self.private_key_path), (temp_public_key, self.public_key_path)])
|
||||
except OSError as e:
|
||||
self.module.fail_json(msg=to_native(e))
|
||||
|
||||
def _generate_temp_keypair(self):
|
||||
temp_private_key = os.path.join(self.module.tmpdir, os.path.basename(self.private_key_path))
|
||||
temp_public_key = temp_private_key + '.pub'
|
||||
|
||||
try:
|
||||
self._generate_keypair(temp_private_key)
|
||||
except (IOError, OSError) as e:
|
||||
self.module.fail_json(msg=to_native(e))
|
||||
|
||||
for f in (temp_private_key, temp_public_key):
|
||||
self.module.add_cleanup_file(f)
|
||||
|
||||
return temp_private_key, temp_public_key
|
||||
|
||||
@abc.abstractmethod
|
||||
def _get_current_key_properties(self):
|
||||
def _generate_keypair(self, private_key_path):
|
||||
pass
|
||||
|
||||
def _public_key_valid(self):
|
||||
if self.original_public_key is None:
|
||||
return False
|
||||
|
||||
valid_public_key = self._get_public_key()
|
||||
valid_public_key.comment = self.comment
|
||||
|
||||
return self.original_public_key == valid_public_key
|
||||
|
||||
@abc.abstractmethod
|
||||
def _get_public_key(self):
|
||||
pass
|
||||
|
||||
@OpensshModule.trigger_change
|
||||
@OpensshModule.skip_if_check_mode
|
||||
def _restore_public_key(self):
|
||||
try:
|
||||
temp_public_key = self._create_temp_public_key(str(self._get_public_key()) + '\n')
|
||||
self._safe_secure_move([
|
||||
(temp_public_key, self.public_key_path)
|
||||
])
|
||||
except (IOError, OSError):
|
||||
self.module.fail_json(
|
||||
msg="The public key is missing or does not match the private key. " +
|
||||
"Unable to regenerate the public key."
|
||||
)
|
||||
|
||||
if self.comment:
|
||||
self._update_comment()
|
||||
|
||||
def _create_temp_public_key(self, content):
|
||||
temp_public_key = os.path.join(self.module.tmpdir, os.path.basename(self.public_key_path))
|
||||
|
||||
default_permissions = 0o644
|
||||
existing_permissions = file_mode(self.public_key_path)
|
||||
|
||||
try:
|
||||
secure_write(temp_public_key, existing_permissions or default_permissions, to_bytes(content))
|
||||
except (IOError, OSError) as e:
|
||||
self.module.fail_json(msg=to_native(e))
|
||||
self.module.add_cleanup_file(temp_public_key)
|
||||
|
||||
return temp_public_key
|
||||
|
||||
@abc.abstractmethod
|
||||
def _update_comment(self):
|
||||
pass
|
||||
|
||||
@abc.abstractmethod
|
||||
def _private_key_loadable(self):
|
||||
pass
|
||||
def _should_remove(self):
|
||||
return self._private_key_exists() or self._public_key_exists()
|
||||
|
||||
@abc.abstractmethod
|
||||
def _check_pass_protected_or_broken_key(self):
|
||||
pass
|
||||
@OpensshModule.trigger_change
|
||||
@OpensshModule.skip_if_check_mode
|
||||
def _remove(self):
|
||||
try:
|
||||
if self._private_key_exists():
|
||||
os.remove(self.private_key_path)
|
||||
if self._public_key_exists():
|
||||
os.remove(self.public_key_path)
|
||||
except (IOError, OSError) as e:
|
||||
self.module.fail_json(msg=to_native(e))
|
||||
|
||||
@property
|
||||
def _result(self):
|
||||
private_key = self.private_key or self.original_private_key
|
||||
public_key = self.public_key or self.original_public_key
|
||||
|
||||
return {
|
||||
'size': self.size,
|
||||
'type': self.type,
|
||||
'filename': self.private_key_path,
|
||||
'fingerprint': private_key.fingerprint if private_key else '',
|
||||
'public_key': str(public_key) if public_key else '',
|
||||
'comment': public_key.comment if public_key else '',
|
||||
}
|
||||
|
||||
@property
|
||||
def diff(self):
|
||||
before = self.original_private_key.to_dict() if self.original_private_key else {}
|
||||
before.update(self.original_public_key.to_dict() if self.original_public_key else {})
|
||||
|
||||
after = self.private_key.to_dict() if self.private_key else {}
|
||||
after.update(self.public_key.to_dict() if self.public_key else {})
|
||||
|
||||
return {
|
||||
'before': before,
|
||||
'after': after,
|
||||
}
|
||||
|
||||
|
||||
class KeypairBackendOpensshBin(KeypairBackend):
|
||||
|
||||
def __init__(self, module):
|
||||
super(KeypairBackendOpensshBin, self).__init__(module)
|
||||
|
||||
self.openssh_bin = module.get_bin_path('ssh-keygen')
|
||||
self.ssh_keygen = KeygenCommand(self.module)
|
||||
|
||||
def _load_privatekey(self):
|
||||
return self.module.run_command([self.openssh_bin, '-lf', self.path])
|
||||
def _generate_keypair(self, private_key_path):
|
||||
self.ssh_keygen.generate_keypair(private_key_path, self.size, self.type, self.comment)
|
||||
|
||||
def _get_publickey_from_privatekey(self):
|
||||
# -P '' is always included as an option to induce the expected standard output for
|
||||
# _check_pass_protected_or_broken_key, but introduces no side-effects when used to
|
||||
# output a matching public key
|
||||
return self.module.run_command([self.openssh_bin, '-P', '', '-yf', self.path])
|
||||
|
||||
def _generate_keypair(self):
|
||||
args = [
|
||||
self.openssh_bin,
|
||||
'-q',
|
||||
'-N', '',
|
||||
'-b', str(self.size),
|
||||
'-t', self.type,
|
||||
'-f', self.path,
|
||||
'-C', self.comment if self.comment else ''
|
||||
]
|
||||
|
||||
# "y" must be entered in response to the "overwrite" prompt
|
||||
stdin_data = 'y' if self.exists() else None
|
||||
|
||||
self.module.run_command(args, data=stdin_data)
|
||||
|
||||
def _get_current_key_properties(self):
|
||||
rc, stdout, stderr = self._load_privatekey()
|
||||
properties = stdout.split()
|
||||
keysize = int(properties[0])
|
||||
fingerprint = properties[1]
|
||||
keytype = properties[-1][1:-1].lower()
|
||||
|
||||
return keysize, keytype, fingerprint
|
||||
def _get_private_key(self):
|
||||
private_key_content = self.ssh_keygen.get_private_key(self.private_key_path)[1]
|
||||
return PrivateKey.from_string(private_key_content)
|
||||
|
||||
def _get_public_key(self):
|
||||
rc, stdout, stderr = self._get_publickey_from_privatekey()
|
||||
return stdout.strip('\n')
|
||||
public_key_content = self.ssh_keygen.get_matching_public_key(self.private_key_path)[1]
|
||||
return PublicKey.from_string(public_key_content)
|
||||
|
||||
def _private_key_readable(self):
|
||||
rc, stdout, stderr = self.ssh_keygen.get_matching_public_key(self.private_key_path)
|
||||
return not (rc == 255 or any_in(stderr, 'is not a public key file', 'incorrect passphrase', 'load failed'))
|
||||
|
||||
def _update_comment(self):
|
||||
return self.module.run_command([self.openssh_bin, '-q', '-o', '-c', '-C', self.comment, '-f', self.path])
|
||||
|
||||
def _private_key_loadable(self):
|
||||
rc, stdout, stderr = self._load_privatekey()
|
||||
return rc == 0
|
||||
|
||||
def _check_pass_protected_or_broken_key(self):
|
||||
rc, stdout, stderr = self._get_publickey_from_privatekey()
|
||||
return rc == 255 or any_in(stderr, 'is not a public key file', 'incorrect passphrase', 'load failed')
|
||||
try:
|
||||
self.ssh_keygen.update_comment(self.private_key_path, self.comment)
|
||||
except (IOError, OSError) as e:
|
||||
self.module.fail_json(msg=to_native(e))
|
||||
|
||||
|
||||
class KeypairBackendCryptography(KeypairBackend):
|
||||
|
||||
def __init__(self, module):
|
||||
super(KeypairBackendCryptography, self).__init__(module)
|
||||
|
||||
if module.params['private_key_format'] == 'auto':
|
||||
ssh = module.get_bin_path('ssh')
|
||||
if ssh:
|
||||
proc = module.run_command([ssh, '-Vq'])
|
||||
ssh_version = parse_openssh_version(proc[2].strip())
|
||||
else:
|
||||
# Default to OpenSSH 7.8 compatibility when OpenSSH is not installed
|
||||
ssh_version = "7.8"
|
||||
if self.type == 'rsa1':
|
||||
self.module.fail_json(msg="RSA1 keys are not supported by the cryptography backend")
|
||||
|
||||
self.private_key_format = 'SSH'
|
||||
self.passphrase = to_bytes(module.params['passphrase']) if module.params['passphrase'] else None
|
||||
self.private_key_format = self._get_key_format(module.params['private_key_format'])
|
||||
|
||||
def _get_key_format(self, key_format):
|
||||
result = 'SSH'
|
||||
|
||||
if key_format == 'auto':
|
||||
# Default to OpenSSH 7.8 compatibility when OpenSSH is not installed
|
||||
ssh_version = self._get_ssh_version() or "7.8"
|
||||
|
||||
if LooseVersion(ssh_version) < LooseVersion("7.8") and self.type != 'ed25519':
|
||||
# OpenSSH made SSH formatted private keys available in version 6.5,
|
||||
# but still defaulted to PKCS1 format with the exception of ed25519 keys
|
||||
self.private_key_format = 'PKCS1'
|
||||
result = 'PKCS1'
|
||||
|
||||
if self.private_key_format == 'SSH' and not HAS_OPENSSH_PRIVATE_FORMAT:
|
||||
module.fail_json(
|
||||
if result == 'SSH' and not HAS_OPENSSH_PRIVATE_FORMAT:
|
||||
self.module.fail_json(
|
||||
msg=missing_required_lib(
|
||||
'cryptography >= 3.0',
|
||||
reason="to load/dump private keys in the default OpenSSH format for OpenSSH >= 7.8 " +
|
||||
@@ -362,96 +371,72 @@ class KeypairBackendCryptography(KeypairBackend):
|
||||
)
|
||||
)
|
||||
|
||||
if self.type == 'rsa1':
|
||||
module.fail_json(msg="RSA1 keys are not supported by the cryptography backend")
|
||||
return result
|
||||
|
||||
self.passphrase = to_bytes(self.passphrase) if self.passphrase else None
|
||||
|
||||
def _load_privatekey(self):
|
||||
return OpensshKeypair.load(path=self.path, passphrase=self.passphrase, no_public_key=True)
|
||||
|
||||
def _generate_keypair(self):
|
||||
def _generate_keypair(self, private_key_path):
|
||||
keypair = OpensshKeypair.generate(
|
||||
keytype=self.type,
|
||||
size=self.size,
|
||||
passphrase=self.passphrase,
|
||||
comment=self.comment if self.comment else "",
|
||||
comment=self.comment or '',
|
||||
)
|
||||
with open(self.path, 'w+b') as f:
|
||||
f.write(
|
||||
OpensshKeypair.encode_openssh_privatekey(
|
||||
keypair.asymmetric_keypair,
|
||||
self.private_key_format
|
||||
)
|
||||
)
|
||||
# ssh-keygen defaults private key permissions to 0600 octal
|
||||
os.chmod(self.path, stat.S_IWUSR + stat.S_IRUSR)
|
||||
with open(self.path + '.pub', 'w+b') as f:
|
||||
f.write(keypair.public_key)
|
||||
# ssh-keygen defaults public key permissions to 0644 octal
|
||||
os.chmod(self.path + ".pub", stat.S_IWUSR + stat.S_IRUSR + stat.S_IRGRP + stat.S_IROTH)
|
||||
|
||||
def _get_current_key_properties(self):
|
||||
keypair = self._load_privatekey()
|
||||
encoded_private_key = OpensshKeypair.encode_openssh_privatekey(
|
||||
keypair.asymmetric_keypair, self.private_key_format
|
||||
)
|
||||
secure_write(private_key_path, 0o600, encoded_private_key)
|
||||
|
||||
return keypair.size, keypair.key_type, keypair.fingerprint
|
||||
public_key_path = private_key_path + '.pub'
|
||||
secure_write(public_key_path, 0o644, keypair.public_key)
|
||||
|
||||
def _get_private_key(self):
|
||||
keypair = OpensshKeypair.load(path=self.private_key_path, passphrase=self.passphrase, no_public_key=True)
|
||||
|
||||
return PrivateKey(
|
||||
size=keypair.size,
|
||||
key_type=keypair.key_type,
|
||||
fingerprint=keypair.fingerprint,
|
||||
)
|
||||
|
||||
def _get_public_key(self):
|
||||
try:
|
||||
keypair = self._load_privatekey()
|
||||
keypair = OpensshKeypair.load(path=self.private_key_path, passphrase=self.passphrase, no_public_key=True)
|
||||
except OpenSSHError:
|
||||
# Simulates the null output of ssh-keygen
|
||||
return ""
|
||||
|
||||
return to_text(keypair.public_key)
|
||||
return PublicKey.from_string(to_text(keypair.public_key))
|
||||
|
||||
def _update_comment(self):
|
||||
keypair = self._load_privatekey()
|
||||
def _private_key_readable(self):
|
||||
try:
|
||||
keypair.comment = self.comment
|
||||
with open(self.path + ".pub", "w+b") as pubkey_file:
|
||||
pubkey_file.write(keypair.public_key + b'\n')
|
||||
except (InvalidCommentError, IOError, OSError) as e:
|
||||
# Return values while unused currently are made to simulate the output of run_command()
|
||||
return 1, "Comment could not be updated", to_native(e)
|
||||
return 0, "Comment updated successfully", ""
|
||||
|
||||
def _private_key_loadable(self):
|
||||
try:
|
||||
self._load_privatekey()
|
||||
except OpenSSHError:
|
||||
return False
|
||||
return True
|
||||
|
||||
def _check_pass_protected_or_broken_key(self):
|
||||
try:
|
||||
OpensshKeypair.load(
|
||||
path=self.path,
|
||||
passphrase=self.passphrase,
|
||||
no_public_key=True,
|
||||
)
|
||||
OpensshKeypair.load(path=self.private_key_path, passphrase=self.passphrase, no_public_key=True)
|
||||
except (InvalidPrivateKeyFileError, InvalidPassphraseError):
|
||||
return True
|
||||
return False
|
||||
|
||||
# Cryptography >= 3.0 uses a SSH key loader which does not raise an exception when a passphrase is provided
|
||||
# when loading an unencrypted key
|
||||
if self.passphrase:
|
||||
try:
|
||||
OpensshKeypair.load(
|
||||
path=self.path,
|
||||
passphrase=None,
|
||||
no_public_key=True,
|
||||
)
|
||||
OpensshKeypair.load(path=self.private_key_path, passphrase=None, no_public_key=True)
|
||||
except (InvalidPrivateKeyFileError, InvalidPassphraseError):
|
||||
return False
|
||||
else:
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
||||
return False
|
||||
return True
|
||||
|
||||
def _update_comment(self):
|
||||
keypair = OpensshKeypair.load(path=self.private_key_path, passphrase=self.passphrase, no_public_key=True)
|
||||
try:
|
||||
keypair.comment = self.comment
|
||||
except InvalidCommentError as e:
|
||||
self.module.fail_json(msg=to_native(e))
|
||||
|
||||
def any_in(sequence, *elements):
|
||||
return any(e in sequence for e in elements)
|
||||
try:
|
||||
temp_public_key = self._create_temp_public_key(keypair.public_key + b'\n')
|
||||
self._safe_secure_move([(temp_public_key, self.public_key_path)])
|
||||
except (IOError, OSError) as e:
|
||||
self.module.fail_json(msg=to_native(e))
|
||||
|
||||
|
||||
def select_backend(module, backend):
|
||||
|
||||
Reference in New Issue
Block a user