feat: add keycloak client provider

src/github_actions/common/__init__.py

@@ -1,6 +1,6 @@
 # github_actions/common/__init__.py
 
-from .actions.init_project import GETTenant, ProjectInitializer, RMKConfigInitCommand, GetRootDomain
+from .actions.init_project import GETTenant, ProjectInitializer, RMKConfigInitCommand, GetRootDomain, RMKClusterSwitchCommand
 from .credentials.cluster_provider_credentials import (
     AWSConfig, AzureConfig, ClusterProviders, Credentials, EnvironmentConfig, GCPConfig
 )
@@ -18,6 +18,8 @@
 from .utils.github_environment_variables import GitHubContext
 from .utils.install_rmk import RMKInstaller
 from .utils.install_kodjin_cli import KodjinCLIInstaller
+from .utils.kubectl import Kubectl
+from .utils.keycloak import KeycloakClientInfoFetcher
 
 
 __all__ = [
@@ -41,11 +43,14 @@
     "GETTenant",
     "GitHubContext",
     "GitHubOutput",
+    "KeycloakClientInfoFetcher",
     "ProjectInitializer",
     "RMKConfigInitCommand",
+    "RMKClusterSwitchCommand",
     "RMKInstaller",
     "S3BucketManager",
     "SlackNotifier",
     "KodjinCLIInstaller",
+    "Kubectl",
     "GetRootDomain",
 ]

src/github_actions/common/actions/init_project.py

@@ -39,6 +39,20 @@ def run(self):
             self.run_command(f"rmk config init --cluster-provider={self.cluster_provider} --progress-bar=false")
 
 
+class RMKClusterSwitchCommand(BaseCommand, CMDInterface):
+    def __init__(self, force: bool = False):
+        super().__init__(environment="")
+        self.force = force
+
+    def execute(self):
+        self.run()
+
+    def run(self):
+        """Switch RMK cluster context."""
+        force_flag = "--force" if self.force else ""
+        self.run_command(f"rmk cluster switch {force_flag}")
+
+
 class GETTenant(BaseCommand, CMDInterface):
     def __init__(self, environment: str):
         super().__init__(environment)

src/github_actions/common/utils/keycloak.py

@@ -0,0 +1,67 @@
+from urllib.parse import urlencode
+import requests
+
+
+class KeycloakClientInfoFetcher:
+    def __init__(
+        self,
+        keycloak_url: str,
+        username: str,
+        admin_client_id: str,
+        client_id: str,
+        password: str,
+        clients_realm: str,
+        token_realm: str = "master",
+    ):
+        self.base_url = keycloak_url.rstrip("/")
+        self.username = username
+        self.admin_client_id = admin_client_id
+        self.client_id = client_id
+        self.password = password
+        self.token_realm = token_realm
+        self.clients_realm = clients_realm
+
+    def get_client_info(self) -> list[dict]:
+        access_token = self._get_access_token()
+        url = f"{self.base_url}/admin/realms/{self.clients_realm}/clients"
+        print(f"Fetching client info from: {url} for client_id: {self.client_id}")  # Print only the first 10 characters for security
+
+        clients_response = requests.get(
+            url,
+            headers={"Authorization": f"Bearer {access_token}"},
+            params={"clientId": self.client_id},
+            timeout=30,
+        )
+        clients_response.raise_for_status()
+        return clients_response.json()
+
+    def _get_access_token(self) -> str:
+        data = urlencode(
+            {
+                "client_id": self.admin_client_id,
+                "username": self.username,
+                "grant_type": "password",
+                "password": self.password,
+            }
+        )
+
+        token_response = requests.post(
+            f"{self.base_url}/realms/{self.token_realm}/protocol/openid-connect/token",
+            headers={"content-type": "application/x-www-form-urlencoded"},
+            data=data,
+            timeout=30,
+        )
+        token_response.raise_for_status()
+
+        response_json = token_response.json()
+
+        # TODO: remove logging before merge!
+        print(f"Retrieved: {response_json}")
+
+        access_token = response_json.get("access_token")
+        print(f"Access token length: {len(access_token) if access_token else 'None'}")  # Print the length of the token for debugging
+
+        if not access_token:
+            raise ValueError("access_token was not returned by Keycloak token endpoint")
+
+        return access_token

src/github_actions/common/utils/kubectl.py

@@ -0,0 +1,52 @@
+from github_actions.common.utils.cmd import BaseCommand, CMDInterface
+
+
+class Kubectl(BaseCommand, CMDInterface):
+    def __init__(self, context=None, kubectl_download_url="https://dl.k8s.io/release/v1.36.2/bin/linux/amd64/kubectl"):
+        super().__init__(None)
+        self.kubectl_download_url = kubectl_download_url
+        self.context = context
+        self.run()
+
+    def run(self):
+        self._install_kubectl()
+
+    def execute(self):
+        return self.run()
+
+    def get_secret(self, secret_name: str, namespace: str, secret_path: str) -> str | None:
+        try:
+            print(f"Getting secret {secret_name} in namespace {namespace}.")
+            context_flag = f"--context {self.context}" if self.context else ""
+
+            print(f"contexts list: {self.run_command('kubectl config get-contexts', capture_output=True)}")
+
+            command = f"kubectl get secret {secret_name} --namespace {namespace} {context_flag} --output yaml | yq '.data.password  | @base64d'"
+            print(f"Executing command: {command}")
+            return self.run_command(command, capture_output=True)
+
+        except Exception as err:
+            raise Exception(f"getting secret {secret_name} in namespace {namespace}:\n{err}")
+
+    def _install_kubectl(self):
+        print("Checking kubectl installation.")
+        try:
+            if self._is_kubectl_installed():
+                version = self.run_command("kubectl version --client", capture_output=True)
+                print(f"kubectl already installed. Version: {version}")
+                return
+
+            print("Installing kubectl.")
+            self.run_command(f"bash -s -- {self.kubectl_download_url}")
+
+            version = self.run_command("kubectl version --client", capture_output=True)
+            print(f"kubectl installed successfully. Version: {version}")
+        except Exception as err:
+            raise Exception(f"installing kubectl:\n{err}")
+
+    def _is_kubectl_installed(self) -> bool:
+        try:
+            self.run_command("command -v kubectl", capture_output=True)
+            return True
+        except ValueError:
+            return False