fleet kubeconfig auto-conversion with kubelogin

Signed-off-by: Wei Weng <Wei.Weng@microsoft.com>

Author: Wei Weng

src/fleet/HISTORY.rst

@@ -168,4 +168,8 @@ Release History
 
 1.8.2
 ++++++
-* Fix fleet namespace get-credentials command with fleet member parameter.
+* Fix fleet namespace get-credentials command with fleet member parameter.
+
+1.8.3
+++++++
+* Add automatic kubelogin conversion to Azure CLI authentication for fleet get-credentials command.

src/fleet/azext_fleet/_helpers.py

@@ -25,12 +25,17 @@
 logger = get_logger(__name__)
 
 
+def is_stdout_path(path):
+    """Check if the path represents stdout."""
+    return path == "-"
+
+
 def print_or_merge_credentials(path, kubeconfig, overwrite_existing, context_name):
     """Merge an unencrypted kubeconfig into the file at the specified path, or print it to
     stdout if the path is "-".
     """
     # Special case for printing to stdout
-    if path == "-":
+    if is_stdout_path(path):
         print(kubeconfig)
         return
 

src/fleet/azext_fleet/_params.py

@@ -65,6 +65,7 @@ def load_arguments(self, _):
         c.argument('context_name', options_list=['--context'], help='If specified, overwrite the default context name.')
         c.argument('path', options_list=['--file', '-f'], type=file_type, completer=FilesCompleter(),
                    default=os.path.join(os.path.expanduser('~'), '.kube', 'config'))
+        c.ignore('skip_kubelogin_conversion')  # Internal parameter, not exposed to users
 
     with self.argument_context('fleet member') as c:
         c.argument('name', options_list=['--name', '-n'], help='Specify the fleet member name.')

src/fleet/azext_fleet/custom.py

@@ -4,8 +4,10 @@
 # --------------------------------------------------------------------------------------------
 
 import os
-import yaml
+import shutil
+import subprocess
 import tempfile
+import yaml
 
 from knack.log import get_logger
 from knack.util import CLIError
@@ -20,6 +22,7 @@
 from azext_fleet._helpers import is_rp_registered, print_or_merge_credentials
 from azext_fleet._helpers import assign_network_contributor_role_to_subnet
 from azext_fleet._helpers import get_msi_object_id
+from azext_fleet._helpers import is_stdout_path
 from azext_fleet.constants import UPGRADE_TYPE_CONTROLPLANEONLY
 from azext_fleet.constants import UPGRADE_TYPE_FULL
 from azext_fleet.constants import UPGRADE_TYPE_NODEIMAGEONLY
@@ -218,14 +221,52 @@ def delete_fleet(cmd,  # pylint: disable=unused-argument
     return sdk_no_wait(no_wait, client.begin_delete, resource_group_name, name, polling_interval=5)
 
 
+def _convert_kubeconfig_to_azurecli(path):
+    """
+    Convert kubeconfig to use Azure CLI authentication if it uses devicecode.
+
+    Args:
+        path: Path to the kubeconfig file to convert
+    """
+    # Skip conversion if path is stdout
+    if is_stdout_path(path):
+        return
+
+    if shutil.which("kubelogin"):
+        try:
+            subprocess.run(
+                ["kubelogin", "convert-kubeconfig", "-l", "azurecli", "--kubeconfig", path],
+                check=True,
+                capture_output=True,
+                text=True,
+                timeout=60,
+            )
+            logger.warning("Converted kubeconfig to use Azure CLI authentication.")
+        except subprocess.CalledProcessError as e:
+            logger.warning("Failed to convert kubeconfig with kubelogin: %s", str(e))
+        except subprocess.TimeoutExpired as e:
+            logger.warning("kubelogin command timed out: %s", str(e))
+        except Exception as e:  # pylint: disable=broad-except
+            logger.warning("Error running kubelogin: %s", str(e))
+    else:
+        logger.warning(
+            "The fleet hub cluster kubeconfig requires kubelogin. "
+            "Please install kubelogin from https://github.com/Azure/kubelogin or run "
+            "'az aks install-cli' to install both kubectl and kubelogin. "
+            "After installing kubelogin, rerun 'az fleet get-credentials' and the "
+            "kubeconfig will be converted automatically."
+        )
+
+
 def get_credentials(cmd,
                     client,
                     resource_group_name,
                     name,
                     path=os.path.join(os.path.expanduser('~'), '.kube', 'config'),
                     overwrite_existing=False,
                     context_name=None,
-                    member_name=None):
+                    member_name=None,
+                    skip_kubelogin_conversion=False):
 
     # If a member name is given, we use the cluster resource ID from the fleet member
     # to get that member cluster's credentials
@@ -273,6 +314,11 @@ def get_credentials(cmd,
         try:
             kubeconfig = credential_results.kubeconfigs[0].value.decode(encoding='UTF-8')
             print_or_merge_credentials(path, kubeconfig, overwrite_existing, context_name)
+            # Fleet hub is always RBAC-enabled and should convert it with kubelogin so that
+            # user doesn't have to manually run kubelogin convert-kubeconfig -l azurecli
+            # every time after az fleet get-credentials
+            if not skip_kubelogin_conversion:
+                _convert_kubeconfig_to_azurecli(path)
         except (IndexError, ValueError) as exc:
             raise CLIError("Fail to find kubeconfig file.") from exc
 
@@ -985,7 +1031,8 @@ def get_namespace_credentials(cmd,
             path=temp_file.name,
             overwrite_existing=overwrite_existing,
             context_name=context_name,
-            member_name=member_name
+            member_name=member_name,
+            skip_kubelogin_conversion=True  # Skip here, we'll convert after namespace modification
         )
 
         with open(temp_file.name, 'r', encoding='utf-8') as f:
@@ -998,3 +1045,6 @@ def get_namespace_credentials(cmd,
         modified_kubeconfig = yaml.dump(kubeconfig, default_flow_style=False)
         print_or_merge_credentials(path, modified_kubeconfig, overwrite_existing, context_name)
         print(f"Default namespace set to '{managed_namespace_name}' for context '{kubeconfig.get('current-context')}'")
+
+        # Apply kubelogin conversion to the final file after namespace modification
+        _convert_kubeconfig_to_azurecli(path)

src/fleet/azext_fleet/tests/latest/test_helpers.py

@@ -0,0 +1,30 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import unittest
+from azext_fleet._helpers import is_stdout_path
+
+
+# Test constants
+STDOUT_PATH = '-'
+SAMPLE_KUBECONFIG_PATH = "/home/user/.kube/config"
+SAMPLE_CONFIG_FILENAME = "config.yaml"
+
+
+class TestHelpers(unittest.TestCase):
+    """Test cases for helper functions."""
+
+    def test_is_stdout_path_with_dash(self):
+        """Test that '-' is recognized as stdout path."""
+        self.assertTrue(is_stdout_path(STDOUT_PATH))
+
+    def test_is_stdout_path_with_regular_path(self):
+        """Test that regular paths are not recognized as stdout."""
+        self.assertFalse(is_stdout_path(SAMPLE_KUBECONFIG_PATH))
+        self.assertFalse(is_stdout_path(SAMPLE_CONFIG_FILENAME))
+
+
+if __name__ == '__main__':
+    unittest.main()

src/fleet/azext_fleet/tests/latest/test_kubelogin_conversion.py

@@ -0,0 +1,169 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import unittest
+import tempfile
+import os
+from unittest.mock import patch, MagicMock, call
+from azext_fleet.custom import _convert_kubeconfig_to_azurecli
+
+
+# Test constants
+KUBELOGIN_BINARY = 'kubelogin'
+MOCK_KUBELOGIN_PATH = '/usr/bin/kubelogin'
+KUBELOGIN_TIMEOUT = 60
+KUBELOGIN_CMD_PREFIX = [KUBELOGIN_BINARY, "convert-kubeconfig", "-l", "azurecli", "--kubeconfig"]
+KUBELOGIN_GITHUB_URL = "https://github.com/Azure/kubelogin"
+KUBECONFIG_SUFFIX = '.kubeconfig'
+STDOUT_PATH = '-'
+CUSTOM_CONFIG_PATH = '/custom/path/to/config'
+
+
+class TestKubeloginConversion(unittest.TestCase):
+    """Test cases for kubelogin automatic conversion functionality."""
+
+    @patch('azext_fleet.custom.shutil.which')
+    @patch('azext_fleet.custom.subprocess.run')
+    @patch('azext_fleet.custom.logger')
+    def test_convert_with_kubelogin_available(self, mock_logger, mock_subprocess, mock_which):
+        """Test conversion when kubelogin is available."""
+        mock_which.return_value = MOCK_KUBELOGIN_PATH
+        mock_subprocess.return_value = MagicMock()
+
+        with tempfile.NamedTemporaryFile(mode='w', suffix=KUBECONFIG_SUFFIX, delete=False) as f:
+            test_path = f.name
+
+        try:
+            _convert_kubeconfig_to_azurecli(test_path)
+
+            # Verify kubelogin was called with correct arguments
+            mock_subprocess.assert_called_once()
+            call_args = mock_subprocess.call_args
+            self.assertEqual(call_args[0][0], KUBELOGIN_CMD_PREFIX + [test_path])
+            self.assertTrue(call_args[1]['check'])
+            self.assertEqual(call_args[1]['timeout'], KUBELOGIN_TIMEOUT)
+
+            # Verify success message was logged
+            mock_logger.warning.assert_called_with("Converted kubeconfig to use Azure CLI authentication.")
+        finally:
+            if os.path.exists(test_path):
+                os.remove(test_path)
+
+    @patch('azext_fleet.custom.shutil.which')
+    @patch('azext_fleet.custom.logger')
+    def test_convert_with_kubelogin_unavailable(self, mock_logger, mock_which):
+        """Test conversion when kubelogin is not available."""
+        mock_which.return_value = None
+
+        with tempfile.NamedTemporaryFile(mode='w', suffix=KUBECONFIG_SUFFIX, delete=False) as f:
+            test_path = f.name
+
+        try:
+            _convert_kubeconfig_to_azurecli(test_path)
+
+            # Verify warning message was logged
+            mock_logger.warning.assert_called_once()
+            warning_msg = mock_logger.warning.call_args[0][0]
+            self.assertIn("kubeconfig requires kubelogin", warning_msg)
+            self.assertIn(KUBELOGIN_GITHUB_URL, warning_msg)
+        finally:
+            if os.path.exists(test_path):
+                os.remove(test_path)
+
+    @patch('azext_fleet.custom.shutil.which')
+    @patch('azext_fleet.custom.subprocess.run')
+    @patch('azext_fleet.custom.logger')
+    def test_convert_handles_subprocess_error(self, mock_logger, mock_subprocess, mock_which):
+        """Test that subprocess errors are handled gracefully."""
+        mock_which.return_value = MOCK_KUBELOGIN_PATH
+        from subprocess import CalledProcessError
+        mock_subprocess.side_effect = CalledProcessError(1, KUBELOGIN_BINARY)
+
+        with tempfile.NamedTemporaryFile(mode='w', suffix=KUBECONFIG_SUFFIX, delete=False) as f:
+            test_path = f.name
+
+        try:
+            _convert_kubeconfig_to_azurecli(test_path)
+
+            # Verify error was logged
+            mock_logger.warning.assert_called_once()
+            warning_msg = mock_logger.warning.call_args[0][0]
+            self.assertIn("Failed to convert kubeconfig with kubelogin", warning_msg)
+        finally:
+            if os.path.exists(test_path):
+                os.remove(test_path)
+
+    @patch('azext_fleet.custom.shutil.which')
+    @patch('azext_fleet.custom.subprocess.run')
+    @patch('azext_fleet.custom.logger')
+    def test_convert_handles_timeout(self, mock_logger, mock_subprocess, mock_which):
+        """Test that timeout errors are handled gracefully."""
+        mock_which.return_value = MOCK_KUBELOGIN_PATH
+        from subprocess import TimeoutExpired
+        mock_subprocess.side_effect = TimeoutExpired(KUBELOGIN_BINARY, KUBELOGIN_TIMEOUT)
+
+        with tempfile.NamedTemporaryFile(mode='w', suffix=KUBECONFIG_SUFFIX, delete=False) as f:
+            test_path = f.name
+
+        try:
+            _convert_kubeconfig_to_azurecli(test_path)
+
+            # Verify timeout was logged
+            mock_logger.warning.assert_called_once()
+            warning_msg = mock_logger.warning.call_args[0][0]
+            self.assertIn("kubelogin command timed out", warning_msg)
+        finally:
+            if os.path.exists(test_path):
+                os.remove(test_path)
+
+    @patch('azext_fleet.custom.shutil.which')
+    @patch('azext_fleet.custom.subprocess.run')
+    @patch('azext_fleet.custom.logger')
+    def test_convert_handles_generic_exception(self, mock_logger, mock_subprocess, mock_which):
+        """Test that generic exceptions are handled gracefully."""
+        mock_which.return_value = MOCK_KUBELOGIN_PATH
+        mock_subprocess.side_effect = Exception("Unexpected error")
+
+        with tempfile.NamedTemporaryFile(mode='w', suffix=KUBECONFIG_SUFFIX, delete=False) as f:
+            test_path = f.name
+
+        try:
+            _convert_kubeconfig_to_azurecli(test_path)
+
+            # Verify error was logged
+            mock_logger.warning.assert_called_once()
+            warning_msg = mock_logger.warning.call_args[0][0]
+            self.assertIn("Error running kubelogin", warning_msg)
+        finally:
+            if os.path.exists(test_path):
+                os.remove(test_path)
+
+    @patch('azext_fleet.custom.shutil.which')
+    @patch('azext_fleet.custom.subprocess.run')
+    def test_convert_skips_stdout_path(self, mock_subprocess, mock_which):
+        """Test that conversion is skipped when path is stdout."""
+        mock_which.return_value = MOCK_KUBELOGIN_PATH
+
+        _convert_kubeconfig_to_azurecli(STDOUT_PATH)
+
+        # Verify subprocess was not called
+        mock_subprocess.assert_not_called()
+
+    @patch('azext_fleet.custom.shutil.which')
+    @patch('azext_fleet.custom.subprocess.run')
+    def test_convert_with_custom_path(self, mock_subprocess, mock_which):
+        """Test conversion with custom kubeconfig path."""
+        mock_which.return_value = MOCK_KUBELOGIN_PATH
+        mock_subprocess.return_value = MagicMock()
+
+        _convert_kubeconfig_to_azurecli(CUSTOM_CONFIG_PATH)
+
+        # Verify kubelogin was called with custom path
+        call_args = mock_subprocess.call_args
+        self.assertEqual(call_args[0][0], KUBELOGIN_CMD_PREFIX + [CUSTOM_CONFIG_PATH])
+
+
+if __name__ == '__main__':
+    unittest.main()

src/fleet/setup.py

@@ -16,7 +16,7 @@
 
 # TODO: Confirm this is the right version number you want and it matches your
 # HISTORY.rst entry.
-VERSION = '1.8.2'
+VERSION = '1.8.3'
 
 # The full list of classifiers is available at
 # https://pypi.python.org/pypi?%3Aaction=list_classifiers