ansible/community.general
1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2024-09-14 20:13:21 +02:00
Code Releases Activity

Keycloak: Authentication required actions management (#6732)

Browse source 
* feat: keycloak required actions

* Update plugins/modules/keycloak_authentication_required_actions.py

Co-authored-by: Felix Fontein <felix@fontein.de>

* Update plugins/modules/keycloak_authentication_required_actions.py

Co-authored-by: Felix Fontein <felix@fontein.de>

* fix: dedent

* fix: unnecessary defaults

* fix: sanity checks

* Update plugins/modules/keycloak_authentication_required_actions.py

Co-authored-by: Felix Fontein <felix@fontein.de>

* fix: ident

---------

Co-authored-by: Skrekulko <Skrekulko@users.noreply.github.com>
Co-authored-by: Felix Fontein <felix@fontein.de>
This commit is contained in:
Skrekulko 2023-06-20 19:18:26 +02:00 committed by GitHub
parent 806f6da16b
commit cd48e818ae
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 1407 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.github/BOTMETA.yml vendored
View file

@ -687,6 +687,8 @@ files:
maintainers: $team_keycloak maintainers: $team_keycloak
$modules/keycloak_authentication.py: $modules/keycloak_authentication.py:
maintainers: elfelip Gaetan2907 maintainers: elfelip Gaetan2907
$modules/keycloak_authentication_required_actions.py:
maintainers: Skrekulko
$modules/keycloak_authz_authorization_scope.py: $modules/keycloak_authz_authorization_scope.py:
maintainers: mattock maintainers: mattock
$modules/keycloak_client_rolemapping.py: $modules/keycloak_client_rolemapping.py:

113
plugins/module_utils/identity/keycloak/keycloak.py
View file

@ -90,6 +90,9 @@ URL_AUTHENTICATION_EXECUTION_CONFIG = "{url}/admin/realms/{realm}/authentication
URL_AUTHENTICATION_EXECUTION_RAISE_PRIORITY = "{url}/admin/realms/{realm}/authentication/executions/{id}/raise-priority" URL_AUTHENTICATION_EXECUTION_RAISE_PRIORITY = "{url}/admin/realms/{realm}/authentication/executions/{id}/raise-priority"
URL_AUTHENTICATION_EXECUTION_LOWER_PRIORITY = "{url}/admin/realms/{realm}/authentication/executions/{id}/lower-priority" URL_AUTHENTICATION_EXECUTION_LOWER_PRIORITY = "{url}/admin/realms/{realm}/authentication/executions/{id}/lower-priority"
URL_AUTHENTICATION_CONFIG = "{url}/admin/realms/{realm}/authentication/config/{id}" URL_AUTHENTICATION_CONFIG = "{url}/admin/realms/{realm}/authentication/config/{id}"
URL_AUTHENTICATION_REGISTER_REQUIRED_ACTION = "{url}/admin/realms/{realm}/authentication/register-required-action"
URL_AUTHENTICATION_REQUIRED_ACTIONS = "{url}/admin/realms/{realm}/authentication/required-actions"
URL_AUTHENTICATION_REQUIRED_ACTIONS_ALIAS = "{url}/admin/realms/{realm}/authentication/required-actions/{alias}"
URL_IDENTITY_PROVIDERS = "{url}/admin/realms/{realm}/identity-provider/instances" URL_IDENTITY_PROVIDERS = "{url}/admin/realms/{realm}/identity-provider/instances"
URL_IDENTITY_PROVIDER = "{url}/admin/realms/{realm}/identity-provider/instances/{alias}" URL_IDENTITY_PROVIDER = "{url}/admin/realms/{realm}/identity-provider/instances/{alias}"
@ -2246,6 +2249,116 @@ class KeycloakAPI(object):
self.module.fail_json(msg='Could not get executions for authentication flow %s in realm %s: %s' self.module.fail_json(msg='Could not get executions for authentication flow %s in realm %s: %s'
% (config["alias"], realm, str(e))) % (config["alias"], realm, str(e)))
def get_required_actions(self, realm='master'):
"""
Get required actions.
:param realm: Realm name (not id).
:return: List of representations of the required actions.
"""
try:
required_actions = json.load(
open_url(
URL_AUTHENTICATION_REQUIRED_ACTIONS.format(
url=self.baseurl,
realm=realm
),
method='GET',
http_agent=self.http_agent, headers=self.restheaders,
timeout=self.connection_timeout,
validate_certs=self.validate_certs
)
)
return required_actions
except Exception:
return None
def register_required_action(self, rep, realm='master'):
"""
Register required action.
:param rep: JSON containing 'providerId', and 'name' attributes.
:param realm: Realm name (not id).
:return: Representation of the required action.
"""
data = {
'name': rep['name'],
'providerId': rep['providerId']
}
try:
return open_url(
URL_AUTHENTICATION_REGISTER_REQUIRED_ACTION.format(
url=self.baseurl,
realm=realm
),
method='POST',
http_agent=self.http_agent, headers=self.restheaders,
data=json.dumps(data),
timeout=self.connection_timeout,
validate_certs=self.validate_certs
)
except Exception as e:
self.module.fail_json(
msg='Unable to register required action %s in realm %s: %s'
% (rep["name"], realm, str(e))
)
def update_required_action(self, alias, rep, realm='master'):
"""
Update required action.
:param alias: Alias of required action.
:param rep: JSON describing new state of required action.
:param realm: Realm name (not id).
:return: HTTPResponse object on success.
"""
try:
return open_url(
URL_AUTHENTICATION_REQUIRED_ACTIONS_ALIAS.format(
url=self.baseurl,
alias=quote(alias),
realm=realm
),
method='PUT',
http_agent=self.http_agent, headers=self.restheaders,
data=json.dumps(rep),
timeout=self.connection_timeout,
validate_certs=self.validate_certs
)
except Exception as e:
self.module.fail_json(
msg='Unable to update required action %s in realm %s: %s'
% (alias, realm, str(e))
)
def delete_required_action(self, alias, realm='master'):
"""
Delete required action.
:param alias: Alias of required action.
:param realm: Realm name (not id).
:return: HTTPResponse object on success.
"""
try:
return open_url(
URL_AUTHENTICATION_REQUIRED_ACTIONS_ALIAS.format(
url=self.baseurl,
alias=quote(alias),
realm=realm
),
method='DELETE',
http_agent=self.http_agent, headers=self.restheaders,
timeout=self.connection_timeout,
validate_certs=self.validate_certs
)
except Exception as e:
self.module.fail_json(
msg='Unable to delete required action %s in realm %s: %s'
% (alias, realm, str(e))
)
def get_identity_providers(self, realm='master'): def get_identity_providers(self, realm='master'):
""" Fetch representations for identity providers in a realm """ Fetch representations for identity providers in a realm
:param realm: realm to be queried :param realm: realm to be queried

457
plugins/modules/keycloak_authentication_required_actions.py Normal file
View file

@ -0,0 +1,457 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
# Copyright (c) 2017, Eike Frost <ei@kefro.st>
# Copyright (c) 2021, Christophe Gilles <christophe.gilles54@gmail.com>
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or
# https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
from __future__ import absolute_import, division, print_function
__metaclass__ = type
DOCUMENTATION = '''
---
module: keycloak_authentication_required_actions
short_description: Allows administration of Keycloak authentication required actions
description:
- This module can register, update and delete required actions.
- It also filters out any duplicate required actions by their alias. The first ocurrence is preserved.
version_added: 7.1.0
attributes:
check_mode:
support: full
diff_mode:
support: full
options:
realm:
description:
- The name of the realm in which are the authentication required actions.
required: true
type: str
required_actions:
elements: dict
description:
- Authentication required action.
suboptions:
alias:
description:
- Unique name of the required action.
required: true
type: str
config:
description:
- Configuration for the required action.
type: dict
defaultAction:
description:
- Indicates, if any new user will have the required action assigned to it.
type: bool
enabled:
description:
- Indicates, if the required action is enabled or not.
type: bool
name:
description:
- Displayed name of the required action. Required for registration.
type: str
priority:
description:
- Priority of the required action.
type: int
providerId:
description:
- Provider ID of the required action. Required for registration.
type: str
type: list
state:
choices: [ "absent", "present" ]
description:
- Control if the realm authentication required actions are going to be registered/updated (V(present)) or deleted (V(absent)).
required: true
type: str
extends_documentation_fragment:
- community.general.keycloak
- community.general.attributes
author:
- Skrekulko (@Skrekulko)
'''
EXAMPLES = '''
- name: Register a new required action.
community.general.keycloak_authentication_required_actions:
auth_client_id: "admin-cli"
auth_keycloak_url: "http://localhost:8080"
auth_password: "password"
auth_realm: "master"
auth_username: "admin"
realm: "master"
required_action:
- alias: "TERMS_AND_CONDITIONS"
name: "Terms and conditions"
providerId: "TERMS_AND_CONDITIONS"
enabled: true
state: "present"
- name: Update the newly registered required action.
community.general.keycloak_authentication_required_actions:
auth_client_id: "admin-cli"
auth_keycloak_url: "http://localhost:8080"
auth_password: "password"
auth_realm: "master"
auth_username: "admin"
realm: "master"
required_action:
- alias: "TERMS_AND_CONDITIONS"
enabled: false
state: "present"
- name: Delete the updated registered required action.
community.general.keycloak_authentication_required_actions:
auth_client_id: "admin-cli"
auth_keycloak_url: "http://localhost:8080"
auth_password: "password"
auth_realm: "master"
auth_username: "admin"
realm: "master"
required_action:
- alias: "TERMS_AND_CONDITIONS"
state: "absent"
'''
RETURN = '''
msg:
description: Message as to what action was taken.
returned: always
type: str
end_state:
description: Representation of the authentication required actions after module execution.
returned: on success
type: complex
contains:
alias:
description:
- Unique name of the required action.
sample: test-provider-id
type: str
config:
description:
- Configuration for the required action.
sample: {}
type: dict
defaultAction:
description:
- Indicates, if any new user will have the required action assigned to it.
sample: false
type: bool
enabled:
description:
- Indicates, if the required action is enabled or not.
sample: false
type: bool
name:
description:
- Displayed name of the required action. Required for registration.
sample: Test provider ID
type: str
priority:
description:
- Priority of the required action.
sample: 90
type: int
providerId:
description:
- Provider ID of the required action. Required for registration.
sample: test-provider-id
type: str
'''
from ansible_collections.community.general.plugins.module_utils.identity.keycloak.keycloak import KeycloakAPI, \
keycloak_argument_spec, get_token, KeycloakError
from ansible.module_utils.basic import AnsibleModule
def sanitize_required_actions(objects):
for obj in objects:
alias = obj['alias']
name = obj['name']
provider_id = obj['providerId']
if not name:
obj['name'] = alias
if provider_id != alias:
obj['providerId'] = alias
return objects
def filter_duplicates(objects):
filtered_objects = {}
for obj in objects:
alias = obj["alias"]
if alias not in filtered_objects:
filtered_objects[alias] = obj
return list(filtered_objects.values())
def main():
"""
Module execution
:return:
"""
argument_spec = keycloak_argument_spec()
meta_args = dict(
realm=dict(type='str', required=True),
required_actions=dict(
type='list',
elements='dict',
options=dict(
alias=dict(type='str', required=True),
config=dict(type='dict'),
defaultAction=dict(type='bool'),
enabled=dict(type='bool'),
name=dict(type='str'),
priority=dict(type='int'),
providerId=dict(type='str')
)
),
state=dict(type='str', choices=['present', 'absent'], required=True)
)
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']]),
required_together=([['auth_realm', 'auth_username', 'auth_password']])
)
result = dict(changed=False, msg='', end_state={}, diff=dict(before={}, after={}))
# 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)
# Convenience variables
realm = module.params.get('realm')
desired_required_actions = module.params.get('required_actions')
state = module.params.get('state')
# Sanitize required actions
desired_required_actions = sanitize_required_actions(desired_required_actions)
# Filter out duplicate required actions
desired_required_actions = filter_duplicates(desired_required_actions)
# Get required actions
before_required_actions = kc.get_required_actions(realm=realm)
if state == 'present':
# Initialize empty lists to hold the required actions that need to be
# registered, updated, and original ones of the updated one
register_required_actions = []
before_updated_required_actions = []
updated_required_actions = []
# Loop through the desired required actions and check if they exist in the before required actions
for desired_required_action in desired_required_actions:
found = False
# Loop through the before required actions and check if the aliases match
for before_required_action in before_required_actions:
if desired_required_action['alias'] == before_required_action['alias']:
update_required = False
# Fill in the parameters
for k, v in before_required_action.items():
if k not in desired_required_action or desired_required_action[k] is None:
desired_required_action[k] = v
# Loop through the keys of the desired and before required actions
# and check if there are any differences between them
for key in desired_required_action.keys():
if key in before_required_action and desired_required_action[key] != before_required_action[key]:
update_required = True
break
# If there are differences, add the before and desired required actions
# to their respective lists for updating
if update_required:
before_updated_required_actions.append(before_required_action)
updated_required_actions.append(desired_required_action)
found = True
break
# If the desired required action is not found in the before required actions,
# add it to the list of required actions to register
if not found:
# Check if name is provided
if 'name' not in desired_required_action or desired_required_action['name'] is None:
module.fail_json(
msg='Unable to register required action %s in realm %s: name not included'
% (desired_required_action['alias'], realm)
)
# Check if provider ID is provided
if 'providerId' not in desired_required_action or desired_required_action['providerId'] is None:
module.fail_json(
msg='Unable to register required action %s in realm %s: providerId not included'
% (desired_required_action['alias'], realm)
)
register_required_actions.append(desired_required_action)
# Handle diff
if module._diff:
diff_required_actions = updated_required_actions.copy()
diff_required_actions.extend(register_required_actions)
result['diff'] = dict(
before=before_updated_required_actions,
after=diff_required_actions
)
# Handle changed
if register_required_actions or updated_required_actions:
result['changed'] = True
# Handle check mode
if module.check_mode:
if register_required_actions or updated_required_actions:
result['change'] = True
result['msg'] = 'Required actions would be registered/updated'
else:
result['change'] = False
result['msg'] = 'Required actions would not be registered/updated'
module.exit_json(**result)
# Register required actions
if register_required_actions:
for register_required_action in register_required_actions:
kc.register_required_action(realm=realm, rep=register_required_action)
kc.update_required_action(alias=register_required_action['alias'], realm=realm, rep=register_required_action)
# Update required actions
if updated_required_actions:
for updated_required_action in updated_required_actions:
kc.update_required_action(alias=updated_required_action['alias'], realm=realm, rep=updated_required_action)
# Initialize the final list of required actions
final_required_actions = []
# Iterate over the before_required_actions
for before_required_action in before_required_actions:
# Check if there is an updated_required_action with the same alias
updated_required_action_found = False
for updated_required_action in updated_required_actions:
if updated_required_action['alias'] == before_required_action['alias']:
# Merge the two dictionaries, favoring the values from updated_required_action
merged_dict = {}
for key in before_required_action.keys():
if key in updated_required_action:
merged_dict[key] = updated_required_action[key]
else:
merged_dict[key] = before_required_action[key]
for key in updated_required_action.keys():
if key not in before_required_action:
merged_dict[key] = updated_required_action[key]
# Add the merged dictionary to the final list of required actions
final_required_actions.append(merged_dict)
# Mark the updated_required_action as found
updated_required_action_found = True
# Stop looking for updated_required_action
break
# If no matching updated_required_action was found, add the before_required_action to the final list of required actions
if not updated_required_action_found:
final_required_actions.append(before_required_action)
# Append any remaining updated_required_actions that were not merged
for updated_required_action in updated_required_actions:
if not any(updated_required_action['alias'] == action['alias'] for action in final_required_actions):
final_required_actions.append(updated_required_action)
# Append newly registered required actions
final_required_actions.extend(register_required_actions)
# Handle message and end state
result['msg'] = 'Required actions registered/updated'
result['end_state'] = final_required_actions
else:
# Filter out the deleted required actions
final_required_actions = []
delete_required_actions = []
for before_required_action in before_required_actions:
delete_action = False
for desired_required_action in desired_required_actions:
if before_required_action['alias'] == desired_required_action['alias']:
delete_action = True
break
if not delete_action:
final_required_actions.append(before_required_action)
else:
delete_required_actions.append(before_required_action)
# Handle diff
if module._diff:
result['diff'] = dict(
before=before_required_actions,
after=final_required_actions
)
# Handle changed
if delete_required_actions:
result['changed'] = True
# Handle check mode
if module.check_mode:
if final_required_actions:
result['change'] = True
result['msg'] = 'Required actions would be deleted'
else:
result['change'] = False
result['msg'] = 'Required actions would not be deleted'
module.exit_json(**result)
# Delete required actions
if delete_required_actions:
for delete_required_action in delete_required_actions:
kc.delete_required_action(alias=delete_required_action['alias'], realm=realm)
# Handle message and end state
result['msg'] = 'Required actions deleted'
result['end_state'] = final_required_actions
module.exit_json(**result)
if __name__ == '__main__':
main()

835
tests/unit/plugins/modules/test_keycloak_authentication_required_actions.py Normal file
View file

@ -0,0 +1,835 @@
# -*- coding: utf-8 -*-
# Copyright (c) 2021, Ansible Project
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
from __future__ import absolute_import, division, print_function
__metaclass__ = type
from contextlib import contextmanager
from ansible_collections.community.general.tests.unit.compat import unittest
from ansible_collections.community.general.tests.unit.compat.mock import patch
from ansible_collections.community.general.tests.unit.plugins.modules.utils import AnsibleExitJson, ModuleTestCase, set_module_args
from ansible_collections.community.general.plugins.modules import keycloak_authentication_required_actions
from itertools import count
from ansible.module_utils.six import StringIO
@contextmanager
def patch_keycloak_api(
get_required_actions=None,
register_required_action=None,
update_required_action=None,
delete_required_action=None,
):
"""
Mock context manager for patching the methods in PwPolicyIPAClient that contact the IPA server
Patches the `login` and `_post_json` methods
Keyword arguments are passed to the mock object that patches `_post_json`
No arguments are passed to the mock object that patches `login` because no tests require it
Example::
with patch_ipa(return_value={}) as (mock_login, mock_post):
...
"""
obj = keycloak_authentication_required_actions.KeycloakAPI
with patch.object(
obj,
'get_required_actions',
side_effect=get_required_actions
) as mock_get_required_actions:
with patch.object(
obj,
'register_required_action',
side_effect=register_required_action
) as mock_register_required_action:
with patch.object(
obj,
'update_required_action',
side_effect=update_required_action
) as mock_update_required_action:
with patch.object(
obj,
'delete_required_action',
side_effect=delete_required_action
) as mock_delete_required_action:
yield (
mock_get_required_actions,
mock_register_required_action,
mock_update_required_action,
mock_delete_required_action,
)
def get_response(object_with_future_response, method, get_id_call_count):
if callable(object_with_future_response):
return object_with_future_response()
if isinstance(object_with_future_response, dict):
return get_response(
object_with_future_response[method], method, get_id_call_count)
if isinstance(object_with_future_response, list):
call_number = next(get_id_call_count)
return get_response(
object_with_future_response[call_number], method, get_id_call_count)
return object_with_future_response
def build_mocked_request(get_id_user_count, response_dict):
def _mocked_requests(*args, **kwargs):
url = args[0]
method = kwargs['method']
future_response = response_dict.get(url, None)
return get_response(future_response, method, get_id_user_count)
return _mocked_requests
def create_wrapper(text_as_string):
"""Allow to mock many times a call to one address.
Without this function, the StringIO is empty for the second call.
"""
def _create_wrapper():
return StringIO(text_as_string)
return _create_wrapper
def mock_good_connection():
token_response = {
'http://keycloak.url/auth/realms/master/protocol/openid-connect/token': create_wrapper('{"access_token": "alongtoken"}'), }
return patch(
'ansible_collections.community.general.plugins.module_utils.identity.keycloak.keycloak.open_url',
side_effect=build_mocked_request(count(), token_response),
autospec=True
)
class TestKeycloakAuthentication(ModuleTestCase):
def setUp(self):
super(TestKeycloakAuthentication, self).setUp()
self.module = keycloak_authentication_required_actions
def test_register_required_action(self):
"""Register a new authentication required action."""
module_args = {
'auth_client_id': 'admin-cli',
'auth_keycloak_url': 'http://keycloak.url/auth',
'auth_password': 'admin',
'auth_realm': 'master',
'auth_username': 'admin',
'realm': 'master',
'required_actions': [
{
'alias': 'test-provider-id',
'name': 'Test provider ID',
'providerId': 'test-provider-id',
},
{
'alias': 'test-provider-id',
'name': 'Test provider ID (DUPLICATE ALIAS)',
'providerId': 'test-provider-id',
},
{
'alias': 'test-provider-id',
'name': 'Test provider ID (DIFFERENT PROVIDER ID)',
'providerId': 'test-provider-id-diff',
},
],
'state': 'present',
}
return_value_required_actions = [
[
{
'alias': 'CONFIGURE_TOTP',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Configure OTP',
'priority': 10,
'providerId': 'CONFIGURE_TOTP'
},
{
'alias': 'TERMS_AND_CONDITIONS',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Terms and conditions',
'priority': 20,
'providerId': 'TERMS_AND_CONDITIONS'
},
{
'alias': 'UPDATE_PASSWORD',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Update Password',
'priority': 30,
'providerId': 'UPDATE_PASSWORD'
},
{
'alias': 'UPDATE_PROFILE',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Update Profile',
'priority': 40,
'providerId': 'UPDATE_PROFILE'
},
{
'alias': 'VERIFY_EMAIL',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Verify Email',
'priority': 50,
'providerId': 'VERIFY_EMAIL'
},
{
'alias': 'delete_account',
'config': {},
'defaultAction': False,
'enabled': False,
'name': 'Delete Account',
'priority': 60,
'providerId': 'delete_account'
},
{
'alias': 'webauthn-register',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Webauthn Register',
'priority': 70,
'providerId': 'webauthn-register'
},
{
'alias': 'webauthn-register-passwordless',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Webauthn Register Passwordless',
'priority': 80,
'providerId': 'webauthn-register-passwordless'
},
{
'alias': 'update_user_locale',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Update User Locale',
'priority': 1000,
'providerId': 'update_user_locale'
}
],
]
changed = True
set_module_args(module_args)
# Run the module
with mock_good_connection():
with patch_keycloak_api(
get_required_actions=return_value_required_actions,
) as (
mock_get_required_actions,
mock_register_required_action,
mock_update_required_action,
mock_delete_required_action,
):
with self.assertRaises(AnsibleExitJson) as exec_info:
self.module.main()
# Verify number of call on each mock
self.assertEqual(len(mock_get_required_actions.mock_calls), 1)
self.assertEqual(len(mock_update_required_action.mock_calls), 1)
self.assertEqual(len(mock_register_required_action.mock_calls), 1)
self.assertEqual(len(mock_delete_required_action.mock_calls), 0)
# Verify that the module's changed status matches what is expected
self.assertIs(exec_info.exception.args[0]['changed'], changed)
def test_register_required_action_idempotency(self):
"""Register an already existing new authentication required action again."""
module_args = {
'auth_client_id': 'admin-cli',
'auth_keycloak_url': 'http://keycloak.url/auth',
'auth_password': 'admin',
'auth_realm': 'master',
'auth_username': 'admin',
'realm': 'master',
'required_actions': [
{
'alias': 'test-provider-id',
'name': 'Test provider ID',
'providerId': 'test-provider-id',
},
{
'alias': 'test-provider-id',
'name': 'Test provider ID (DUPLICATE ALIAS)',
'providerId': 'test-provider-id',
},
{
'alias': 'test-provider-id',
'name': 'Test provider ID (DIFFERENT PROVIDER ID)',
'providerId': 'test-provider-id-diff',
},
],
'state': 'present',
}
return_value_required_actions = [
[
{
'alias': 'CONFIGURE_TOTP',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Configure OTP',
'priority': 10,
'providerId': 'CONFIGURE_TOTP'
},
{
'alias': 'TERMS_AND_CONDITIONS',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Terms and conditions',
'priority': 20,
'providerId': 'TERMS_AND_CONDITIONS'
},
{
'alias': 'UPDATE_PASSWORD',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Update Password',
'priority': 30,
'providerId': 'UPDATE_PASSWORD'
},
{
'alias': 'UPDATE_PROFILE',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Update Profile',
'priority': 40,
'providerId': 'UPDATE_PROFILE'
},
{
'alias': 'VERIFY_EMAIL',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Verify Email',
'priority': 50,
'providerId': 'VERIFY_EMAIL'
},
{
'alias': 'delete_account',
'config': {},
'defaultAction': False,
'enabled': False,
'name': 'Delete Account',
'priority': 60,
'providerId': 'delete_account'
},
{
'alias': 'webauthn-register',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Webauthn Register',
'priority': 70,
'providerId': 'webauthn-register'
},
{
'alias': 'webauthn-register-passwordless',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Webauthn Register Passwordless',
'priority': 80,
'providerId': 'webauthn-register-passwordless'
},
{
'alias': 'test-provider-id',
'config': {},
'defaultAction': False,
'enabled': False,
'name': 'Test provider ID',
'priority': 90,
'providerId': 'test-provider-id'
},
{
'alias': 'update_user_locale',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Update User Locale',
'priority': 1000,
'providerId': 'update_user_locale'
}
],
]
changed = False
set_module_args(module_args)
# Run the module
with mock_good_connection():
with patch_keycloak_api(
get_required_actions=return_value_required_actions,
) as (
mock_get_required_actions,
mock_register_required_action,
mock_update_required_action,
mock_delete_required_action,
):
with self.assertRaises(AnsibleExitJson) as exec_info:
self.module.main()
# Verify number of call on each mock
self.assertEqual(len(mock_get_required_actions.mock_calls), 1)
self.assertEqual(len(mock_update_required_action.mock_calls), 0)
self.assertEqual(len(mock_register_required_action.mock_calls), 0)
self.assertEqual(len(mock_delete_required_action.mock_calls), 0)
# Verify that the module's changed status matches what is expected
self.assertIs(exec_info.exception.args[0]['changed'], changed)
def test_update_required_actions(self):
"""Update an authentication required action."""
module_args = {
'auth_client_id': 'admin-cli',
'auth_keycloak_url': 'http://keycloak.url/auth',
'auth_password': 'admin',
'auth_realm': 'master',
'auth_username': 'admin',
'realm': 'master',
'required_actions': [
{
'alias': 'test-provider-id',
'name': 'Test provider ID UPDATED',
'providerId': 'test-provider-id',
},
{
'alias': 'test-provider-id',
'name': 'Test provider ID UPDATED (DUPLICATE ALIAS)',
'providerId': 'test-provider-id',
},
{
'alias': 'test-provider-id',
'name': 'Test provider ID UPDATED (DIFFERENT PROVIDER ID)',
'providerId': 'test-provider-id-diff',
},
],
'state': 'present',
}
return_value_required_actions = [
[
{
'alias': 'CONFIGURE_TOTP',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Configure OTP',
'priority': 10,
'providerId': 'CONFIGURE_TOTP'
},
{
'alias': 'TERMS_AND_CONDITIONS',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Terms and conditions',
'priority': 20,
'providerId': 'TERMS_AND_CONDITIONS'
},
{
'alias': 'UPDATE_PASSWORD',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Update Password',
'priority': 30,
'providerId': 'UPDATE_PASSWORD'
},
{
'alias': 'UPDATE_PROFILE',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Update Profile',
'priority': 40,
'providerId': 'UPDATE_PROFILE'
},
{
'alias': 'VERIFY_EMAIL',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Verify Email',
'priority': 50,
'providerId': 'VERIFY_EMAIL'
},
{
'alias': 'delete_account',
'config': {},
'defaultAction': False,
'enabled': False,
'name': 'Delete Account',
'priority': 60,
'providerId': 'delete_account'
},
{
'alias': 'webauthn-register',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Webauthn Register',
'priority': 70,
'providerId': 'webauthn-register'
},
{
'alias': 'webauthn-register-passwordless',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Webauthn Register Passwordless',
'priority': 80,
'providerId': 'webauthn-register-passwordless'
},
{
'alias': 'test-provider-id',
'config': {},
'defaultAction': False,
'enabled': False,
'name': 'Test provider ID',
'priority': 90,
'providerId': 'test-provider-id'
},
{
'alias': 'update_user_locale',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Update User Locale',
'priority': 1000,
'providerId': 'update_user_locale'
}
],
]
changed = True
set_module_args(module_args)
# Run the module
with mock_good_connection():
with patch_keycloak_api(
get_required_actions=return_value_required_actions,
) as (
mock_get_required_actions,
mock_register_required_action,
mock_update_required_action,
mock_delete_required_action,
):
with self.assertRaises(AnsibleExitJson) as exec_info:
self.module.main()
# Verify number of call on each mock
self.assertEqual(len(mock_get_required_actions.mock_calls), 1)
self.assertEqual(len(mock_update_required_action.mock_calls), 1)
self.assertEqual(len(mock_register_required_action.mock_calls), 0)
self.assertEqual(len(mock_delete_required_action.mock_calls), 0)
# Verify that the module's changed status matches what is expected
self.assertIs(exec_info.exception.args[0]['changed'], changed)
def test_delete_required_action(self):
"""Delete a registered authentication required action."""
module_args = {
'auth_client_id': 'admin-cli',
'auth_keycloak_url': 'http://keycloak.url/auth',
'auth_password': 'admin',
'auth_realm': 'master',
'auth_username': 'admin',
'realm': 'master',
'required_actions': [
{
'alias': 'test-provider-id',
},
],
'state': 'absent',
}
return_value_required_actions = [
[
{
'alias': 'CONFIGURE_TOTP',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Configure OTP',
'priority': 10,
'providerId': 'CONFIGURE_TOTP'
},
{
'alias': 'TERMS_AND_CONDITIONS',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Terms and conditions',
'priority': 20,
'providerId': 'TERMS_AND_CONDITIONS'
},
{
'alias': 'UPDATE_PASSWORD',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Update Password',
'priority': 30,
'providerId': 'UPDATE_PASSWORD'
},
{
'alias': 'UPDATE_PROFILE',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Update Profile',
'priority': 40,
'providerId': 'UPDATE_PROFILE'
},
{
'alias': 'VERIFY_EMAIL',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Verify Email',
'priority': 50,
'providerId': 'VERIFY_EMAIL'
},
{
'alias': 'delete_account',
'config': {},
'defaultAction': False,
'enabled': False,
'name': 'Delete Account',
'priority': 60,
'providerId': 'delete_account'
},
{
'alias': 'webauthn-register',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Webauthn Register',
'priority': 70,
'providerId': 'webauthn-register'
},
{
'alias': 'webauthn-register-passwordless',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Webauthn Register Passwordless',
'priority': 80,
'providerId': 'webauthn-register-passwordless'
},
{
'alias': 'test-provider-id',
'config': {},
'defaultAction': False,
'enabled': False,
'name': 'Test provider ID',
'priority': 90,
'providerId': 'test-provider-id'
},
{
'alias': 'update_user_locale',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Update User Locale',
'priority': 1000,
'providerId': 'update_user_locale'
}
],
]
changed = True
set_module_args(module_args)
# Run the module
with mock_good_connection():
with patch_keycloak_api(
get_required_actions=return_value_required_actions,
) as (
mock_get_required_actions,
mock_register_required_action,
mock_update_required_action,
mock_delete_required_action,
):
with self.assertRaises(AnsibleExitJson) as exec_info:
self.module.main()
# Verify number of call on each mock
self.assertEqual(len(mock_get_required_actions.mock_calls), 1)
self.assertEqual(len(mock_update_required_action.mock_calls), 0)
self.assertEqual(len(mock_register_required_action.mock_calls), 0)
self.assertEqual(len(mock_delete_required_action.mock_calls), 1)
# Verify that the module's changed status matches what is expected
self.assertIs(exec_info.exception.args[0]['changed'], changed)
def test_delete_required_action_idempotency(self):
"""Delete an already deleted authentication required action."""
module_args = {
'auth_client_id': 'admin-cli',
'auth_keycloak_url': 'http://keycloak.url/auth',
'auth_password': 'admin',
'auth_realm': 'master',
'auth_username': 'admin',
'realm': 'master',
'required_actions': [
{
'alias': 'test-provider-id',
},
],
'state': 'absent',
}
return_value_required_actions = [
[
{
'alias': 'CONFIGURE_TOTP',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Configure OTP',
'priority': 10,
'providerId': 'CONFIGURE_TOTP'
},
{
'alias': 'TERMS_AND_CONDITIONS',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Terms and conditions',
'priority': 20,
'providerId': 'TERMS_AND_CONDITIONS'
},
{
'alias': 'UPDATE_PASSWORD',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Update Password',
'priority': 30,
'providerId': 'UPDATE_PASSWORD'
},
{
'alias': 'UPDATE_PROFILE',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Update Profile',
'priority': 40,
'providerId': 'UPDATE_PROFILE'
},
{
'alias': 'VERIFY_EMAIL',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Verify Email',
'priority': 50,
'providerId': 'VERIFY_EMAIL'
},
{
'alias': 'delete_account',
'config': {},
'defaultAction': False,
'enabled': False,
'name': 'Delete Account',
'priority': 60,
'providerId': 'delete_account'
},
{
'alias': 'webauthn-register',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Webauthn Register',
'priority': 70,
'providerId': 'webauthn-register'
},
{
'alias': 'webauthn-register-passwordless',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Webauthn Register Passwordless',
'priority': 80,
'providerId': 'webauthn-register-passwordless'
},
{
'alias': 'update_user_locale',
'config': {},
'defaultAction': False,
'enabled': True,
'name': 'Update User Locale',
'priority': 1000,
'providerId': 'update_user_locale'
}
],
]
changed = False
set_module_args(module_args)
# Run the module
with mock_good_connection():
with patch_keycloak_api(
get_required_actions=return_value_required_actions,
) as (
mock_get_required_actions,
mock_register_required_action,
mock_update_required_action,
mock_delete_required_action,
):
with self.assertRaises(AnsibleExitJson) as exec_info:
self.module.main()
# Verify number of call on each mock
self.assertEqual(len(mock_get_required_actions.mock_calls), 1)
self.assertEqual(len(mock_update_required_action.mock_calls), 0)
self.assertEqual(len(mock_register_required_action.mock_calls), 0)
self.assertEqual(len(mock_delete_required_action.mock_calls), 0)
# Verify that the module's changed status matches what is expected
self.assertIs(exec_info.exception.args[0]['changed'], changed)
if __name__ == '__main__':
unittest.main()