Coverage for plugins/modules/purefb_kmip.py: 75%
105 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-10-30 17:06 -0400
1#!/usr/bin/python
2# -*- coding: utf-8 -*-
4# (c) 2025, Simon Dodsley (simon@everpuredata.com)
5# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
7from __future__ import absolute_import, division, print_function
9__metaclass__ = type
11ANSIBLE_METADATA = {
12 "metadata_version": "1.1",
13 "status": ["preview"],
14 "supported_by": "community",
15}
17DOCUMENTATION = r"""
18---
19module: purefb_kmip
20version_added: '1.22.0'
21short_description: Manage FlashBlade KMIP server objects
22description:
23- Manage FlashBlade KMIP Server objects
24- A KMIP server object names one or more KMIP servers and the CA certificate,
25 or certificate group, used to validate their authenticity.
26author:
27- Everpure Ansible Team (@sdodsley) <pure-ansible-team@everpuredata.com>
28options:
29 name:
30 description:
31 - Name of the KMIP server object
32 type: str
33 required: true
34 state:
35 description:
36 - Action for the module to perform
37 - I(test) reports the results of the array's KMIP connectivity tests and
38 makes no changes
39 default: present
40 choices: [ absent, present, test ]
41 type: str
42 ca_certificate:
43 type: str
44 description:
45 - Name of an existing certificate on the array, used to validate the
46 authenticity of the configured KMIP servers.
47 - Use the M(everpure.flashblade.purefb_certs) module to create certificates.
48 - One of I(ca_certificate) or I(ca_certificate_group) is required when
49 creating a new KMIP object.
50 ca_certificate_group:
51 type: str
52 description:
53 - Name of an existing certificate group on the array, containing CA
54 certificates that can validate the authenticity of the configured
55 KMIP servers.
56 - The group must contain at least one certificate.
57 - Use the M(everpure.flashblade.purefb_certgrp) module to create
58 certificate groups.
59 - One of I(ca_certificate) or I(ca_certificate_group) is required when
60 creating a new KMIP object.
61 uris:
62 type: list
63 elements: str
64 description:
65 - A list of URIs for the configured KMIP servers, in the format
66 C([protocol://]hostname:port).
67 - Required when creating a new KMIP server object.
68extends_documentation_fragment:
69- everpure.flashblade.everpure.fb
70"""
72EXAMPLES = r"""
73- name: Create KMIP object
74 everpure.flashblade.purefb_kmip:
75 name: foo
76 ca_certificate: kmip_ca_cert
77 uris:
78 - 1.1.1.1:8888
79 - 2.3.3.3:9999
80 fb_url: 10.10.10.2
81 api_token: T-9f276a18-50ab-446e-8a0c-666a3529a1b6
83- name: Update the servers in an existing KMIP object
84 everpure.flashblade.purefb_kmip:
85 name: foo
86 uris:
87 - 3.3.3.3:8888
88 - 4.4.4.4:9999
89 fb_url: 10.10.10.2
90 api_token: T-9f276a18-50ab-446e-8a0c-666a3529a1b6
92- name: Point a KMIP object at a certificate group instead
93 everpure.flashblade.purefb_kmip:
94 name: foo
95 ca_certificate_group: kmip_ca_group
96 fb_url: 10.10.10.2
97 api_token: T-9f276a18-50ab-446e-8a0c-666a3529a1b6
99- name: Test KMIP object connectivity
100 everpure.flashblade.purefb_kmip:
101 name: foo
102 state: test
103 fb_url: 10.10.10.2
104 api_token: T-9f276a18-50ab-446e-8a0c-666a3529a1b6
106- name: Delete KMIP object
107 everpure.flashblade.purefb_kmip:
108 name: foo
109 state: absent
110 fb_url: 10.10.10.2
111 api_token: T-9f276a18-50ab-446e-8a0c-666a3529a1b6
112"""
114RETURN = r"""
115"""
117HAS_PYPURECLIENT = True
118try:
119 from pypureclient.flashblade import KmipServer, Reference
120except ImportError:
121 HAS_PYPURECLIENT = False
123from ansible.module_utils.basic import AnsibleModule
124from ansible_collections.everpure.flashblade.plugins.module_utils.purefb import (
125 get_system,
126 purefb_argument_spec,
127)
128from ansible_collections.everpure.flashblade.plugins.module_utils.common import (
129 get_error_message,
130)
132# Module option -> KmipServer field. Both are references to an object that
133# must already exist on the array, so the module sends Reference(name=...)
134# and compares against the name on the current object.
135REFERENCE_PARAMS = {
136 "ca_certificate": "ca_certificate",
137 "ca_certificate_group": "ca_certificate_group",
138}
141def _reference_name(obj, field):
142 """Name held by a reference field on a KMIP object, or None."""
143 return getattr(getattr(obj, field, None), "name", None)
146def get_kmip(module, blade):
147 """Return the named KMIP object, or None if it does not exist."""
148 res = blade.get_kmip(names=[module.params["name"]])
149 if res.status_code != 200:
150 return None
151 items = list(res.items)
152 return items[0] if items else None
155def _check_certificate(module, blade, param):
156 """Fail unless the named certificate or certificate group exists."""
157 name = module.params[param]
158 if not name:
159 return
160 if param == "ca_certificate_group":
161 res = blade.get_certificate_groups(names=[name])
162 kind = "Certificate group"
163 else:
164 res = blade.get_certificates(names=[name])
165 kind = "Certificate"
166 if res.status_code != 200:
167 module.fail_json(msg="{0} {1} does not exist.".format(kind, name))
170def report_kmip_test(module, blade):
171 """Report the array's KMIP connectivity tests. Makes no changes."""
172 test_response = []
173 response = list(blade.get_kmip_test(names=[module.params["name"]]).items)
174 for component in response:
175 test_response.append(
176 {
177 "component_address": component.component_address,
178 "component_name": component.component_name,
179 "description": component.description,
180 "destination": component.destination,
181 "enabled": bool(component.enabled),
182 "result_details": getattr(component, "result_details", ""),
183 "success": bool(component.success),
184 "test_type": component.test_type,
185 "resource_name": _reference_name(component, "resource"),
186 }
187 )
188 module.exit_json(changed=False, test_response=test_response)
191def update_kmip(module, blade, current_kmip):
192 """PATCH only the settings that differ from the current object."""
193 patch_kwargs = {}
194 if module.params["uris"] is not None:
195 current_uris = sorted(current_kmip.uris or [])
196 wanted_uris = sorted(module.params["uris"])
197 if current_uris != wanted_uris:
198 patch_kwargs["uris"] = wanted_uris
199 for param, field in REFERENCE_PARAMS.items():
200 wanted = module.params[param]
201 if wanted is None:
202 continue
203 if wanted != _reference_name(current_kmip, field):
204 _check_certificate(module, blade, param)
205 patch_kwargs[field] = Reference(name=wanted)
207 changed = bool(patch_kwargs)
208 if changed and not module.check_mode:
209 res = blade.patch_kmip(
210 names=[module.params["name"]],
211 kmip_server=KmipServer(**patch_kwargs),
212 )
213 if res.status_code != 200:
214 module.fail_json(
215 msg="Updating existing KMIP object {0} failed. Error: {1}".format(
216 module.params["name"], get_error_message(res)
217 )
218 )
219 module.exit_json(changed=changed)
222def create_kmip(module, blade):
223 """Create a new KMIP object."""
224 if not module.params["uris"]:
225 module.fail_json(msg="uris is required to create a new KMIP object")
226 if not (module.params["ca_certificate"] or module.params["ca_certificate_group"]):
227 # The array rejects a KMIP object with neither: "Either a CA
228 # certificate or a CA certificate group must be specified."
229 module.fail_json(
230 msg="One of ca_certificate or ca_certificate_group is required "
231 "to create a new KMIP object"
232 )
233 post_kwargs = {"uris": sorted(module.params["uris"])}
234 for param, field in REFERENCE_PARAMS.items():
235 if module.params[param]:
236 _check_certificate(module, blade, param)
237 post_kwargs[field] = Reference(name=module.params[param])
239 changed = True
240 if not module.check_mode:
241 res = blade.post_kmip(
242 names=[module.params["name"]],
243 kmip_server=KmipServer(**post_kwargs),
244 )
245 if res.status_code != 200:
246 module.fail_json(
247 msg="Creating KMIP object {0} failed. Error: {1}".format(
248 module.params["name"], get_error_message(res)
249 )
250 )
251 module.exit_json(changed=changed)
254def delete_kmip(module, blade):
255 """Delete an existing KMIP object."""
256 changed = True
257 if not module.check_mode:
258 res = blade.delete_kmip(names=[module.params["name"]])
259 if res.status_code != 200:
260 module.fail_json(
261 msg="Failed to delete {0} KMIP object. Error: {1}".format(
262 module.params["name"], get_error_message(res)
263 )
264 )
265 module.exit_json(changed=changed)
268def main():
269 argument_spec = purefb_argument_spec()
270 argument_spec.update(
271 dict(
272 state=dict(
273 type="str",
274 default="present",
275 choices=["absent", "present", "test"],
276 ),
277 name=dict(type="str", required=True),
278 ca_certificate=dict(type="str"),
279 ca_certificate_group=dict(type="str"),
280 uris=dict(type="list", elements="str"),
281 )
282 )
284 module = AnsibleModule(
285 argument_spec,
286 supports_check_mode=True,
287 )
289 if not HAS_PYPURECLIENT:
290 module.fail_json(msg="py-pure-client sdk is required for this module")
292 blade = get_system(module)
293 state = module.params["state"]
294 current_kmip = get_kmip(module, blade)
296 if state == "test":
297 if not current_kmip:
298 module.fail_json(
299 msg="KMIP object {0} does not exist.".format(module.params["name"])
300 )
301 report_kmip_test(module, blade)
302 elif state == "present" and not current_kmip:
303 create_kmip(module, blade)
304 elif state == "present" and current_kmip:
305 update_kmip(module, blade, current_kmip)
306 elif state == "absent" and current_kmip:
307 delete_kmip(module, blade)
309 module.exit_json(changed=False)
312if __name__ == "__main__":
313 main()