dont expose api key in configmap

Author: mainred

src/aks-agent/azext_aks_agent/agent/k8s/aks_agent_manager.py

@@ -3,6 +3,7 @@
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
 
+import base64
 import json
 import os
 import time
@@ -80,14 +81,13 @@ def __init__(self, namespace: str = AGENT_NAMESPACE, kubeconfig_path: Optional[s
 
         self.kubeconfig_path = kubeconfig_path
 
+        # Initialize Kubernetes client
+        self._init_k8s_client()
         # Use provided helm manager or create a new one with kubeconfig
         self.helm_manager = helm_manager or HelmManager(kubeconfig_path=self.kubeconfig_path)
 
         self._load_existing_helm_release_config()
 
-        # Initialize Kubernetes client
-        self._init_k8s_client()
-
     def set_aks_context(self, resource_group_name: Optional[str] = None,
                         cluster_name: Optional[str] = None,
                         subscription_id: Optional[str] = None):
@@ -159,7 +159,11 @@ def _load_existing_helm_release_config(self):
                 self.llm_config_manager.model_list = model_list
                 if not model_list:
                     logger.warning("No modelList found in Helm values")
-                logger.debug("LLM configuration loaded from Helm values: %d models found", len(model_list))
+                else:
+                    logger.debug("LLM configuration loaded from Helm values: %d models found", len(model_list))
+
+                    # Read API keys from Kubernetes secret and populate model_list
+                    self._populate_api_keys_from_secret()
 
                 # Load managed identity client ID if present
                 mcp_addons = helm_values.get("mcpAddons", {})
@@ -183,6 +187,59 @@ def _load_existing_helm_release_config(self):
             logger.error("Failed to load LLM config from Helm values: %s", e)
             raise AzCLIError(f"Failed to load LLM config from Helm values: {e}")
 
+    def _populate_api_keys_from_secret(self):
+        """
+        Read API keys from Kubernetes secret and populate them into model_list.
+
+        The model_list from Helm values contains environment variable references like
+        '{{ env.AZURE_GPT_4_API_KEY }}'. This method reads the actual API keys from
+        the Kubernetes secret and replaces those references with actual values.
+        """
+        try:
+            # Try to read the secret
+            secret = self.core_v1.read_namespaced_secret(
+                name=self.llm_secret_name,
+                namespace=self.namespace
+            )
+
+            if not secret.data:
+                logger.warning("Secret '%s' exists but has no data", self.llm_secret_name)
+                return
+
+            # Decode secret data (base64 encoded)
+            secret_data = {}
+            for key, value in secret.data.items():
+                decoded_value = base64.b64decode(value).decode("utf-8")
+                secret_data[key] = decoded_value
+
+            logger.debug("Read %d API keys from secret '%s'", len(secret_data), self.llm_secret_name)
+
+            from azext_aks_agent.agent.llm_providers.base import LLMProvider
+
+            # Populate API keys into model_list
+
+            for model_name, model_config in self.llm_config_manager.model_list.items():
+                # Get the expected secret key for this model
+                secret_key = LLMProvider.sanitize_k8s_secret_key(model_config)
+
+                # If the secret contains this key, populate it
+                if secret_key in secret_data:
+                    model_config["api_key"] = secret_data[secret_key]
+                    logger.debug("Populated API key for model '%s' from secret key '%s'",
+                                 model_name, secret_key)
+                else:
+                    logger.warning("API key is not found for model '%s', please update the model '%s' API key.",
+                                   model_name, model_name)
+
+        except ApiException as e:
+            if e.status == 404:
+                logger.debug("Secret '%s' not found in namespace '%s', skipping API key population",
+                             self.llm_secret_name, self.namespace)
+            else:
+                logger.warning("Failed to read secret '%s': %s", self.llm_secret_name, e)
+        except Exception as e:  # pylint: disable=broad-exception-caught
+            logger.warning("Unexpected error reading API keys from secret: %s", e)
+
     def get_agent_pods(self) -> Tuple[bool, Union[List[str], str]]:
         """
         Get running AKS agent pods from the Kubernetes cluster.
@@ -738,7 +795,7 @@ def _create_helm_values(self):
         env_vars = self.llm_config_manager.get_env_vars(self.llm_secret_name)
 
         helm_values = {
-            "modelList": self.llm_config_manager.model_list,
+            "modelList": self.llm_config_manager.secured_model_list(),
             "additionalEnvVars": env_vars,
             "nodeSelector": {"kubernetes.io/os": "linux"},
         }

src/aks-agent/azext_aks_agent/agent/llm_config_manager.py

@@ -24,6 +24,12 @@ def save(self, provider: LLMProvider, params: dict):
         params["model"] = model_name
         self.model_list[model_name] = params
 
+    def secured_model_list(self) -> Dict[str, dict]:
+        secured_config = {}
+        for model_name, model_config in self.model_list.items():
+            secured_config[model_name] = LLMProvider.to_secured_model_list_config(model_config)
+        return secured_config
+
     def get_llm_model_secret_data(self) -> Dict[str, str]:
         """
         Get Kubernetes secret data for all LLM models in the configuration.

src/aks-agent/azext_aks_agent/agent/llm_providers/base.py

@@ -202,14 +202,14 @@ def sanitize_key_part(part):
         return secret_key
 
     @classmethod
-    def to_model_list_config(cls, params: dict) -> Dict[str, dict]:
+    def to_secured_model_list_config(cls, params: dict) -> Dict[str, dict]:
         """Create a model config dictionary for the model list from the provider parameters.
+        Returns a copy of params with the api_key replaced by environment variable reference.
         """
         secret_key = cls.sanitize_k8s_secret_key(params)
-        model_name = params.get("model")
-        model_list_config = {model_name: params}
-        model_list_config[model_name].update({"api_key": f"{{{{ env.{secret_key} }}}}"})
-        return model_list_config
+        secured_params = params.copy()
+        secured_params.update({"api_key": f"{{{{ env.{secret_key} }}}}"})
+        return secured_params
 
     @classmethod
     def to_env_vars(cls, secret_name, params: dict) -> Dict[str, str]:

src/aks-agent/azext_aks_agent/tests/latest/test_llm_config_manager.py

@@ -21,13 +21,10 @@ def setUp(self):
         """Set up test fixtures."""
         self.manager = LLMConfigManager()
         self.test_model_config = {
-            "provider": "azure",
             "model": "gpt-4",
             "api_key": "test-key",
             "api_base": "https://test.openai.azure.com",
             "api_version": "2023-05-15",
-            "DEPLOYMENT_NAME": "gpt-4-deployment",
-            "MODEL_NAME": "gpt-4"
         }
 
     def test_init_empty(self):
@@ -85,6 +82,76 @@ def test_get_env_vars(self, mock_to_env):
         self.assertIsNotNone(result)
         mock_to_env.assert_called_once()
 
+    @patch('azext_aks_agent.agent.llm_config_manager.LLMProvider.to_secured_model_list_config')
+    def test_secured_model_list_multiple_models(self, mock_to_secured):
+        """Test secured_model_list with multiple models."""
+        gpt4_secured = {
+            "model": "azure/gpt-4",
+            "api_key": "{{ env.AZURE_GPT_4_API_KEY }}"
+        }
+        gpt35_secured = {
+            "model": "azure/gpt-35-turbo",
+            "api_key": "{{ env.AZURE_GPT_35_TURBO_API_KEY }}"
+        }
+
+        mock_to_secured.side_effect = [gpt4_secured, gpt35_secured]
+
+        self.manager.model_list = {
+            "azure/gpt-4": {"model": "azure/gpt-4", "api_key": "key1"},
+            "azure/gpt-35-turbo": {"model": "azure/gpt-35-turbo", "api_key": "key2"}
+        }
+
+        result = self.manager.secured_model_list()
+
+        self.assertEqual(len(result), 2)
+        self.assertIn("azure/gpt-4", result)
+        self.assertIn("azure/gpt-35-turbo", result)
+        self.assertEqual(result["azure/gpt-4"], gpt4_secured)
+        self.assertEqual(result["azure/gpt-35-turbo"], gpt35_secured)
+        self.assertEqual(mock_to_secured.call_count, 2)
+
+    @patch('azext_aks_agent.agent.llm_config_manager.LLMProvider.to_secured_model_list_config')
+    def test_secured_model_list_does_not_modify_original(self, mock_to_secured):
+        """Test that secured_model_list doesn't modify the original model_list."""
+        original_api_key = "test-key"
+        original_config = {
+            "model": "azure/gpt-4",
+            "api_key": original_api_key
+        }
+
+        secured_config = {
+            "model": "azure/gpt-4",
+            "api_key": "{{ env.AZURE_GPT_4_API_KEY }}"
+        }
+        mock_to_secured.return_value = secured_config
+
+        self.manager.model_list = {
+            "azure/gpt-4": original_config
+        }
+
+        result = self.manager.secured_model_list()
+
+        # Verify original model_list is unchanged
+        self.assertEqual(self.manager.model_list["azure/gpt-4"]["api_key"], original_api_key)
+
+        # Verify result has secured api_key
+        self.assertEqual(result["azure/gpt-4"]["api_key"], "{{ env.AZURE_GPT_4_API_KEY }}")
+
+    @patch('azext_aks_agent.agent.llm_config_manager.LLMProvider.to_secured_model_list_config')
+    def test_secured_model_list_preserves_model_names(self, mock_to_secured):
+        """Test that secured_model_list preserves model names as keys."""
+        mock_to_secured.return_value = {"secured": "config"}
+
+        model_names = ["azure/gpt-4", "openai/gpt-4-turbo", "anthropic/claude-3"]
+        for model_name in model_names:
+            self.manager.model_list[model_name] = {"model": model_name}
+
+        result = self.manager.secured_model_list()
+
+        # Verify all model names are preserved as keys
+        for model_name in model_names:
+            self.assertIn(model_name, result)
+
 
 if __name__ == '__main__':
     unittest.main()