community.general/lib/ansible/modules/crypto/openssl_csr.py

#!/usr/bin/python
# -*- coding: utf-8 -*-
#
# (c) 2017, Yanis Guenane <yanis+ansible@guenane.org>
# 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.0',
                    'status': ['preview'],
                    'supported_by': 'community'}


DOCUMENTATION = '''
---
module: openssl_csr
author: "Yanis Guenane (@Spredzy)"
version_added: "2.4"
short_description: Generate OpenSSL Certificate Signing Request (CSR)
description:
    - "This module allows one to (re)generates OpenSSL certificate signing requests.
       It uses the pyOpenSSL python library to interact with openssl. This module support
       the subjectAltName extension. Note: At least one of commonName or subjectAltName must
       be specified. This module uses file common arguments to specify generated file permissions."
requirements:
    - "python-pyOpenSSL"
options:
    state:
        required: false
        default: "present"
        choices: [ present, absent ]
        description:
            - Whether the certificate signing request should exist or not, taking action if the state is different from what is stated.
    digest:
        required: false
        default: "sha256"
        description:
            - Digest used when signing the certificate signing request with the private key
    privatekey_path:
        required: true
        description:
            - Path to the privatekey to use when signing the certificate signing request
    privatekey_passphrase:
        required: false
        description:
            - The passphrase for the privatekey.
        version_added: "2.4"
    version:
        required: false
        default: 3
        description:
            - Version of the certificate signing request
    force:
        required: false
        default: False
        choices: [ True, False ]
        description:
            - Should the certificate signing request be forced regenerated by this ansible module
    path:
        required: true
        description:
            - Name of the folder in which the generated OpenSSL certificate signing request will be written
    subjectAltName:
        required: false
        description:
            - SAN extension to attach to the certificate signing request
    countryName:
        required: false
        aliases: [ 'C' ]
        description:
            - countryName field of the certificate signing request subject
    stateOrProvinceName:
        required: false
        aliases: [ 'ST' ]
        description:
            - stateOrProvinceName field of the certificate signing request subject
    localityName:
        required: false
        aliases: [ 'L' ]
        description:
            - localityName field of the certificate signing request subject
    organizationName:
        required: false
        aliases: [ 'O' ]
        description:
            - organizationName field of the certificate signing request subject
    organizationalUnitName:
        required: false
        aliases: [ 'OU' ]
        description:
            - organizationalUnitName field of the certificate signing request subject
    commonName:
        required: false
        aliases: [ 'CN' ]
        description:
            - commonName field of the certificate signing request subject
    emailAddress:
        required: false
        aliases: [ 'E' ]
        description:
            - emailAddress field of the certificate signing request subject
'''


EXAMPLES = '''
# Generate an OpenSSL Certificate Signing Request
- openssl_csr:
    path: /etc/ssl/csr/www.ansible.com.csr
    privatekey_path: /etc/ssl/private/ansible.com.pem
    commonName: www.ansible.com

# Generate an OpenSSL Certificate Signing Request with a
# passphrase protected private key
- openssl_csr:
    path: /etc/ssl/csr/www.ansible.com.csr
    privatekey_path: /etc/ssl/private/ansible.com.pem
    privatekey_passphrase: ansible
    commonName: www.ansible.com

# Generate an OpenSSL Certificate Signing Request with Subject information
- openssl_csr:
    path: /etc/ssl/csr/www.ansible.com.csr
    privatekey_path: /etc/ssl/private/ansible.com.pem
    countryName: FR
    organizationName: Ansible
    emailAddress: jdoe@ansible.com
    commonName: www.ansible.com

# Generate an OpenSSL Certificate Signing Request with subjectAltName extension
- openssl_csr:
    path: /etc/ssl/csr/www.ansible.com.csr
    privatekey_path: /etc/ssl/private/ansible.com.pem
    subjectAltName: 'DNS:www.ansible.com,DNS:m.ansible.com'

# Force re-generate an OpenSSL Certificate Signing Request
- openssl_csr:
    path: /etc/ssl/csr/www.ansible.com.csr
    privatekey_path: /etc/ssl/private/ansible.com.pem
    force: True
    commonName: www.ansible.com
'''


RETURN = '''
csr:
    description: Path to the generated Certificate Signing Request
    returned: changed or success
    type: string
    sample: /etc/ssl/csr/www.ansible.com.csr
subject:
    description: A dictionnary of the subject attached to the CSR
    returned: changed or success
    type: list
    sample: {'CN': 'www.ansible.com', 'O': 'Ansible'}
subjectAltName:
    description: The alternative names this CSR is valid for
    returned: changed or success
    type: string
    sample: 'DNS:www.ansible.com,DNS:m.ansible.com'
'''

import errno
import os

try:
    from OpenSSL import crypto
except ImportError:
    pyopenssl_found = False
else:
    pyopenssl_found = True

from ansible.module_utils import crypto as crypto_utils
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils._text import to_native


class CertificateSigningRequestError(Exception):
    pass


class CertificateSigningRequest(object):

    def __init__(self, module):
        self.state = module.params['state']
        self.digest = module.params['digest']
        self.force = module.params['force']
        self.subjectAltName = module.params['subjectAltName']
        self.path = module.params['path']
        self.privatekey_path = module.params['privatekey_path']
        self.privatekey_passphrase = module.params['privatekey_passphrase']
        self.version = module.params['version']
        self.changed = True
        self.request = None
        self.privatekey = None

        self.subject = {
            'C': module.params['countryName'],
            'ST': module.params['stateOrProvinceName'],
            'L': module.params['localityName'],
            'O': module.params['organizationName'],
            'OU': module.params['organizationalUnitName'],
            'CN': module.params['commonName'],
            'emailAddress': module.params['emailAddress'],
        }

        if self.subjectAltName is None:
            self.subjectAltName = 'DNS:%s' % self.subject['CN']

        self.subject = dict((k, v) for k, v in self.subject.items() if v)

    def generate(self, module):
        '''Generate the certificate signing request.'''

        if not os.path.exists(self.path) or self.force:
            req = crypto.X509Req()
            req.set_version(self.version)
            subject = req.get_subject()
            for (key, value) in self.subject.items():
                if value is not None:
                    setattr(subject, key, value)

            if self.subjectAltName is not None:
                req.add_extensions([crypto.X509Extension(b"subjectAltName", False, self.subjectAltName.encode('ascii'))])

            self.privatekey = crypto_utils.load_privatekey(
                self.privatekey_path,
                self.privatekey_passphrase
            )

            req.set_pubkey(self.privatekey)
            req.sign(self.privatekey, self.digest)
            self.request = req

            try:
                csr_file = open(self.path, 'wb')
                csr_file.write(crypto.dump_certificate_request(crypto.FILETYPE_PEM, self.request))
                csr_file.close()
            except (IOError, OSError) as exc:
                raise CertificateSigningRequestError(exc)
        else:
            self.changed = False

        file_args = module.load_file_common_arguments(module.params)
        if module.set_fs_attributes_if_different(file_args, False):
            self.changed = True

    def remove(self):
        '''Remove the Certificate Signing Request.'''

        try:
            os.remove(self.path)
        except OSError as exc:
            if exc.errno != errno.ENOENT:
                raise CertificateSigningRequestError(exc)
            else:
                self.changed = False

    def dump(self):
        '''Serialize the object into a dictionary.'''

        result = {
            'csr': self.path,
            'subject': self.subject,
            'subjectAltName': self.subjectAltName,
            'changed': self.changed
        }

        return result


def main():
    module = AnsibleModule(
        argument_spec=dict(
            state=dict(default='present', choices=['present', 'absent'], type='str'),
            digest=dict(default='sha256', type='str'),
            privatekey_path=dict(require=True, type='path'),
            privatekey_passphrase=dict(type='str', no_log=True),
            version=dict(default='3', type='int'),
            force=dict(default=False, type='bool'),
            subjectAltName=dict(aliases=['subjectAltName'], type='str'),
            path=dict(required=True, type='path'),
            countryName=dict(aliases=['C'], type='str'),
            stateOrProvinceName=dict(aliases=['ST'], type='str'),
            localityName=dict(aliases=['L'], type='str'),
            organizationName=dict(aliases=['O'], type='str'),
            organizationalUnitName=dict(aliases=['OU'], type='str'),
            commonName=dict(aliases=['CN'], type='str'),
            emailAddress=dict(aliases=['E'], type='str'),
        ),
        add_file_common_args=True,
        supports_check_mode=True,
        required_one_of=[['commonName', 'subjectAltName']],
    )

    if not pyopenssl_found:
        module.fail_json(msg='the python pyOpenSSL module is required')

    path = module.params['path']
    base_dir = os.path.dirname(module.params['path'])

    if not os.path.isdir(base_dir):
        module.fail_json(name=path, msg='The directory %s does not exist' % path)

    csr = CertificateSigningRequest(module)

    if module.params['state'] == 'present':

        if module.check_mode:
            result = csr.dump()
            result['changed'] = module.params['force'] or not os.path.exists(path)
            module.exit_json(**result)

        try:
            csr.generate(module)
        except CertificateSigningRequestError as exc:
            module.fail_json(msg=to_native(exc))

    else:

        if module.check_mode:
            result = csr.dump()
            result['changed'] = os.path.exists(path)
            module.exit_json(**result)

        try:
            csr.remove()
        except CertificateSigningRequestError as exc:
            module.fail_json(msg=to_native(exc))

    result = csr.dump()

    module.exit_json(**result)


if __name__ == "__main__":
    main()