[AKS] Fix #34091: az aks get-credentials: Fix kubelogin conversion for explicit kubeconfig paths

src/azure-cli/azure/cli/command_modules/acs/custom.py

@@ -2064,13 +2064,12 @@ def aks_get_credentials(cmd, client, resource_group_name, name, admin=False,
             path, kubeconfig, overwrite_existing, context_name)
 
         # Check if kubeconfig requires kubelogin with devicecode and convert it
-        if uses_kubelogin_devicecode(kubeconfig):
+        if path != "-" and uses_kubelogin_devicecode(kubeconfig):
             if which("kubelogin"):
                 try:
-                    # Run kubelogin convert-kubeconfig -l azurecli
+                    # Select the merged file without changing how relative paths are resolved.
                     subprocess.run(
-                        ["kubelogin", "convert-kubeconfig", "-l", "azurecli"],
-                        cwd=os.path.dirname(path),
+                        ["kubelogin", "convert-kubeconfig", "-l", "azurecli", "--kubeconfig", path],
                         check=True,
                     )
                     logger.warning("Converted kubeconfig to use Azure CLI authentication.")

src/azure-cli/azure/cli/command_modules/acs/tests/latest/test_custom.py

@@ -8,9 +8,12 @@
 import hashlib
 import io
 import shutil
+import stat
 import subprocess
+import sys
 import tarfile
 import tempfile
+import textwrap
 import unittest
 from unittest import mock
 from urllib.error import HTTPError, URLError
@@ -57,6 +60,7 @@
     aks_agentpool_upgrade,
     aks_disable_addons,
     aks_enable_addons,
+    aks_get_credentials,
     aks_stop,
     aks_upgrade,
     is_monitoring_addon_enabled,
@@ -94,6 +98,241 @@
 )
 
 
+class AksGetCredentialsTest(unittest.TestCase):
+    def setUp(self):
+        directory = tempfile.TemporaryDirectory()
+        self.addCleanup(directory.cleanup)
+        self.directory = directory.name
+        self.addCleanup(os.chdir, os.getcwd())
+        os.chdir(self.directory)
+        environment = dict(os.environ, HOME=self.directory, USERPROFILE=self.directory)
+        environment.pop('KUBECONFIG', None)
+        patcher = mock.patch.dict(os.environ, environment, clear=True)
+        patcher.start()
+        self.addCleanup(patcher.stop)
+
+        self.cmd = mock.Mock()
+        self.cmd.cli_ctx.cloud.profile = 'latest'
+        self.client = mock.Mock()
+        self.kubeconfig = yaml.safe_dump({
+            'apiVersion': 'v1',
+            'kind': 'Config',
+            'clusters': [{'name': 'test-cluster', 'cluster': {'server': 'https://test-cluster'}}],
+            'contexts': [{'name': 'test-cluster', 'context': {
+                'cluster': 'test-cluster', 'user': 'clusterUser_rg_test-cluster'}}],
+            'current-context': 'test-cluster',
+            'users': [{'name': 'clusterUser_rg_test-cluster', 'user': {'exec': {
+                'apiVersion': 'client.authentication.k8s.io/v1beta1',
+                'command': 'kubelogin', 'args': ['get-token', '--login', 'devicecode']}}}],
+        })
+        self._set_credentials(self.kubeconfig)
+        self.unrelated_config = self.kubeconfig.replace('test-cluster', 'other-cluster')
+        self.default_file = os.path.join(self.directory, '.kube', 'config')
+        self.other_file = os.path.join(self.directory, 'other.config')
+        self._write_config(self.default_file, self.unrelated_config)
+        self._write_config(self.other_file, self.unrelated_config)
+
+    def _set_credentials(self, kubeconfig):
+        result = mock.Mock(kubeconfigs=[mock.Mock(value=kubeconfig.encode('utf-8'))])
+        self.client.list_cluster_user_credentials.return_value = result
+        self.client.list_cluster_admin_credentials.return_value = result
+
+    @staticmethod
+    def _write_config(path, content):
+        os.makedirs(os.path.dirname(path) or '.', exist_ok=True)
+        with open(path, 'w', encoding='utf-8') as stream:
+            stream.write(content)
+        os.chmod(path, 0o600)
+
+    @staticmethod
+    def _read_config(path):
+        with open(path, encoding='utf-8') as stream:
+            return yaml.safe_load(stream)
+
+    def _get_credentials(self, path, **kwargs):
+        aks_get_credentials(self.cmd, self.client, 'rg', 'test-cluster', path=path, **kwargs)
+
+    def _conversion_cases(self):
+        # The CLI default is passed explicitly: the handler's default is evaluated at import time.
+        return [
+            ('cluster.config', None, 'cluster.config'),
+            (os.path.join('.', 'dot.config'), None, os.path.join('.', 'dot.config')),
+            (os.path.join('nested', 'cluster.config'), None, os.path.join('nested', 'cluster.config')),
+            (os.path.join(self.directory, 'absolute.config'), None,
+             os.path.join(self.directory, 'absolute.config')),
+            (os.path.join('directory with spaces', 'cluster config'), None,
+             os.path.join('directory with spaces', 'cluster config')),
+            ('explicit.config', self.other_file, 'explicit.config'),
+            (self.default_file, None, self.default_file),
+            (self.default_file, os.pathsep.join(['env.config', self.other_file]), 'env.config'),
+            (self.default_file, os.pathsep + self.other_file, self.default_file),
+        ]
+
+    def _assert_merged(self, path, login):
+        config = self._read_config(path)
+        self.assertEqual(config['current-context'], 'test-cluster')
+        user = next(user for user in config['users'] if user['name'] == 'clusterUser_rg_test-cluster')
+        self.assertEqual(user['user']['exec']['args'], ['get-token', '--login', login])
+        for unrelated in (self.default_file, self.other_file):
+            if unrelated != path:
+                with open(unrelated, encoding='utf-8') as stream:
+                    self.assertEqual(stream.read(), self.unrelated_config)
+
+    @mock.patch('azure.cli.command_modules.acs.custom.which', return_value='/bin/kubelogin')
+    @mock.patch('azure.cli.command_modules.acs.custom.subprocess.run')
+    def test_get_credentials_conversion_paths(self, run, which):
+        for path, kubeconfig_env, target in self._conversion_cases():
+            environment = {'KUBECONFIG': kubeconfig_env} if kubeconfig_env is not None else {}
+            with self.subTest(path=path, kubeconfig=kubeconfig_env), mock.patch.dict(os.environ, environment):
+                self._write_config(self.default_file, self.unrelated_config)
+                self._get_credentials(path, overwrite_existing=True)
+                run.assert_called_once_with(
+                    ['kubelogin', 'convert-kubeconfig', '-l', 'azurecli', '--kubeconfig', target], check=True)
+                self._assert_merged(target, 'devicecode')
+                run.reset_mock()
+        which.assert_called_with('kubelogin')
+        self.client.list_cluster_user_credentials.assert_called_with(
+            'rg', 'test-cluster', server_fqdn=None, format=None)
+        self.client.list_cluster_admin_credentials.assert_not_called()
+
+    def _install_kubelogin(self, exit_code=0):
+        """Use a real subprocess, with only the converter's CLI/file boundary implemented."""
+        bin_directory = os.path.join(self.directory, 'bin')
+        os.makedirs(bin_directory, exist_ok=True)
+        executable = os.path.join(bin_directory, 'kubelogin')
+        script = textwrap.dedent('''\
+            import argparse
+            from pathlib import Path
+
+            parser = argparse.ArgumentParser()
+            parser.add_argument('command', choices=['convert-kubeconfig'])
+            parser.add_argument('-l', choices=['azurecli'], required=True)
+            parser.add_argument('--kubeconfig', required=True)
+            args = parser.parse_args()
+            path = Path(args.kubeconfig)
+            content = path.read_text(encoding='utf-8')
+            path.write_text(content.replace('- devicecode', '- azurecli'), encoding='utf-8')
+        ''')
+        if exit_code:
+            script = 'raise SystemExit({})\n'.format(exit_code)
+        with open(executable, 'w', encoding='utf-8') as stream:
+            stream.write('#!{}\n{}'.format(sys.executable, script))
+        os.chmod(executable, 0o700)
+        os.environ['PATH'] = bin_directory + os.pathsep + os.environ.get('PATH', '')
+
+    @unittest.skipIf(os.name == 'nt', 'Executable fixture requires POSIX shebang support')
+    @mock.patch('azure.cli.command_modules.acs.custom.logger')
+    def test_get_credentials_converts_selected_kubeconfig(self, logger):
+        self._install_kubelogin()
+        for path, kubeconfig_env, target in self._conversion_cases():
+            environment = {'KUBECONFIG': kubeconfig_env} if kubeconfig_env is not None else {}
+            with self.subTest(path=path, kubeconfig=kubeconfig_env), mock.patch.dict(os.environ, environment):
+                self._write_config(self.default_file, self.unrelated_config)
+                self._get_credentials(path, overwrite_existing=True)
+                self._assert_merged(target, 'azurecli')
+                self.assertEqual(stat.S_IMODE(os.stat(target).st_mode), 0o600)
+                logger.warning.assert_any_call('Converted kubeconfig to use Azure CLI authentication.')
+                logger.warning.assert_any_call('Merged "test-cluster" as current context in {}'.format(target))
+                logger.reset_mock()
+
+    @mock.patch('azure.cli.command_modules.acs.custom.which')
+    @mock.patch('azure.cli.command_modules.acs.custom.subprocess.run')
+    @mock.patch('azure.cli.command_modules.acs.custom.logger')
+    def test_get_credentials_stdout_skips_conversion(self, logger, run, which):
+        with mock.patch.dict(os.environ, {'KUBECONFIG': self.other_file}), \
+                mock.patch('sys.stdout', new_callable=io.StringIO) as stdout:
+            self._get_credentials('-')
+        self.assertEqual(stdout.getvalue(), self.kubeconfig + '\n')
+        self.assertFalse(os.path.exists('-'))
+        for path in (self.default_file, self.other_file):
+            with open(path, encoding='utf-8') as stream:
+                self.assertEqual(stream.read(), self.unrelated_config)
+        which.assert_not_called()
+        run.assert_not_called()
+        logger.warning.assert_not_called()
+
+    @mock.patch('azure.cli.command_modules.acs.custom.which')
+    @mock.patch('azure.cli.command_modules.acs.custom.subprocess.run')
+    def test_get_credentials_without_devicecode_skips_conversion(self, run, which):
+        token_config = yaml.safe_load(self.kubeconfig)
+        token_config['users'][0]['user'] = {'token': 'test-token'}
+        for kubeconfig in (self.kubeconfig.replace('devicecode', 'azurecli'),
+                           self.kubeconfig.replace('kubelogin', 'other-command'), yaml.safe_dump(token_config)):
+            with self.subTest(kubeconfig=kubeconfig):
+                self._set_credentials(kubeconfig)
+                self._get_credentials('non-devicecode.config', overwrite_existing=True)
+                self.assertEqual(self._read_config('non-devicecode.config'), yaml.safe_load(kubeconfig))
+                which.assert_not_called()
+                run.assert_not_called()
+
+    @mock.patch('azure.cli.command_modules.acs.custom.which', return_value=None)
+    @mock.patch('azure.cli.command_modules.acs.custom.subprocess.run')
+    @mock.patch('azure.cli.command_modules.acs.custom.logger')
+    def test_get_credentials_missing_kubelogin_warns(self, logger, run, which):
+        self._get_credentials('cluster.config')
+        self._assert_merged('cluster.config', 'devicecode')
+        which.assert_called_once_with('kubelogin')
+        run.assert_not_called()
+        self.assertIn('Please install kubelogin', logger.warning.call_args[0][0])
+        logger.warning.assert_any_call('Merged "test-cluster" as current context in cluster.config')
+
+    @mock.patch('azure.cli.command_modules.acs.custom.which', return_value='/bin/kubelogin')
+    @mock.patch('azure.cli.command_modules.acs.custom.subprocess.run')
+    @mock.patch('azure.cli.command_modules.acs.custom.logger')
+    def test_get_credentials_conversion_failure_warns(self, logger, run, _which):
+        for error, warning in (
+                (subprocess.CalledProcessError(17, 'kubelogin'), 'Failed to convert kubeconfig with kubelogin: %s'),
+                (FileNotFoundError(2, 'not found', 'kubelogin'), 'Error running kubelogin: %s')):
+            with self.subTest(error=error):
+                run.side_effect = error
+                self._get_credentials('cluster.config', overwrite_existing=True)
+                self._assert_merged('cluster.config', 'devicecode')
+                logger.warning.assert_any_call(warning, str(error))
+                logger.warning.assert_any_call('Merged "test-cluster" as current context in cluster.config')
+
+    @unittest.skipIf(os.name == 'nt', 'Executable fixture requires POSIX shebang support')
+    @mock.patch('azure.cli.command_modules.acs.custom.logger')
+    def test_get_credentials_nonzero_kubelogin_exit(self, logger):
+        self._install_kubelogin(exit_code=17)
+        self._get_credentials('cluster.config')
+        self._assert_merged('cluster.config', 'devicecode')
+        self.assertEqual(logger.warning.call_args[0][0], 'Failed to convert kubeconfig with kubelogin: %s')
+        self.assertIn('17', logger.warning.call_args[0][1])
+
+    @mock.patch('azure.cli.command_modules.acs.custom.which', return_value='/bin/kubelogin')
+    @mock.patch('azure.cli.command_modules.acs.custom.subprocess.run')
+    def test_get_credentials_preserves_overwrite_context_and_admin(self, _run, _which):
+        for admin in (False, True):
+            with self.subTest(admin=admin):
+                kubeconfig = self.kubeconfig.replace('clusterUser', 'clusterAdmin') if admin else self.kubeconfig
+                self._set_credentials(kubeconfig)
+                self._write_config('cluster.config', kubeconfig.replace('devicecode', 'azurecli'))
+                self._get_credentials('cluster.config', admin=admin, overwrite_existing=True,
+                                      context_name='custom-context', public_fqdn=True)
+                config = self._read_config('cluster.config')
+                self.assertEqual(config['current-context'], 'custom-context-admin' if admin else 'custom-context')
+                self.assertEqual(config['users'][0]['user']['exec']['args'][-1], 'devicecode')
+                self.assertIn('custom-context', [cluster['name'] for cluster in config['clusters']])
+                if admin:
+                    self.client.list_cluster_admin_credentials.assert_called_once_with(
+                        'rg', 'test-cluster', server_fqdn='public')
+                else:
+                    self.client.list_cluster_user_credentials.assert_called_once_with(
+                        'rg', 'test-cluster', server_fqdn='public', format=None)
+
+    @unittest.skipIf(os.name == 'nt', 'Symlink test not applicable on Windows')
+    @mock.patch('azure.cli.command_modules.acs.custom.which')
+    @mock.patch('azure.cli.command_modules.acs.custom.subprocess.run')
+    def test_get_credentials_rejects_symlink_before_conversion(self, run, which):
+        os.symlink(self.default_file, 'linked.config')
+        with self.assertRaisesRegex(CLIError, 'symbolic link'):
+            self._get_credentials('linked.config')
+        with open(self.default_file, encoding='utf-8') as stream:
+            self.assertEqual(stream.read(), self.unrelated_config)
+        which.assert_not_called()
+        run.assert_not_called()
+
+
 class AcsCustomCommandTest(unittest.TestCase):
     def setUp(self):
         self.cli = MockCLI()