populate cluster config from client.authentication.k8s.io/exec extension

Author: wkloucek

kubernetes/base/config/exec_provider.py

@@ -58,7 +58,7 @@ def __init__(self, exec_config, cwd, cluster=None):
         else:
             self.cluster = None
         self.cwd = cwd or None
-    
+
     @property
     def shell(self):
         # for windows systems `shell` should be `True`
@@ -81,6 +81,10 @@ def run(self, previous_response=None):
             kubernetes_exec_info['spec']['response'] = previous_response
         if self.cluster:
             kubernetes_exec_info['spec']['cluster'] = self.cluster.value
+            for extension in self.cluster.value["extensions"]:
+                if extension["name"] == "client.authentication.k8s.io/exec":
+                    kubernetes_exec_info["spec"]["cluster"]["config"] = extension["extension"]
+                    break
 
         self.env['KUBERNETES_EXEC_INFO'] = json.dumps(kubernetes_exec_info)
         process = subprocess.Popen(

kubernetes/base/config/exec_provider_test.py

@@ -183,6 +183,39 @@ def test_with_cluster_info(self, mock):
         obj = json.loads(mock.call_args.kwargs['env']['KUBERNETES_EXEC_INFO'])
         self.assertEqual(obj['spec']['cluster']['server'], 'name.company.com')
 
+    @mock.patch("subprocess.Popen")
+    def test_with_cluster_info_from_exec_extension(self, mock):
+        instance = mock.return_value
+        instance.wait.return_value = 0
+        instance.communicate.return_value = (self.output_ok, "")
+        ep = ExecProvider(
+            self.input_with_cluster,
+            None,
+            ConfigNode(
+                "cluster",
+                {
+                    "server": "name.company.com",
+                    "extensions": [
+                        {
+                            "name": "client.authentication.k8s.io/exec",
+                            "extension": {
+                                "namespace": "myproject",
+                                "name": "mycluster",
+                            },
+                        },
+                    ],
+                },
+            ),
+        )
+        result = ep.run()
+        self.assertTrue(isinstance(result, dict))
+        self.assertTrue("token" in result)
+
+        obj = json.loads(mock.call_args.kwargs["env"]["KUBERNETES_EXEC_INFO"])
+        self.assertEqual(obj["spec"]["cluster"]["server"], "name.company.com")
+        self.assertEqual(obj["spec"]["cluster"]["config"]["namespace"], "myproject")
+        self.assertEqual(obj["spec"]["cluster"]["config"]["name"], "mycluster")
+
 
 if __name__ == '__main__':
     unittest.main()