internet/ansible-freeipa
4
0
Fork 0
mirror of https://github.com/freeipa/ansible-freeipa.git synced 2026-05-06 13:23:14 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #671 from rjeffman/baseclass_playground

Browse Source 
Deprecate FreeIPABaseModule in favor of IPAAnsibleModule.
This commit is contained in:
Thomas Woerner
2021-12-13 14:03:14 +01:00
committed by GitHub
parent a1bfa608e2 2f9791f6c5
commit e7f902ca48
4 changed files with 239 additions and 208 deletions
Download Patch File Download Diff File Expand all files Collapse all files

233
plugins/module_utils/ansible_freeipa_module.py
View File

@@ -566,10 +566,76 @@ else:
print(jsonify(kwargs))
sys.exit(0)
class AnsibleFreeIPAParams(Mapping):
def __init__(self, ansible_module):
class IPAParamMapping(Mapping):
"""
Provides IPA API mapping to playbook parameters or computed values.
It can be used to define a mapping of playbook parameters
or methods that provide computed values to IPA API arguments.
Playbook parameters can be retrieved as properties,
and the set of IPA arguments for a command can be
retrived with ``get_ipa_command_args()``. The keys for
``param_mapping`` are also the keys of the argument set.
The values of ``param_mapping`` can be either:
* a str representing a key of ``AnsibleModule.params``.
* a callable.
In case of an ``AnsibleModule.param`` the value of the playbook
param will be used for that argument. If it is a ``callable``,
the value returned by the execution of it will be used.
Example:
-------
def check_params(ipa_params):
# Module parameters can be accessed as properties.
if len(ipa_params.name) == 0:
ipa_params.ansible_module.fail_json(msg="No given name.")
def define_ipa_commands(self):
# Create the argument dict from the defined mapping.
args = self.get_ipa_command_args()
_commands = [("obj-name", "some_ipa_command", args)]
return _commands
def a_method_for_a_computed_param():
return "Some computed value"
def main():
ansible_module = SomeIPAModule(argument_spec=dict(
name=dict(type="list", aliases=["cn"], required=True),
state=dict(type="str", default="present",
choices=["present", "absent"]),
module_param=(type="str", required=False),
)
)
# Define the playbook to IPA API mapping
ipa_param_mapping = {
"arg_to_be_passed_to_ipa_command": "module_param",
"another_arg": a_method_for_a_computed_param,
}
ipa_params = IPAParamMapping(
ansible_module,
param_mapping=ipa_param_mapping
)
check_params(ipa_params)
comands = define_ipa_commands(ipa_params)
ansible_module.execute_ipa_commands(commands)
"""
def __init__(self, ansible_module, param_mapping=None):
self.mapping = ansible_module.params
self.ansible_module = ansible_module
self.param_mapping = param_mapping or {}
def __getitem__(self, key):
param = self.mapping[key]
@@ -590,6 +656,36 @@ else:
def __getattr__(self, name):
return self.get(name)
def get_ipa_command_args(self, **kwargs):
"""Return a dict to be passed to an IPA command."""
args = {}
for ipa_param_name, param_name in self.param_mapping.items():
# Check if param_name is actually a param
if param_name in self.ansible_module.params:
value = self.ansible_module.params_get(param_name)
if isinstance(value, bool):
value = "TRUE" if value else "FALSE"
# Since param wasn't a param check if it's a method name
elif callable(param_name):
value = param_name(**kwargs)
# We don't have a way to guess the value so fail.
else:
self.ansible_module.fail_json(
msg=(
"Couldn't get a value for '%s'. Option '%s' is "
"not a module argument neither a defined method."
)
% (ipa_param_name, param_name)
)
if value is not None:
args[ipa_param_name] = value
return args
class IPAAnsibleModule(AnsibleModule):
"""
IPA Ansible Module.
@@ -1034,6 +1130,10 @@ else:
# pylint: disable=super-with-arguments
super(FreeIPABaseModule, self).__init__(*args, **kwargs)
self.deprecate(
msg="FreeIPABaseModule is deprecated. Use IPAAnsibleModule.",
)
# Status of an execution. Will be changed to True
# if something is actually peformed.
self.changed = False
@@ -1049,11 +1149,6 @@ else:
# Module exit arguments.
self.exit_args = {}
# Wrapper around the AnsibleModule.params.
# Return the actual params but performing transformations
# when needed.
self.ipa_params = AnsibleFreeIPAParams(self)
def get_ipa_command_args(self, **kwargs):
"""
Return a dict to be passed to an IPA command.
@@ -1074,97 +1169,73 @@ else:
server).
"""
args = {}
for ipa_param_name, param_name in self.ipa_param_mapping.items():
# Check if param_name is actually a param
if param_name in self.ipa_params:
value = self.ipa_params.get(param_name)
if isinstance(value, bool):
value = "TRUE" if value else "FALSE"
# Since param wasn't a param check if it's a method name
elif hasattr(self, param_name):
method = getattr(self, param_name)
if callable(method):
value = method(**kwargs)
# We don't have a way to guess the value so fail.
else:
self.fail_json(
msg=(
"Couldn't get a value for '%s'. Option '%s' is "
"not a module argument neither a defined method."
)
% (ipa_param_name, param_name)
)
if value is not None:
args[ipa_param_name] = value
return args
self.deprecate(
msg=(
"FreeIPABaseModule is deprecated. Use IPAAnsibleModule. "
"Use 'AnsibleFreeIPAParams.get_ipa_command_args()', "
"Instantiate it using the class 'ipa_params_mapping'."
)
)
mapping = IPAParamMapping(self, self.ipa_param_mapping)
return mapping.get_ipa_command_args(**kwargs)
def check_ipa_params(self):
"""Validate ipa_params before command is called."""
self.deprecate(
msg=(
"FreeIPABaseModule is deprecated. Use IPAAnsibleModule. "
),
)
pass # pylint: disable=unnecessary-pass
def define_ipa_commands(self):
"""Define commands that will be run in IPA server."""
raise NotImplementedError
def get_command_errors(self, command, result):
"""Look for erros into command results."""
# Get all errors
# All "already a member" and "not a member" failures in the
# result are ignored. All others are reported.
errors = []
for item in result.get("failed", tuple()):
failed_item = result["failed"][item]
for member_type in failed_item:
for member, failure in failed_item[member_type]:
if (
"already a member" in failure
or "not a member" in failure
):
continue
errors.append(
"%s: %s %s: %s"
% (command, member_type, member, failure)
)
if len(errors) > 0:
self.fail_json(", ".join("errors")) # pylint: disable=E1121
def add_ipa_command(self, command, name=None, args=None):
"""Add a command to the list of commands to be executed."""
self.ipa_commands.append((name, command, args or {}))
def _run_ipa_commands(self):
"""Execute commands in self.ipa_commands."""
if self.check_mode:
self.changed = len(self.ipa_commands) > 0
return
self.changed = self.execute_ipa_commands(
self.ipa_commands,
result_handler=self.process_results.__func__,
exit_args=self.exit_args
)
result = None
def process_results(
self, result, command, name, args, exit_args
): # pylint: disable=unused-argument
"""
Process an API command result.
for name, command, args in self.ipa_commands:
try:
result = self.ipa_command(command, name, args)
except Exception as excpt:
self.fail_json(msg="%s: %s: %s" % (command, name,
str(excpt)))
else:
self.process_command_result(name, command, args, result)
self.get_command_errors(command, result)
This method must be overriden in subclasses if 'exit_args'
is to be modified.
"""
self.deprecate(
msg=(
"FreeIPABaseModule is deprecated. Use IPAAnsibleModule. "
),
)
self.process_command_result(name, command, args, result)
def process_command_result(self, _name, _command, _args, result):
"""
Process an API command result.
This method can be overriden in subclasses, and
change self.exit_values
to return data in the result for the controller.
change self.exit_values to return data in the
result for the controller.
"""
self.deprecate(
msg=(
"FreeIPABaseModule is deprecated. Use IPAAnsibleModule. "
"To aid in porting to IPAAnsibleModule, change to "
"'FreeIPABaseModule.process_results'."
),
)
if "completed" in result:
if result["completed"] > 0:
self.changed = True
@@ -1178,12 +1249,24 @@ else:
Returns True in case current IPA object attributes differ from
args passed to the module.
"""
self.deprecate(
msg=(
"FreeIPABaseModule is deprecated. Use IPAAnsibleModule. "
"FreeIPABaseModule require_ipa_attrs_change() is "
"deprecated. Use ansible_freeipa_module.compare_args()."
),
)
equal = compare_args_ipa(self, command_args, ipa_attrs)
return not equal
def ipa_run(self):
"""Execute module actions."""
ipaapi_context = self.ipa_params.get("ipaapi_context")
self.deprecate(
msg=(
"FreeIPABaseModule is deprecated. Use IPAAnsibleModule."
),
)
ipaapi_context = self.params_get("ipaapi_context")
with self.ipa_connect(context=ipaapi_context):
self.check_ipa_params()
self.define_ipa_commands()

67
plugins/modules/README.md
View File

@@ -1,6 +1,5 @@
# Writing a new Ansible FreeIPA module
## Minimum requirements
A ansible-freeipa module should have:
* Code:
@@ -13,68 +12,4 @@ A ansible-freeipa module should have:
* Tests:
* Test cases (also playbooks) defined in `tests/<module_name>/test_<something>.yml`. It's ok to have multiple files in this directory.
## Code
The module file have to start with the python shebang line, license header and definition of the constants `ANSIBLE_METADATA`, `DOCUMENTATION`, `EXAMPLES` and `RETURNS`. Those constants need to be defined before the code (even imports). See https://docs.ansible.com/ansible/latest/dev_guide/developing_modules_general.html#starting-a-new-module for more information.
Although it's use is not yet required, ansible-freeipa provides `FreeIPABaseModule` as a helper class for the implementation of new modules. See the example bellow:
```python
from ansible.module_utils.ansible_freeipa_module import FreeIPABaseModule
class SomeIPAModule(FreeIPABaseModule):
ipa_param_mapping = {
"arg_to_be_passed_to_ipa_command": "module_param",
"another_arg": "get_another_module_param",
}
def get_another_module_param(self):
another_module_param = self.ipa_params.another_module_param
# Validate or modify another_module_param ...
return another_module_param
def check_ipa_params(self):
# Validate your params here ...
# Example:
if not self.ipa_params.module_param in VALID_OPTIONS:
self.fail_json(msg="Invalid value for argument module_param")
def define_ipa_commands(self):
args = self.get_ipa_command_args()
self.add_ipa_command("some_ipa_command", name="obj-name", args=args)
def main():
ipa_module = SomeIPAModule(argument_spec=dict(
module_param=dict(type="str", default=None, required=False),
another_module_param=dict(type="str", default=None, required=False),
))
ipa_module.ipa_run()
if __name__ == "__main__":
main()
```
In the example above, the module will call the command `some_ipa_command`, using "obj-name" as name and, `arg_to_be_passed_to_ipa_command` and `another_arg` as arguments.
The values of the arguments will be determined by the class attribute `ipa_param_mapping`.
In the case of `arg_to_be_passed_to_ipa_command` the key (`module_param`) is defined in the module `argument_specs` so the value of the argument is actually used.
On the other hand, `another_arg` as mapped to something else: a callable method. In this case the method will be called and it's result used as value for `another_arg`.
**NOTE**: Keep mind that to take advantage of the parameters mapping defined in `ipa_param_mapping` you will have to call `args = self.get_ipa_command_args()` and use `args` in your command. There is no implicit call of this method.
## Disclaimer
The `FreeIPABaseModule` is new and might not be suitable to all cases and every module yet. In case you need to extend it's functionality for a new module please open an issue or PR and we'll be happy to discuss it.
Use the script `utils/new_module` to create the stub files for a new module.

46
plugins/modules/ipaautomountlocation.py
View File

@@ -68,13 +68,16 @@ RETURN = '''
'''
from ansible.module_utils.ansible_freeipa_module import (
FreeIPABaseModule, ipalib_errors
IPAAnsibleModule, ipalib_errors
)
class AutomountLocation(FreeIPABaseModule):
class AutomountLocation(IPAAnsibleModule):
ipa_param_mapping = {}
def __init__(self, *args, **kwargs):
# pylint: disable=super-with-arguments
super(AutomountLocation, self).__init__(*args, **kwargs)
self.commands = []
def get_location(self, location):
try:
@@ -87,40 +90,28 @@ class AutomountLocation(FreeIPABaseModule):
return response.get("result", None)
def check_ipa_params(self):
if len(self.ipa_params.name) == 0:
if len(self.params_get("name")) == 0:
self.fail_json(msg="At least one location must be provided.")
def define_ipa_commands(self):
state = self.params_get("state")
for location_name in self.ipa_params.name:
for location_name in self.params_get("name"):
location = self.get_location(location_name)
if not location and self.ipa_params.state == "present":
if not location and state == "present":
# does not exist and is wanted
self.add_ipa_command(
"automountlocation_add",
name=location_name,
args=None,
)
elif location and self.ipa_params.state == "absent":
self.commands.append(
(location_name, "automountlocation_add", {}))
elif location and state == "absent":
# exists and is not wanted
self.add_ipa_command(
"automountlocation_del",
name=location_name,
args=None,
)
self.commands.append(
(location_name, "automountlocation_del", {}))
def main():
ipa_module = AutomountLocation(
argument_spec=dict(
ipaadmin_principal=dict(type="str",
default="admin"
),
ipaadmin_password=dict(type="str",
required=False,
no_log=True
),
state=dict(type='str',
default='present',
choices=['present', 'absent']
@@ -132,7 +123,12 @@ def main():
),
),
)
ipa_module.ipa_run()
ipaapi_context = ipa_module.params_get("ipaapi_context")
with ipa_module.ipa_connect(context=ipaapi_context):
ipa_module.check_ipa_params()
ipa_module.define_ipa_commands()
changed = ipa_module.execute_ipa_commands(ipa_module.commands)
ipa_module.exit_json(changed=changed)
if __name__ == "__main__":

101
plugins/modules/ipadnszone.py
View File

@@ -203,11 +203,13 @@ dnszone:
from ipapython.dnsutil import DNSName # noqa: E402
from ansible.module_utils.ansible_freeipa_module import (
FreeIPABaseModule,
IPAAnsibleModule,
is_ip_address,
is_ip_network_address,
is_valid_port,
ipalib_errors
ipalib_errors,
compare_args_ipa,
IPAParamMapping,
) # noqa: E402
import netaddr
from ansible.module_utils import six
@@ -217,31 +219,39 @@ if six.PY3:
unicode = str
class DNSZoneModule(FreeIPABaseModule):
class DNSZoneModule(IPAAnsibleModule):
ipa_param_mapping = {
# Direct Mapping
"idnsforwardpolicy": "forward_policy",
"idnssoarefresh": "refresh",
"idnssoaretry": "retry",
"idnssoaexpire": "expire",
"idnssoaminimum": "minimum",
"dnsttl": "ttl",
"dnsdefaultttl": "default_ttl",
"idnsallowsyncptr": "allow_sync_ptr",
"idnsallowdynupdate": "dynamic_update",
"idnssecinlinesigning": "dnssec",
"idnsupdatepolicy": "update_policy",
# Mapping by method
"idnsforwarders": "get_ipa_idnsforwarders",
"idnsallowtransfer": "get_ipa_idnsallowtransfer",
"idnsallowquery": "get_ipa_idnsallowquery",
"idnssoamname": "get_ipa_idnssoamname",
"idnssoarname": "get_ipa_idnssoarname",
"skip_nameserver_check": "get_ipa_skip_nameserver_check",
"skip_overlap_check": "get_ipa_skip_overlap_check",
"nsec3paramrecord": "get_ipa_nsec3paramrecord",
}
def __init__(self, *args, **kwargs):
# pylint: disable=super-with-arguments
super(DNSZoneModule, self).__init__(*args, **kwargs)
ipa_param_mapping = {
# Direct Mapping
"idnsforwardpolicy": "forward_policy",
"idnssoarefresh": "refresh",
"idnssoaretry": "retry",
"idnssoaexpire": "expire",
"idnssoaminimum": "minimum",
"dnsttl": "ttl",
"dnsdefaultttl": "default_ttl",
"idnsallowsyncptr": "allow_sync_ptr",
"idnsallowdynupdate": "dynamic_update",
"idnssecinlinesigning": "dnssec",
"idnsupdatepolicy": "update_policy",
# Mapping by method
"idnsforwarders": self.get_ipa_idnsforwarders,
"idnsallowtransfer": self.get_ipa_idnsallowtransfer,
"idnsallowquery": self.get_ipa_idnsallowquery,
"idnssoamname": self.get_ipa_idnssoamname,
"idnssoarname": self.get_ipa_idnssoarname,
"skip_nameserver_check": self.get_ipa_skip_nameserver_check,
"skip_overlap_check": self.get_ipa_skip_overlap_check,
"nsec3paramrecord": self.get_ipa_nsec3paramrecord,
}
self.commands = []
self.ipa_params = IPAParamMapping(self, ipa_param_mapping)
self.exit_args = {}
def validate_ips(self, ips, error_msg):
invalid_ips = [
@@ -441,39 +451,34 @@ class DNSZoneModule(FreeIPABaseModule):
for zone_name in self.get_zone_names():
# Look for existing zone in IPA
zone, is_zone_active = self.get_zone(zone_name)
args = self.get_ipa_command_args(zone=zone)
args = self.ipa_params.get_ipa_command_args(zone=zone)
if self.ipa_params.state in ["present", "enabled", "disabled"]:
if not zone:
# Since the zone doesn't exist we just create it
# with given args
self.add_ipa_command("dnszone_add", zone_name, args)
self.commands.append((zone_name, "dnszone_add", args))
is_zone_active = True
# just_added = True
else:
# Zone already exist so we need to verify if given args
# matches the current config. If not we updated it.
if self.require_ipa_attrs_change(args, zone):
self.add_ipa_command("dnszone_mod", zone_name, args)
if not compare_args_ipa(self, args, zone):
self.commands.append((zone_name, "dnszone_mod", args))
if self.ipa_params.state == "enabled" and not is_zone_active:
self.add_ipa_command("dnszone_enable", zone_name)
self.commands.append((zone_name, "dnszone_enable", {}))
if self.ipa_params.state == "disabled" and is_zone_active:
self.add_ipa_command("dnszone_disable", zone_name)
self.commands.append((zone_name, "dnszone_disable", {}))
if self.ipa_params.state == "absent" and zone is not None:
self.add_ipa_command("dnszone_del", zone_name)
self.commands.append((zone_name, "dnszone_del", {}))
def process_command_result(self, name, command, args, result):
# pylint: disable=super-with-arguments
super(DNSZoneModule, self).process_command_result(
name, command, args, result
)
def process_results(self, _result, command, name, _args, exit_args):
if command == "dnszone_add" and self.ipa_params.name_from_ip:
dnszone_exit_args = self.exit_args.setdefault('dnszone', {})
dnszone_exit_args['name'] = name
exit_args.setdefault('dnszone', {})["name"] = name
def get_argument_spec():
@@ -532,12 +537,24 @@ def get_argument_spec():
def main():
DNSZoneModule(
ansible_module = DNSZoneModule(
argument_spec=get_argument_spec(),
mutually_exclusive=[["name", "name_from_ip"]],
required_one_of=[["name", "name_from_ip"]],
supports_check_mode=True,
).ipa_run()
)
exit_args = {}
ipaapi_context = ansible_module.params_get("ipaapi_context")
with ansible_module.ipa_connect(context=ipaapi_context):
ansible_module.check_ipa_params()
ansible_module.define_ipa_commands()
changed = ansible_module.execute_ipa_commands(
ansible_module.commands,
result_handler=DNSZoneModule.process_results,
exit_args=exit_args
)
ansible_module.exit_json(changed=changed, **exit_args)
if __name__ == "__main__":