community.general/plugins/modules/keycloak_user.py

#!/usr/bin/python

# Copyright (c) 2019, INSPQ (@elfelip)
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later

from __future__ import annotations

DOCUMENTATION = r"""
module: keycloak_user
short_description: Create and configure a user in Keycloak
description:
  - This module creates, removes, or updates Keycloak users.
version_added: 7.1.0
options:
  auth_username:
    aliases: []
  realm:
    description:
      - The name of the realm in which is the client.
    default: master
    type: str
  username:
    description:
      - Username for the user.
    required: true
    type: str
  id:
    description:
      - ID of the user on the Keycloak server if known.
    type: str
  enabled:
    description:
      - Enabled user.
    type: bool
  email_verified:
    description:
      - Set or reset the C(emailVerified) flag of the user.
      - When O(email_verified_behavior=no_defaults), the default value of this option becomes C(null) and
        that causes the module not to change any existing value for that attribute.
    type: bool
    aliases:
      - emailVerified
  first_name:
    description:
      - The user's first name.
    type: str
    aliases:
      - firstName
  last_name:
    description:
      - The user's last name.
    type: str
    aliases:
      - lastName
  email:
    description:
      - User email.
    type: str
  federation_link:
    description:
      - Federation Link.
    type: str
    aliases:
      - federationLink
  service_account_client_id:
    description:
      - Description of the client Application.
    type: str
    aliases:
      - serviceAccountClientId
  client_consents:
    description:
      - Client Authenticator Type.
    type: list
    elements: dict
    default: []
    aliases:
      - clientConsents
    suboptions:
      client_id:
        description:
          - Client ID of the client role. Not the technical ID of the client.
        type: str
        required: true
        aliases:
          - clientId
      roles:
        description:
          - List of client roles to assign to the user.
        type: list
        required: true
        elements: str
  groups:
    description:
      - List of groups for the user.
      - Groups can be referenced by their name, like V(staff), or their path, like V(/staff/engineering). The path syntax
        allows you to reference subgroups, which is not possible otherwise.
      - Using the path is possible since community.general 10.6.0.
    type: list
    elements: dict
    default: []
    suboptions:
      name:
        description:
          - Name of the group.
        type: str
      state:
        description:
          - Control whether the user must be member of this group or not.
        choices: ["present", "absent"]
        default: present
        type: str
  credentials:
    description:
      - User credentials.
    default: []
    type: list
    elements: dict
    suboptions:
      type:
        description:
          - Credential type.
        type: str
        required: true
      value:
        description:
          - Value of the credential.
        type: str
        required: true
      temporary:
        description:
          - If V(true), the users are required to reset their credentials at next login.
        type: bool
        default: false
  required_actions:
    description:
      - Set or reset a user's required actions.
    type: list
    elements: str
    aliases:
      - requiredActions
  federated_identities:
    description:
      - List of IDPs of user.
    default: []
    type: list
    elements: str
    aliases:
      - federatedIdentities
  attributes:
    description:
      - List of user attributes.
    type: list
    elements: dict
    suboptions:
      name:
        description:
          - Name of the attribute.
        type: str
      values:
        description:
          - Values for the attribute as list.
        type: list
        elements: str
      state:
        description:
          - Control whether the attribute must exists or not.
        choices: ["present", "absent"]
        default: present
        type: str
  access:
    description:
      - List user access.
    type: dict
  disableable_credential_types:
    description:
      - List user Credential Type.
    default: []
    type: list
    elements: str
    aliases:
      - disableableCredentialTypes
  origin:
    description:
      - User origin.
    type: str
  self:
    description:
      - User self administration.
    type: str
  state:
    description:
      - Control whether the user should exists or not.
    choices: ["present", "absent"]
    default: present
    type: str
  force:
    description:
      - If V(true), allows to remove user and recreate it.
    type: bool
    default: false
  email_verified_behavior:
    description:
      - The O(email_verified) option used to have a default value. This caused problems when the
        user expects different behavior from keycloak by default.
      - The default value of this option is V(compatibility), which will ensure that the old default value
        for O(email_verified) is used.
      - When set to V(no_defaults), the module will not change existing values of O(email_verified) if no value is specified.
    type: str
    choices:
      - compatibility
      - no_defaults
    default: compatibility
    version_added: "13.1.0"
extends_documentation_fragment:
  - community.general._keycloak
  - community.general._keycloak.actiongroup_keycloak
  - community.general._attributes
attributes:
  check_mode:
    support: full
  diff_mode:
    support: full
  action_group:
    version_added: 10.2.0
notes:
  - The module does not modify the user ID of an existing user.
author:
  - Philippe Gauthier (@elfelip)
"""

EXAMPLES = r"""
- name: Create a user user1
  community.general.keycloak_user:
    auth_keycloak_url: http://localhost:8080/auth
    auth_username: admin
    auth_password: password
    realm: master
    username: user1
    firstName: user1
    lastName: user1
    email: user1
    enabled: true
    emailVerified: false
    credentials:
      - type: password
        value: password
        temporary: false
    attributes:
      - name: attr1
        values:
          - value1
        state: present
      - name: attr2
        values:
          - value2
        state: absent
    groups:
      - name: group1
        state: present
    state: present

- name: Re-create a User
  community.general.keycloak_user:
    auth_keycloak_url: http://localhost:8080/auth
    auth_username: admin
    auth_password: password
    realm: master
    username: user1
    firstName: user1
    lastName: user1
    email: user1
    enabled: true
    emailVerified: false
    credentials:
      - type: password
        value: password
        temporary: false
    attributes:
      - name: attr1
        values:
          - value1
        state: present
      - name: attr2
        values:
          - value2
        state: absent
    groups:
      - name: group1
        state: present
    state: present

- name: Re-create a User
  community.general.keycloak_user:
    auth_keycloak_url: http://localhost:8080/auth
    auth_username: admin
    auth_password: password
    realm: master
    username: user1
    firstName: user1
    lastName: user1
    email: user1
    enabled: true
    emailVerified: false
    credentials:
      - type: password
        value: password
        temporary: false
    attributes:
      - name: attr1
        values:
          - value1
        state: present
      - name: attr2
        values:
          - value2
        state: absent
    groups:
      - name: group1
        state: present
    state: present
    force: true

- name: Remove User
  community.general.keycloak_user:
    auth_keycloak_url: http://localhost:8080/auth
    auth_username: admin
    auth_password: password
    realm: master
    username: user1
    state: absent
"""

RETURN = r"""
proposed:
  description: Representation of the proposed user.
  returned: on success
  type: dict
existing:
  description: Representation of the existing user.
  returned: on success
  type: dict
end_state:
  description: Representation of the user after module execution.
  returned: on success
  type: dict
user_created:
  description: Indicates whether a user was created.
  returned: in success
  type: bool
  version_added: 12.0.0
"""

import copy

from ansible.module_utils.basic import AnsibleModule

from ansible_collections.community.general.plugins.module_utils._keycloak import (
    KeycloakAPI,
    KeycloakError,
    camel,
    get_token,
    is_struct_included,
    keycloak_argument_spec,
)


def main():
    argument_spec = keycloak_argument_spec()
    argument_spec["auth_username"]["aliases"] = []
    credential_spec = dict(
        type=dict(type="str", required=True),
        value=dict(type="str", required=True, no_log=True),
        temporary=dict(type="bool", default=False),
    )
    client_consents_spec = dict(
        client_id=dict(type="str", required=True, aliases=["clientId"]),
        roles=dict(type="list", elements="str", required=True),
    )
    attributes_spec = dict(
        name=dict(type="str"),
        values=dict(type="list", elements="str"),
        state=dict(type="str", choices=["present", "absent"], default="present"),
    )
    groups_spec = dict(name=dict(type="str"), state=dict(type="str", choices=["present", "absent"], default="present"))
    meta_args = dict(
        realm=dict(type="str", default="master"),
        self=dict(type="str"),
        id=dict(type="str"),
        username=dict(type="str", required=True),
        first_name=dict(type="str", aliases=["firstName"]),
        last_name=dict(type="str", aliases=["lastName"]),
        email=dict(type="str"),
        enabled=dict(type="bool"),
        email_verified=dict(type="bool", aliases=["emailVerified"]),
        federation_link=dict(type="str", aliases=["federationLink"]),
        service_account_client_id=dict(type="str", aliases=["serviceAccountClientId"]),
        attributes=dict(type="list", elements="dict", options=attributes_spec),
        access=dict(type="dict"),
        groups=dict(type="list", default=[], elements="dict", options=groups_spec),
        disableable_credential_types=dict(
            type="list", default=[], aliases=["disableableCredentialTypes"], elements="str"
        ),
        required_actions=dict(type="list", aliases=["requiredActions"], elements="str"),
        credentials=dict(type="list", default=[], elements="dict", options=credential_spec),
        federated_identities=dict(type="list", default=[], aliases=["federatedIdentities"], elements="str"),
        client_consents=dict(
            type="list", default=[], aliases=["clientConsents"], elements="dict", options=client_consents_spec
        ),
        origin=dict(type="str"),
        state=dict(choices=["absent", "present"], default="present"),
        force=dict(type="bool", default=False),
        email_verified_behavior=dict(type="str", choices=["compatibility", "no_defaults"], default="compatibility"),
    )
    argument_spec.update(meta_args)

    module = AnsibleModule(
        argument_spec=argument_spec,
        supports_check_mode=True,
        required_one_of=(
            [["token", "auth_realm", "auth_username", "auth_password", "auth_client_id", "auth_client_secret"]]
        ),
        required_together=([["auth_username", "auth_password"]]),
        required_by={"refresh_token": "auth_realm"},
    )

    result = dict(changed=False, msg="", diff={}, proposed={}, existing={}, end_state={})

    # Obtain access token, initialize API
    try:
        connection_header = get_token(module.params)
    except KeycloakError as e:
        module.fail_json(msg=str(e))

    kc = KeycloakAPI(module, connection_header)

    realm = module.params.get("realm")
    state = module.params.get("state")
    force = module.params.get("force")
    username = module.params.get("username")
    groups = module.params.get("groups")

    # If there is no value for email_verified, check if we should to set the old default
    if module.params["email_verified"] is None and module.params["email_verified_behavior"] == "compatibility":
        module.params["email_verified"] = False

    ignored_arguments = list(keycloak_argument_spec().keys()) + [
        "state",
        "realm",
        "force",
        "groups",
        "email_verified_behavior",
    ]

    # Filter and map the parameters names that apply to the user
    user_params = [x for x in module.params if x not in ignored_arguments and module.params[x] is not None]

    before_user = kc.get_user_by_username(username=username, realm=realm)

    if before_user is None:
        before_user = {}

    changeset = {}

    for param in user_params:
        new_param_value = module.params.get(param)
        if param == "attributes" and param in before_user:
            old_value = kc.convert_keycloak_user_attributes_dict_to_module_list(attributes=before_user["attributes"])
        else:
            old_value = before_user[param] if param in before_user else None
        if new_param_value != old_value:
            if old_value is not None and param == "attributes":
                for old_attribute in old_value:
                    old_attribute_found = False
                    for new_attribute in new_param_value:
                        if new_attribute["name"] == old_attribute["name"]:
                            old_attribute_found = True
                    if not old_attribute_found:
                        new_param_value.append(copy.deepcopy(old_attribute))
            if isinstance(new_param_value, dict):
                changeset[camel(param)] = copy.deepcopy(new_param_value)
            else:
                changeset[camel(param)] = new_param_value
    # Prepare the desired values using the existing values (non-existence results in a dict that is save to use as a basis)
    desired_user = copy.deepcopy(before_user)
    desired_user.update(changeset)

    if before_user:
        before_groups = kc.get_user_groups(user_id=before_user["id"], realm=realm)
        before_user["groups"] = before_groups
    else:
        before_groups = []

    result["proposed"] = changeset
    result["existing"] = before_user
    # Default values for user_created
    result["user_created"] = False
    changed = False
    after_user = {}

    if state == "absent":
        if not before_user:
            # Do nothing and exit
            result["msg"] = "User does not exist, doing nothing."
        else:
            # Delete user
            if not module.check_mode:
                kc.delete_user(user_id=before_user["id"], realm=realm)
            result["msg"] = f"User {before_user['username']} deleted"
            changed = True
    else:
        if (not before_user or force) and username is None:
            module.fail_json(msg="username must be specified when creating a new user")

        if force and before_user and not module.check_mode:  # If the force option is set to true
            # Delete the existing user
            kc.delete_user(user_id=before_user["id"], realm=realm)

        if not before_user or force:
            # Create a new user
            if not module.check_mode:
                # Create the user
                after_user = kc.create_user(userrep=desired_user, realm=realm)
                # Add user ID to desired_user for group updates
                desired_user["id"] = after_user["id"]
            else:
                after_user = desired_user

            result["msg"] = f"User {desired_user['username']} created"
            result["user_created"] = True
            changed = True
        else:
            # Update an existing user
            excludes = [
                "access",
                "notBefore",
                "createdTimestamp",
                "totp",
                "credentials",
                "disableableCredentialTypes",
                "groups",
                "clientConsents",
                "federatedIdentities",
            ]
            # Compare users
            if not (
                is_struct_included(desired_user, before_user, excludes, empty_list_result=False)
            ):  # If the new user introduces a change to the existing user
                # Update the user
                if not module.check_mode:
                    after_user = kc.update_user(userrep=desired_user, realm=realm)
                changed = True

            if not after_user:
                # no change
                after_user = desired_user

        # set user groups
        if not module.check_mode:
            changed |= kc.update_user_groups_membership(userrep=desired_user, groups=groups, realm=realm)

        present_groups = [g["name"] for g in groups if g["state"] == "present"]
        absent_groups = [g["name"] for g in groups if g["state"] == "absent"]

        desired_user["groups"] = (set(before_groups) | set(present_groups)) - set(absent_groups)

        if module.check_mode:
            # check if group meberships would have changed
            changed |= not is_struct_included(
                desired_user["groups"], before_user["groups"], excludes, empty_list_result=False
            )
        else:
            after_user["groups"] = kc.get_user_groups(user_id=desired_user["id"], realm=realm)

    if not result["msg"]:
        if changed:
            result["msg"] = f"User {desired_user['username']} updated"
        else:
            result["msg"] = f"No changes made for user {desired_user['username']}"
    result["end_state"] = after_user
    result["changed"] = changed
    result["diff"] = dict(before=before_user, after=after_user)
    module.exit_json(**result)


if __name__ == "__main__":
    main()