diff --git a/conf/rhbk.yaml.template b/conf/rhbk.yaml.template new file mode 100755 index 00000000000..7fa1da22c9a --- /dev/null +++ b/conf/rhbk.yaml.template @@ -0,0 +1,15 @@ +# section for RHBK integration +RHBK: + # RHBK Hostname + HOST_NAME: # update with rhbk host name + # RHBK port, 8443 by default + HOST_PORT: # update with rhbk host name + # RHBK Host Url + HOST_URL: # update with rhbk environment url + # RHBK Host Admin of Realm + RHBK_USER: sat_admin + # RHBK Host Admin Password + RHBK_PASSWORD: # update with password + # RHBK Host Realm + REALM: satqe + TOTP_SECRET: # update with the totp secret token diff --git a/pytest_fixtures/component/satellite_auth.py b/pytest_fixtures/component/satellite_auth.py index 5a21ff6a75a..465f85147a7 100644 --- a/pytest_fixtures/component/satellite_auth.py +++ b/pytest_fixtures/component/satellite_auth.py @@ -22,8 +22,10 @@ @pytest.fixture(scope='module') -def default_sso_host(module_target_sat): +def default_sso_host(request, module_target_sat): """Returns default sso host""" + if hasattr(request, 'param'): + return SSOHost(module_target_sat, rhbk=request.param) return SSOHost(module_target_sat) @@ -287,8 +289,16 @@ def auth_data(request, ad_data, ipa_data): @pytest.fixture(scope='module') -def enroll_configure_rhsso_external_auth(module_target_sat): +def enroll_configure_rhsso_external_auth(request, module_target_sat): """Enroll the Satellite6 Server to an RHSSO Server.""" + if hasattr(request, 'param') and request.param: + uri = f'https://{settings.rhbk.host_name}:{settings.rhbk.host_port}' + password = settings.rhbk.rhbk_password + realm = settings.rhbk.realm + else: + uri = f'https://{settings.rhsso.host_name}:443' + password = settings.rhsso.rhsso_password + realm = settings.rhsso.realm if settings.robottelo.rhel_source == "ga": module_target_sat.register_to_cdn() # keycloak-httpd-client-install needs lxml but it's not an rpm dependency + is not documented @@ -302,18 +312,18 @@ def enroll_configure_rhsso_external_auth(module_target_sat): # if target directory not given it is installing in /usr/local/lib64 assert ( module_target_sat.execute( - f'openssl s_client -connect {settings.rhsso.host_name}:443 -showcerts /dev/null| ' + f'openssl s_client -connect {uri} -showcerts /dev/null| ' f'sed "/BEGIN CERTIFICATE/,/END CERTIFICATE/!d" > {CERT_PATH}/rh-sso.crt' ).status == 0 ) assert ( module_target_sat.execute( - f'echo {settings.rhsso.rhsso_password} | keycloak-httpd-client-install \ + f'echo {password} | keycloak-httpd-client-install \ --app-name foreman-openidc \ - --keycloak-server-url {settings.rhsso.host_url} \ + --keycloak-server-url {uri} \ --keycloak-admin-username "admin" \ - --keycloak-realm "{settings.rhsso.realm}" \ + --keycloak-realm "{realm}" \ --keycloak-admin-realm master \ --keycloak-auth-role root-admin -t openidc -l /users/extlogin --force' ).status @@ -323,7 +333,7 @@ def enroll_configure_rhsso_external_auth(module_target_sat): module_target_sat.execute( f'satellite-installer --foreman-keycloak true ' f"--foreman-keycloak-app-name 'foreman-openidc' " - f"--foreman-keycloak-realm '{settings.rhsso.realm}' ", + f"--foreman-keycloak-realm '{realm}' ", timeout=1000000, ).status == 0 @@ -380,17 +390,22 @@ def configure_realm(module_target_sat, default_ipa_host): @pytest.fixture(scope="module") -def rhsso_setting_setup(module_target_sat): +def rhsso_setting_setup(request, module_target_sat): """Update the RHSSO setting and revert it in cleanup""" + if hasattr(request, 'param') and request.param: + uri = f'{settings.rhbk.host_url}' + realm = settings.rhbk.realm + else: + uri = settings.rhsso.host_url + realm = settings.rhsso.realm rhhso_settings = { 'authorize_login_delegation': True, 'authorize_login_delegation_auth_source_user_autocreate': 'External', 'login_delegation_logout_url': f'https://{module_target_sat.hostname}/users/extlogout', 'oidc_algorithm': 'RS256', 'oidc_audience': [f'{module_target_sat.hostname}-foreman-openidc'], - 'oidc_issuer': f'{settings.rhsso.host_url}/auth/realms/{settings.rhsso.realm}', - 'oidc_jwks_url': f'{settings.rhsso.host_url}/auth/realms' - f'/{settings.rhsso.realm}/protocol/openid-connect/certs', + 'oidc_issuer': f'{uri}/auth/realms/{realm}', + 'oidc_jwks_url': f'{uri}/auth/realms' f'/{realm}/protocol/openid-connect/certs', } for setting_name, setting_value in rhhso_settings.items(): # replace entietes field with targetsat.api diff --git a/robottelo/constants/__init__.py b/robottelo/constants/__init__.py index ba7e7498492..c9ef69afa2b 100644 --- a/robottelo/constants/__init__.py +++ b/robottelo/constants/__init__.py @@ -198,6 +198,8 @@ VALID_GPG_KEY_BETA_FILE = "valid_gpg_key_beta.txt" KEY_CLOAK_CLI = "/opt/rh/rh-sso7/root/usr/share/keycloak/bin/kcadm.sh" +# this symlink needs to be created manually on the RHBK instance; default path is something version-specific like /opt/rhbk-24.0.6/bin/kcadm.sh +RHBK_CLI = "/bin/kcadm.sh" RPM_TO_UPLOAD = "which-2.19-6.el6.x86_64.rpm" SRPM_TO_UPLOAD = "which-2.19-6.el6.src.rpm" diff --git a/robottelo/hosts.py b/robottelo/hosts.py index 31a01dab624..f42f2d49065 100644 --- a/robottelo/hosts.py +++ b/robottelo/hosts.py @@ -42,6 +42,7 @@ CUSTOM_PUPPET_MODULE_REPOS_VERSION, HAMMER_CONFIG, KEY_CLOAK_CLI, + RHBK_CLI, RHSSO_NEW_GROUP, RHSSO_NEW_USER, RHSSO_RESET_PASSWORD, @@ -2422,24 +2423,44 @@ def run_orphan_cleanup(self, smart_proxy_id=None): class SSOHost(Host): """Class for RHSSO functions and setup""" - def __init__(self, sat_obj, **kwargs): + def __init__(self, sat_obj, rhbk=False, **kwargs): self.satellite = sat_obj - kwargs['hostname'] = kwargs.get('hostname', settings.rhsso.host_name) + if rhbk: + self.rhbk = True + self.host_url = settings.rhbk.host_url + self.host_name = settings.rhbk.host_name + self.host_port = settings.rhbk.host_port + self.realm = settings.rhbk.realm + self.user = settings.rhbk.rhbk_user + self.password = settings.rhbk.rhbk_password + self.kcadm = RHBK_CLI + kwargs['hostname'] = kwargs.get('hostname', settings.rhbk.host_name) + else: + self.rhbk = False + self.host_url = settings.rhsso.host_url + self.host_name = settings.rhsso.host_name + self.host_port = 443 + self.realm = settings.rhsso.realm + self.user = settings.rhsso.rhsso_user + self.password = settings.rhsso.rhsso_password + self.kcadm = KEY_CLOAK_CLI + kwargs['hostname'] = kwargs.get('hostname', settings.rhsso.host_name) kwargs['ipv6'] = kwargs.get('ipv6', settings.server.is_ipv6) super().__init__(**kwargs) def get_rhsso_client_id(self): """getter method for fetching the client id and can be used other functions""" client_name = f'{self.satellite.hostname}-foreman-openidc' + uri = self.host_url if self.rhbk else self.host_url.replace("https://", "http://") self.execute( - f'{KEY_CLOAK_CLI} config credentials ' - f'--server {settings.rhsso.host_url.replace("https://", "http://")}/auth ' - f'--realm {settings.rhsso.realm} ' - f'--user {settings.rhsso.rhsso_user} ' - f'--password {settings.rhsso.rhsso_password}' + f'{self.kcadm} config credentials ' + f'--server {uri}/auth ' + f'--realm {self.realm} ' + f'--user {self.user} ' + f'--password {self.password}' ) - result = self.execute(f'{KEY_CLOAK_CLI} get clients --fields id,clientId') + result = self.execute(f'{self.kcadm} get clients --fields id,clientId') result_json = json.loads(result.stdout) client_id = None for client in result_json: @@ -2451,16 +2472,14 @@ def get_rhsso_client_id(self): @lru_cache def get_rhsso_user_details(self, username): """Getter method to receive the user id""" - result = self.execute( - f"{KEY_CLOAK_CLI} get users -r {settings.rhsso.realm} -q username={username}" - ) + result = self.execute(f"{self.kcadm} get users -r {self.realm} -q username={username}") result_json = json.loads(result.stdout) return result_json[0] @lru_cache def get_rhsso_groups_details(self, group_name): """Getter method to receive the group id""" - result = self.execute(f"{KEY_CLOAK_CLI} get groups -r {settings.rhsso.realm}") + result = self.execute(f"{self.kcadm} get groups -r {self.realm}") group_list = json.loads(result.stdout) query_group = [group for group in group_list if group['name'] == group_name] return query_group[0] @@ -2482,8 +2501,8 @@ def create_mapper(self, json_content, client_id): """Helper method to create the RH-SSO Client Mapper""" self.upload_rhsso_entity(json_content, "mapper_file") self.execute( - f'{KEY_CLOAK_CLI} create clients/{client_id}/protocol-mappers/models -r ' - f'{settings.rhsso.realm} -f {"mapper_file"}' + f'{self.kcadm} create clients/{client_id}/protocol-mappers/models -r ' + f'{self.realm} -f {"mapper_file"}' ) def create_new_rhsso_user(self, username=None): @@ -2494,13 +2513,13 @@ def create_new_rhsso_user(self, username=None): username = gen_string('alphanumeric') update_data_user.username = username update_data_user.email = username + random.choice(valid_emails_list()) - update_data_pass.value = settings.rhsso.rhsso_password + update_data_pass.value = self.password self.upload_rhsso_entity(update_data_user, "create_user") self.upload_rhsso_entity(update_data_pass, "reset_password") - self.execute(f"{KEY_CLOAK_CLI} create users -r {settings.rhsso.realm} -f create_user") + self.execute(f"{self.kcadm} create users -r {self.realm} -f create_user") user_details = self.get_rhsso_user_details(update_data_user.username) self.execute( - f'{KEY_CLOAK_CLI} update -r {settings.rhsso.realm} ' + f'{self.kcadm} update -r {self.realm} ' f'users/{user_details["id"]}/reset-password -f {"reset_password"}' ) return update_data_user @@ -2508,21 +2527,19 @@ def create_new_rhsso_user(self, username=None): def update_rhsso_user(self, username, group_name=None): update_data_user = Box(RHSSO_USER_UPDATE) user_details = self.get_rhsso_user_details(username) - update_data_user.realm = settings.rhsso.realm + update_data_user.realm = self.realm update_data_user.userId = f"{user_details['id']}" if group_name: group_details = self.get_rhsso_groups_details(group_name=group_name) update_data_user['groupId'] = f"{group_details['id']}" self.upload_rhsso_entity(update_data_user, "update_user") group_path = f"users/{user_details['id']}/groups/{group_details['id']}" - self.execute( - f"{KEY_CLOAK_CLI} update -r {settings.rhsso.realm} {group_path} -f update_user" - ) + self.execute(f"{self.kcadm} update -r {self.realm} {group_path} -f update_user") def delete_rhsso_user(self, username): """Delete the RHSSO user""" user_details = self.get_rhsso_user_details(username) - self.execute(f"{KEY_CLOAK_CLI} delete -r {settings.rhsso.realm} users/{user_details['id']}") + self.execute(f"{self.kcadm} delete -r {settings.rhsso.realm} users/{user_details['id']}") def create_group(self, group_name=None): """Create the RHSSO group""" @@ -2532,23 +2549,21 @@ def create_group(self, group_name=None): update_user_group.name = group_name self.upload_rhsso_entity(update_user_group, "create_group") result = self.execute( - f"{KEY_CLOAK_CLI} create groups -r {settings.rhsso.realm} -f create_group" + f"{self.kcadm} create groups -r {settings.rhsso.realm} -f create_group" ) return result.stdout def delete_rhsso_group(self, group_name): """Delete the RHSSO group""" group_details = self.get_rhsso_groups_details(group_name) - self.execute( - f"{KEY_CLOAK_CLI} delete -r {settings.rhsso.realm} groups/{group_details['id']}" - ) + self.execute(f"{self.kcadm} delete -r {settings.rhsso.realm} groups/{group_details['id']}") def update_client_configuration(self, json_content): """Update the client configuration""" client_id = self.get_rhsso_client_id() self.upload_rhsso_entity(json_content, "update_client_info") update_cmd = ( - f"{KEY_CLOAK_CLI} update clients/{client_id} " # EOL space important + f"{self.kcadm} update clients/{client_id} " # EOL space important "-f update_client_info -s enabled=true --merge" ) assert self.execute(update_cmd).status == 0 @@ -2557,8 +2572,8 @@ def update_client_configuration(self, json_content): def oidc_token_endpoint(self): """getter oidc token endpoint""" return ( - f"https://{settings.rhsso.host_name}/auth/realms/" - f"{settings.rhsso.realm}/protocol/openid-connect/token" + f"https://{self.host_name}:{self.host_port}/auth/realms/" + f"{self.realm}/protocol/openid-connect/token" ) def get_oidc_client_id(self): @@ -2568,16 +2583,13 @@ def get_oidc_client_id(self): @cached_property def oidc_authorization_endpoint(self): """getter for the oidc authorization endpoint""" - return ( - f"https://{settings.rhsso.host_name}/auth/realms/" - f"{settings.rhsso.realm}/protocol/openid-connect/auth" - ) + return f"https://{self.host_name}/auth/realms/" f"{self.realm}/protocol/openid-connect/auth" def get_two_factor_token_rh_sso_url(self): """getter for the two factor token rh_sso url""" return ( - f"https://{settings.rhsso.host_name}/auth/realms/" - f"{settings.rhsso.realm}/protocol/openid-connect/" + f"https://{self.host_name}/auth/realms/" + f"{self.realm}/protocol/openid-connect/" f"auth?response_type=code&client_id={self.satellite.hostname}-foreman-openidc&" "redirect_uri=urn:ietf:wg:oauth:2.0:oob&scope=openid" ) diff --git a/tests/foreman/destructive/test_ldapauthsource.py b/tests/foreman/destructive/test_ldapauthsource.py index 27ba4386d66..8a55b97a262 100644 --- a/tests/foreman/destructive/test_ldapauthsource.py +++ b/tests/foreman/destructive/test_ldapauthsource.py @@ -43,6 +43,12 @@ def rh_sso_hammer_auth_setup(module_target_sat, default_sso_host, request): @pytest.mark.pit_server +@pytest.mark.parametrize( + ('default_sso_host', 'rhsso_setting_setup', 'enroll_configure_rhsso_external_auth'), + [(True, True, True), (False, False, False)], + ids=['RHBK', 'RHSSO'], + indirect=True, +) def test_rhsso_login_using_hammer( module_target_sat, enable_external_auth_rhsso, @@ -58,30 +64,28 @@ def test_rhsso_login_using_hammer( :CaseImportance: High """ + if default_sso_host.rhbk: + user = settings.rhbk.rhbk_user + password = settings.rhbk.rhbk_password + else: + user = settings.rhsso.rhsso_user + password = settings.rhsso.rhsso_password + result = module_target_sat.cli.AuthLogin.oauth( { 'oidc-token-endpoint': default_sso_host.oidc_token_endpoint, 'oidc-client-id': default_sso_host.get_oidc_client_id(), - 'username': settings.rhsso.rhsso_user, - 'password': settings.rhsso.rhsso_password, + 'username': user, + 'password': password, } ) - assert f"Successfully logged in as '{settings.rhsso.rhsso_user}'." == result[0]['message'] - result = module_target_sat.cli.Auth.with_user( - username=settings.rhsso.rhsso_user, password=settings.rhsso.rhsso_password - ).status() - assert ( - f"Session exists, currently logged in as '{settings.rhsso.rhsso_user}'." - in result[0]['message'] - ) - task_list = module_target_sat.cli.Task.with_user( - username=settings.rhsso.rhsso_user, password=settings.rhsso.rhsso_password - ).list() + assert f"Successfully logged in as '{user}'." == result[0]['message'] + result = module_target_sat.cli.Auth.with_user(username=user, password=password).status() + assert f"Session exists, currently logged in as '{user}'." in result[0]['message'] + task_list = module_target_sat.cli.Task.with_user(username=user, password=password).list() assert len(task_list) >= 0 with pytest.raises(CLIReturnCodeError) as error: - module_target_sat.cli.Role.with_user( - username=settings.rhsso.rhsso_user, password=settings.rhsso.rhsso_password - ).list() + module_target_sat.cli.Role.with_user(username=user, password=password).list() assert 'Missing one of the required permissions' in error.value.message