# -*- coding: utf-8 -*- """ansible-freeipa iparole module implementation.""" # Authors: # Rafael Guterres Jeffman # Thomas Woerner # # Copyright (C) 2020-2022 Red Hat # see file 'COPYING' for use and warranty information # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see . from __future__ import (absolute_import, division, print_function) __metaclass__ = type ANSIBLE_METADATA = { "metadata_version": "1.0", "supported_by": "community", "status": ["preview"], } DOCUMENTATION = """ --- module: iparole short_description: Manage FreeIPA role description: Manage FreeIPA role extends_documentation_fragment: - ipamodule_base_docs options: name: description: The list of role name strings. type: list elements: str required: true aliases: ["cn"] description: description: A description for the role. type: str required: false rename: description: Rename the role object. type: str required: false aliases: ["new_name"] privilege: description: List of privileges type: list elements: str required: false user: description: List of users. type: list elements: str required: false group: description: List of groups. type: list elements: str required: false host: description: List of hosts. type: list elements: str required: false hostgroup: description: List of hostgroups. type: list elements: str required: false service: description: List of services. type: list elements: str required: false sysaccount: description: List of sysaccounts. type: list elements: str required: false action: description: Work on role or member level. type: str choices: ["role", "member"] default: role required: false state: description: The state to ensure. type: str choices: ["present", "absent", "renamed"] default: present required: false author: - Rafael Guterres Jeffman (@rjeffman) - Thomas Woerner (@t-woerner) """ EXAMPLES = """ - name: Ensure a role named `somerole` is present. iparole: ipaadmin_password: SomeADMINpassword name: somerole - name: Ensure user `pinky` is a memmer of role `somerole`. iparole: ipaadmin_password: SomeADMINpassword name: somerole user: - pinky action: member - name: Ensure a role named `somerole` is absent. iparole: ipaadmin_password: SomeADMINpassword name: somerole state: absent """ # pylint: disable=wrong-import-position # pylint: disable=import-error # pylint: disable=no-name-in-module from ansible.module_utils._text import to_text from ansible.module_utils.ansible_freeipa_module import \ IPAAnsibleModule, gen_add_del_lists, compare_args_ipa, \ gen_intersection_list, ensure_fqdn, ipalib_errors from ansible.module_utils import six if six.PY3: unicode = str def find_role(module, name): """Find if a role with the given name already exist.""" try: _result = module.ipa_command("role_show", name, {"all": True}) except ipalib_errors.NotFound: # An exception is raised if role name is not found. return None return _result["result"] def gen_args(module): """Generate arguments for executing commands.""" arg_map = { "description": "description", "rename": "rename", } args = {} for param, arg in arg_map.items(): value = module.params_get(param) if value is not None: args[arg] = value return args def check_parameters(module): """Check if parameters passed for module processing are valid.""" action = module.params_get("action") state = module.params_get("state") invalid = [] if state == "renamed": if action == "member": module.fail_json( msg="Invalid action 'member' with state 'renamed'.") invalid = [ "description", "user", "group", "host", "hostgroup", "service", "sysaccount", "privilege", ] if state == "present": invalid = ["rename"] if action == "member": invalid.extend(['description']) if state == "absent": invalid.extend(['description', 'rename']) if action != "member": invalid.extend(['privilege']) module.params_fail_used_invalid(invalid, state, action) def get_member_host_with_fqdn_lowercase(module, mod_member): """Retrieve host members from module, as FQDN, lowercase.""" default_domain = module.ipa_get_domain() hosts = module.params_get(mod_member) return ( [ensure_fqdn(host, default_domain).lower() for host in hosts] if hosts else hosts ) def ensure_absent_state(module, name, action, res_find): """Define commands to ensure absent state.""" commands = [] if action == "role": commands.append([name, 'role_del', {}]) if action == "member": _members = module.params_get_lowercase("privilege") if _members is not None: del_list = gen_intersection_list( _members, result_get_value_lowercase(res_find, "memberof_privilege") ) if del_list: commands.append([name, "role_remove_privilege", {"privilege": del_list}]) member_args = {} for key in ['user', 'group', 'hostgroup', 'sysaccount']: _members = module.params_get_lowercase(key) if _members: del_list = gen_intersection_list( _members, result_get_value_lowercase(res_find, "member_%s" % key) ) if del_list: member_args[key] = del_list # ensure hosts are FQDN. _members = get_member_host_with_fqdn_lowercase(module, "host") if _members: del_list = gen_intersection_list( _members, res_find.get('member_host')) if del_list: member_args["host"] = del_list _services = get_service_param(module, "service") if _services: _existing = result_get_value_lowercase(res_find, "member_service") items = gen_intersection_list(_services.keys(), _existing) if items: member_args["service"] = [_services[key] for key in items] # Only add remove command if there's at least one member no manage. if member_args: commands.append([name, "role_remove_member", member_args]) return commands def get_service_param(module, key): """ Retrieve dict of services, with realm, from the module parameters. As the services are compared in a case insensitive manner, but are recorded in a case preserving way, a dict mapping the services in lowercase to the provided module parameter is generated, so that dict keys can be used for comparison and the values are used with IPA API. """ _services = module.params_get(key) if _services is not None: ipa_realm = module.ipa_get_realm() _services = [ to_text(svc) if '@' in svc else ('%s@%s' % (svc, ipa_realm)) for svc in _services ] if _services: _services = {svc.lower(): svc for svc in _services} return _services def result_get_value_lowercase(res_find, key, default=None): """ Retrieve a member of a dictionary converted to lowercase. If field data is a string it is returned in lowercase. If field data is a list or tuple, it is assumed that all values are strings and the result is a list of strings in lowercase. If 'key' is not found in the dictionary, returns 'default'. """ existing = res_find.get(key) if existing is not None: if isinstance(existing, (list, tuple)): existing = [to_text(item).lower() for item in existing] if isinstance(existing, (str, unicode)): # pylint: disable=W0012,E0606 existing = existing.lower() else: existing = default return existing def gen_services_add_del_lists(module, mod_member, res_find, res_member): """Generate add/del lists for service principals.""" add_list, del_list = None, None _services = get_service_param(module, mod_member) if _services is not None: _existing = result_get_value_lowercase(res_find, res_member) add_list, del_list = gen_add_del_lists(_services.keys(), _existing) if add_list: add_list = [_services[key] for key in add_list] if del_list: del_list = [to_text(item) for item in del_list] return add_list, del_list def ensure_role_with_members_is_present(module, name, res_find, action): """Define commands to ensure member are present for action `role`.""" commands = [] _members = module.params_get_lowercase("privilege") if _members: add_list, del_list = gen_add_del_lists( _members, result_get_value_lowercase(res_find, "memberof_privilege") ) if add_list: commands.append([name, "role_add_privilege", {"privilege": add_list}]) if action == "role" and del_list: commands.append([name, "role_remove_privilege", {"privilege": del_list}]) add_members = {} del_members = {} for key in ["user", "group", "hostgroup", "sysaccount"]: _members = module.params_get_lowercase(key) if _members is not None: add_list, del_list = gen_add_del_lists( _members, result_get_value_lowercase(res_find, "member_%s" % key) ) if add_list: add_members[key] = add_list if del_list: del_members[key] = del_list # ensure hosts are FQDN. _members = get_member_host_with_fqdn_lowercase(module, "host") if _members: add_list, del_list = gen_add_del_lists( _members, res_find.get('member_host')) if add_list: add_members["host"] = add_list if del_list: del_members["host"] = del_list (add_services, del_services) = gen_services_add_del_lists( module, "service", res_find, "member_service") if add_services: add_members["service"] = add_services if del_services: del_members["service"] = del_services if add_members: commands.append([name, "role_add_member", add_members]) # Only remove members if ensuring role, not acting on members. if action == "role" and del_members: commands.append([name, "role_remove_member", del_members]) return commands def role_commands_for_name(module, state, action, name): """Define commands for the Role module.""" commands = [] res_find = find_role(module, name) if state == "renamed": args = gen_args(module) if res_find is None: module.fail_json(msg="No role '%s'" % name) else: commands.append([name, 'role_mod', args]) if state == "present": args = gen_args(module) if action == "role": if res_find is None: commands.append([name, 'role_add', args]) res_find = {} else: if not compare_args_ipa(module, args, res_find): commands.append([name, 'role_mod', args]) if action == "member": if res_find is None: module.fail_json(msg="No role '%s'" % name) cmds = ensure_role_with_members_is_present( module, name, res_find, action ) commands.extend(cmds) if state == "absent" and res_find is not None: cmds = ensure_absent_state(module, name, action, res_find) commands.extend(cmds) return commands def create_module(): """Create module description.""" ansible_module = IPAAnsibleModule( argument_spec=dict( # generalgroups name=dict(type="list", elements="str", aliases=["cn"], required=True), # present description=dict(required=False, type="str", default=None), rename=dict(required=False, type="str", default=None, aliases=["new_name"]), # members privilege=dict(required=False, type='list', elements="str", default=None), user=dict(required=False, type='list', elements="str", default=None), group=dict(required=False, type='list', elements="str", default=None), host=dict(required=False, type='list', elements="str", default=None), hostgroup=dict(required=False, type='list', elements="str", default=None), service=dict(required=False, type='list', elements="str", default=None), sysaccount=dict(required=False, type='list', elements="str", default=None), # state action=dict(type="str", default="role", choices=["role", "member"]), state=dict(type="str", default="present", choices=["present", "absent", "renamed"]), ), supports_check_mode=True, mutually_exclusive=[], required_one_of=[] ) ansible_module._ansible_debug = True # pylint: disable=protected-access return ansible_module def main(): """Process role module script.""" ansible_module = create_module() check_parameters(ansible_module) # Init # Connect to IPA API with ansible_module.ipa_connect(): state = ansible_module.params_get("state") action = ansible_module.params_get("action") names = ansible_module.params_get("name") sysaccount = ansible_module.params_get("sysaccount") commands = [] has_sysaccount_member = ansible_module.ipa_command_param_exists( "role_add_member", "sysaccount") if not has_sysaccount_member and sysaccount is not None: ansible_module.fail_json( msg="sysaccount members are not supported by your IPA version") for name in names: cmds = role_commands_for_name(ansible_module, state, action, name) commands.extend(cmds) exit_args = {} # Execute commands changed = ansible_module.execute_ipa_commands( commands, fail_on_member_errors=True) # Done ansible_module.exit_json(changed=changed, **exit_args) if __name__ == "__main__": main()