ansible-collections/community.general
1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2025-08-06 06:04:24 -07:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Inspq keycloak role composites (#6469)

Browse source 
* Add composites to keycloak_role module

* Add composites support for realm role in keycloak module_utils

* Clean f.write from keycloak_role module

* keycloak_role support state for realm role composites

* Add support for composites in client role for keycloak_role module

* Add changelog fragment for keycloak role composites PR

* Fix pep8 and validate-modules tests errors

* Update changelogs/fragments/6469-add-composites-support-for-keycloak-role.yml

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

* Update plugins/modules/keycloak_role.py

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

* Update plugins/modules/keycloak_role.py

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

* Update plugins/modules/keycloak_role.py

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

* Update plugins/modules/keycloak_role.py

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

* Update plugins/modules/keycloak_role.py

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

* Update plugins/modules/keycloak_role.py

I will try it

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

* Fix test_keycloak_role assertion

* Fix role composite compare before update in keycloak_role module

* Fix realm problem with update_role_composites in keycloak.py module_utils

* Add units tests for composites and client roles in keycloak_role module

* Update plugins/module_utils/identity/keycloak/keycloak.py

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

* Update plugins/module_utils/identity/keycloak/keycloak.py

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

* Change try in is_struct_included and add unit tests for keycloak.py module_utils

* Add integration tests for composites roles and fix bug with non master roles in keycloak_role module

* Update plugins/modules/keycloak_role.py

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

* Update plugins/modules/keycloak_role.py

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

* Update plugins/modules/keycloak_role.py

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

* Update plugins/module_utils/identity/keycloak/keycloak.py

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

* Update plugins/module_utils/identity/keycloak/keycloak.py

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

* is_struct_included refactor

---------

Co-authored-by: Felix Fontein <felix@fontein.de>
This commit is contained in:
Philippe Gauthier 2023-06-15 00:57:30 -04:00 • committed by GitHub
commit 9395df1c6f
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 860 additions and 17 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -1680,6 +1680,9 @@ class KeycloakAPI(object):
"""
roles_url = URL_REALM_ROLES.format(url=self.baseurl, realm=realm)
try:
if "composites" in rolerep:
keycloak_compatible_composites = self.convert_role_composites(rolerep["composites"])
rolerep["composites"] = keycloak_compatible_composites
return open_url(roles_url, method='POST', http_agent=self.http_agent, headers=self.restheaders, timeout=self.connection_timeout,
data=json.dumps(rolerep), validate_certs=self.validate_certs)
except Exception as e:
@ -1694,12 +1697,124 @@ class KeycloakAPI(object):
"""
role_url = URL_REALM_ROLE.format(url=self.baseurl, realm=realm, name=quote(rolerep['name']))
try:
return open_url(role_url, method='PUT', http_agent=self.http_agent, headers=self.restheaders, timeout=self.connection_timeout,
data=json.dumps(rolerep), validate_certs=self.validate_certs)
composites = None
if "composites" in rolerep:
composites = copy.deepcopy(rolerep["composites"])
del rolerep["composites"]
role_response = open_url(role_url, method='PUT', http_agent=self.http_agent, headers=self.restheaders, timeout=self.connection_timeout,
data=json.dumps(rolerep), validate_certs=self.validate_certs)
if composites is not None:
self.update_role_composites(rolerep=rolerep, composites=composites, realm=realm)
return role_response
except Exception as e:
self.module.fail_json(msg='Could not update role %s in realm %s: %s'
% (rolerep['name'], realm, str(e)))
def get_role_composites(self, rolerep, clientid=None, realm='master'):
composite_url = ''
try:
if clientid is not None:
client = self.get_client_by_clientid(client_id=clientid, realm=realm)
cid = client['id']
composite_url = URL_CLIENT_ROLE_COMPOSITES.format(url=self.baseurl, realm=realm, id=cid, name=quote(rolerep["name"]))
else:
composite_url = URL_REALM_ROLE_COMPOSITES.format(url=self.baseurl, realm=realm, name=quote(rolerep["name"]))
# Get existing composites
return json.loads(to_native(open_url(
composite_url,
method='GET',
http_agent=self.http_agent,
headers=self.restheaders,
timeout=self.connection_timeout,
validate_certs=self.validate_certs).read()))
except Exception as e:
self.module.fail_json(msg='Could not get role %s composites in realm %s: %s'
% (rolerep['name'], realm, str(e)))
def create_role_composites(self, rolerep, composites, clientid=None, realm='master'):
composite_url = ''
try:
if clientid is not None:
client = self.get_client_by_clientid(client_id=clientid, realm=realm)
cid = client['id']
composite_url = URL_CLIENT_ROLE_COMPOSITES.format(url=self.baseurl, realm=realm, id=cid, name=quote(rolerep["name"]))
else:
composite_url = URL_REALM_ROLE_COMPOSITES.format(url=self.baseurl, realm=realm, name=quote(rolerep["name"]))
# Get existing composites
# create new composites
return open_url(composite_url, method='POST', http_agent=self.http_agent, headers=self.restheaders, timeout=self.connection_timeout,
data=json.dumps(composites), validate_certs=self.validate_certs)
except Exception as e:
self.module.fail_json(msg='Could not create role %s composites in realm %s: %s'
% (rolerep['name'], realm, str(e)))
def delete_role_composites(self, rolerep, composites, clientid=None, realm='master'):
composite_url = ''
try:
if clientid is not None:
client = self.get_client_by_clientid(client_id=clientid, realm=realm)
cid = client['id']
composite_url = URL_CLIENT_ROLE_COMPOSITES.format(url=self.baseurl, realm=realm, id=cid, name=quote(rolerep["name"]))
else:
composite_url = URL_REALM_ROLE_COMPOSITES.format(url=self.baseurl, realm=realm, name=quote(rolerep["name"]))
# Get existing composites
# create new composites
return open_url(composite_url, method='DELETE', http_agent=self.http_agent, headers=self.restheaders, timeout=self.connection_timeout,
data=json.dumps(composites), validate_certs=self.validate_certs)
except Exception as e:
self.module.fail_json(msg='Could not create role %s composites in realm %s: %s'
% (rolerep['name'], realm, str(e)))
def update_role_composites(self, rolerep, composites, clientid=None, realm='master'):
# Get existing composites
existing_composites = self.get_role_composites(rolerep=rolerep, clientid=clientid, realm=realm)
composites_to_be_created = []
composites_to_be_deleted = []
for composite in composites:
composite_found = False
existing_composite_client = None
for existing_composite in existing_composites:
if existing_composite["clientRole"]:
existing_composite_client = self.get_client_by_id(existing_composite["containerId"], realm=realm)
if ("client_id" in composite
and composite['client_id'] is not None
and existing_composite_client["clientId"] == composite["client_id"]
and composite["name"] == existing_composite["name"]):
composite_found = True
break
else:
if (("client_id" not in composite or composite['client_id'] is None)
and composite["name"] == existing_composite["name"]):
composite_found = True
break
if (not composite_found and ('state' not in composite or composite['state'] == 'present')):
if "client_id" in composite and composite['client_id'] is not None:
client_roles = self.get_client_roles(clientid=composite['client_id'], realm=realm)
for client_role in client_roles:
if client_role['name'] == composite['name']:
composites_to_be_created.append(client_role)
break
else:
realm_role = self.get_realm_role(name=composite["name"], realm=realm)
composites_to_be_created.append(realm_role)
elif composite_found and 'state' in composite and composite['state'] == 'absent':
if "client_id" in composite and composite['client_id'] is not None:
client_roles = self.get_client_roles(clientid=composite['client_id'], realm=realm)
for client_role in client_roles:
if client_role['name'] == composite['name']:
composites_to_be_deleted.append(client_role)
break
else:
realm_role = self.get_realm_role(name=composite["name"], realm=realm)
composites_to_be_deleted.append(realm_role)
if len(composites_to_be_created) > 0:
# create new composites
self.create_role_composites(rolerep=rolerep, composites=composites_to_be_created, clientid=clientid, realm=realm)
if len(composites_to_be_deleted) > 0:
# delete new composites
self.delete_role_composites(rolerep=rolerep, composites=composites_to_be_deleted, clientid=clientid, realm=realm)
def delete_realm_role(self, name, realm='master'):
""" Delete a realm role.
@ -1778,12 +1893,30 @@ class KeycloakAPI(object):
% (clientid, realm))
roles_url = URL_CLIENT_ROLES.format(url=self.baseurl, realm=realm, id=cid)
try:
if "composites" in rolerep:
keycloak_compatible_composites = self.convert_role_composites(rolerep["composites"])
rolerep["composites"] = keycloak_compatible_composites
return open_url(roles_url, method='POST', http_agent=self.http_agent, headers=self.restheaders, timeout=self.connection_timeout,
data=json.dumps(rolerep), validate_certs=self.validate_certs)
except Exception as e:
self.module.fail_json(msg='Could not create role %s for client %s in realm %s: %s'
% (rolerep['name'], clientid, realm, str(e)))
def convert_role_composites(self, composites):
keycloak_compatible_composites = {
'client': {},
'realm': []
}
for composite in composites:
if 'state' not in composite or composite['state'] == 'present':
if "client_id" in composite and composite["client_id"] is not None:
if composite["client_id"] not in keycloak_compatible_composites["client"]:
keycloak_compatible_composites["client"][composite["client_id"]] = []
keycloak_compatible_composites["client"][composite["client_id"]].append(composite["name"])
else:
keycloak_compatible_composites["realm"].append(composite["name"])
return keycloak_compatible_composites
def update_client_role(self, rolerep, clientid, realm="master"):
""" Update an existing client role.
@ -1798,8 +1931,15 @@ class KeycloakAPI(object):
% (clientid, realm))
role_url = URL_CLIENT_ROLE.format(url=self.baseurl, realm=realm, id=cid, name=quote(rolerep['name']))
try:
return open_url(role_url, method='PUT', http_agent=self.http_agent, headers=self.restheaders, timeout=self.connection_timeout,
data=json.dumps(rolerep), validate_certs=self.validate_certs)
composites = None
if "composites" in rolerep:
composites = copy.deepcopy(rolerep["composites"])
del rolerep['composites']
update_role_response = open_url(role_url, method='PUT', http_agent=self.http_agent, headers=self.restheaders, timeout=self.connection_timeout,
data=json.dumps(rolerep), validate_certs=self.validate_certs)
if composites is not None:
self.update_role_composites(rolerep=rolerep, clientid=clientid, composites=composites, realm=realm)
return update_role_response
except Exception as e:
self.module.fail_json(msg='Could not update role %s for client %s in realm %s: %s'
% (rolerep['name'], clientid, realm, str(e)))

77
plugins/modules/keycloak_role.py
View file

@ -77,6 +77,42 @@ options:
description:
- A dict of key/value pairs to set as custom attributes for the role.
- Values may be single values (e.g. a string) or a list of strings.
composite:
description:
- If V(true), the role is a composition of other realm and/or client role.
default: false
type: bool
version_added: 7.1.0
composites:
description:
- List of roles to include to the composite realm role.
- If the composite role is a client role, the C(clientId) (not ID of the client) must be specified.
default: []
type: list
elements: dict
version_added: 7.1.0
suboptions:
name:
description:
- Name of the role. This can be the name of a REALM role or a client role.
type: str
required: true
client_id:
description:
- Client ID if the role is a client role. Do not include this option for a REALM role.
- Use the client ID you can see in the Keycloak console, not the technical ID of the client.
type: str
required: false
aliases:
- clientId
state:
description:
- Create the composite if present, remove it if absent.
type: str
choices:
- present
- absent
default: present
extends_documentation_fragment:
- community.general.keycloak
@ -198,8 +234,9 @@ end_state:
'''
from ansible_collections.community.general.plugins.module_utils.identity.keycloak.keycloak import KeycloakAPI, camel, \
keycloak_argument_spec, get_token, KeycloakError
keycloak_argument_spec, get_token, KeycloakError, is_struct_included
from ansible.module_utils.basic import AnsibleModule
import copy
def main():
@ -210,6 +247,12 @@ def main():
"""
argument_spec = keycloak_argument_spec()
composites_spec = dict(
name=dict(type='str', required=True),
client_id=dict(type='str', aliases=['clientId'], required=False),
state=dict(type='str', default='present', choices=['present', 'absent'])
)
meta_args = dict(
state=dict(type='str', default='present', choices=['present', 'absent']),
name=dict(type='str', required=True),
@ -217,6 +260,8 @@ def main():
realm=dict(type='str', default='master'),
client_id=dict(type='str'),
attributes=dict(type='dict'),
composites=dict(type='list', default=[], options=composites_spec, elements='dict'),
composite=dict(type='bool', default=False),
)
argument_spec.update(meta_args)
@ -250,7 +295,7 @@ def main():
# Filter and map the parameters names that apply to the role
role_params = [x for x in module.params
if x not in list(keycloak_argument_spec().keys()) + ['state', 'realm', 'client_id', 'composites'] and
if x not in list(keycloak_argument_spec().keys()) + ['state', 'realm', 'client_id'] and
module.params.get(x) is not None]
# See if it already exists in Keycloak
@ -269,10 +314,10 @@ def main():
new_param_value = module.params.get(param)
old_value = before_role[param] if param in before_role else None
if new_param_value != old_value:
changeset[camel(param)] = new_param_value
changeset[camel(param)] = copy.deepcopy(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_role = before_role.copy()
desired_role = copy.deepcopy(before_role)
desired_role.update(changeset)
result['proposed'] = changeset
@ -309,6 +354,9 @@ def main():
kc.create_client_role(desired_role, clientid, realm)
after_role = kc.get_client_role(name, clientid, realm)
if after_role['composite']:
after_role['composites'] = kc.get_role_composites(rolerep=after_role, clientid=clientid, realm=realm)
result['end_state'] = after_role
result['msg'] = 'Role {name} has been created'.format(name=name)
@ -316,10 +364,25 @@ def main():
else:
if state == 'present':
compare_exclude = []
if 'composites' in desired_role and isinstance(desired_role['composites'], list) and len(desired_role['composites']) > 0:
composites = kc.get_role_composites(rolerep=before_role, clientid=clientid, realm=realm)
before_role['composites'] = []
for composite in composites:
before_composite = {}
if composite['clientRole']:
composite_client = kc.get_client_by_id(id=composite['containerId'], realm=realm)
before_composite['client_id'] = composite_client['clientId']
else:
before_composite['client_id'] = None
before_composite['name'] = composite['name']
before_composite['state'] = 'present'
before_role['composites'].append(before_composite)
else:
compare_exclude.append('composites')
# Process an update
# no changes
if desired_role == before_role:
if is_struct_included(desired_role, before_role, exclude=compare_exclude):
result['changed'] = False
result['end_state'] = desired_role
result['msg'] = "No changes required to role {name}.".format(name=name)
@ -341,6 +404,8 @@ def main():
else:
kc.update_client_role(desired_role, clientid, realm)
after_role = kc.get_client_role(name, clientid, realm)
if after_role['composite']:
after_role['composites'] = kc.get_role_composites(rolerep=after_role, clientid=clientid, realm=realm)
result['end_state'] = after_role