RHBK support

Make changes so RHSSO tests become parametrized and work for both RHSSO and RHBK

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

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 2>/dev/null| '
+            f'openssl s_client -connect {uri} -showcerts </dev/null 2>/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

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"

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,35 +2513,33 @@ 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
 
     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"
         )

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