a-c-m
/
community.crypto
mirror of https://github.com/ansible-collections/community.crypto.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Update openssl_signature module (#63) 

* Use module_utils from collection, clean up code a bit

 * add DSA keys, because why not...

 * sign/verify was added in pyOpenSSL 0.11 apparently

 * Add signing capability detection to module_utils.crypto.basic

 * Rework feature detection of signature types.

 * Rename parameters to match other modules

 * Add initial version of integration tests

 * fix whitespace in tests

 * More whitespace fixes

 * small fixes for issues in testing

 * Organize integration tests as test matrix

 * another indentation fix to make pep8 happy

 * use openssl pkeyutl when possible, otherwise fall back to openssl dgst

 * More linter fixes

 * openssl pkeyutl -help can apparently return 1

 * ignore errors on openssl call and another try at formatting

 * Remove the OpenSSL calls in tests

 * Add collection name to deprecation notice and deprecate at version 2.0.0

 * Exclude Ed448/25519 tests on pyopenssl

 * revert the collection name in the deprecation notice (breaks 2.9)

 * limit test platforms even more

 * disable FreeBSD DSA and ECC tests

 * Add module name to README

 * rewrite and split into 2 modules instead

 * add module to README and fix whitespace issue

 * remove duplicated tests

 * address review remarks

 * resolve another comment
pull/106/head
Markus Teufelberger 2020-06-06 11:20:00 +02:00 committed by Felix Fontein
parent 128991c3dc
commit 346c2f55ff
8 changed files with 705 additions and 166 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
README.md
View File

@ -24,6 +24,8 @@ Most modules require a recent enough version of [the Python cryptography library
- openssl_privatekey_info
- openssl_privatekey
- openssl_publickey
- openssl_signature_info
- openssl_signature
- x509_certificate_info
- x509_certificate
- x509_crl_info

92
plugins/module_utils/crypto/basic.py
View File

@ -65,11 +65,78 @@ try:
x509.RFC822Name.__hash__ = simple_hash
x509.UniformResourceIdentifier.__hash__ = simple_hash
# Test whether we have support for X25519, X448, Ed25519 and/or Ed448
# Test whether we have support for DSA, EC, Ed25519, Ed448, RSA, X25519 and/or X448
try:
# added in 0.5 - https://cryptography.io/en/latest/hazmat/primitives/asymmetric/dsa/
import cryptography.hazmat.primitives.asymmetric.dsa
CRYPTOGRAPHY_HAS_DSA = True
try:
# added later in 1.5
cryptography.hazmat.primitives.asymmetric.dsa.DSAPrivateKey.sign
CRYPTOGRAPHY_HAS_DSA_SIGN = True
except AttributeError:
CRYPTOGRAPHY_HAS_DSA_SIGN = False
except ImportError:
CRYPTOGRAPHY_HAS_DSA = False
CRYPTOGRAPHY_HAS_DSA_SIGN = False
try:
# added in 2.6 - https://cryptography.io/en/latest/hazmat/primitives/asymmetric/ed25519/
import cryptography.hazmat.primitives.asymmetric.ed25519
CRYPTOGRAPHY_HAS_ED25519 = True
try:
# added with the primitive in 2.6
cryptography.hazmat.primitives.asymmetric.ed25519.Ed25519PrivateKey.sign
CRYPTOGRAPHY_HAS_ED25519_SIGN = True
except AttributeError:
CRYPTOGRAPHY_HAS_ED25519_SIGN = False
except ImportError:
CRYPTOGRAPHY_HAS_ED25519 = False
CRYPTOGRAPHY_HAS_ED25519_SIGN = False
try:
# added in 2.6 - https://cryptography.io/en/latest/hazmat/primitives/asymmetric/ed448/
import cryptography.hazmat.primitives.asymmetric.ed448
CRYPTOGRAPHY_HAS_ED448 = True
try:
# added with the primitive in 2.6
cryptography.hazmat.primitives.asymmetric.ed448.Ed448PrivateKey.sign
CRYPTOGRAPHY_HAS_ED448_SIGN = True
except AttributeError:
CRYPTOGRAPHY_HAS_ED448_SIGN = False
except ImportError:
CRYPTOGRAPHY_HAS_ED448 = False
CRYPTOGRAPHY_HAS_ED448_SIGN = False
try:
# added in 0.5 - https://cryptography.io/en/latest/hazmat/primitives/asymmetric/ec/
import cryptography.hazmat.primitives.asymmetric.ec
CRYPTOGRAPHY_HAS_EC = True
try:
# added later in 1.5
cryptography.hazmat.primitives.asymmetric.ec.EllipticCurvePrivateKey.sign
CRYPTOGRAPHY_HAS_EC_SIGN = True
except AttributeError:
CRYPTOGRAPHY_HAS_EC_SIGN = False
except ImportError:
CRYPTOGRAPHY_HAS_EC = False
CRYPTOGRAPHY_HAS_EC_SIGN = False
try:
# added in 0.5 - https://cryptography.io/en/latest/hazmat/primitives/asymmetric/rsa/
import cryptography.hazmat.primitives.asymmetric.rsa
CRYPTOGRAPHY_HAS_RSA = True
try:
# added later in 1.4
cryptography.hazmat.primitives.asymmetric.rsa.RSAPrivateKey.sign
CRYPTOGRAPHY_HAS_RSA_SIGN = True
except AttributeError:
CRYPTOGRAPHY_HAS_RSA_SIGN = False
except ImportError:
CRYPTOGRAPHY_HAS_RSA = False
CRYPTOGRAPHY_HAS_RSA_SIGN = False
try:
# added in 2.0 - https://cryptography.io/en/latest/hazmat/primitives/asymmetric/x25519/
import cryptography.hazmat.primitives.asymmetric.x25519
CRYPTOGRAPHY_HAS_X25519 = True
try:
# added later in 2.5
cryptography.hazmat.primitives.asymmetric.x25519.X25519PrivateKey.private_bytes
CRYPTOGRAPHY_HAS_X25519_FULL = True
except AttributeError:
@ -78,29 +145,28 @@ try:
CRYPTOGRAPHY_HAS_X25519 = False
CRYPTOGRAPHY_HAS_X25519_FULL = False
try:
# added in 2.5 - https://cryptography.io/en/latest/hazmat/primitives/asymmetric/x448/
import cryptography.hazmat.primitives.asymmetric.x448
CRYPTOGRAPHY_HAS_X448 = True
except ImportError:
CRYPTOGRAPHY_HAS_X448 = False
try:
import cryptography.hazmat.primitives.asymmetric.ed25519
CRYPTOGRAPHY_HAS_ED25519 = True
except ImportError:
CRYPTOGRAPHY_HAS_ED25519 = False
try:
import cryptography.hazmat.primitives.asymmetric.ed448
CRYPTOGRAPHY_HAS_ED448 = True
except ImportError:
CRYPTOGRAPHY_HAS_ED448 = False
HAS_CRYPTOGRAPHY = True
except ImportError:
# Error handled in the calling module.
CRYPTOGRAPHY_HAS_EC = False
CRYPTOGRAPHY_HAS_EC_SIGN = False
CRYPTOGRAPHY_HAS_ED25519 = False
CRYPTOGRAPHY_HAS_ED25519_SIGN = False
CRYPTOGRAPHY_HAS_ED448 = False
CRYPTOGRAPHY_HAS_ED448_SIGN = False
CRYPTOGRAPHY_HAS_DSA = False
CRYPTOGRAPHY_HAS_DSA_SIGN = False
CRYPTOGRAPHY_HAS_RSA = False
CRYPTOGRAPHY_HAS_RSA_SIGN = False
CRYPTOGRAPHY_HAS_X25519 = False
CRYPTOGRAPHY_HAS_X25519_FULL = False
CRYPTOGRAPHY_HAS_X448 = False
CRYPTOGRAPHY_HAS_ED25519 = False
CRYPTOGRAPHY_HAS_ED448 = False
HAS_CRYPTOGRAPHY = False

282
plugins/modules/openssl_signature.py
View File

@ -1,54 +1,52 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
# Copyright: (c) 2016, Yanis Guenane <yanis+ansible@guenane.org>
# Copyright: (c) 2019, Patrick Pichler <ppichler+ansible@mgit.at>
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
ANSIBLE_METADATA = {'metadata_version': '1.1'}
DOCUMENTATION = r'''
---
module: openssl_signature
short_description: Sign and verify data with openssl
description: This module allows one to sign and verify data via certificate and private key
version_added: 1.1.0
short_description: Sign data with openssl
description:
- This module allows one to sign data using a private key.
- The module can use the cryptography Python library, or the pyOpenSSL Python
library. By default, it tries to detect which one is available. This can be
overridden with the I(select_crypto_backend) option. Please note that the PyOpenSSL backend
was deprecated in Ansible 2.9 and will be removed in community.crypto 2.0.0.
requirements:
- Either cryptography >= 1.2.3 (older versions might work as well)
- Or pyOpenSSL
- Either cryptography >= 1.4 (some key types require newer versions)
- Or pyOpenSSL >= 0.11 (Ed25519 and Ed448 keys are not supported with this backend)
author:
- Patrick Pichler (@aveexy)
- Markus Teufelberger (@MarkusTeufelberger)
options:
action:
description: Action to be executed
type: str
required: true
choices: [ sign, verify ]
private_key:
description: Private key required for sign action
privatekey_path:
description:
- The path to the private key to use when signing.
- Either I(privatekey_path) or I(privatekey_content) must be specified, but not both.
type: path
certificate:
description: Certificate required for verify action
type: path
passphrase:
description: Passphrase for private_key
privatekey_content:
description:
- The content of the private key to use when signing the certificate signing request.
- Either I(privatekey_path) or I(privatekey_content) must be specified, but not both.
type: str
privatekey_passphrase:
description:
- The passphrase for the private key.
- This is required if the private key is password protected.
type: str
path:
description: file to sign/verify
description:
- The file to sign.
- This file will only be read and not modified.
type: path
required: true
signature:
description: base64 encoded signature required for verify action
type: str
select_crypto_backend:
description:
- Determines which crypto backend to use.
@ -58,28 +56,41 @@ options:
type: str
default: auto
choices: [ auto, cryptography, pyopenssl ]
notes:
- |
When using the C(cryptography) backend, the following key types require at least the following C(cryptography) version:
RSA keys: C(cryptography) >= 1.4
DSA and ECDSA keys: C(cryptography) >= 1.5
ed448 and ed25519 keys: C(cryptography) >= 2.6
seealso:
- module: community.crypto.openssl_signature_info
- module: community.crypto.openssl_privatekey
'''
EXAMPLES = r'''
- name: Sign example file
openssl_signature:
action: sign
private_key: private.key
community.crypto.openssl_signature:
privatekey_path: private.key
path: /tmp/example_file
register: sig
- name: Verify signature of example file
openssl_signature:
action: verify
certificate: cert.pem
community.crypto.openssl_signature_info:
certificate_path: cert.pem
path: /tmp/example_file
signature: sig.signature
signature: "{{ sig.signature }}"
register: verify
- name: Make sure the signature is valid
assert:
that:
- verify.valid
'''
RETURN = r'''
signature:
description: base64 encoded signature
returned: changed or success
description: Base64 encoded signature.
returned: success
type: str
'''
@ -88,14 +99,13 @@ import traceback
from distutils.version import LooseVersion
import base64
MINIMAL_PYOPENSSL_VERSION = '0.6'
MINIMAL_CRYPTOGRAPHY_VERSION = '1.2.3'
MINIMAL_PYOPENSSL_VERSION = '0.11'
MINIMAL_CRYPTOGRAPHY_VERSION = '1.4'
PYOPENSSL_IMP_ERR = None
try:
import OpenSSL
from OpenSSL import crypto
PYOPENSSL_VERSION = LooseVersion(OpenSSL.__version__)
except ImportError:
PYOPENSSL_IMP_ERR = traceback.format_exc()
@ -106,12 +116,8 @@ else:
CRYPTOGRAPHY_IMP_ERR = None
try:
import cryptography
import cryptography.hazmat.primitives.asymmetric.rsa
import cryptography.hazmat.primitives.asymmetric.dsa
import cryptography.hazmat.primitives.asymmetric.ec
import cryptography.hazmat.primitives.asymmetric.padding
import cryptography.hazmat.primitives.hashes
CRYPTOGRAPHY_VERSION = LooseVersion(cryptography.__version__)
except ImportError:
CRYPTOGRAPHY_IMP_ERR = traceback.format_exc()
@ -119,35 +125,48 @@ except ImportError:
else:
CRYPTOGRAPHY_FOUND = True
from ansible.module_utils import crypto as crypto_utils
from ansible_collections.community.crypto.plugins.module_utils.crypto.basic import (
CRYPTOGRAPHY_HAS_DSA_SIGN,
CRYPTOGRAPHY_HAS_EC_SIGN,
CRYPTOGRAPHY_HAS_ED25519_SIGN,
CRYPTOGRAPHY_HAS_ED448_SIGN,
CRYPTOGRAPHY_HAS_RSA_SIGN,
OpenSSLObjectError,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.support import (
OpenSSLObject,
load_privatekey,
)
from ansible.module_utils._text import to_native, to_bytes
from ansible.module_utils.basic import AnsibleModule, missing_required_lib
class SignatureBase(crypto_utils.OpenSSLObject):
class SignatureBase(OpenSSLObject):
def __init__(self, module, backend):
super(SignatureBase, self).__init__(
module.params['path'],
'present',
False,
module.check_mode
path=module.params['path'],
state='present',
force=False,
check_mode=module.check_mode
)
self.backend = backend
self.action = module.params['action']
self.signature = module.params['signature']
self.passphrase = module.params['passphrase']
self.private_key = module.params['private_key']
self.certificate = module.params['certificate']
self.privatekey_path = module.params['privatekey_path']
self.privatekey_content = module.params['privatekey_content']
if self.privatekey_content is not None:
self.privatekey_content = self.privatekey_content.encode('utf-8')
self.privatekey_passphrase = module.params['privatekey_passphrase']
def generate(self):
# Empty method because crypto_utils.OpenSSLObject wants this
# Empty method because OpenSSLObject wants this
pass
def dump(self):
# Empty method because crypto_utils.OpenSSLObject wants this
# Empty method because OpenSSLObject wants this
pass
@ -158,36 +177,25 @@ class SignaturePyOpenSSL(SignatureBase):
super(SignaturePyOpenSSL, self).__init__(module, backend)
def run(self):
try:
result = dict()
result = dict()
try:
with open(self.path, "rb") as f:
_in = f.read()
if self.action == "verify":
_signature = base64.b64decode(self.signature)
certificate = crypto_utils.load_certificate(self.certificate, backend=self.backend)
try:
OpenSSL.crypto.verify(certificate, _signature, _in, 'sha256')
except Exception:
self.module.fail_json(
msg="Verification failed"
)
elif self.action == "sign":
private_key = crypto_utils.load_privatekey(
self.private_key,
None if self.passphrase is None else to_bytes(self.passphrase),
backend=self.backend
)
out = OpenSSL.crypto.sign(private_key, _in, "sha256")
result['signature'] = base64.b64encode(out)
private_key = load_privatekey(
path=self.privatekey_path,
content=self.privatekey_content,
passphrase=self.privatekey_passphrase,
backend=self.backend,
)
signature = OpenSSL.crypto.sign(private_key, _in, "sha256")
result['signature'] = base64.b64encode(signature)
return result
except Exception as e:
raise crypto_utils.OpenSSLObjectError(e)
raise OpenSSLObjectError(e)
# Implementation with using cryptography
@ -206,97 +214,65 @@ class SignatureCryptography(SignatureBase):
with open(self.path, "rb") as f:
_in = f.read()
if self.action == "verify":
_signature = base64.b64decode(self.signature)
public_key = crypto_utils.load_certificate(self.certificate, backend=self.backend).public_key()
private_key = load_privatekey(
path=self.privatekey_path,
content=self.privatekey_content,
passphrase=self.privatekey_passphrase,
backend=self.backend,
)
try:
if isinstance(public_key, cryptography.hazmat.primitives.asymmetric.rsa.RSAPublicKey):
public_key.verify(_signature, _in, _padding, _hash)
signature = None
elif isinstance(public_key, cryptography.hazmat.primitives.asymmetric.ec.EllipticCurvePublicKey):
public_key.verify(_signature, _in, cryptography.hazmat.primitives.asymmetric.ec.ECDSA(_hash))
if CRYPTOGRAPHY_HAS_DSA_SIGN:
if isinstance(private_key, cryptography.hazmat.primitives.asymmetric.dsa.DSAPrivateKey):
signature = private_key.sign(_in, _hash)
elif (isinstance(public_key, cryptography.hazmat.primitives.asymmetric.ed25519.Ed25519PublicKey) or
isinstance(public_key, cryptography.hazmat.primitives.asymmetric.ed448.Ed448PublicKey)):
public_key.verify(_signature, _in)
if CRYPTOGRAPHY_HAS_EC_SIGN:
if isinstance(private_key, cryptography.hazmat.primitives.asymmetric.ec.EllipticCurvePrivateKey):
signature = private_key.sign(_in, cryptography.hazmat.primitives.asymmetric.ec.ECDSA(_hash))
else:
self.module.fail_json(
msg="Unsupported algorithm"
)
if CRYPTOGRAPHY_HAS_ED25519_SIGN:
if isinstance(private_key, cryptography.hazmat.primitives.asymmetric.ed25519.Ed25519PrivateKey):
signature = private_key.sign(_in)
except Exception:
self.module.fail_json(
msg="Verification failed"
)
if CRYPTOGRAPHY_HAS_ED448_SIGN:
if isinstance(private_key, cryptography.hazmat.primitives.asymmetric.ed448.Ed448PrivateKey):
signature = private_key.sign(_in)
elif self.action == "sign":
private_key = crypto_utils.load_privatekey(
self.private_key,
None if self.passphrase is None else to_bytes(self.passphrase),
backend=self.backend
if CRYPTOGRAPHY_HAS_RSA_SIGN:
if isinstance(private_key, cryptography.hazmat.primitives.asymmetric.rsa.RSAPrivateKey):
signature = private_key.sign(_in, _padding, _hash)
if signature is None:
self.module.fail_json(
msg="Unsupported key type. Your cryptography version is {0}".format(CRYPTOGRAPHY_VERSION)
)
if isinstance(private_key, cryptography.hazmat.primitives.asymmetric.rsa.RSAPrivateKey):
out = private_key.sign(_in, _padding, _hash)
elif isinstance(private_key, cryptography.hazmat.primitives.asymmetric.ec.EllipticCurvePrivateKey):
out = private_key.sign(_in, cryptography.hazmat.primitives.asymmetric.ec.ECDSA(_hash))
elif (isinstance(private_key, cryptography.hazmat.primitives.asymmetric.ed25519.Ed25519PrivateKey) or
isinstance(private_key, cryptography.hazmat.primitives.asymmetric.ed448.Ed448PrivateKey)):
out = private_key.sign(_in)
else:
self.module.fail_json(
msg="Unsupported algorithm"
)
result['signature'] = base64.b64encode(out)
result['signature'] = base64.b64encode(signature)
return result
except Exception as e:
raise crypto_utils.OpenSSLObjectError(e)
raise OpenSSLObjectError(e)
def main():
module = AnsibleModule(
argument_spec=dict(
action=dict(type='str', choices=[
'sign', 'verify'
]),
private_key=dict(type='path'),
certificate=dict(type='path'),
passphrase=dict(type='str', no_log=True),
privatekey_path=dict(type='path'),
privatekey_content=dict(type='str'),
privatekey_passphrase=dict(type='str', no_log=True),
path=dict(type='path', required=True),
signature=dict(type='path'),
select_crypto_backend=dict(type='str', choices=['auto', 'pyopenssl', 'cryptography'], default='auto'),
),
supports_check_mode=False,
mutually_exclusive=(
['privatekey_path', 'privatekey_content'],
),
required_one_of=(
['privatekey_path', 'privatekey_content'],
),
supports_check_mode=True,
)
if module.params['private_key'] is not None and module.params['certificate'] is not None:
module.fail_json(
msg="private_key and certificate are mutually exclusive"
)
if module.params['private_key'] is None and module.params['action'] == "sign":
module.fail_json(
msg="Private key missing"
)
if module.params['certificate'] is None and module.params['action'] == "verify":
module.fail_json(
msg="Certificate missing"
)
if module.params['action'] == "verify" and module.params['signature'] is None:
module.fail_json(
msg="Can't verify without a signature"
)
if not os.path.isfile(module.params['path']):
module.fail_json(
name=module.params['path'],
@ -327,7 +303,7 @@ def main():
module.fail_json(msg=missing_required_lib('pyOpenSSL >= {0}'.format(MINIMAL_PYOPENSSL_VERSION)),
exception=PYOPENSSL_IMP_ERR)
module.deprecate('The module is using the PyOpenSSL backend. This backend has been deprecated',
version='2.13')
version='2.0.0', collection_name='community.crypto')
_sign = SignaturePyOpenSSL(module, backend)
elif backend == 'cryptography':
if not CRYPTOGRAPHY_FOUND:
@ -338,7 +314,7 @@ def main():
result = _sign.run()
module.exit_json(**result)
except crypto_utils.OpenSSLObjectError as exc:
except OpenSSLObjectError as exc:
module.fail_json(msg=to_native(exc))

354
plugins/modules/openssl_signature_info.py Normal file
View File

@ -0,0 +1,354 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
# Copyright: (c) 2019, Patrick Pichler <ppichler+ansible@mgit.at>
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
DOCUMENTATION = r'''
---
module: openssl_signature_info
version_added: 1.1.0
short_description: Verify signatures with openssl
description:
- This module allows one to verify a signature for a file via a certificate.
- The module can use the cryptography Python library, or the pyOpenSSL Python
library. By default, it tries to detect which one is available. This can be
overridden with the I(select_crypto_backend) option. Please note that the PyOpenSSL backend
was deprecated in Ansible 2.9 and will be removed in community.crypto 2.0.0.
requirements:
- Either cryptography >= 1.4 (some key types require newer versions)
- Or pyOpenSSL >= 0.11 (Ed25519 and Ed448 keys are not supported with this backend)
author:
- Patrick Pichler (@aveexy)
- Markus Teufelberger (@MarkusTeufelberger)
options:
path:
description:
- The signed file to verify.
- This file will only be read and not modified.
type: path
required: true
certificate_path:
description:
- The path to the certificate used to verify the signature.
- Either I(certificate_path) or I(certificate_content) must be specified, but not both.
type: path
certificate_content:
description:
- The content of the certificate used to verify the signature.
- Either I(certificate_path) or I(certificate_content) must be specified, but not both.
type: str
signature:
description: Base64 encoded signature.
type: str
required: true
select_crypto_backend:
description:
- Determines which crypto backend to use.
- The default choice is C(auto), which tries to use C(cryptography) if available, and falls back to C(pyopenssl).
- If set to C(pyopenssl), will try to use the L(pyOpenSSL,https://pypi.org/project/pyOpenSSL/) library.
- If set to C(cryptography), will try to use the L(cryptography,https://cryptography.io/) library.
type: str
default: auto
choices: [ auto, cryptography, pyopenssl ]
notes:
- |
When using the C(cryptography) backend, the following key types require at least the following C(cryptography) version:
RSA keys: C(cryptography) >= 1.4
DSA and ECDSA keys: C(cryptography) >= 1.5
ed448 and ed25519 keys: C(cryptography) >= 2.6
seealso:
- module: community.crypto.openssl_signature
- module: community.crypto.x509_certificate
'''
EXAMPLES = r'''
- name: Sign example file
community.crypto.openssl_signature:
privatekey_path: private.key
path: /tmp/example_file
register: sig
- name: Verify signature of example file
community.crypto.openssl_signature_info:
certificate_path: cert.pem
path: /tmp/example_file
signature: "{{ sig.signature }}"
register: verify
- name: Make sure the signature is valid
assert:
that:
- verify.valid
'''
RETURN = r'''
valid:
description: C(true) means the signature was valid for the given file, C(false) means it wasn't.
returned: success
type: bool
'''
import os
import traceback
from distutils.version import LooseVersion
import base64
MINIMAL_PYOPENSSL_VERSION = '0.11'
MINIMAL_CRYPTOGRAPHY_VERSION = '1.4'
PYOPENSSL_IMP_ERR = None
try:
import OpenSSL
from OpenSSL import crypto
PYOPENSSL_VERSION = LooseVersion(OpenSSL.__version__)
except ImportError:
PYOPENSSL_IMP_ERR = traceback.format_exc()
PYOPENSSL_FOUND = False
else:
PYOPENSSL_FOUND = True
CRYPTOGRAPHY_IMP_ERR = None
try:
import cryptography
import cryptography.hazmat.primitives.asymmetric.padding
import cryptography.hazmat.primitives.hashes
CRYPTOGRAPHY_VERSION = LooseVersion(cryptography.__version__)
except ImportError:
CRYPTOGRAPHY_IMP_ERR = traceback.format_exc()
CRYPTOGRAPHY_FOUND = False
else:
CRYPTOGRAPHY_FOUND = True
from ansible_collections.community.crypto.plugins.module_utils.crypto.basic import (
CRYPTOGRAPHY_HAS_DSA_SIGN,
CRYPTOGRAPHY_HAS_EC_SIGN,
CRYPTOGRAPHY_HAS_ED25519_SIGN,
CRYPTOGRAPHY_HAS_ED448_SIGN,
CRYPTOGRAPHY_HAS_RSA_SIGN,
OpenSSLObjectError,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.support import (
OpenSSLObject,
load_certificate,
)
from ansible.module_utils._text import to_native, to_bytes
from ansible.module_utils.basic import AnsibleModule, missing_required_lib
class SignatureInfoBase(OpenSSLObject):
def __init__(self, module, backend):
super(SignatureInfoBase, self).__init__(
path=module.params['path'],
state='present',
force=False,
check_mode=module.check_mode
)
self.backend = backend
self.signature = module.params['signature']
self.certificate_path = module.params['certificate_path']
self.certificate_content = module.params['certificate_content']
if self.certificate_content is not None:
self.certificate_content = self.certificate_content.encode('utf-8')
def generate(self):
# Empty method because OpenSSLObject wants this
pass
def dump(self):
# Empty method because OpenSSLObject wants this
pass
# Implementation with using pyOpenSSL
class SignatureInfoPyOpenSSL(SignatureInfoBase):
def __init__(self, module, backend):
super(SignatureInfoPyOpenSSL, self).__init__(module, backend)
def run(self):
result = dict()
try:
with open(self.path, "rb") as f:
_in = f.read()
_signature = base64.b64decode(self.signature)
certificate = load_certificate(
path=self.certificate_path,
content=self.certificate_content,
backend=self.backend,
)
try:
OpenSSL.crypto.verify(certificate, _signature, _in, 'sha256')
result['valid'] = True
except Exception:
result['valid'] = False
return result
except Exception as e:
raise OpenSSLObjectError(e)
# Implementation with using cryptography
class SignatureInfoCryptography(SignatureInfoBase):
def __init__(self, module, backend):
super(SignatureInfoCryptography, self).__init__(module, backend)
def run(self):
_padding = cryptography.hazmat.primitives.asymmetric.padding.PKCS1v15()
_hash = cryptography.hazmat.primitives.hashes.SHA256()
result = dict()
try:
with open(self.path, "rb") as f:
_in = f.read()
_signature = base64.b64decode(self.signature)
certificate = load_certificate(
path=self.certificate_path,
content=self.certificate_content,
backend=self.backend,
)
public_key = certificate.public_key()
verified = False
valid = False
if CRYPTOGRAPHY_HAS_DSA_SIGN:
try:
if isinstance(public_key, cryptography.hazmat.primitives.asymmetric.dsa.DSAPublicKey):
public_key.verify(_signature, _in, _hash)
verified = True
valid = True
except cryptography.exceptions.InvalidSignature:
verified = True
valid = False
if CRYPTOGRAPHY_HAS_EC_SIGN:
try:
if isinstance(public_key, cryptography.hazmat.primitives.asymmetric.ec.EllipticCurvePublicKey):
public_key.verify(_signature, _in, cryptography.hazmat.primitives.asymmetric.ec.ECDSA(_hash))
verified = True
valid = True
except cryptography.exceptions.InvalidSignature:
verified = True
valid = False
if CRYPTOGRAPHY_HAS_ED25519_SIGN:
try:
if isinstance(public_key, cryptography.hazmat.primitives.asymmetric.ed25519.Ed25519PublicKey):
public_key.verify(_signature, _in)
verified = True
valid = True
except cryptography.exceptions.InvalidSignature:
verified = True
valid = False
if CRYPTOGRAPHY_HAS_ED448_SIGN:
try:
if isinstance(public_key, cryptography.hazmat.primitives.asymmetric.ed448.Ed448PublicKey):
public_key.verify(_signature, _in)
verified = True
valid = True
except cryptography.exceptions.InvalidSignature:
verified = True
valid = False
if CRYPTOGRAPHY_HAS_RSA_SIGN:
try:
if isinstance(public_key, cryptography.hazmat.primitives.asymmetric.rsa.RSAPublicKey):
public_key.verify(_signature, _in, _padding, _hash)
verified = True
valid = True
except cryptography.exceptions.InvalidSignature:
verified = True
valid = False
if not verified:
self.module.fail_json(
msg="Unsupported key type. Your cryptography version is {0}".format(CRYPTOGRAPHY_VERSION)
)
result['valid'] = valid
return result
except Exception as e:
raise OpenSSLObjectError(e)
def main():
module = AnsibleModule(
argument_spec=dict(
certificate_path=dict(type='path'),
certificate_content=dict(type='str'),
path=dict(type='path', required=True),
signature=dict(type='str', required=True),
select_crypto_backend=dict(type='str', choices=['auto', 'pyopenssl', 'cryptography'], default='auto'),
),
mutually_exclusive=(
['certificate_path', 'certificate_content'],
),
required_one_of=(
['certificate_path', 'certificate_content'],
),
supports_check_mode=True,
)
if not os.path.isfile(module.params['path']):
module.fail_json(
name=module.params['path'],
msg='The file {0} does not exist'.format(module.params['path'])
)
backend = module.params['select_crypto_backend']
if backend == 'auto':
# Detection what is possible
can_use_cryptography = CRYPTOGRAPHY_FOUND and CRYPTOGRAPHY_VERSION >= LooseVersion(MINIMAL_CRYPTOGRAPHY_VERSION)
can_use_pyopenssl = PYOPENSSL_FOUND and PYOPENSSL_VERSION >= LooseVersion(MINIMAL_PYOPENSSL_VERSION)
# Decision
if can_use_cryptography:
backend = 'cryptography'
elif can_use_pyopenssl:
backend = 'pyopenssl'
# Success?
if backend == 'auto':
module.fail_json(msg=("Can't detect any of the required Python libraries "
"cryptography (>= {0}) or PyOpenSSL (>= {1})").format(
MINIMAL_CRYPTOGRAPHY_VERSION,
MINIMAL_PYOPENSSL_VERSION))
try:
if backend == 'pyopenssl':
if not PYOPENSSL_FOUND:
module.fail_json(msg=missing_required_lib('pyOpenSSL >= {0}'.format(MINIMAL_PYOPENSSL_VERSION)),
exception=PYOPENSSL_IMP_ERR)
module.deprecate('The module is using the PyOpenSSL backend. This backend has been deprecated',
version='2.0.0', collection_name='community.crypto')
_sign = SignatureInfoPyOpenSSL(module, backend)
elif backend == 'cryptography':
if not CRYPTOGRAPHY_FOUND:
module.fail_json(msg=missing_required_lib('cryptography >= {0}'.format(MINIMAL_CRYPTOGRAPHY_VERSION)),
exception=CRYPTOGRAPHY_IMP_ERR)
_sign = SignatureInfoCryptography(module, backend)
result = _sign.run()
module.exit_json(**result)
except OpenSSLObjectError as exc:
module.fail_json(msg=to_native(exc))
if __name__ == '__main__':
main()

3
tests/integration/targets/openssl_signature/aliases Normal file
View File

@ -0,0 +1,3 @@
shippable/posix/group1
openssl_signature_info
destructive

2
tests/integration/targets/openssl_signature/meta/main.yml Normal file
View File

@ -0,0 +1,2 @@
dependencies:
- setup_openssl

28
tests/integration/targets/openssl_signature/tasks/loop.yml Normal file
View File

@ -0,0 +1,28 @@
---
# This file is intended to be included in a loop statement
- name: Sign statement with {{ item.type }} key - {{ item.passwd }} using {{ item.backend }}
openssl_signature:
privatekey_path: '{{ output_dir }}/{{item.backend}}_privatekey_{{ item.type }}_{{ item.passwd }}.pem'
privatekey_passphrase: '{{ item.privatekey_passphrase | default(omit) }}'
path: '{{ output_dir }}/statement.txt'
select_crypto_backend: '{{ item.backend }}'
register: sign_result
- debug:
var: sign_result
- name: Verify {{ item.type }} signature - {{ item.passwd }} using {{ item.backend }}
openssl_signature_info:
certificate_path: '{{ output_dir }}/{{item.backend}}_certificate_{{ item.type }}_{{ item.passwd }}.pem'
path: '{{ output_dir }}/statement.txt'
signature: '{{ sign_result.signature }}'
select_crypto_backend: '{{ item.backend }}'
register: verify_result
- name: Make sure the signature is valid
assert:
that:
- verify_result.valid
- debug:
var: verify_result

108
tests/integration/targets/openssl_signature/tasks/main.yml Normal file
View File

@ -0,0 +1,108 @@
---
# Test matrix:
# * pyopenssl or cryptography
# * DSA or ECC or ...
# * password protected private key or not
- name: Set up test combinations
set_fact:
all_tests: []
backends: []
key_types: []
key_password:
- passwd: nopasswd
- passwd: passwd
privatekey_passphrase: hunter2
privatekey_cipher: auto
- name: Add cryptography backend
set_fact:
backends: "{{ backends + [ { 'backend': 'cryptography' } ] }}"
when: cryptography_version.stdout is version('1.4', '>=')
- name: Add pyopenssl backend
set_fact:
backends: "{{ backends + [ { 'backend': 'pyopenssl' } ] }}"
when: pyopenssl_version.stdout is version('0.11', '>=')
- name: Add RSA tests
set_fact:
key_types: "{{ key_types + [ { 'type': 'RSA' } ] }}"
when: cryptography_version.stdout is version('1.4', '>=')
- name: Add DSA + ECDSA tests
set_fact:
key_types: "{{ key_types + [ { 'type': 'DSA', 'size': 2048 }, { 'type': 'ECC', 'curve': 'secp256r1' } ] }}"
when:
- cryptography_version.stdout is version('1.5', '>=')
# FreeBSD 11 fails on secp256r1 keys
- not ansible_os_family == 'FreeBSD'
- name: Add Ed25519 + Ed448 tests
set_fact:
key_types: "{{ key_types + [ { 'type': 'Ed25519' }, { 'type': 'Ed448' } ] }}"
when:
# The module under tests works with >= 2.6, but we also need to be able to create a certificate which requires 2.8
- cryptography_version.stdout is version('2.8', '>=')
# FreeBSD doesn't have support for Ed448/25519
- not ansible_os_family == 'FreeBSD'
- name: Create all test combinations
set_fact:
# Explanation: see https://serverfault.com/a/1004124
all_tests: >-
[
{% for b in backends %}
{% for kt in key_types %}
{% for kp in key_password %}
{# Exclude Ed25519 and Ed448 tests on pyopenssl #}
{% if not (b.backend == 'pyopenssl' and (kt.type == 'Ed25519' or kt.type == 'Ed448')) %}
{{ b | combine (kt) | combine(kp) }},
{% endif %}
{% endfor %}
{% endfor %}
{% endfor %}
]
- name: Generate private keys
openssl_privatekey:
path: '{{ output_dir }}/{{item.backend}}_privatekey_{{ item.type }}_{{ item.passwd }}.pem'
type: '{{ item.type }}'
curve: '{{ item.curve | default(omit) }}'
size: '{{ item.size | default(omit) }}'
passphrase: '{{ item.privatekey_passphrase | default(omit) }}'
cipher: '{{ item.privatekey_cipher | default(omit) }}'
select_crypto_backend: cryptography
loop: '{{ all_tests }}'
- name: Generate public keys
openssl_publickey:
path: '{{ output_dir }}/{{item.backend}}_publickey_{{ item.type }}_{{ item.passwd }}.pem'
privatekey_path: '{{ output_dir }}/{{item.backend}}_privatekey_{{ item.type }}_{{ item.passwd }}.pem'
privatekey_passphrase: '{{ item.privatekey_passphrase | default(omit) }}'
loop: '{{ all_tests }}'
- name: Generate CSRs
openssl_csr:
path: '{{ output_dir }}/{{item.backend}}_{{ item.type }}_{{ item.passwd }}.csr'
privatekey_path: '{{ output_dir }}/{{item.backend}}_privatekey_{{ item.type }}_{{ item.passwd }}.pem'
privatekey_passphrase: '{{ item.privatekey_passphrase | default(omit) }}'
loop: '{{ all_tests }}'
- name: Generate selfsigned certificates
x509_certificate:
provider: selfsigned
path: '{{ output_dir }}/{{item.backend}}_certificate_{{ item.type }}_{{ item.passwd }}.pem'
privatekey_path: '{{ output_dir }}/{{item.backend}}_privatekey_{{ item.type }}_{{ item.passwd }}.pem'
privatekey_passphrase: '{{ item.privatekey_passphrase | default(omit) }}'
csr_path: '{{ output_dir }}/{{item.backend}}_{{ item.type }}_{{ item.passwd }}.csr'
loop: '{{ all_tests }}'
- name: Create statement to be signed
copy:
content: "Erst wenn der Subwoofer die Katze inhaliert, fickt der Bass richtig übel. -- W.A. Mozart"
dest: '{{ output_dir }}/statement.txt'
- name: Loop over all variants
include_tasks: loop.yml
loop: '{{ all_tests }}'