ansible-collections/community.general
1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2025-07-25 22:30:22 -07:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

[PR #6469/9395df1c backport][stable-7] Inspq keycloak role composites (#6693)

Browse source 
Inspq keycloak role composites (#6469)

* 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>
(cherry picked from commit 9395df1c6f)

Co-authored-by: Philippe Gauthier <philippe.gauthier@inspq.qc.ca>
This commit is contained in:
patchback[bot] 2023-06-15 08:42:10 +02:00 • committed by GitHub
commit 41ba810463
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)))