load certs from oc secrets

Author: mastastny

config/settings.yaml.tpl

@@ -92,6 +92,25 @@ default:
       token: # token for openshift where the testenv tools are deployed (unnecessary for for default openshift)
     private_base_url:
       default: echo_api # tool name to be used by default for backend
+  shared_certs:
+    # config for obtaining certificates which are signed by same CA as tools.
+    # If this config is not set, testsuite will use same openshift and namespace as tools
+    namespace: tools # openshift namespace/project where the secrets with certificates are deployed
+    server_url: # openshift url where the secrets with certificates are deployed
+    token: # token for openshift where the secrets with certificates are deployed
+    # certs can be also added manually
+    client_certs:
+        valid:
+            - name: client1
+              crt: "cert1"
+              key: "key1"
+            - name: client2
+              crt: "cert2"
+              key: "key2"
+        invalid:
+            - name: client3
+              crt: "cert3"
+              key: "key3"
   warn_and_skip:
     # section to control how warn_and_skip should behave for particular tests
     # works just for tests and fixture that use warn_and_skip

testsuite/dynaconf_loader.py

@@ -145,6 +145,43 @@ def _rhsso_password(tools_config, rhsso_config):
     return None
 
 
+def _shared_tool_certs(tools_config, shared_certs_config):
+    """Search for SSO clients certificates"""
+    if not shared_certs_config:
+        shared_certs_config = tools_config
+    server_url = shared_certs_config.get("server_url")
+    project = shared_certs_config.get("namespace")
+    token = shared_certs_config.get("token")
+    client = OpenShiftClient(project_name=project, server_url=server_url, token=token)
+    try:
+        iter(client.secrets)
+    except OpenShiftPythonException:
+        return {"valid": [], "invalid": []}
+
+    valid_certs = []
+    invalid_certs = []
+    for secret in client.secrets:
+        try:
+            name = secret["metadata"]["name"]
+            annotations = secret["metadata"]["annotations"]
+            shared_cert = annotations["shared_cert"]
+            valid = annotations["valid"]
+        except KeyError:
+            continue
+        if shared_cert != "True":
+            continue
+        cert = {
+            "name": name,
+            "crt": client.secrets[name]["tls.crt"].decode("utf-8"),
+            "key": client.secrets[name]["tls.key"].decode("utf-8"),
+        }
+        if valid == "True":
+            valid_certs.append(cert)
+        if valid == "False":
+            invalid_certs.append(cert)
+    return {"valid": valid_certs, "invalid": invalid_certs}
+
+
 def _threescale_operator_ocp(ocp):
     try:
         ocp.threescale_operator  # pylint: disable=pointless-statement
@@ -247,6 +284,7 @@ def load(obj, env=None, silent=None, key=None):
             ocp_tools_setup = ocp_setup
 
         rhsso_setup = obj.get("rhsso", {})
+        shared_certs_setup = obj.get("shared_certs", {})
 
         ocp = OpenShiftClient(
             project_name=project, server_url=ocp_setup.get("server_url"), token=ocp_setup.get("token")
@@ -337,6 +375,7 @@ def load(obj, env=None, silent=None, key=None):
                 "apicast": {"openshift": apicast_operator_ocp},
             },
             "rhsso": {"password": _rhsso_password(ocp_tools_setup, rhsso_setup)},
+            "shared_certs": {"client_certs": _shared_tool_certs(ocp_tools_setup, shared_certs_setup)},
         }
 
         # this overwrites what's already in settings to ensure NAMESPACE is propagated

testsuite/rhsso/__init__.py

@@ -1,6 +1,7 @@
 """Utility classes for working with RHSSO server"""
 
 import functools
+from typing import cast
 
 import backoff
 from keycloak import KeycloakOpenID
@@ -26,13 +27,13 @@ def __init__(self, rhsso: RHSSO, realm: Realm, client: Client, user, username, p
         self.client = client
         self.username = username
         self.password = password
-        self._oidc_client = None
+        self._oidc_client: KeycloakOpenID | None = None
 
     @property
     def oidc_client(self) -> KeycloakOpenID:
         """OIDCClient for the created client"""
         if not self._oidc_client:
-            self._oidc_client = self.client.oidc_client
+            self._oidc_client = cast(KeycloakOpenID, self.client.oidc_client)
         return self._oidc_client
 
     def issuer_url(self) -> str:

testsuite/rhsso/objects.py

@@ -1,5 +1,5 @@
 """This module contains object wrappers on top of python-keycloak API, since that is not object based"""
-from enum import verify
+
 from urllib.parse import urlparse
 
 from keycloak import KeycloakAdmin, KeycloakOpenID, KeycloakPostError

testsuite/tests/apicast/policy/fapi/__init__.py

@@ -1,30 +1,42 @@
 """Test Financial-Grade API policy with oauth2 certificate bound access token (advanced profile)"""
 
-import os
 import pytest
 import requests
 
 from testsuite import rawobj
-from testsuite.utils import blame
+from testsuite.utils import blame, warn_and_skip
 from testsuite.rhsso import OIDCClientAuth, OIDCClientAuthHook
-# pylint: disable=reimported,unused-import
-from testsuite.tests.apicast.policy.tls.conftest import (manager, superdomain, server_authority, staging_gateway,
-                                                         gateway_options, gateway_environment)
-# pylint: disable=reimported,unused-import
-from  testsuite.tests.apicast.policy.tls.conftest import (certificate, superdomain, valid_authority, create_cert,
-                                                          superdomain, manager)
+from testsuite.certificates import Certificate
+
+# pylint: disable=reimported, unused-import
+# flake8: noqa
+from testsuite.tests.apicast.policy.tls.conftest import (
+    certificate,
+    manager,
+    superdomain,
+    server_authority,
+    staging_gateway,
+    gateway_options,
+    gateway_environment,
+    valid_authority,
+    create_cert,
+)
 
 
-@pytest.fixture()
-def mtls_client_cert():
+@pytest.fixture(scope="module")
+def mtls_client_cert(request, testconfig):
     """Clients certificate. CA of this cert needs to be trusted by the sso"""
-    # todo create certs dynamically
-    script_dir = os.path.dirname(os.path.abspath(__file__))
-    crt_path = os.path.join(script_dir, "certificates", "client.crt" )
-    key_path = os.path.join(script_dir, "certificates", "client.key" )
-    return crt_path, key_path
+    try:
+        crt = testconfig["shared_certs"]["client_certs"]["valid"][0]
+    except IndexError:
+        warn_and_skip("Valid tools cert is not available, skipping fapi advanced tests")
+    cert = Certificate(crt["key"], crt["crt"])
+    if not testconfig["skip_cleanup"]:
+        request.addfinalizer(cert.delete_files)
+    return cert.files["certificate"], cert.files["key"]
 
 
+# flake8: noqa
 @pytest.fixture()
 def unknown_cert(certificate):
     """Cert which won't be used for obtaining the token"""
@@ -40,10 +52,13 @@ def service(service):
         3scale APIcast
         Fapi
     """
-    fapi_policy = rawobj.PolicyConfig("fapi", configuration = {
-        "validate_x_fapi_customer_ip_address": True,
-        "validate_oauth2_certificate_bound_access_token": True
-      })
+    fapi_policy = rawobj.PolicyConfig(
+        "fapi",
+        configuration={
+            "validate_x_fapi_customer_ip_address": True,
+            "validate_oauth2_certificate_bound_access_token": True,
+        },
+    )
     service.proxy.list().policies.insert(1, fapi_policy)
     return service
 
@@ -56,16 +71,15 @@ def fapi_sso_client_id():
 
 # pylint: disable=too-many-arguments
 @pytest.fixture(scope="module")
-def application(service, custom_application, custom_app_plan, lifecycle_hooks, request, fapi_sso_client_id,
-                rhsso_service_info):
+def application(
+    service, custom_application, custom_app_plan, lifecycle_hooks, request, fapi_sso_client_id, rhsso_service_info
+):
     """application bound to the account and service existing over whole testing session"""
     plan = custom_app_plan(rawobj.ApplicationPlan(blame(request, blame(request, "fapi"))), service)
     app = custom_application(
-        rawobj.Application(
-            name=fapi_sso_client_id,
-            app_id=fapi_sso_client_id,
-            application_plan=plan),
-        hooks=lifecycle_hooks)
+        rawobj.Application(name=fapi_sso_client_id, app_id=fapi_sso_client_id, application_plan=plan),
+        hooks=lifecycle_hooks,
+    )
     service.proxy.deploy()
     app.register_auth("oidc", OIDCClientAuth.partial(rhsso_service_info))
     return app
@@ -88,14 +102,15 @@ def fapi_sso_client(rhsso_service_info, fapi_sso_client_id):
             "tls.client.certificate.bound.access.tokens": "true",
             "client.authentication.type": "tls",
             "x509.allow.regex.pattern.comparison": "true",
-        }
+        },
     }
     return rhsso_service_info.realm.create_client(fapi_sso_client_id, **client_config)
 
 
 # pylint: disable=too-few-public-methods
 class FapiClient:
     """Temporary class for sending requests to apicast. This should be replaced by using fixture api_client"""
+
     def __init__(self, base_url, verify):
         self.base_url = base_url
         self.http_session = requests.Session()