feat(core): support AICORE_SERVICE_KEY env var for credentials and rename env vars

packages/core/ai_core_sdk/credentials.py

@@ -8,13 +8,12 @@
 from dataclasses import dataclass
 
 from ai_core_sdk.helpers import get_home
-from ai_core_sdk.helpers.constants import (AI_CORE_PREFIX, AUTH_ENDPOINT_SUFFIX, CONFIG_FILE_ENV_VAR, PROFILE_ENV_VAR,
-                                           VCAP_AICORE_SERVICE_NAME, VCAP_SERVICES_ENV_VAR)
+from ai_core_sdk.helpers.constants import (AI_CORE_PREFIX, AUTH_ENDPOINT_SUFFIX, ENV_VAR_AICORE_CONFIG_FILE, ENV_VAR_AICORE_PROFILE,
+                                           ENV_VAR_AICORE_SERVICE_KEY, VCAP_AICORE_SERVICE_NAME, ENV_VAR_VCAP_SERVICES)
 from ai_core_sdk.helpers.logging import get_logger
 
 logger = get_logger()
 
-
 def get_nested_value(data_dict, keys: List[str]):
     """
     Retrieve a nested value from a dictionary using a list of strings.
@@ -28,14 +27,20 @@ def get_nested_value(data_dict, keys: List[str]):
         current_value = current_value[key]
     return current_value
 
+def _get_nested_value_safe(data: Dict, keys) -> Optional[Any]:
+    try:
+        return get_nested_value(data, keys)
+    except KeyError:
+        logger.debug("Key path %s not found in service key.", keys)
+        return None
 
 @dataclass
 class VCAPEnvironment:
     services: List[Service]
 
     @classmethod
     def from_env(cls, env_var: Optional[str] = None):
-        env_var = env_var or VCAP_SERVICES_ENV_VAR
+        env_var = env_var or ENV_VAR_VCAP_SERVICES
         env = json.loads(os.environ.get(env_var, '{}'))
         return cls.from_dict(env)
 
@@ -148,9 +153,9 @@ class Source:
 def init_conf(profile: str = None):
     # Read configuration from ${AICORE_HOME}/config_<profile>.json.
     home = pathlib.Path(get_home())
-    profile = profile or os.environ.get(PROFILE_ENV_VAR)
+    profile = profile or os.environ.get(ENV_VAR_AICORE_PROFILE)
     profile_config_file = f'config_{profile}.json'
-    direct_config_file = pathlib.Path(os.getenv(CONFIG_FILE_ENV_VAR)) if os.getenv(CONFIG_FILE_ENV_VAR) else None
+    direct_config_file = pathlib.Path(os.getenv(ENV_VAR_AICORE_CONFIG_FILE)) if os.getenv(ENV_VAR_AICORE_CONFIG_FILE) else None
     path_to_config = (direct_config_file or
                       (home / ('config.json' if profile in ('default', '', None) else profile_config_file)))
     config = {}
@@ -241,12 +246,44 @@ def _str_or_none(value) -> Optional[str]:
     return str(value) if value else None
 
 
+def _load_service_key() -> Dict[str, Any]:
+    """Read and parse AICORE_SERVICE_KEY from the environment.
+
+    :return: Parsed service key dict, or an empty dict if the env var is not set.
+    :raises ValueError: If the env var is set but contains invalid JSON.
+    """
+    service_key_json_string = os.environ.get(ENV_VAR_AICORE_SERVICE_KEY)
+    if not service_key_json_string:
+        return {}
+    try:
+        return json.loads(service_key_json_string)
+    except json.JSONDecodeError as exc:
+        raise ValueError(
+            f"{ENV_VAR_AICORE_SERVICE_KEY} is set but contains invalid JSON: {exc}"
+        ) from exc
+
+def _load_vcap_service_key() -> Optional[Dict[str, Any]]:
+    """Read the AI Core service credentials from the ``VCAP_SERVICES`` environment variable.
+
+    Parses ``VCAP_SERVICES``, looks up the entry named ``'aicore'``, and returns
+    its credentials dict.  Returns ``None`` if ``VCAP_SERVICES`` is not set or
+    does not contain an ``'aicore'`` binding.
+
+    :return: Credentials dict from the ``aicore`` VCAP binding, or ``None`` if absent.
+    :rtype: Optional[Dict[str, Any]]
+    """
+    try:
+        vcap_service = VCAPEnvironment.from_env()[VCAP_AICORE_SERVICE_NAME]
+    except KeyError:
+        vcap_service = None
+    return vcap_service
+
 def fetch_credentials(profile: str = None, credential_values: List[CredentialsValue] = CORE_CREDENTIAL_VALUES,
                       validate: bool = True, **kwargs) -> Dict[str, str]:
     """
     Fetch credentials from a single source based on precedence.
 
-    Precedence order: kwargs > environment variables > config file > VCAP service
+    Precedence order: kwargs > separate environment variables > service key > config file > VCAP service
 
     Once a source is selected (first one with any credential), all credentials
     come from that source only. Resource group is an exception and follows
@@ -256,20 +293,20 @@ def fetch_credentials(profile: str = None, credential_values: List[CredentialsVa
     """
     config = init_conf(profile=profile)
 
-    try:
-        vcap_service = VCAPEnvironment.from_env()[VCAP_AICORE_SERVICE_NAME]
-    except KeyError:
-        vcap_service = None
-
+    # `cv.vcap_key` describes the full path inside a VCAP_SERVICES entry, starting with
+    # `credentials` (e.g. `('credentials', 'clientid')`)
     sources = [
         Source("kwargs",
                lambda cv: _str_or_none(kwargs.get(cv.name))),
         Source("environment variables",
                lambda cv: _str_or_none(os.environ.get(f'{AI_CORE_PREFIX}_{cv.name.upper()}'))),
+        # A service key is already the inner credentials object, so the leading `credentials` segment is stripped.
+        Source("service key",
+               lambda cv, service_key = _load_service_key(): _str_or_none(_get_nested_value_safe(service_key, cv.vcap_key[1:])) if cv.vcap_key else None),
         Source("config file",
                lambda cv: _str_or_none(config.get(f'{AI_CORE_PREFIX}_{cv.name.upper()}'))),
         Source("VCAP service",
-               lambda cv: _str_or_none(vcap_service.get(cv.vcap_key, None) if vcap_service and cv.vcap_key else None)),
+               lambda cv, vcap_service = _load_vcap_service_key(): _str_or_none(vcap_service.get(cv.vcap_key, None) if vcap_service and cv.vcap_key else None)),
     ]
 
     credentials = _resolve_credentials(sources, credential_values)

packages/core/ai_core_sdk/helpers/__init__.py

@@ -2,7 +2,7 @@
 from typing import Dict
 
 from ai_api_client_sdk.helpers.authenticator import Authenticator
-from .constants import DEFAULT_HOME_PATH, HOME_PATH_ENV_VAR
+from .constants import DEFAULT_HOME_PATH, ENV_VAR_AICORE_HOME_PATH
 
 
 def form_top_skip_params(top: int = None, skip: int = None) -> Dict[str, int]:
@@ -36,4 +36,4 @@ def is_within_aicore() -> bool:
 
 
 def get_home() -> str:
-    return os.environ.get(HOME_PATH_ENV_VAR, DEFAULT_HOME_PATH)
+    return os.environ.get(ENV_VAR_AICORE_HOME_PATH, DEFAULT_HOME_PATH)

packages/core/ai_core_sdk/helpers/constants.py

@@ -3,13 +3,14 @@
 
 AI_CORE_PREFIX = 'AICORE'
 AUTH_ENDPOINT_SUFFIX = '/oauth/token'
-CONFIG_FILE_ENV_VAR = f'{AI_CORE_PREFIX}_CONFIG'
+ENV_VAR_AICORE_CONFIG_FILE = f'{AI_CORE_PREFIX}_CONFIG'
 DEBUG_ENV_VAR_NAME = "DEBUG"
 DEFAULT_HOME_PATH = os.path.join(os.path.expanduser('~'), '.aicore')
-HOME_PATH_ENV_VAR = f'{AI_CORE_PREFIX}_HOME'
-PROFILE_ENV_VAR = f'{AI_CORE_PREFIX}_PROFILE'
+ENV_VAR_AICORE_HOME_PATH = f'{AI_CORE_PREFIX}_HOME'
+ENV_VAR_AICORE_PROFILE = f'{AI_CORE_PREFIX}_PROFILE'
+ENV_VAR_AICORE_SERVICE_KEY = f'{AI_CORE_PREFIX}_SERVICE_KEY'
 VCAP_AICORE_SERVICE_NAME = 'aicore'
-VCAP_SERVICES_ENV_VAR = 'VCAP_SERVICES'
+ENV_VAR_VCAP_SERVICES = 'VCAP_SERVICES'
 
 
 class Timeouts(Enum):

packages/core/integration_tests/test_e2e_x509.py

@@ -9,8 +9,8 @@
 from . import write_x509_credentials_into_files, remove_x509_credentials
 from .ai_core_v2_client_e2e_test_base import AICoreV2ClientE2ETestBase
 from ai_core_sdk.ai_core_v2_client import AICoreV2Client
-from ai_core_sdk.helpers.constants import (AI_CORE_PREFIX, HOME_PATH_ENV_VAR, VCAP_AICORE_SERVICE_NAME,
-                                           VCAP_SERVICES_ENV_VAR)
+from ai_core_sdk.helpers.constants import (AI_CORE_PREFIX, ENV_VAR_AICORE_HOME_PATH, VCAP_AICORE_SERVICE_NAME,
+                                           ENV_VAR_VCAP_SERVICES)
 from ai_core_sdk.models import Scenario
 
 
@@ -72,7 +72,7 @@ def _query_and_assert_scenarios(self, client: AICoreV2Client):
         self.assertIsNotNone(scenario.id)
         self.assertIsNotNone(scenario.name)
 
-    @patch.dict(os.environ, {VCAP_SERVICES_ENV_VAR: VCAP_SERVICE_X509_ENV_VALUE, 'AICORE_RESOURCE_GROUP': RESOURCE_GROUP_ID})
+    @patch.dict(os.environ, {ENV_VAR_VCAP_SERVICES: VCAP_SERVICE_X509_ENV_VALUE, 'AICORE_RESOURCE_GROUP': RESOURCE_GROUP_ID})
     def test_x509_from_vcap(self):
         client = AICoreV2Client.from_env()
         self._query_and_assert_scenarios(client)
@@ -86,7 +86,7 @@ def test_x509_from_profile(self):
                 json.dump({}, f)
             with open(profile_config_path, 'w') as f:
                 json.dump(self.valid_x509_config, f)
-            with patch.dict(os.environ, {HOME_PATH_ENV_VAR: str(temp_dir)}):
+            with patch.dict(os.environ, {ENV_VAR_AICORE_HOME_PATH: str(temp_dir)}):
                 client = AICoreV2Client.from_env(profile_name=profile)
                 self._query_and_assert_scenarios(client)
 

packages/core/tests/ai_core_client/test_ai_core_v2_client.py

@@ -5,7 +5,7 @@
 from unittest import TestCase
 from unittest.mock import MagicMock, patch
 from ai_core_sdk.ai_core_v2_client import AICoreV2Client
-from ai_core_sdk.helpers.constants import (AI_CORE_PREFIX, HOME_PATH_ENV_VAR, VCAP_SERVICES_ENV_VAR,
+from ai_core_sdk.helpers.constants import (AI_CORE_PREFIX, ENV_VAR_AICORE_HOME_PATH, ENV_VAR_VCAP_SERVICES,
                                            VCAP_AICORE_SERVICE_NAME)
 from ai_core_sdk.exception import AIAPIAuthenticatorException
 from ai_core_sdk.resource_clients.internal_rest_client import InternalRestClient
@@ -192,7 +192,7 @@ def test_x509_file_path_from_config(self):
             config[k] = f'cfg_{v}'
 
         with tempfile.TemporaryDirectory() as temp_dir:
-            with patch.dict(os.environ, {HOME_PATH_ENV_VAR: temp_dir}):
+            with patch.dict(os.environ, {ENV_VAR_AICORE_HOME_PATH: temp_dir}):
                 config_file_path = os.path.join(temp_dir, 'config.json')
                 with open(config_file_path, 'w') as f:
                     json.dump(config, f)
@@ -205,7 +205,7 @@ def test_x509_file_path_from_config(self):
 
         AICoreV2Client.__init__ = aicv2c_init
 
-    @patch.dict(os.environ, {VCAP_SERVICES_ENV_VAR: VCAP_SERVICE_X509_ENV_VALUE})
+    @patch.dict(os.environ, {ENV_VAR_VCAP_SERVICES: VCAP_SERVICE_X509_ENV_VALUE})
     def test_x509_from_vcap(self):
         vcap_dict_credentials = VCAP_SERVICE_X509_DICT[VCAP_AICORE_SERVICE_NAME][0]['credentials']
         init_mock = MagicMock(return_value=None)

packages/core/tests/ai_core_client/test_cli.py

@@ -8,7 +8,7 @@
 
 from ai_core_sdk.ai_core_v2_client import AICoreV2Client
 from ai_core_sdk.helpers import get_home
-from ai_core_sdk.helpers.constants import HOME_PATH_ENV_VAR
+from ai_core_sdk.helpers.constants import ENV_VAR_AICORE_HOME_PATH
 
 from click.testing import CliRunner
 
@@ -30,7 +30,7 @@ class TestAICoreCLI(TestCase):
 
     def test_from_env(self):
         with tempfile.TemporaryDirectory() as temp_dir:
-            with patch.dict(os.environ, {HOME_PATH_ENV_VAR: temp_dir}):
+            with patch.dict(os.environ, {ENV_VAR_AICORE_HOME_PATH: temp_dir}):
                 from ai_core_sdk.cli import cli
                 runner = CliRunner()
                 temp_dir = pathlib.Path(temp_dir)
@@ -52,7 +52,7 @@ def test_from_env(self):
 
     def test_from_input(self):
         with tempfile.TemporaryDirectory() as temp_dir:
-            with patch.dict(os.environ, {HOME_PATH_ENV_VAR: temp_dir}):
+            with patch.dict(os.environ, {ENV_VAR_AICORE_HOME_PATH: temp_dir}):
                 from ai_core_sdk.cli import cli
                 runner = CliRunner()
                 result = runner.invoke(cli, [f'configure', '-s', AICORE_DUMMY_KEY['clientsecret'],