Merge pull request #74 from pehala/rhsso_rewrite

Rework RHSSO into a more general OIDCProvider interface

README.md

@@ -51,6 +51,8 @@ To run all tests you can then use ```make test```
 
 For just running tests, the container image is the easiest option, you can log in to OpenShift and then run it like this
 
+If you omit any options, Testsuite will run only subset of tests that don't require that variable e.g. not providing Auth0 will result in skipping Auth0 tests.
+
 NOTE: For binding kubeconfig file, the "others" need to have permission to read, otherwise it will not work.
 The results and reports will be saved in `/test-run-results` in the container.
 
@@ -61,6 +63,9 @@ podman run \
 	-v $HOME/.kube/config:/run/kubeconfig:z \
 	-e KUADRANT_OPENSHIFT__project=authorino \
 	-e KUADRANT_OPENSHIFT2__project=authorino2 \
+	-e KUADRANT_AUTH0__url="AUTH0_URL" \
+	-e KUADRANT_AUTH0__client_id="AUTH0_CLIENT_ID" \
+	-e KUADRANT_AUTH0__client_secret="AUTH0_CLIENT_SECRET" \	
 	quay.io/kuadrant/testsuite:latest
 ```
 
@@ -74,5 +79,8 @@ podman run \
 	-e KUADRANT_RHSSO__url="https://my-sso.net" \
 	-e KUADRANT_RHSSO__password="ADMIN_PASSWORD" \
 	-e KUADRANT_RHSSO__username="ADMIN_USERNAME" \
+	-e KUADRANT_AUTH0__url="AUTH0_URL" \
+	-e KUADRANT_AUTH0__client_id="AUTH0_CLIENT_ID" \
+	-e KUADRANT_AUTH0__client_secret="AUTH0_CLIENT_SECRET" \
 	quay.io/kuadrant/testsuite:latest
 ```

config/settings.local.yaml.tpl

@@ -16,9 +16,9 @@
 #      username: "testUser"
 #      password: "testPassword"
 #  auth0:
-#    client: "CLIENT_ID"
-#    client-secret: "CLIENT_SECRET"
-#    domain: "AUTH0_DOMAIN"
+#    client_id: "CLIENT_ID"
+#    client_secret: "CLIENT_SECRET"
+#    url: "AUTH0_URL"
 #  cfssl: "cfssl"  # Path to the CFSSL library for TLS tests
 #  authorino:
 #    image: "quay.io/kuadrant/authorino:latest"  # If specified will override the authorino image

testsuite/httpx/auth.py

@@ -1,20 +1,26 @@
 """Auth Classes for HttpX"""
-import typing
-from typing import Generator
+from functools import cached_property
+from typing import Generator, Callable, Union
 
 from httpx import Auth, Request, URL, Response
 
 from testsuite.openshift.objects.api_key import APIKey
-from testsuite.rhsso import Client
+from testsuite.oidc import Token
 
 
 class HttpxOidcClientAuth(Auth):
     """Auth class for Httpx client for product secured by oidc"""
 
-    def __init__(self, client: Client, location, username=None, password=None) -> None:
+    def __init__(self, token: Union[Token, Callable[[], Token]], location="authorization") -> None:
         self.location = location
-        self.oidc_client = client.oidc_client
-        self.token = self.oidc_client.token(username, password)
+        self._token = token
+
+    @cached_property
+    def token(self):
+        """Lazily retrieves token from OIDC provider"""
+        if callable(self._token):
+            return self._token()
+        return self._token
 
     def _add_credentials(self, request: Request, token):
         if self.location == 'authorization':
@@ -27,13 +33,13 @@ def _add_credentials(self, request: Request, token):
             raise ValueError(f"Unknown credentials location '{self.location}'")
 
     def auth_flow(self, request: Request) -> Generator[Request, Response, None]:
-        self._add_credentials(request, self.token["access_token"])
+        self._add_credentials(request, self.token.access_token)
         response = yield request
 
         if response.status_code == 403:
             # Renew access token and try again
-            self.token = self.oidc_client.refresh_token(self.token["refresh_token"])
-            self._add_credentials(request, self.token["access_token"])
+            self.token.refresh()
+            self._add_credentials(request, self.token.access_token)
             yield request
 
 
@@ -45,18 +51,6 @@ def __init__(self, api_key: APIKey, prefix: str = "APIKEY") -> None:
         self.api_key = str(api_key)
         self.prefix = prefix
 
-    def auth_flow(self, request: Request) -> typing.Generator[Request, Response, None]:
+    def auth_flow(self, request: Request) -> Generator[Request, Response, None]:
         request.headers["Authorization"] = f"{self.prefix} {self.api_key}"
         yield request
-
-
-class Auth0Auth(Auth):
-    """Auth class for authentication with Auth0 token"""
-
-    def __init__(self, token: str) -> None:
-        super().__init__()
-        self.token = token
-
-    def auth_flow(self, request: Request) -> typing.Generator[Request, Response, None]:
-        request.headers["Authorization"] = f"Bearer {self.token}"
-        yield request

testsuite/oidc/__init__.py

@@ -0,0 +1,32 @@
+"""Common classes for OIDC provider"""
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from typing import Callable, Tuple
+
+
+@dataclass
+class Token:
+    """Token class"""
+    access_token: str
+    refresh_function: Callable[[str], "Token"]
+    refresh_token: str
+
+    def refresh(self) -> "Token":
+        """Refreshes token"""
+        return self.refresh_function(self.refresh_token)
+
+    def __str__(self) -> str:
+        return self.access_token
+
+
+class OIDCProvider(ABC):
+    """Interface for all methods we need for OIDCProvider"""
+
+    @property
+    @abstractmethod
+    def well_known(self):
+        """Dict (or a dict-like structure) access to all well_known URLS"""
+
+    @abstractmethod
+    def get_token(self, username=None, password=None) -> Token:
+        """Returns Token wrapper class with current access token and ability to refresh it"""

testsuite/oidc/auth0.py

@@ -0,0 +1,42 @@
+"""Module containing all classes necessary to work with Auth0"""
+from functools import cached_property
+
+import httpx
+
+from testsuite.oidc import OIDCProvider, Token
+
+
+class Auth0Provider(OIDCProvider):
+    """Auth0 OIDC provider"""
+
+    def __init__(self, domain, client_id, client_secret) -> None:
+        super().__init__()
+        self.domain = domain
+        self.client_id = client_id
+        self.client_secret = client_secret
+
+    @property
+    def token_endpoint(self):
+        """Returns token_endpoint URL"""
+        return self.well_known["token_endpoint"]
+
+    @cached_property
+    def well_known(self):
+        response = httpx.get(self.domain + "/.well-known/openid-configuration")
+        return response.json()
+
+    # pylint: disable=unused-argument
+    def refresh_token(self, refresh_token):
+        """Refresh tokens are not yet implemented for Auth0, will attempt to acquire new token instead"""
+        return self.get_token()
+
+    def get_token(self, username=None, password=None) -> Token:
+        response = httpx.post(self.token_endpoint, json={
+            "client_id": self.client_id,
+            "client_secret": self.client_secret,
+            "grant_type": "client_credentials",
+            "audience": self.domain + "/api/v2/"
+        })
+        data = response.json()
+        assert response.status_code == 200, f"Unable to acquire token from Auth0, reason: {data}"
+        return Token(data["access_token"], self.refresh_token, "None")

testsuite/oidc/rhsso/__init__.py

@@ -0,0 +1,87 @@
+"""Objects for RHSSO"""
+from functools import cached_property
+from urllib.parse import urlparse
+
+from keycloak import KeycloakOpenID, KeycloakAdmin, KeycloakPostError
+
+from testsuite.oidc import OIDCProvider, Token
+from testsuite.objects import LifecycleObject
+from .objects import Realm, Client
+
+
+# pylint: disable=too-many-instance-attributes
+class RHSSO(OIDCProvider, LifecycleObject):
+    """
+    OIDCProvider implementation for RHSSO. It creates Realm, client and user.
+    """
+
+    def __init__(self, server_url, username, password, realm_name, client_name,
+                 test_username="testUser", test_password="testPassword") -> None:
+        self.test_username = test_username
+        self.test_password = test_password
+        self.realm_name = realm_name
+        self.client_name = client_name
+        self.realm = None
+        self.user = None
+        self.client = None
+
+        try:
+            self.master = KeycloakAdmin(server_url=server_url,
+                                        username=username,
+                                        password=password,
+                                        realm_name="master",
+                                        verify=False,
+                                        auto_refresh_token=['get', 'put', 'post', 'delete'])
+            self.server_url = server_url
+        except KeycloakPostError:
+            # Older Keycloaks versions (and RHSSO) needs requires url to be pointed at auth/ endpoint
+            # pylint: disable=protected-access
+            self.server_url = urlparse(server_url)._replace(path="auth/").geturl()
+            self.master = KeycloakAdmin(server_url=self.server_url,
+                                        username=username,
+                                        password=password,
+                                        realm_name="master",
+                                        verify=False,
+                                        auto_refresh_token=['get', 'put', 'post', 'delete'])
+
+    def create_realm(self, name: str, **kwargs) -> Realm:
+        """Creates new realm"""
+        self.master.create_realm(payload={
+            "realm": name,
+            "enabled": True,
+            "sslRequired": "None",
+            **kwargs
+        })
+        return Realm(self.master, name)
+
+    def commit(self):
+        self.realm: Realm = self.create_realm(self.realm_name, accessTokenLifespan=24 * 60 * 60)
+
+        self.client = self.realm.create_client(
+            name=self.client_name,
+            directAccessGrantsEnabled=True,
+            publicClient=False,
+            protocol="openid-connect",
+            standardFlowEnabled=False)
+        self.user = self.realm.create_user(self.test_username, self.test_password)
+
+    def delete(self):
+        self.realm.delete()
+
+    @property
+    def oidc_client(self) -> KeycloakOpenID:
+        """OIDCClient for the created client"""
+        return self.client.oidc_client  # type: ignore
+
+    @cached_property
+    def well_known(self):
+        return self.oidc_client.well_known()
+
+    def refresh_token(self, refresh_token):
+        """Refreshes token"""
+        data = self.oidc_client.refresh_token(refresh_token)
+        return Token(data["access_token"], self.refresh_token, data["refresh_token"])
+
+    def get_token(self, username=None, password=None) -> Token:
+        data = self.oidc_client.token(username or self.test_username, password or self.test_password)
+        return Token(data["access_token"], self.refresh_token, data["refresh_token"])

testsuite/rhsso/objects.py → testsuite/oidc/rhsso/objects.py

@@ -1,7 +1,7 @@
 """Object wrappers of RHSSO resources"""
-from urllib.parse import urlparse
+from functools import cached_property
 
-from keycloak import KeycloakOpenID, KeycloakAdmin, KeycloakPostError
+from keycloak import KeycloakOpenID, KeycloakAdmin
 
 
 class Realm:
@@ -80,51 +80,10 @@ def assign_role(self, role_name):
         role = self.admin.get_client_role(realm_management, role_name)
         self.admin.assign_client_role(user["id"], realm_management, role)
 
-    @property
+    @cached_property
     def oidc_client(self):
         """OIDC client"""
         # Note This is different clientId (clientId) than self.client_id (Id), because RHSSO
         client_id = self.admin.get_client(self.client_id)["clientId"]
         secret = self.admin.get_client_secrets(self.client_id)["value"]
         return self.realm.oidc_client(client_id, secret)
-
-
-class RHSSO:
-    """Helper class for RHSSO server"""
-
-    def __init__(self, server_url, username, password) -> None:
-        try:
-            self.master = KeycloakAdmin(server_url=server_url,
-                                        username=username,
-                                        password=password,
-                                        realm_name="master",
-                                        verify=False,
-                                        auto_refresh_token=['get', 'put', 'post', 'delete'])
-            self.server_url = server_url
-        except KeycloakPostError:
-            # Older Keycloaks versions (and RHSSO) needs requires url to be pointed at auth/ endpoint
-            # pylint: disable=protected-access
-            self.server_url = urlparse(server_url)._replace(path="auth/").geturl()
-            self.master = KeycloakAdmin(server_url=self.server_url,
-                                        username=username,
-                                        password=password,
-                                        realm_name="master",
-                                        verify=False,
-                                        auto_refresh_token=['get', 'put', 'post', 'delete'])
-
-    def create_realm(self, name: str, **kwargs) -> Realm:
-        """Creates new realm"""
-        self.master.create_realm(payload={
-            "realm": name,
-            "enabled": True,
-            "sslRequired": "None",
-            **kwargs
-        })
-        return Realm(self.master, name)
-
-    def create_oidc_client(self, realm, client_id, secret) -> KeycloakOpenID:
-        """Creates OIDC client"""
-        return KeycloakOpenID(server_url=self.master.server_url,
-                              client_id=client_id,
-                              realm_name=realm,
-                              client_secret_key=secret)