cmueller/ansible
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Remove incidental_consul tests (#71811)

Browse source 
* Add explicit intg tests for argspec functionality

* ci_complete ci_coverage

* Remove incidental_consul and incidental_setup_openssl

* ci_complete ci_coverage
This commit is contained in:
Matt Martz 2020-09-22 10:33:59 -05:00 committed by GitHub
parent 00b22ab55e
commit a99212464c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
19 changed files with 68 additions and 6351 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
test/integration/targets/argspec/aliases Normal file
View file

@ -0,0 +1 @@
shippable/posix/group5

34
test/integration/targets/argspec/library/argspec.py Normal file
View file

@ -0,0 +1,34 @@
#!/usr/bin/python
# Copyright: (c) 2020, Matt Martz <matt@sivel.net>
# 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
from ansible.module_utils.basic import AnsibleModule
def main():
module = AnsibleModule(
{
'state': {
'default': 'present',
'type': 'str',
'choices': ['absent', 'present'],
},
'path': {},
'content': {},
},
required_if=(
('state', 'present', ('path', 'content'), True),
),
mutually_exclusive=(
('path', 'content'),
),
)
module.exit_json(**module.params)
if __name__ == '__main__':
main()

33
test/integration/targets/argspec/tasks/main.yml Normal file
View file

@ -0,0 +1,33 @@
- argspec:
state: absent
register: argspec_simple_good
- argspec:
state: present
register: argspec_required_if_fail
ignore_errors: true
- argspec:
state: present
path: foo
register: argspec_required_if_good1
- argspec:
state: present
content: foo
register: argspec_required_if_good2
- argspec:
state: present
content: foo
path: foo
register: argspec_mutually_exclusive_fail
ignore_errors: true
- assert:
that:
- argspec_simple_good is successful
- argspec_required_if_fail is failed
- argspec_required_if_good1 is successful
- argspec_required_if_good2 is successful
- argspec_mutually_exclusive_fail is failed

4
test/integration/targets/incidental_consul/aliases
View file

@ -1,4 +0,0 @@
shippable/posix/incidental
destructive
skip/aix
skip/power/centos

3
test/integration/targets/incidental_consul/meta/main.yml
View file

@ -1,3 +0,0 @@
---
dependencies:
- incidental_setup_openssl

162
test/integration/targets/incidental_consul/tasks/consul_session.yml
View file

@ -1,162 +0,0 @@
- name: list sessions
consul_session:
state: list
register: result
- assert:
that:
- result is changed
- "'sessions' in result"
- name: create a session
consul_session:
state: present
name: testsession
register: result
- assert:
that:
- result is changed
- result['name'] == 'testsession'
- "'session_id' in result"
- set_fact:
session_id: "{{ result['session_id'] }}"
- name: list sessions after creation
consul_session:
state: list
register: result
- set_fact:
session_count: "{{ result['sessions'] | length }}"
- assert:
that:
- result is changed
# selectattr not available on Jinja 2.2 provided by CentOS 6
# hence the two following tasks (set_fact/assert) are used
# - (result['sessions'] | selectattr('ID', 'match', '^' ~ session_id ~ '$') | first)['Name'] == 'testsession'
- name: search created session
set_fact:
test_session_found: True
loop: "{{ result['sessions'] }}"
when: "item.get('ID') == session_id and item.get('Name') == 'testsession'"
- name: ensure session was created
assert:
that:
- test_session_found|default(False)
- name: fetch info about a session
consul_session:
state: info
id: '{{ session_id }}'
register: result
- assert:
that:
- result is changed
- name: ensure 'id' parameter is required when state=info
consul_session:
state: info
name: test
register: result
ignore_errors: True
- assert:
that:
- result is failed
- name: ensure unknown scheme fails
consul_session:
state: info
id: '{{ session_id }}'
scheme: non_existent
register: result
ignore_errors: True
- assert:
that:
- result is failed
- when: pyopenssl_version.stdout is version('0.15', '>=')
block:
- name: ensure SSL certificate is checked
consul_session:
state: info
id: '{{ session_id }}'
port: 8501
scheme: https
register: result
ignore_errors: True
- name: previous task should fail since certificate is not known
assert:
that:
- result is failed
- "'certificate verify failed' in result.msg"
- name: ensure SSL certificate isn't checked when validate_certs is disabled
consul_session:
state: info
id: '{{ session_id }}'
port: 8501
scheme: https
validate_certs: False
register: result
- name: previous task should succeed since certificate isn't checked
assert:
that:
- result is changed
- name: ensure a secure connection is possible
consul_session:
state: info
id: '{{ session_id }}'
port: 8501
scheme: https
environment:
REQUESTS_CA_BUNDLE: '{{ remote_dir }}/cert.pem'
register: result
- assert:
that:
- result is changed
- name: delete a session
consul_session:
state: absent
id: '{{ session_id }}'
register: result
- assert:
that:
- result is changed
- name: list sessions after deletion
consul_session:
state: list
register: result
- assert:
that:
- result is changed
# selectattr and equalto not available on Jinja 2.2 provided by CentOS 6
# hence the two following tasks (command/assert) are used
# - (result['sessions'] | selectattr('ID', 'equalto', session_id) | list | length) == 0
- name: search deleted session
command: echo 'session found'
loop: "{{ result['sessions'] }}"
when: "item.get('ID') == session_id and item.get('Name') == 'testsession'"
register: search_deleted
- name: ensure session was deleted
assert:
that:
- search_deleted is skipped # each iteration is skipped
- search_deleted is not changed # and then unchanged

97
test/integration/targets/incidental_consul/tasks/main.yml
View file

@ -1,97 +0,0 @@
---
- name: Install Consul and test
vars:
consul_version: '1.5.0'
consul_uri: https://s3.amazonaws.com/ansible-ci-files/test/integration/targets/consul/consul_{{ consul_version }}_{{ ansible_system | lower }}_{{ consul_arch }}.zip
consul_cmd: '{{ output_dir }}/consul'
block:
- name: register pyOpenSSL version
command: "{{ ansible_python_interpreter }} -c 'import OpenSSL; print(OpenSSL.__version__)'"
register: pyopenssl_version
- name: Install requests<2.20 (CentOS/RHEL 6)
pip:
name: requests<2.20
register: result
until: result is success
when: ansible_distribution_file_variety|default() == 'RedHat' and ansible_distribution_major_version is version('6', '<=')
- name: Install python-consul
pip:
name: python-consul
register: result
until: result is success
- when: pyopenssl_version.stdout is version('0.15', '>=')
block:
- name: Generate privatekey
openssl_privatekey:
path: '{{ output_dir }}/privatekey.pem'
- name: Generate CSR
openssl_csr:
path: '{{ output_dir }}/csr.csr'
privatekey_path: '{{ output_dir }}/privatekey.pem'
subject:
commonName: localhost
- name: Generate selfsigned certificate
openssl_certificate:
path: '{{ output_dir }}/cert.pem'
csr_path: '{{ output_dir }}/csr.csr'
privatekey_path: '{{ output_dir }}/privatekey.pem'
provider: selfsigned
selfsigned_digest: sha256
register: selfsigned_certificate
- name: 'Install unzip'
package:
name: unzip
register: result
until: result is success
when: ansible_distribution != "MacOSX" # unzip already installed
- assert:
# Linux: x86_64, FreeBSD: amd64
that: ansible_architecture in ['i386', 'x86_64', 'amd64']
- set_fact:
consul_arch: '386'
when: ansible_architecture == 'i386'
- set_fact:
consul_arch: amd64
when: ansible_architecture in ['x86_64', 'amd64']
- name: 'Download consul binary'
unarchive:
src: '{{ consul_uri }}'
dest: '{{ output_dir }}'
remote_src: true
register: result
until: result is success
- vars:
remote_dir: '{{ echo_output_dir.stdout }}'
block:
- command: 'echo {{ output_dir }}'
register: echo_output_dir
- name: 'Create configuration file'
template:
src: consul_config.hcl.j2
dest: '{{ output_dir }}/consul_config.hcl'
- name: 'Start Consul (dev mode enabled)'
shell: 'nohup {{ consul_cmd }} agent -dev -config-file {{ output_dir }}/consul_config.hcl </dev/null >/dev/null 2>&1 &'
- name: 'Create some data'
command: '{{ consul_cmd }} kv put data/value{{ item }} foo{{ item }}'
loop: [1, 2, 3]
- import_tasks: consul_session.yml
always:
- name: 'Kill consul process'
shell: "kill $(cat {{ output_dir }}/consul.pid)"
ignore_errors: true

13
test/integration/targets/incidental_consul/templates/consul_config.hcl.j2
View file

@ -1,13 +0,0 @@
# {{ ansible_managed }}
server = true
pid_file = "{{ remote_dir }}/consul.pid"
ports {
http = 8500
{% if pyopenssl_version.stdout is version('0.15', '>=') %}
https = 8501
{% endif %}
}
{% if pyopenssl_version.stdout is version('0.15', '>=') %}
key_file = "{{ remote_dir }}/privatekey.pem"
cert_file = "{{ remote_dir }}/cert.pem"
{% endif %}

2
test/integration/targets/incidental_setup_openssl/aliases
View file

@ -1,2 +0,0 @@
hidden

48
test/integration/targets/incidental_setup_openssl/tasks/main.yml
View file

@ -1,48 +0,0 @@
---
- name: Include OS-specific variables
include_vars: '{{ ansible_os_family }}.yml'
when: not ansible_os_family == "Darwin"
- name: Install OpenSSL
become: True
package:
name: '{{ openssl_package_name }}'
when: not ansible_os_family == 'Darwin'
- name: Install pyOpenSSL (Python 3)
become: True
package:
name: '{{ pyopenssl_package_name_python3 }}'
when: not ansible_os_family == 'Darwin' and ansible_python_version is version('3.0', '>=')
- name: Install pyOpenSSL (Python 2)
become: True
package:
name: '{{ pyopenssl_package_name }}'
when: not ansible_os_family == 'Darwin' and ansible_python_version is version('3.0', '<')
- name: Install pyOpenSSL (Darwin)
become: True
pip:
name:
- pyOpenSSL==19.1.0
# dependencies for pyOpenSSL
- cffi==1.14.2
- cryptography==3.1
- enum34==1.1.10
- ipaddress==1.0.23
- pycparser==2.20
- six==1.15.0
when: ansible_os_family == 'Darwin'
- name: register pyOpenSSL version
command: "{{ ansible_python.executable }} -c 'import OpenSSL; print(OpenSSL.__version__)'"
register: pyopenssl_version
- name: register openssl version
shell: "openssl version | cut -d' ' -f2"
register: openssl_version
- name: register cryptography version
command: "{{ ansible_python.executable }} -c 'import cryptography; print(cryptography.__version__)'"
register: cryptography_version

3
test/integration/targets/incidental_setup_openssl/vars/Debian.yml
View file

@ -1,3 +0,0 @@
pyopenssl_package_name: python-openssl
pyopenssl_package_name_python3: python3-openssl
openssl_package_name: openssl

3
test/integration/targets/incidental_setup_openssl/vars/FreeBSD.yml
View file

@ -1,3 +0,0 @@
pyopenssl_package_name: py27-openssl
pyopenssl_package_name_python3: py36-openssl
openssl_package_name: openssl

3
test/integration/targets/incidental_setup_openssl/vars/RedHat.yml
View file

@ -1,3 +0,0 @@
pyopenssl_package_name: pyOpenSSL
pyopenssl_package_name_python3: python3-pyOpenSSL
openssl_package_name: openssl

3
test/integration/targets/incidental_setup_openssl/vars/Suse.yml
View file

@ -1,3 +0,0 @@
pyopenssl_package_name: python-pyOpenSSL
pyopenssl_package_name_python3: python3-pyOpenSSL
openssl_package_name: openssl

284
test/support/integration/plugins/modules/consul_session.py
View file

@ -1,284 +0,0 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
# Copyright: (c) 2015, Steve Gargan <steve.gargan@gmail.com>
# 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',
'status': ['preview'],
'supported_by': 'community'}
DOCUMENTATION = """
module: consul_session
short_description: Manipulate consul sessions
description:
- Allows the addition, modification and deletion of sessions in a consul
cluster. These sessions can then be used in conjunction with key value pairs
to implement distributed locks. In depth documentation for working with
sessions can be found at http://www.consul.io/docs/internals/sessions.html
requirements:
- python-consul
- requests
version_added: "2.0"
author:
- Steve Gargan (@sgargan)
options:
id:
description:
- ID of the session, required when I(state) is either C(info) or
C(remove).
type: str
state:
description:
- Whether the session should be present i.e. created if it doesn't
exist, or absent, removed if present. If created, the I(id) for the
session is returned in the output. If C(absent), I(id) is
required to remove the session. Info for a single session, all the
sessions for a node or all available sessions can be retrieved by
specifying C(info), C(node) or C(list) for the I(state); for C(node)
or C(info), the node I(name) or session I(id) is required as parameter.
choices: [ absent, info, list, node, present ]
type: str
default: present
name:
description:
- The name that should be associated with the session. Required when
I(state=node) is used.
type: str
delay:
description:
- The optional lock delay that can be attached to the session when it
is created. Locks for invalidated sessions ar blocked from being
acquired until this delay has expired. Durations are in seconds.
type: int
default: 15
node:
description:
- The name of the node that with which the session will be associated.
by default this is the name of the agent.
type: str
datacenter:
description:
- The name of the datacenter in which the session exists or should be
created.
type: str
checks:
description:
- Checks that will be used to verify the session health. If
all the checks fail, the session will be invalidated and any locks
associated with the session will be release and can be acquired once
the associated lock delay has expired.
type: list
host:
description:
- The host of the consul agent defaults to localhost.
type: str
default: localhost
port:
description:
- The port on which the consul agent is running.
type: int
default: 8500
scheme:
description:
- The protocol scheme on which the consul agent is running.
type: str
default: http
version_added: "2.1"
validate_certs:
description:
- Whether to verify the TLS certificate of the consul agent.
type: bool
default: True
version_added: "2.1"
behavior:
description:
- The optional behavior that can be attached to the session when it
is created. This controls the behavior when a session is invalidated.
choices: [ delete, release ]
type: str
default: release
version_added: "2.2"
"""
EXAMPLES = '''
- name: register basic session with consul
consul_session:
name: session1
- name: register a session with an existing check
consul_session:
name: session_with_check
checks:
- existing_check_name
- name: register a session with lock_delay
consul_session:
name: session_with_delay
delay: 20s
- name: retrieve info about session by id
consul_session:
id: session_id
state: info
- name: retrieve active sessions
consul_session:
state: list
'''
try:
import consul
from requests.exceptions import ConnectionError
python_consul_installed = True
except ImportError:
python_consul_installed = False
from ansible.module_utils.basic import AnsibleModule
def execute(module):
state = module.params.get('state')
if state in ['info', 'list', 'node']:
lookup_sessions(module)
elif state == 'present':
update_session(module)
else:
remove_session(module)
def lookup_sessions(module):
datacenter = module.params.get('datacenter')
state = module.params.get('state')
consul_client = get_consul_api(module)
try:
if state == 'list':
sessions_list = consul_client.session.list(dc=datacenter)
# Ditch the index, this can be grabbed from the results
if sessions_list and len(sessions_list) >= 2:
sessions_list = sessions_list[1]
module.exit_json(changed=True,
sessions=sessions_list)
elif state == 'node':
node = module.params.get('node')
sessions = consul_client.session.node(node, dc=datacenter)
module.exit_json(changed=True,
node=node,
sessions=sessions)
elif state == 'info':
session_id = module.params.get('id')
session_by_id = consul_client.session.info(session_id, dc=datacenter)
module.exit_json(changed=True,
session_id=session_id,
sessions=session_by_id)
except Exception as e:
module.fail_json(msg="Could not retrieve session info %s" % e)
def update_session(module):
name = module.params.get('name')
delay = module.params.get('delay')
checks = module.params.get('checks')
datacenter = module.params.get('datacenter')
node = module.params.get('node')
behavior = module.params.get('behavior')
consul_client = get_consul_api(module)
try:
session = consul_client.session.create(
name=name,
behavior=behavior,
node=node,
lock_delay=delay,
dc=datacenter,
checks=checks
)
module.exit_json(changed=True,
session_id=session,
name=name,
behavior=behavior,
delay=delay,
checks=checks,
node=node)
except Exception as e:
module.fail_json(msg="Could not create/update session %s" % e)
def remove_session(module):
session_id = module.params.get('id')
consul_client = get_consul_api(module)
try:
consul_client.session.destroy(session_id)
module.exit_json(changed=True,
session_id=session_id)
except Exception as e:
module.fail_json(msg="Could not remove session with id '%s' %s" % (
session_id, e))
def get_consul_api(module):
return consul.Consul(host=module.params.get('host'),
port=module.params.get('port'),
scheme=module.params.get('scheme'),
verify=module.params.get('validate_certs'))
def test_dependencies(module):
if not python_consul_installed:
module.fail_json(msg="python-consul required for this module. "
"see https://python-consul.readthedocs.io/en/latest/#installation")
def main():
argument_spec = dict(
checks=dict(type='list'),
delay=dict(type='int', default='15'),
behavior=dict(type='str', default='release', choices=['release', 'delete']),
host=dict(type='str', default='localhost'),
port=dict(type='int', default=8500),
scheme=dict(type='str', default='http'),
validate_certs=dict(type='bool', default=True),
id=dict(type='str'),
name=dict(type='str'),
node=dict(type='str'),
state=dict(type='str', default='present', choices=['absent', 'info', 'list', 'node', 'present']),
datacenter=dict(type='str'),
)
module = AnsibleModule(
argument_spec=argument_spec,
required_if=[
('state', 'node', ['name']),
('state', 'info', ['id']),
('state', 'remove', ['id']),
],
supports_check_mode=False
)
test_dependencies(module)
try:
execute(module)
except ConnectionError as e:
module.fail_json(msg='Could not connect to consul agent at %s:%s, error was %s' % (
module.params.get('host'), module.params.get('port'), e))
except Exception as e:
module.fail_json(msg=str(e))
if __name__ == '__main__':
main()

2757
test/support/integration/plugins/modules/openssl_certificate.py
View file

File diff suppressed because it is too large Load diff

864
test/support/integration/plugins/modules/openssl_certificate_info.py
View file

@ -1,864 +0,0 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
# Copyright: (c) 2016-2017, Yanis Guenane <yanis+ansible@guenane.org>
# Copyright: (c) 2017, Markus Teufelberger <mteufelberger+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',
'status': ['preview'],
'supported_by': 'community'}
DOCUMENTATION = r'''
---
module: openssl_certificate_info
version_added: '2.8'
short_description: Provide information of OpenSSL X.509 certificates
description:
- This module allows one to query information on OpenSSL certificates.
- It uses the pyOpenSSL or cryptography python library to interact with OpenSSL. If both the
cryptography and PyOpenSSL libraries are available (and meet the minimum version requirements)
cryptography will be preferred as a backend over PyOpenSSL (unless the backend is forced with
C(select_crypto_backend)). Please note that the PyOpenSSL backend was deprecated in Ansible 2.9
and will be removed in Ansible 2.13.
requirements:
- PyOpenSSL >= 0.15 or cryptography >= 1.6
author:
- Felix Fontein (@felixfontein)
- Yanis Guenane (@Spredzy)
- Markus Teufelberger (@MarkusTeufelberger)
options:
path:
description:
- Remote absolute path where the certificate file is loaded from.
- Either I(path) or I(content) must be specified, but not both.
type: path
content:
description:
- Content of the X.509 certificate in PEM format.
- Either I(path) or I(content) must be specified, but not both.
type: str
version_added: "2.10"
valid_at:
description:
- A dict of names mapping to time specifications. Every time specified here
will be checked whether the certificate is valid at this point. See the
C(valid_at) return value for informations on the result.
- Time can be specified either as relative time or as absolute timestamp.
- Time will always be interpreted as UTC.
- Valid format is C([+-]timespec | ASN.1 TIME) where timespec can be an integer
+ C([w | d | h | m | s]) (e.g. C(+32w1d2h), and ASN.1 TIME (i.e. pattern C(YYYYMMDDHHMMSSZ)).
Note that all timestamps will be treated as being in UTC.
type: dict
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.
- Please note that the C(pyopenssl) backend has been deprecated in Ansible 2.9, and will be removed in Ansible 2.13.
From that point on, only the C(cryptography) backend will be available.
type: str
default: auto
choices: [ auto, cryptography, pyopenssl ]
notes:
- All timestamp values are provided in ASN.1 TIME format, i.e. following the C(YYYYMMDDHHMMSSZ) pattern.
They are all in UTC.
seealso:
- module: openssl_certificate
'''
EXAMPLES = r'''
- name: Generate a Self Signed OpenSSL certificate
openssl_certificate:
path: /etc/ssl/crt/ansible.com.crt
privatekey_path: /etc/ssl/private/ansible.com.pem
csr_path: /etc/ssl/csr/ansible.com.csr
provider: selfsigned
# Get information on the certificate
- name: Get information on generated certificate
openssl_certificate_info:
path: /etc/ssl/crt/ansible.com.crt
register: result
- name: Dump information
debug:
var: result
# Check whether the certificate is valid or not valid at certain times, fail
# if this is not the case. The first task (openssl_certificate_info) collects
# the information, and the second task (assert) validates the result and
# makes the playbook fail in case something is not as expected.
- name: Test whether that certificate is valid tomorrow and/or in three weeks
openssl_certificate_info:
path: /etc/ssl/crt/ansible.com.crt
valid_at:
point_1: "+1d"
point_2: "+3w"
register: result
- name: Validate that certificate is valid tomorrow, but not in three weeks
assert:
that:
- result.valid_at.point_1 # valid in one day
- not result.valid_at.point_2 # not valid in three weeks
'''
RETURN = r'''
expired:
description: Whether the certificate is expired (i.e. C(notAfter) is in the past)
returned: success
type: bool
basic_constraints:
description: Entries in the C(basic_constraints) extension, or C(none) if extension is not present.
returned: success
type: list
elements: str
sample: "[CA:TRUE, pathlen:1]"
basic_constraints_critical:
description: Whether the C(basic_constraints) extension is critical.
returned: success
type: bool
extended_key_usage:
description: Entries in the C(extended_key_usage) extension, or C(none) if extension is not present.
returned: success
type: list
elements: str
sample: "[Biometric Info, DVCS, Time Stamping]"
extended_key_usage_critical:
description: Whether the C(extended_key_usage) extension is critical.
returned: success
type: bool
extensions_by_oid:
description: Returns a dictionary for every extension OID
returned: success
type: dict
contains:
critical:
description: Whether the extension is critical.
returned: success
type: bool
value:
description: The Base64 encoded value (in DER format) of the extension
returned: success
type: str
sample: "MAMCAQU="
sample: '{"1.3.6.1.5.5.7.1.24": { "critical": false, "value": "MAMCAQU="}}'
key_usage:
description: Entries in the C(key_usage) extension, or C(none) if extension is not present.
returned: success
type: str
sample: "[Key Agreement, Data Encipherment]"
key_usage_critical:
description: Whether the C(key_usage) extension is critical.
returned: success
type: bool
subject_alt_name:
description: Entries in the C(subject_alt_name) extension, or C(none) if extension is not present.
returned: success
type: list
elements: str
sample: "[DNS:www.ansible.com, IP:1.2.3.4]"
subject_alt_name_critical:
description: Whether the C(subject_alt_name) extension is critical.
returned: success
type: bool
ocsp_must_staple:
description: C(yes) if the OCSP Must Staple extension is present, C(none) otherwise.
returned: success
type: bool
ocsp_must_staple_critical:
description: Whether the C(ocsp_must_staple) extension is critical.
returned: success
type: bool
issuer:
description:
- The certificate's issuer.
- Note that for repeated values, only the last one will be returned.
returned: success
type: dict
sample: '{"organizationName": "Ansible", "commonName": "ca.example.com"}'
issuer_ordered:
description: The certificate's issuer as an ordered list of tuples.
returned: success
type: list
elements: list
sample: '[["organizationName", "Ansible"], ["commonName": "ca.example.com"]]'
version_added: "2.9"
subject:
description:
- The certificate's subject as a dictionary.
- Note that for repeated values, only the last one will be returned.
returned: success
type: dict
sample: '{"commonName": "www.example.com", "emailAddress": "test@example.com"}'
subject_ordered:
description: The certificate's subject as an ordered list of tuples.
returned: success
type: list
elements: list
sample: '[["commonName", "www.example.com"], ["emailAddress": "test@example.com"]]'
version_added: "2.9"
not_after:
description: C(notAfter) date as ASN.1 TIME
returned: success
type: str
sample: 20190413202428Z
not_before:
description: C(notBefore) date as ASN.1 TIME
returned: success
type: str
sample: 20190331202428Z
public_key:
description: Certificate's public key in PEM format
returned: success
type: str
sample: "-----BEGIN PUBLIC KEY-----\nMIICIjANBgkqhkiG9w0BAQEFAAOCAg8A..."
public_key_fingerprints:
description:
- Fingerprints of certificate's public key.
- For every hash algorithm available, the fingerprint is computed.
returned: success
type: dict
sample: "{'sha256': 'd4:b3:aa:6d:c8:04:ce:4e:ba:f6:29:4d:92:a3:94:b0:c2:ff:bd:bf:33:63:11:43:34:0f:51:b0:95:09:2f:63',
'sha512': 'f7:07:4a:f0:b0:f0:e6:8b:95:5f:f9:e6:61:0a:32:68:f1..."
signature_algorithm:
description: The signature algorithm used to sign the certificate.
returned: success
type: str
sample: sha256WithRSAEncryption
serial_number:
description: The certificate's serial number.
returned: success
type: int
sample: 1234
version:
description: The certificate version.
returned: success
type: int
sample: 3
valid_at:
description: For every time stamp provided in the I(valid_at) option, a
boolean whether the certificate is valid at that point in time
or not.
returned: success
type: dict
subject_key_identifier:
description:
- The certificate's subject key identifier.
- The identifier is returned in hexadecimal, with C(:) used to separate bytes.
- Is C(none) if the C(SubjectKeyIdentifier) extension is not present.
returned: success and if the pyOpenSSL backend is I(not) used
type: str
sample: '00:11:22:33:44:55:66:77:88:99:aa:bb:cc:dd:ee:ff:00:11:22:33'
version_added: "2.9"
authority_key_identifier:
description:
- The certificate's authority key identifier.
- The identifier is returned in hexadecimal, with C(:) used to separate bytes.
- Is C(none) if the C(AuthorityKeyIdentifier) extension is not present.
returned: success and if the pyOpenSSL backend is I(not) used
type: str
sample: '00:11:22:33:44:55:66:77:88:99:aa:bb:cc:dd:ee:ff:00:11:22:33'
version_added: "2.9"
authority_cert_issuer:
description:
- The certificate's authority cert issuer as a list of general names.
- Is C(none) if the C(AuthorityKeyIdentifier) extension is not present.
returned: success and if the pyOpenSSL backend is I(not) used
type: list
elements: str
sample: "[DNS:www.ansible.com, IP:1.2.3.4]"
version_added: "2.9"
authority_cert_serial_number:
description:
- The certificate's authority cert serial number.
- Is C(none) if the C(AuthorityKeyIdentifier) extension is not present.
returned: success and if the pyOpenSSL backend is I(not) used
type: int
sample: '12345'
version_added: "2.9"
ocsp_uri:
description: The OCSP responder URI, if included in the certificate. Will be
C(none) if no OCSP responder URI is included.
returned: success
type: str
version_added: "2.9"
'''
import abc
import binascii
import datetime
import os
import re
import traceback
from distutils.version import LooseVersion
from ansible.module_utils import crypto as crypto_utils
from ansible.module_utils.basic import AnsibleModule, missing_required_lib
from ansible.module_utils.six import string_types
from ansible.module_utils._text import to_native, to_text, to_bytes
from ansible.module_utils.compat import ipaddress as compat_ipaddress
MINIMAL_CRYPTOGRAPHY_VERSION = '1.6'
MINIMAL_PYOPENSSL_VERSION = '0.15'
PYOPENSSL_IMP_ERR = None
try:
import OpenSSL
from OpenSSL import crypto
PYOPENSSL_VERSION = LooseVersion(OpenSSL.__version__)
if OpenSSL.SSL.OPENSSL_VERSION_NUMBER >= 0x10100000:
# OpenSSL 1.1.0 or newer
OPENSSL_MUST_STAPLE_NAME = b"tlsfeature"
OPENSSL_MUST_STAPLE_VALUE = b"status_request"
else:
# OpenSSL 1.0.x or older
OPENSSL_MUST_STAPLE_NAME = b"1.3.6.1.5.5.7.1.24"
OPENSSL_MUST_STAPLE_VALUE = b"DER:30:03:02:01:05"
except ImportError:
PYOPENSSL_IMP_ERR = traceback.format_exc()
PYOPENSSL_FOUND = False
else:
PYOPENSSL_FOUND = True
CRYPTOGRAPHY_IMP_ERR = None
try:
import cryptography
from cryptography import x509
from cryptography.hazmat.primitives import serialization
CRYPTOGRAPHY_VERSION = LooseVersion(cryptography.__version__)
except ImportError:
CRYPTOGRAPHY_IMP_ERR = traceback.format_exc()
CRYPTOGRAPHY_FOUND = False
else:
CRYPTOGRAPHY_FOUND = True
TIMESTAMP_FORMAT = "%Y%m%d%H%M%SZ"
class CertificateInfo(crypto_utils.OpenSSLObject):
def __init__(self, module, backend):
super(CertificateInfo, self).__init__(
module.params['path'] or '',
'present',
False,
module.check_mode,
)
self.backend = backend
self.module = module
self.content = module.params['content']
if self.content is not None:
self.content = self.content.encode('utf-8')
self.valid_at = module.params['valid_at']
if self.valid_at:
for k, v in self.valid_at.items():
if not isinstance(v, string_types):
self.module.fail_json(
msg='The value for valid_at.{0} must be of type string (got {1})'.format(k, type(v))
)
self.valid_at[k] = crypto_utils.get_relative_time_option(v, 'valid_at.{0}'.format(k))
def generate(self):
# Empty method because crypto_utils.OpenSSLObject wants this
pass
def dump(self):
# Empty method because crypto_utils.OpenSSLObject wants this
pass
@abc.abstractmethod
def _get_signature_algorithm(self):
pass
@abc.abstractmethod
def _get_subject_ordered(self):
pass
@abc.abstractmethod
def _get_issuer_ordered(self):
pass
@abc.abstractmethod
def _get_version(self):
pass
@abc.abstractmethod
def _get_key_usage(self):
pass
@abc.abstractmethod
def _get_extended_key_usage(self):
pass
@abc.abstractmethod
def _get_basic_constraints(self):
pass
@abc.abstractmethod
def _get_ocsp_must_staple(self):
pass
@abc.abstractmethod
def _get_subject_alt_name(self):
pass
@abc.abstractmethod
def _get_not_before(self):
pass
@abc.abstractmethod
def _get_not_after(self):
pass
@abc.abstractmethod
def _get_public_key(self, binary):
pass
@abc.abstractmethod
def _get_subject_key_identifier(self):
pass
@abc.abstractmethod
def _get_authority_key_identifier(self):
pass
@abc.abstractmethod
def _get_serial_number(self):
pass
@abc.abstractmethod
def _get_all_extensions(self):
pass
@abc.abstractmethod
def _get_ocsp_uri(self):
pass
def get_info(self):
result = dict()
self.cert = crypto_utils.load_certificate(self.path, content=self.content, backend=self.backend)
result['signature_algorithm'] = self._get_signature_algorithm()
subject = self._get_subject_ordered()
issuer = self._get_issuer_ordered()
result['subject'] = dict()
for k, v in subject:
result['subject'][k] = v
result['subject_ordered'] = subject
result['issuer'] = dict()
for k, v in issuer:
result['issuer'][k] = v
result['issuer_ordered'] = issuer
result['version'] = self._get_version()
result['key_usage'], result['key_usage_critical'] = self._get_key_usage()
result['extended_key_usage'], result['extended_key_usage_critical'] = self._get_extended_key_usage()
result['basic_constraints'], result['basic_constraints_critical'] = self._get_basic_constraints()
result['ocsp_must_staple'], result['ocsp_must_staple_critical'] = self._get_ocsp_must_staple()
result['subject_alt_name'], result['subject_alt_name_critical'] = self._get_subject_alt_name()
not_before = self._get_not_before()
not_after = self._get_not_after()
result['not_before'] = not_before.strftime(TIMESTAMP_FORMAT)
result['not_after'] = not_after.strftime(TIMESTAMP_FORMAT)
result['expired'] = not_after < datetime.datetime.utcnow()
result['valid_at'] = dict()
if self.valid_at:
for k, v in self.valid_at.items():
result['valid_at'][k] = not_before <= v <= not_after
result['public_key'] = self._get_public_key(binary=False)
pk = self._get_public_key(binary=True)
result['public_key_fingerprints'] = crypto_utils.get_fingerprint_of_bytes(pk) if pk is not None else dict()
if self.backend != 'pyopenssl':
ski = self._get_subject_key_identifier()
if ski is not None:
ski = to_native(binascii.hexlify(ski))
ski = ':'.join([ski[i:i + 2] for i in range(0, len(ski), 2)])
result['subject_key_identifier'] = ski
aki, aci, acsn = self._get_authority_key_identifier()
if aki is not None:
aki = to_native(binascii.hexlify(aki))
aki = ':'.join([aki[i:i + 2] for i in range(0, len(aki), 2)])
result['authority_key_identifier'] = aki
result['authority_cert_issuer'] = aci
result['authority_cert_serial_number'] = acsn
result['serial_number'] = self._get_serial_number()
result['extensions_by_oid'] = self._get_all_extensions()
result['ocsp_uri'] = self._get_ocsp_uri()
return result
class CertificateInfoCryptography(CertificateInfo):
"""Validate the supplied cert, using the cryptography backend"""
def __init__(self, module):
super(CertificateInfoCryptography, self).__init__(module, 'cryptography')
def _get_signature_algorithm(self):
return crypto_utils.cryptography_oid_to_name(self.cert.signature_algorithm_oid)
def _get_subject_ordered(self):
result = []
for attribute in self.cert.subject:
result.append([crypto_utils.cryptography_oid_to_name(attribute.oid), attribute.value])
return result
def _get_issuer_ordered(self):
result = []
for attribute in self.cert.issuer:
result.append([crypto_utils.cryptography_oid_to_name(attribute.oid), attribute.value])
return result
def _get_version(self):
if self.cert.version == x509.Version.v1:
return 1
if self.cert.version == x509.Version.v3:
return 3
return "unknown"
def _get_key_usage(self):
try:
current_key_ext = self.cert.extensions.get_extension_for_class(x509.KeyUsage)
current_key_usage = current_key_ext.value
key_usage = dict(
digital_signature=current_key_usage.digital_signature,
content_commitment=current_key_usage.content_commitment,
key_encipherment=current_key_usage.key_encipherment,
data_encipherment=current_key_usage.data_encipherment,
key_agreement=current_key_usage.key_agreement,
key_cert_sign=current_key_usage.key_cert_sign,
crl_sign=current_key_usage.crl_sign,
encipher_only=False,
decipher_only=False,
)
if key_usage['key_agreement']:
key_usage.update(dict(
encipher_only=current_key_usage.encipher_only,
decipher_only=current_key_usage.decipher_only
))
key_usage_names = dict(
digital_signature='Digital Signature',
content_commitment='Non Repudiation',
key_encipherment='Key Encipherment',
data_encipherment='Data Encipherment',
key_agreement='Key Agreement',
key_cert_sign='Certificate Sign',
crl_sign='CRL Sign',
encipher_only='Encipher Only',
decipher_only='Decipher Only',
)
return sorted([
key_usage_names[name] for name, value in key_usage.items() if value
]), current_key_ext.critical
except cryptography.x509.ExtensionNotFound:
return None, False
def _get_extended_key_usage(self):
try:
ext_keyusage_ext = self.cert.extensions.get_extension_for_class(x509.ExtendedKeyUsage)
return sorted([
crypto_utils.cryptography_oid_to_name(eku) for eku in ext_keyusage_ext.value
]), ext_keyusage_ext.critical
except cryptography.x509.ExtensionNotFound:
return None, False
def _get_basic_constraints(self):
try:
ext_keyusage_ext = self.cert.extensions.get_extension_for_class(x509.BasicConstraints)
result = []
result.append('CA:{0}'.format('TRUE' if ext_keyusage_ext.value.ca else 'FALSE'))
if ext_keyusage_ext.value.path_length is not None:
result.append('pathlen:{0}'.format(ext_keyusage_ext.value.path_length))
return sorted(result), ext_keyusage_ext.critical
except cryptography.x509.ExtensionNotFound:
return None, False
def _get_ocsp_must_staple(self):
try:
try:
# This only works with cryptography >= 2.1
tlsfeature_ext = self.cert.extensions.get_extension_for_class(x509.TLSFeature)
value = cryptography.x509.TLSFeatureType.status_request in tlsfeature_ext.value
except AttributeError as dummy:
# Fallback for cryptography < 2.1
oid = x509.oid.ObjectIdentifier("1.3.6.1.5.5.7.1.24")
tlsfeature_ext = self.cert.extensions.get_extension_for_oid(oid)
value = tlsfeature_ext.value.value == b"\x30\x03\x02\x01\x05"
return value, tlsfeature_ext.critical
except cryptography.x509.ExtensionNotFound:
return None, False
def _get_subject_alt_name(self):
try:
san_ext = self.cert.extensions.get_extension_for_class(x509.SubjectAlternativeName)
result = [crypto_utils.cryptography_decode_name(san) for san in san_ext.value]
return result, san_ext.critical
except cryptography.x509.ExtensionNotFound:
return None, False
def _get_not_before(self):
return self.cert.not_valid_before
def _get_not_after(self):
return self.cert.not_valid_after
def _get_public_key(self, binary):
return self.cert.public_key().public_bytes(
serialization.Encoding.DER if binary else serialization.Encoding.PEM,
serialization.PublicFormat.SubjectPublicKeyInfo
)
def _get_subject_key_identifier(self):
try:
ext = self.cert.extensions.get_extension_for_class(x509.SubjectKeyIdentifier)
return ext.value.digest
except cryptography.x509.ExtensionNotFound:
return None
def _get_authority_key_identifier(self):
try:
ext = self.cert.extensions.get_extension_for_class(x509.AuthorityKeyIdentifier)
issuer = None
if ext.value.authority_cert_issuer is not None:
issuer = [crypto_utils.cryptography_decode_name(san) for san in ext.value.authority_cert_issuer]
return ext.value.key_identifier, issuer, ext.value.authority_cert_serial_number
except cryptography.x509.ExtensionNotFound:
return None, None, None
def _get_serial_number(self):
return self.cert.serial_number
def _get_all_extensions(self):
return crypto_utils.cryptography_get_extensions_from_cert(self.cert)
def _get_ocsp_uri(self):
try:
ext = self.cert.extensions.get_extension_for_class(x509.AuthorityInformationAccess)
for desc in ext.value:
if desc.access_method == x509.oid.AuthorityInformationAccessOID.OCSP:
if isinstance(desc.access_location, x509.UniformResourceIdentifier):
return desc.access_location.value
except x509.ExtensionNotFound as dummy:
pass
return None
class CertificateInfoPyOpenSSL(CertificateInfo):
"""validate the supplied certificate."""
def __init__(self, module):
super(CertificateInfoPyOpenSSL, self).__init__(module, 'pyopenssl')
def _get_signature_algorithm(self):
return to_text(self.cert.get_signature_algorithm())
def __get_name(self, name):
result = []
for sub in name.get_components():
result.append([crypto_utils.pyopenssl_normalize_name(sub[0]), to_text(sub[1])])
return result
def _get_subject_ordered(self):
return self.__get_name(self.cert.get_subject())
def _get_issuer_ordered(self):
return self.__get_name(self.cert.get_issuer())
def _get_version(self):
# Version numbers in certs are off by one:
# v1: 0, v2: 1, v3: 2 ...
return self.cert.get_version() + 1
def _get_extension(self, short_name):
for extension_idx in range(0, self.cert.get_extension_count()):
extension = self.cert.get_extension(extension_idx)
if extension.get_short_name() == short_name:
result = [
crypto_utils.pyopenssl_normalize_name(usage.strip()) for usage in to_text(extension, errors='surrogate_or_strict').split(',')
]
return sorted(result), bool(extension.get_critical())
return None, False
def _get_key_usage(self):
return self._get_extension(b'keyUsage')
def _get_extended_key_usage(self):
return self._get_extension(b'extendedKeyUsage')
def _get_basic_constraints(self):
return self._get_extension(b'basicConstraints')
def _get_ocsp_must_staple(self):
extensions = [self.cert.get_extension(i) for i in range(0, self.cert.get_extension_count())]
oms_ext = [
ext for ext in extensions
if to_bytes(ext.get_short_name()) == OPENSSL_MUST_STAPLE_NAME and to_bytes(ext) == OPENSSL_MUST_STAPLE_VALUE
]
if OpenSSL.SSL.OPENSSL_VERSION_NUMBER < 0x10100000:
# Older versions of libssl don't know about OCSP Must Staple
oms_ext.extend([ext for ext in extensions if ext.get_short_name() == b'UNDEF' and ext.get_data() == b'\x30\x03\x02\x01\x05'])
if oms_ext:
return True, bool(oms_ext[0].get_critical())
else:
return None, False
def _normalize_san(self, san):
if san.startswith('IP Address:'):
san = 'IP:' + san[len('IP Address:'):]
if san.startswith('IP:'):
ip = compat_ipaddress.ip_address(san[3:])
san = 'IP:{0}'.format(ip.compressed)
return san
def _get_subject_alt_name(self):
for extension_idx in range(0, self.cert.get_extension_count()):
extension = self.cert.get_extension(extension_idx)
if extension.get_short_name() == b'subjectAltName':
result = [self._normalize_san(altname.strip()) for altname in
to_text(extension, errors='surrogate_or_strict').split(', ')]
return result, bool(extension.get_critical())
return None, False
def _get_not_before(self):
time_string = to_native(self.cert.get_notBefore())
return datetime.datetime.strptime(time_string, "%Y%m%d%H%M%SZ")
def _get_not_after(self):
time_string = to_native(self.cert.get_notAfter())
return datetime.datetime.strptime(time_string, "%Y%m%d%H%M%SZ")
def _get_public_key(self, binary):
try:
return crypto.dump_publickey(
crypto.FILETYPE_ASN1 if binary else crypto.FILETYPE_PEM,
self.cert.get_pubkey()
)
except AttributeError:
try:
# pyOpenSSL < 16.0:
bio = crypto._new_mem_buf()
if binary:
rc = crypto._lib.i2d_PUBKEY_bio(bio, self.cert.get_pubkey()._pkey)
else:
rc = crypto._lib.PEM_write_bio_PUBKEY(bio, self.cert.get_pubkey()._pkey)
if rc != 1:
crypto._raise_current_error()
return crypto._bio_to_string(bio)
except AttributeError:
self.module.warn('Your pyOpenSSL version does not support dumping public keys. '
'Please upgrade to version 16.0 or newer, or use the cryptography backend.')
def _get_subject_key_identifier(self):
# Won't be implemented
return None
def _get_authority_key_identifier(self):
# Won't be implemented
return None, None, None
def _get_serial_number(self):
return self.cert.get_serial_number()
def _get_all_extensions(self):
return crypto_utils.pyopenssl_get_extensions_from_cert(self.cert)
def _get_ocsp_uri(self):
for i in range(self.cert.get_extension_count()):
ext = self.cert.get_extension(i)
if ext.get_short_name() == b'authorityInfoAccess':
v = str(ext)
m = re.search('^OCSP - URI:(.*)$', v, flags=re.MULTILINE)
if m:
return m.group(1)
return None
def main():
module = AnsibleModule(
argument_spec=dict(
path=dict(type='path'),
content=dict(type='str'),
valid_at=dict(type='dict'),
select_crypto_backend=dict(type='str', default='auto', choices=['auto', 'cryptography', 'pyopenssl']),
),
required_one_of=(
['path', 'content'],
),
mutually_exclusive=(
['path', 'content'],
),
supports_check_mode=True,
)
try:
if module.params['path'] is not None:
base_dir = os.path.dirname(module.params['path']) or '.'
if not os.path.isdir(base_dir):
module.fail_json(
name=base_dir,
msg='The directory %s does not exist or the file is not a directory' % base_dir
)
backend = module.params['select_crypto_backend']
if backend == 'auto':
# Detect what backend we can use
can_use_cryptography = CRYPTOGRAPHY_FOUND and CRYPTOGRAPHY_VERSION >= LooseVersion(MINIMAL_CRYPTOGRAPHY_VERSION)
can_use_pyopenssl = PYOPENSSL_FOUND and PYOPENSSL_VERSION >= LooseVersion(MINIMAL_PYOPENSSL_VERSION)
# If cryptography is available we'll use it
if can_use_cryptography:
backend = 'cryptography'
elif can_use_pyopenssl:
backend = 'pyopenssl'
# Fail if no backend has been found
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))
if backend == 'pyopenssl':
if not PYOPENSSL_FOUND:
module.fail_json(msg=missing_required_lib('pyOpenSSL >= {0}'.format(MINIMAL_PYOPENSSL_VERSION)),
exception=PYOPENSSL_IMP_ERR)
try:
getattr(crypto.X509Req, 'get_extensions')
except AttributeError:
module.fail_json(msg='You need to have PyOpenSSL>=0.15')
module.deprecate('The module is using the PyOpenSSL backend. This backend has been deprecated',
version='2.13', collection_name='ansible.builtin')
certificate = CertificateInfoPyOpenSSL(module)
elif backend == 'cryptography':
if not CRYPTOGRAPHY_FOUND:
module.fail_json(msg=missing_required_lib('cryptography >= {0}'.format(MINIMAL_CRYPTOGRAPHY_VERSION)),
exception=CRYPTOGRAPHY_IMP_ERR)
certificate = CertificateInfoCryptography(module)
result = certificate.get_info()
module.exit_json(**result)
except crypto_utils.OpenSSLObjectError as exc:
module.fail_json(msg=to_native(exc))
if __name__ == "__main__":
main()

1161
test/support/integration/plugins/modules/openssl_csr.py
View file

File diff suppressed because it is too large Load diff

944
test/support/integration/plugins/modules/openssl_privatekey.py
View file

@ -1,944 +0,0 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
# Copyright: (c) 2016, 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.1',
'status': ['preview'],
'supported_by': 'community'}
DOCUMENTATION = r'''
---
module: openssl_privatekey
version_added: "2.3"
short_description: Generate OpenSSL private keys
description:
- This module allows one to (re)generate OpenSSL private keys.
- One can generate L(RSA,https://en.wikipedia.org/wiki/RSA_%28cryptosystem%29),
L(DSA,https://en.wikipedia.org/wiki/Digital_Signature_Algorithm),
L(ECC,https://en.wikipedia.org/wiki/Elliptic-curve_cryptography) or
L(EdDSA,https://en.wikipedia.org/wiki/EdDSA) private keys.
- Keys are generated in PEM format.
- "Please note that the module regenerates private keys if they don't match
the module's options. In particular, if you provide another passphrase
(or specify none), change the keysize, etc., the private key will be
regenerated. If you are concerned that this could **overwrite your private key**,
consider using the I(backup) option."
- 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 Ansible 2.13."
requirements:
- Either cryptography >= 1.2.3 (older versions might work as well)
- Or pyOpenSSL
author:
- Yanis Guenane (@Spredzy)
- Felix Fontein (@felixfontein)
options:
state:
description:
- Whether the private key should exist or not, taking action if the state is different from what is stated.
type: str
default: present
choices: [ absent, present ]
size:
description:
- Size (in bits) of the TLS/SSL key to generate.
type: int
default: 4096
type:
description:
- The algorithm used to generate the TLS/SSL private key.
- Note that C(ECC), C(X25519), C(X448), C(Ed25519) and C(Ed448) require the C(cryptography) backend.
C(X25519) needs cryptography 2.5 or newer, while C(X448), C(Ed25519) and C(Ed448) require
cryptography 2.6 or newer. For C(ECC), the minimal cryptography version required depends on the
I(curve) option.
type: str
default: RSA
choices: [ DSA, ECC, Ed25519, Ed448, RSA, X25519, X448 ]
curve:
description:
- Note that not all curves are supported by all versions of C(cryptography).
- For maximal interoperability, C(secp384r1) or C(secp256r1) should be used.
- We use the curve names as defined in the
L(IANA registry for TLS,https://www.iana.org/assignments/tls-parameters/tls-parameters.xhtml#tls-parameters-8).
type: str
choices:
- secp384r1
- secp521r1
- secp224r1
- secp192r1
- secp256r1
- secp256k1
- brainpoolP256r1
- brainpoolP384r1
- brainpoolP512r1
- sect571k1
- sect409k1
- sect283k1
- sect233k1
- sect163k1
- sect571r1
- sect409r1
- sect283r1
- sect233r1
- sect163r2
version_added: "2.8"
force:
description:
- Should the key be regenerated even if it already exists.
type: bool
default: no
path:
description:
- Name of the file in which the generated TLS/SSL private key will be written. It will have 0600 mode.
type: path
required: true
passphrase:
description:
- The passphrase for the private key.
type: str
version_added: "2.4"
cipher:
description:
- The cipher to encrypt the private key. (Valid values can be found by
running `openssl list -cipher-algorithms` or `openssl list-cipher-algorithms`,
depending on your OpenSSL version.)
- When using the C(cryptography) backend, use C(auto).
type: str
version_added: "2.4"
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.
- Please note that the C(pyopenssl) backend has been deprecated in Ansible 2.9, and will be removed in Ansible 2.13.
From that point on, only the C(cryptography) backend will be available.
type: str
default: auto
choices: [ auto, cryptography, pyopenssl ]
version_added: "2.8"
format:
description:
- Determines which format the private key is written in. By default, PKCS1 (traditional OpenSSL format)
is used for all keys which support it. Please note that not every key can be exported in any format.
- The value C(auto) selects a fromat based on the key format. The value C(auto_ignore) does the same,
but for existing private key files, it will not force a regenerate when its format is not the automatically
selected one for generation.
- Note that if the format for an existing private key mismatches, the key is *regenerated* by default.
To change this behavior, use the I(format_mismatch) option.
- The I(format) option is only supported by the C(cryptography) backend. The C(pyopenssl) backend will
fail if a value different from C(auto_ignore) is used.
type: str
default: auto_ignore
choices: [ pkcs1, pkcs8, raw, auto, auto_ignore ]
version_added: "2.10"
format_mismatch:
description:
- Determines behavior of the module if the format of a private key does not match the expected format, but all
other parameters are as expected.
- If set to C(regenerate) (default), generates a new private key.
- If set to C(convert), the key will be converted to the new format instead.
- Only supported by the C(cryptography) backend.
type: str
default: regenerate
choices: [ regenerate, convert ]
version_added: "2.10"
backup:
description:
- Create a backup file including a timestamp so you can get
the original private key back if you overwrote it with a new one by accident.
type: bool
default: no
version_added: "2.8"
return_content:
description:
- If set to C(yes), will return the (current or generated) private key's content as I(privatekey).
- Note that especially if the private key is not encrypted, you have to make sure that the returned
value is treated appropriately and not accidentally written to logs etc.! Use with care!
type: bool
default: no
version_added: "2.10"
regenerate:
description:
- Allows to configure in which situations the module is allowed to regenerate private keys.
The module will always generate a new key if the destination file does not exist.
- By default, the key will be regenerated when it doesn't match the module's options,
except when the key cannot be read or the passphrase does not match. Please note that
this B(changed) for Ansible 2.10. For Ansible 2.9, the behavior was as if C(full_idempotence)
is specified.
- If set to C(never), the module will fail if the key cannot be read or the passphrase
isn't matching, and will never regenerate an existing key.
- If set to C(fail), the module will fail if the key does not correspond to the module's
options.
- If set to C(partial_idempotence), the key will be regenerated if it does not conform to
the module's options. The key is B(not) regenerated if it cannot be read (broken file),
the key is protected by an unknown passphrase, or when they key is not protected by a
passphrase, but a passphrase is specified.
- If set to C(full_idempotence), the key will be regenerated if it does not conform to the
module's options. This is also the case if the key cannot be read (broken file), the key
is protected by an unknown passphrase, or when they key is not protected by a passphrase,
but a passphrase is specified. Make sure you have a B(backup) when using this option!
- If set to C(always), the module will always regenerate the key. This is equivalent to
setting I(force) to C(yes).
- Note that if I(format_mismatch) is set to C(convert) and everything matches except the
format, the key will always be converted, except if I(regenerate) is set to C(always).
type: str
choices:
- never
- fail
- partial_idempotence
- full_idempotence
- always
default: full_idempotence
version_added: '2.10'
extends_documentation_fragment:
- files
seealso:
- module: openssl_certificate
- module: openssl_csr
- module: openssl_dhparam
- module: openssl_pkcs12
- module: openssl_publickey
'''
EXAMPLES = r'''
- name: Generate an OpenSSL private key with the default values (4096 bits, RSA)
openssl_privatekey:
path: /etc/ssl/private/ansible.com.pem
- name: Generate an OpenSSL private key with the default values (4096 bits, RSA) and a passphrase
openssl_privatekey:
path: /etc/ssl/private/ansible.com.pem
passphrase: ansible
cipher: aes256
- name: Generate an OpenSSL private key with a different size (2048 bits)
openssl_privatekey:
path: /etc/ssl/private/ansible.com.pem
size: 2048
- name: Force regenerate an OpenSSL private key if it already exists
openssl_privatekey:
path: /etc/ssl/private/ansible.com.pem
force: yes
- name: Generate an OpenSSL private key with a different algorithm (DSA)
openssl_privatekey:
path: /etc/ssl/private/ansible.com.pem
type: DSA
'''
RETURN = r'''
size:
description: Size (in bits) of the TLS/SSL private key.
returned: changed or success
type: int
sample: 4096
type:
description: Algorithm used to generate the TLS/SSL private key.
returned: changed or success
type: str
sample: RSA
curve:
description: Elliptic curve used to generate the TLS/SSL private key.
returned: changed or success, and I(type) is C(ECC)
type: str
sample: secp256r1
filename:
description: Path to the generated TLS/SSL private key file.
returned: changed or success
type: str
sample: /etc/ssl/private/ansible.com.pem
fingerprint:
description:
- The fingerprint of the public key. Fingerprint will be generated for each C(hashlib.algorithms) available.
- The PyOpenSSL backend requires PyOpenSSL >= 16.0 for meaningful output.
returned: changed or success
type: dict
sample:
md5: "84:75:71:72:8d:04:b5:6c:4d:37:6d:66:83:f5:4c:29"
sha1: "51:cc:7c:68:5d:eb:41:43:88:7e:1a:ae:c7:f8:24:72:ee:71:f6:10"
sha224: "b1:19:a6:6c:14:ac:33:1d:ed:18:50:d3:06:5c:b2:32:91:f1:f1:52:8c:cb:d5:75:e9:f5:9b:46"
sha256: "41:ab:c7:cb:d5:5f:30:60:46:99:ac:d4:00:70:cf:a1:76:4f:24:5d:10:24:57:5d:51:6e:09:97:df:2f:de:c7"
sha384: "85:39:50:4e:de:d9:19:33:40:70:ae:10:ab:59:24:19:51:c3:a2:e4:0b:1c:b1:6e:dd:b3:0c:d9:9e:6a:46:af:da:18:f8:ef:ae:2e:c0:9a:75:2c:9b:b3:0f:3a:5f:3d"
sha512: "fd:ed:5e:39:48:5f:9f:fe:7f:25:06:3f:79:08:cd:ee:a5:e7:b3:3d:13:82:87:1f:84:e1:f5:c7:28:77:53:94:86:56:38:69:f0:d9:35:22:01:1e:a6:60:...:0f:9b"
backup_file:
description: Name of backup file created.
returned: changed and if I(backup) is C(yes)
type: str
sample: /path/to/privatekey.pem.2019-03-09@11:22~
privatekey:
description:
- The (current or generated) private key's content.
- Will be Base64-encoded if the key is in raw format.
returned: if I(state) is C(present) and I(return_content) is C(yes)
type: str
version_added: "2.10"
'''
import abc
import base64
import os
import traceback
from distutils.version import LooseVersion
MINIMAL_PYOPENSSL_VERSION = '0.6'
MINIMAL_CRYPTOGRAPHY_VERSION = '1.2.3'
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.exceptions
import cryptography.hazmat.backends
import cryptography.hazmat.primitives.serialization
import cryptography.hazmat.primitives.asymmetric.rsa
import cryptography.hazmat.primitives.asymmetric.dsa
import cryptography.hazmat.primitives.asymmetric.ec
import cryptography.hazmat.primitives.asymmetric.utils
CRYPTOGRAPHY_VERSION = LooseVersion(cryptography.__version__)
except ImportError:
CRYPTOGRAPHY_IMP_ERR = traceback.format_exc()
CRYPTOGRAPHY_FOUND = False
else:
CRYPTOGRAPHY_FOUND = True
from ansible.module_utils.crypto import (
CRYPTOGRAPHY_HAS_X25519,
CRYPTOGRAPHY_HAS_X25519_FULL,
CRYPTOGRAPHY_HAS_X448,
CRYPTOGRAPHY_HAS_ED25519,
CRYPTOGRAPHY_HAS_ED448,
)
from ansible.module_utils import crypto as crypto_utils
from ansible.module_utils._text import to_native, to_bytes
from ansible.module_utils.basic import AnsibleModule, missing_required_lib
class PrivateKeyError(crypto_utils.OpenSSLObjectError):
pass
class PrivateKeyBase(crypto_utils.OpenSSLObject):
def __init__(self, module):
super(PrivateKeyBase, self).__init__(
module.params['path'],
module.params['state'],
module.params['force'],
module.check_mode
)
self.size = module.params['size']
self.passphrase = module.params['passphrase']
self.cipher = module.params['cipher']
self.privatekey = None
self.fingerprint = {}
self.format = module.params['format']
self.format_mismatch = module.params['format_mismatch']
self.privatekey_bytes = None
self.return_content = module.params['return_content']
self.regenerate = module.params['regenerate']
if self.regenerate == 'always':
self.force = True
self.backup = module.params['backup']
self.backup_file = None
if module.params['mode'] is None:
module.params['mode'] = '0600'
@abc.abstractmethod
def _generate_private_key(self):
"""(Re-)Generate private key."""
pass
@abc.abstractmethod
def _ensure_private_key_loaded(self):
"""Make sure that the private key has been loaded."""
pass
@abc.abstractmethod
def _get_private_key_data(self):
"""Return bytes for self.privatekey"""
pass
@abc.abstractmethod
def _get_fingerprint(self):
pass
def generate(self, module):
"""Generate a keypair."""
if not self.check(module, perms_required=False, ignore_conversion=True) or self.force:
# Regenerate
if self.backup:
self.backup_file = module.backup_local(self.path)
self._generate_private_key()
privatekey_data = self._get_private_key_data()
if self.return_content:
self.privatekey_bytes = privatekey_data
crypto_utils.write_file(module, privatekey_data, 0o600)
self.changed = True
elif not self.check(module, perms_required=False, ignore_conversion=False):
# Convert
if self.backup:
self.backup_file = module.backup_local(self.path)
self._ensure_private_key_loaded()
privatekey_data = self._get_private_key_data()
if self.return_content:
self.privatekey_bytes = privatekey_data
crypto_utils.write_file(module, privatekey_data, 0o600)
self.changed = True
self.fingerprint = self._get_fingerprint()
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, module):
if self.backup:
self.backup_file = module.backup_local(self.path)
super(PrivateKeyBase, self).remove(module)
@abc.abstractmethod
def _check_passphrase(self):
pass
@abc.abstractmethod
def _check_size_and_type(self):
pass
@abc.abstractmethod
def _check_format(self):
pass
def check(self, module, perms_required=True, ignore_conversion=True):
"""Ensure the resource is in its desired state."""
state_and_perms = super(PrivateKeyBase, self).check(module, perms_required=False)
if not state_and_perms:
# key does not exist
return False
if not self._check_passphrase():
if self.regenerate in ('full_idempotence', 'always'):
return False
module.fail_json(msg='Unable to read the key. The key is protected with a another passphrase / no 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.regenerate != 'never':
if not self._check_size_and_type():
if self.regenerate in ('partial_idempotence', 'full_idempotence', 'always'):
return False
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`.')
if not self._check_format():
# During conversion step, convert if format does not match and format_mismatch == 'convert'
if not ignore_conversion and self.format_mismatch == 'convert':
return False
# During generation step, regenerate if format does not match and format_mismatch == 'regenerate'
if ignore_conversion and self.format_mismatch == 'regenerate' and self.regenerate != 'never':
if not ignore_conversion or self.regenerate in ('partial_idempotence', 'full_idempotence', 'always'):
return False
module.fail_json(msg='Key has wrong format.'
' Will not proceed. To force regeneration, call the module with `generate`'
' set to `partial_idempotence`, `full_idempotence` or `always`, or with `force=yes`.'
' To convert the key, set `format_mismatch` to `convert`.')
# check whether permissions are correct (in case that needs to be checked)
return not perms_required or super(PrivateKeyBase, self).check(module, perms_required=perms_required)
def dump(self):
"""Serialize the object into a dictionary."""
result = {
'size': self.size,
'filename': self.path,
'changed': self.changed,
'fingerprint': self.fingerprint,
}
if self.backup_file:
result['backup_file'] = self.backup_file
if self.return_content:
if self.privatekey_bytes is None:
self.privatekey_bytes = crypto_utils.load_file_if_exists(self.path, ignore_errors=True)
if self.privatekey_bytes:
if crypto_utils.identify_private_key_format(self.privatekey_bytes) == 'raw':
result['privatekey'] = base64.b64encode(self.privatekey_bytes)
else:
result['privatekey'] = self.privatekey_bytes.decode('utf-8')
else:
result['privatekey'] = None
return result
# Implementation with using pyOpenSSL
class PrivateKeyPyOpenSSL(PrivateKeyBase):
def __init__(self, module):
super(PrivateKeyPyOpenSSL, self).__init__(module)
if module.params['type'] == 'RSA':
self.type = crypto.TYPE_RSA
elif module.params['type'] == 'DSA':
self.type = crypto.TYPE_DSA
else:
module.fail_json(msg="PyOpenSSL backend only supports RSA and DSA keys.")
if self.format != 'auto_ignore':
module.fail_json(msg="PyOpenSSL backend only supports auto_ignore format.")
def _generate_private_key(self):
"""(Re-)Generate private key."""
self.privatekey = crypto.PKey()
try:
self.privatekey.generate_key(self.type, self.size)
except (TypeError, ValueError) as exc:
raise PrivateKeyError(exc)
def _ensure_private_key_loaded(self):
"""Make sure that the private key has been loaded."""
if self.privatekey is None:
try:
self.privatekey = privatekey = crypto_utils.load_privatekey(self.path, self.passphrase)
except crypto_utils.OpenSSLBadPassphraseError as exc:
raise PrivateKeyError(exc)
def _get_private_key_data(self):
"""Return bytes for self.privatekey"""
if self.cipher and self.passphrase:
return crypto.dump_privatekey(crypto.FILETYPE_PEM, self.privatekey,
self.cipher, to_bytes(self.passphrase))
else:
return crypto.dump_privatekey(crypto.FILETYPE_PEM, self.privatekey)
def _get_fingerprint(self):
return crypto_utils.get_fingerprint(self.path, self.passphrase)
def _check_passphrase(self):
try:
crypto_utils.load_privatekey(self.path, self.passphrase)
return True
except Exception as dummy:
return False
def _check_size_and_type(self):
def _check_size(privatekey):
return self.size == privatekey.bits()
def _check_type(privatekey):
return self.type == privatekey.type()
self._ensure_private_key_loaded()
return _check_size(self.privatekey) and _check_type(self.privatekey)
def _check_format(self):
# Not supported by this backend
return True
def dump(self):
"""Serialize the object into a dictionary."""
result = super(PrivateKeyPyOpenSSL, self).dump()
if self.type == crypto.TYPE_RSA:
result['type'] = 'RSA'
else:
result['type'] = 'DSA'
return result
# Implementation with using cryptography
class PrivateKeyCryptography(PrivateKeyBase):
def _get_ec_class(self, ectype):
ecclass = cryptography.hazmat.primitives.asymmetric.ec.__dict__.get(ectype)
if ecclass is None:
self.module.fail_json(msg='Your cryptography version does not support {0}'.format(ectype))
return ecclass
def _add_curve(self, name, ectype, deprecated=False):
def create(size):
ecclass = self._get_ec_class(ectype)
return ecclass()
def verify(privatekey):
ecclass = self._get_ec_class(ectype)
return isinstance(privatekey.private_numbers().public_numbers.curve, ecclass)
self.curves[name] = {
'create': create,
'verify': verify,
'deprecated': deprecated,
}
def __init__(self, module):
super(PrivateKeyCryptography, self).__init__(module)
self.curves = dict()
self._add_curve('secp384r1', 'SECP384R1')
self._add_curve('secp521r1', 'SECP521R1')
self._add_curve('secp224r1', 'SECP224R1')
self._add_curve('secp192r1', 'SECP192R1')
self._add_curve('secp256r1', 'SECP256R1')
self._add_curve('secp256k1', 'SECP256K1')
self._add_curve('brainpoolP256r1', 'BrainpoolP256R1', deprecated=True)
self._add_curve('brainpoolP384r1', 'BrainpoolP384R1', deprecated=True)
self._add_curve('brainpoolP512r1', 'BrainpoolP512R1', deprecated=True)
self._add_curve('sect571k1', 'SECT571K1', deprecated=True)
self._add_curve('sect409k1', 'SECT409K1', deprecated=True)
self._add_curve('sect283k1', 'SECT283K1', deprecated=True)
self._add_curve('sect233k1', 'SECT233K1', deprecated=True)
self._add_curve('sect163k1', 'SECT163K1', deprecated=True)
self._add_curve('sect571r1', 'SECT571R1', deprecated=True)
self._add_curve('sect409r1', 'SECT409R1', deprecated=True)
self._add_curve('sect283r1', 'SECT283R1', deprecated=True)
self._add_curve('sect233r1', 'SECT233R1', deprecated=True)
self._add_curve('sect163r2', 'SECT163R2', deprecated=True)
self.module = module
self.cryptography_backend = cryptography.hazmat.backends.default_backend()
self.type = module.params['type']
self.curve = module.params['curve']
if not CRYPTOGRAPHY_HAS_X25519 and self.type == 'X25519':
self.module.fail_json(msg='Your cryptography version does not support X25519')
if not CRYPTOGRAPHY_HAS_X25519_FULL and self.type == 'X25519':
self.module.fail_json(msg='Your cryptography version does not support X25519 serialization')
if not CRYPTOGRAPHY_HAS_X448 and self.type == 'X448':
self.module.fail_json(msg='Your cryptography version does not support X448')
if not CRYPTOGRAPHY_HAS_ED25519 and self.type == 'Ed25519':
self.module.fail_json(msg='Your cryptography version does not support Ed25519')
if not CRYPTOGRAPHY_HAS_ED448 and self.type == 'Ed448':
self.module.fail_json(msg='Your cryptography version does not support Ed448')
def _get_wanted_format(self):
if self.format not in ('auto', 'auto_ignore'):
return self.format
if self.type in ('X25519', 'X448', 'Ed25519', 'Ed448'):
return 'pkcs8'
else:
return 'pkcs1'
def _generate_private_key(self):
"""(Re-)Generate private key."""
try:
if self.type == 'RSA':
self.privatekey = cryptography.hazmat.primitives.asymmetric.rsa.generate_private_key(
public_exponent=65537, # OpenSSL always uses this
key_size=self.size,
backend=self.cryptography_backend
)
if self.type == 'DSA':
self.privatekey = cryptography.hazmat.primitives.asymmetric.dsa.generate_private_key(
key_size=self.size,
backend=self.cryptography_backend
)
if CRYPTOGRAPHY_HAS_X25519_FULL and self.type == 'X25519':
self.privatekey = cryptography.hazmat.primitives.asymmetric.x25519.X25519PrivateKey.generate()
if CRYPTOGRAPHY_HAS_X448 and self.type == 'X448':
self.privatekey = cryptography.hazmat.primitives.asymmetric.x448.X448PrivateKey.generate()
if CRYPTOGRAPHY_HAS_ED25519 and self.type == 'Ed25519':
self.privatekey = cryptography.hazmat.primitives.asymmetric.ed25519.Ed25519PrivateKey.generate()
if CRYPTOGRAPHY_HAS_ED448 and self.type == 'Ed448':
self.privatekey = cryptography.hazmat.primitives.asymmetric.ed448.Ed448PrivateKey.generate()
if self.type == 'ECC' and self.curve in self.curves:
if self.curves[self.curve]['deprecated']:
self.module.warn('Elliptic curves of type {0} should not be used for new keys!'.format(self.curve))
self.privatekey = cryptography.hazmat.primitives.asymmetric.ec.generate_private_key(
curve=self.curves[self.curve]['create'](self.size),
backend=self.cryptography_backend
)
except cryptography.exceptions.UnsupportedAlgorithm as dummy:
self.module.fail_json(msg='Cryptography backend does not support the algorithm required for {0}'.format(self.type))
def _ensure_private_key_loaded(self):
"""Make sure that the private key has been loaded."""
if self.privatekey is None:
self.privatekey = self._load_privatekey()
def _get_private_key_data(self):
"""Return bytes for self.privatekey"""
# Select export format and encoding
try:
export_format = self._get_wanted_format()
export_encoding = cryptography.hazmat.primitives.serialization.Encoding.PEM
if export_format == 'pkcs1':
# "TraditionalOpenSSL" format is PKCS1
export_format = cryptography.hazmat.primitives.serialization.PrivateFormat.TraditionalOpenSSL
elif export_format == 'pkcs8':
export_format = cryptography.hazmat.primitives.serialization.PrivateFormat.PKCS8
elif export_format == 'raw':
export_format = cryptography.hazmat.primitives.serialization.PrivateFormat.Raw
export_encoding = cryptography.hazmat.primitives.serialization.Encoding.Raw
except AttributeError:
self.module.fail_json(msg='Cryptography backend does not support the selected output format "{0}"'.format(self.format))
# Select key encryption
encryption_algorithm = cryptography.hazmat.primitives.serialization.NoEncryption()
if self.cipher and self.passphrase:
if self.cipher == 'auto':
encryption_algorithm = cryptography.hazmat.primitives.serialization.BestAvailableEncryption(to_bytes(self.passphrase))
else:
self.module.fail_json(msg='Cryptography backend can only use "auto" for cipher option.')
# Serialize key
try:
return self.privatekey.private_bytes(
encoding=export_encoding,
format=export_format,
encryption_algorithm=encryption_algorithm
)
except ValueError as dummy:
self.module.fail_json(
msg='Cryptography backend cannot serialize the private key in the required format "{0}"'.format(self.format)
)
except Exception as dummy:
self.module.fail_json(
msg='Error while serializing the private key in the required format "{0}"'.format(self.format),
exception=traceback.format_exc()
)
def _load_privatekey(self):
try:
# Read bytes
with open(self.path, 'rb') as f:
data = f.read()
# Interpret bytes depending on format.
format = crypto_utils.identify_private_key_format(data)
if format == 'raw':
if len(data) == 56 and CRYPTOGRAPHY_HAS_X448:
return cryptography.hazmat.primitives.asymmetric.x448.X448PrivateKey.from_private_bytes(data)
if len(data) == 57 and CRYPTOGRAPHY_HAS_ED448:
return cryptography.hazmat.primitives.asymmetric.ed448.Ed448PrivateKey.from_private_bytes(data)
if len(data) == 32:
if CRYPTOGRAPHY_HAS_X25519 and (self.type == 'X25519' or not CRYPTOGRAPHY_HAS_ED25519):
return cryptography.hazmat.primitives.asymmetric.x25519.X25519PrivateKey.from_private_bytes(data)
if CRYPTOGRAPHY_HAS_ED25519 and (self.type == 'Ed25519' or not CRYPTOGRAPHY_HAS_X25519):
return cryptography.hazmat.primitives.asymmetric.ed25519.Ed25519PrivateKey.from_private_bytes(data)
if CRYPTOGRAPHY_HAS_X25519 and CRYPTOGRAPHY_HAS_ED25519:
try:
return cryptography.hazmat.primitives.asymmetric.x25519.X25519PrivateKey.from_private_bytes(data)
except Exception:
return cryptography.hazmat.primitives.asymmetric.ed25519.Ed25519PrivateKey.from_private_bytes(data)
raise PrivateKeyError('Cannot load raw key')
else:
return cryptography.hazmat.primitives.serialization.load_pem_private_key(
data,
None if self.passphrase is None else to_bytes(self.passphrase),
backend=self.cryptography_backend
)
except Exception as e:
raise PrivateKeyError(e)
def _get_fingerprint(self):
# Get bytes of public key
private_key = self._load_privatekey()
public_key = private_key.public_key()
public_key_bytes = public_key.public_bytes(
cryptography.hazmat.primitives.serialization.Encoding.DER,
cryptography.hazmat.primitives.serialization.PublicFormat.SubjectPublicKeyInfo
)
# Get fingerprints of public_key_bytes
return crypto_utils.get_fingerprint_of_bytes(public_key_bytes)
def _check_passphrase(self):
try:
with open(self.path, 'rb') as f:
data = f.read()
format = crypto_utils.identify_private_key_format(data)
if format == 'raw':
# Raw keys cannot be encrypted. To avoid incompatibilities, we try to
# actually load the key (and return False when this fails).
self._load_privatekey()
# Loading the key succeeded. Only return True when no passphrase was
# provided.
return self.passphrase is None
else:
return cryptography.hazmat.primitives.serialization.load_pem_private_key(
data,
None if self.passphrase is None else to_bytes(self.passphrase),
backend=self.cryptography_backend
)
except Exception as dummy:
return False
def _check_size_and_type(self):
self._ensure_private_key_loaded()
if isinstance(self.privatekey, cryptography.hazmat.primitives.asymmetric.rsa.RSAPrivateKey):
return self.type == 'RSA' and self.size == self.privatekey.key_size
if isinstance(self.privatekey, cryptography.hazmat.primitives.asymmetric.dsa.DSAPrivateKey):
return self.type == 'DSA' and self.size == self.privatekey.key_size
if CRYPTOGRAPHY_HAS_X25519 and isinstance(self.privatekey, cryptography.hazmat.primitives.asymmetric.x25519.X25519PrivateKey):
return self.type == 'X25519'
if CRYPTOGRAPHY_HAS_X448 and isinstance(self.privatekey, cryptography.hazmat.primitives.asymmetric.x448.X448PrivateKey):
return self.type == 'X448'
if CRYPTOGRAPHY_HAS_ED25519 and isinstance(self.privatekey, cryptography.hazmat.primitives.asymmetric.ed25519.Ed25519PrivateKey):
return self.type == 'Ed25519'
if CRYPTOGRAPHY_HAS_ED448 and isinstance(self.privatekey, cryptography.hazmat.primitives.asymmetric.ed448.Ed448PrivateKey):
return self.type == 'Ed448'
if isinstance(self.privatekey, cryptography.hazmat.primitives.asymmetric.ec.EllipticCurvePrivateKey):
if self.type != 'ECC':
return False
if self.curve not in self.curves:
return False
return self.curves[self.curve]['verify'](self.privatekey)
return False
def _check_format(self):
if self.format == 'auto_ignore':
return True
try:
with open(self.path, 'rb') as f:
content = f.read()
format = crypto_utils.identify_private_key_format(content)
return format == self._get_wanted_format()
except Exception as dummy:
return False
def dump(self):
"""Serialize the object into a dictionary."""
result = super(PrivateKeyCryptography, self).dump()
result['type'] = self.type
if self.type == 'ECC':
result['curve'] = self.curve
return result
def main():
module = AnsibleModule(
argument_spec=dict(
state=dict(type='str', default='present', choices=['present', 'absent']),
size=dict(type='int', default=4096),
type=dict(type='str', default='RSA', choices=[
'DSA', 'ECC', 'Ed25519', 'Ed448', 'RSA', 'X25519', 'X448'
]),
curve=dict(type='str', choices=[
'secp384r1', 'secp521r1', 'secp224r1', 'secp192r1', 'secp256r1',
'secp256k1', 'brainpoolP256r1', 'brainpoolP384r1', 'brainpoolP512r1',
'sect571k1', 'sect409k1', 'sect283k1', 'sect233k1', 'sect163k1',
'sect571r1', 'sect409r1', 'sect283r1', 'sect233r1', 'sect163r2',
]),
force=dict(type='bool', default=False),
path=dict(type='path', required=True),
passphrase=dict(type='str', no_log=True),
cipher=dict(type='str'),
backup=dict(type='bool', default=False),
format=dict(type='str', default='auto_ignore', choices=['pkcs1', 'pkcs8', 'raw', 'auto', 'auto_ignore']),
format_mismatch=dict(type='str', default='regenerate', choices=['regenerate', 'convert']),
select_crypto_backend=dict(type='str', choices=['auto', 'pyopenssl', 'cryptography'], default='auto'),
return_content=dict(type='bool', default=False),
regenerate=dict(
type='str',
default='full_idempotence',
choices=['never', 'fail', 'partial_idempotence', 'full_idempotence', 'always']
),
),
supports_check_mode=True,
add_file_common_args=True,
required_together=[
['cipher', 'passphrase']
],
required_if=[
['type', 'ECC', ['curve']],
],
)
base_dir = os.path.dirname(module.params['path']) or '.'
if not os.path.isdir(base_dir):
module.fail_json(
name=base_dir,
msg='The directory %s does not exist or the file is not a directory' % base_dir
)
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 module.params['cipher'] and module.params['passphrase'] and module.params['cipher'] != 'auto':
# First try pyOpenSSL, then cryptography
if can_use_pyopenssl:
backend = 'pyopenssl'
elif can_use_cryptography:
backend = 'cryptography'
else:
# First try cryptography, then pyOpenSSL
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.13', collection_name='ansible.builtin')
private_key = PrivateKeyPyOpenSSL(module)
elif backend == 'cryptography':
if not CRYPTOGRAPHY_FOUND:
module.fail_json(msg=missing_required_lib('cryptography >= {0}'.format(MINIMAL_CRYPTOGRAPHY_VERSION)),
exception=CRYPTOGRAPHY_IMP_ERR)
private_key = PrivateKeyCryptography(module)
if private_key.state == 'present':
if module.check_mode:
result = private_key.dump()
result['changed'] = private_key.force \
or not private_key.check(module, ignore_conversion=True) \
or not private_key.check(module, ignore_conversion=False)
module.exit_json(**result)
private_key.generate(module)
else:
if module.check_mode:
result = private_key.dump()
result['changed'] = os.path.exists(module.params['path'])
module.exit_json(**result)
private_key.remove(module)
result = private_key.dump()
module.exit_json(**result)
except crypto_utils.OpenSSLObjectError as exc:
module.fail_json(msg=to_native(exc))
if __name__ == '__main__':
main()