Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 37 additions & 7 deletions packages/core/ai_core_sdk/credentials.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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)

Expand Down Expand Up @@ -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 = {}
Expand Down Expand Up @@ -241,12 +246,30 @@ def _str_or_none(value) -> Optional[str]:
return str(value) if value else None

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

q: not strictly part of this PR but is it intended that this eg converts from empty string to None (and 0 to None)? If yes maybe we could make the method name more meaningful



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 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
Expand All @@ -261,11 +284,18 @@ def fetch_credentials(profile: str = None, credential_values: List[CredentialsVa
except KeyError:
vcap_service = None

service_key = _load_service_key()

# `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: _str_or_none(_get_nested_value_safe(service_key, cv.vcap_key[1:])) if cv.vcap_key else None),

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[pp] writing
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)
could be a way avoid the extra service_key = _load_service_key() line.

Source("config file",
lambda cv: _str_or_none(config.get(f'{AI_CORE_PREFIX}_{cv.name.upper()}'))),
Source("VCAP service",
Expand Down
4 changes: 2 additions & 2 deletions packages/core/ai_core_sdk/helpers/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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]:
Expand Down Expand Up @@ -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)
9 changes: 5 additions & 4 deletions packages/core/ai_core_sdk/helpers/constants.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down
8 changes: 4 additions & 4 deletions packages/core/integration_tests/test_e2e_x509.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand Down Expand Up @@ -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)
Expand All @@ -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)

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand All @@ -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)
Expand Down
6 changes: 3 additions & 3 deletions packages/core/tests/ai_core_client/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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)
Expand All @@ -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'],
Expand Down
95 changes: 84 additions & 11 deletions packages/core/tests/ai_core_client/test_credentials.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,8 +14,8 @@
fetch_credentials,
init_conf, CORE_CREDENTIAL_VALUES,
)
from ai_core_sdk.helpers.constants import (AI_CORE_PREFIX, HOME_PATH_ENV_VAR, PROFILE_ENV_VAR, VCAP_SERVICES_ENV_VAR,
VCAP_AICORE_SERVICE_NAME, CONFIG_FILE_ENV_VAR)
from ai_core_sdk.helpers.constants import (AI_CORE_PREFIX, ENV_VAR_AICORE_HOME_PATH, ENV_VAR_AICORE_PROFILE, ENV_VAR_VCAP_SERVICES,
VCAP_AICORE_SERVICE_NAME, ENV_VAR_AICORE_CONFIG_FILE, ENV_VAR_AICORE_SERVICE_KEY)

VCAP_SERVICE_DICT = {
VCAP_AICORE_SERVICE_NAME: [{
Expand Down Expand Up @@ -60,7 +60,7 @@ def setUpClass(cls):
cls.vcap_dict = VCAP_SERVICE_DICT[VCAP_AICORE_SERVICE_NAME][0]

def test_vcap_services(self):
with patch.dict(os.environ, {VCAP_SERVICES_ENV_VAR: VCAP_SERVICE_ENV_VALUE}):
with patch.dict(os.environ, {ENV_VAR_VCAP_SERVICES: VCAP_SERVICE_ENV_VALUE}):
vcap_services = VCAPEnvironment.from_env()
self.assertTrue(all(isinstance(srv, Service) for srv in vcap_services.services))
self.assertEqual(len(vcap_services.services), 1)
Expand Down Expand Up @@ -141,34 +141,34 @@ def test_init_conf(self, mock_logger):
conf = init_conf('MOCK_LLM')

# load default config
with patch.dict(os.environ, {HOME_PATH_ENV_VAR: str(self.temp_dir)}):
with patch.dict(os.environ, {ENV_VAR_AICORE_HOME_PATH: str(self.temp_dir)}):
conf = init_conf()
self.assertDictEqual(conf, self.default_config)
mock_logger.debug.assert_called_with('Config file path %s', self.temp_dir / 'config.json')

# load profile config
with patch.dict(os.environ, {HOME_PATH_ENV_VAR: str(self.temp_dir), PROFILE_ENV_VAR: self.profile}):
with patch.dict(os.environ, {ENV_VAR_AICORE_HOME_PATH: str(self.temp_dir), ENV_VAR_AICORE_PROFILE: self.profile}):
conf = init_conf()
self.assertDictEqual(conf, self.profile_config)
mock_logger.debug.assert_called_with('Config file path %s', self.temp_dir / self.file_name_profile)

# load profile config with profile param
with patch.dict(os.environ, {HOME_PATH_ENV_VAR: str(self.temp_dir)}):
with patch.dict(os.environ, {ENV_VAR_AICORE_HOME_PATH: str(self.temp_dir)}):
conf = init_conf(profile=self.profile)
self.assertDictEqual(conf, self.profile_config)
mock_logger.debug.assert_called_with('Config file path %s', self.temp_dir / self.file_name_profile)

# load profile config via env variable
with patch.dict(os.environ, {f'{AI_CORE_PREFIX}_PROFILE': self.profile,
HOME_PATH_ENV_VAR: str(self.temp_dir)}):
ENV_VAR_AICORE_HOME_PATH: str(self.temp_dir)}):
conf = init_conf()
self.assertDictEqual(conf, self.profile_config)
# overwrite env variable with explicit profile
conf = init_conf(profile='default')
self.assertDictEqual(conf, self.default_config)
mock_logger.debug.assert_called_with('Config file path %s', self.temp_dir / 'config.json')

@patch.dict(os.environ, {VCAP_SERVICES_ENV_VAR: VCAP_SERVICE_ENV_VALUE})
@patch.dict(os.environ, {ENV_VAR_VCAP_SERVICES: VCAP_SERVICE_ENV_VALUE})
@patch('ai_core_sdk.credentials.logger')
def test_fetch_credentials_from_vcap_services(self, mock_logger):
mock_logger.debug = MagicMock()
Expand All @@ -183,7 +183,7 @@ def test_fetch_credentials_from_vcap_services(self, mock_logger):
mock_logger.debug.assert_any_call("Using credentials from: VCAP service")
mock_logger.debug.assert_any_call("No resource_group found in any source")

@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})
@patch('ai_core_sdk.credentials.logger')
def test_fetch_credentials_from_vcap_services_with_x509_env_var(self, mock_logger):
mock_logger.debug = MagicMock()
Expand Down Expand Up @@ -224,7 +224,7 @@ def test_fetch_credentials_from_env(self, mock_logger):
@patch('ai_core_sdk.credentials.logger')
def test_fetch_credentials_from_config_file(self, mock_logger):
mock_logger.debug = MagicMock()
with patch.dict(os.environ, {CONFIG_FILE_ENV_VAR: str(self.temp_dir / 'config.json')}):
with patch.dict(os.environ, {ENV_VAR_AICORE_CONFIG_FILE: str(self.temp_dir / 'config.json')}):
fetch_credentials()

mock_logger.debug.assert_any_call("Using credentials from: config file")
Expand Down Expand Up @@ -276,7 +276,7 @@ def test_init_conf_permission_error(self, mock_logger):
config_file.chmod(0o000)

try:
with patch.dict(os.environ, {CONFIG_FILE_ENV_VAR: str(config_file)}):
with patch.dict(os.environ, {ENV_VAR_AICORE_CONFIG_FILE: str(config_file)}):
conf = init_conf()
# Should return empty config when permission is denied
self.assertDictEqual(conf, {})
Expand All @@ -290,6 +290,79 @@ def test_init_conf_permission_error(self, mock_logger):
# Restore permissions for cleanup in teardown
config_file.chmod(0o644)

@patch('ai_core_sdk.credentials.logger')
def test_fetch_credentials_from_service_key(self, mock_logger):
mock_logger.debug = MagicMock()

service_key = {
'clientid': 'sk-client-id',
'clientsecret': 'sk-client-secret',
'url': 'https://sk-auth-url',
'serviceurls': {'AI_API_URL': 'https://sk-api-url'},
}
with patch.dict(os.environ, {ENV_VAR_AICORE_SERVICE_KEY: json.dumps(service_key)}):
credentials = fetch_credentials()

self.assertEqual(credentials['client_id'], 'sk-client-id')
self.assertEqual(credentials['client_secret'], 'sk-client-secret')
self.assertEqual(credentials['auth_url'], 'https://sk-auth-url/oauth/token')
self.assertEqual(credentials['base_url'], 'https://sk-api-url/v2')
mock_logger.debug.assert_any_call("Using credentials from: service key")

@patch('ai_core_sdk.credentials.logger')
def test_fetch_credentials_from_service_key_x509(self, mock_logger):
mock_logger.debug = MagicMock()

service_key = {
'clientid': 'sk-client-id',
'certurl': 'https://sk-cert-url',
'certificate': 'sk-cert-content',
'key': 'sk-key-content',
'serviceurls': {'AI_API_URL': 'https://sk-api-url'},
}
with patch.dict(os.environ, {ENV_VAR_AICORE_SERVICE_KEY: json.dumps(service_key)}):
credentials = fetch_credentials()

self.assertEqual(credentials['client_id'], 'sk-client-id')
self.assertEqual(credentials['cert_str'], 'sk-cert-content')
self.assertEqual(credentials['key_str'], 'sk-key-content')
self.assertEqual(credentials['auth_url'], 'https://sk-cert-url/oauth/token')
self.assertEqual(credentials['base_url'], 'https://sk-api-url/v2')
mock_logger.debug.assert_any_call("Using credentials from: service key")

@patch('ai_core_sdk.credentials.logger')
def test_service_key_lower_precedence_than_env_vars(self, mock_logger):
mock_logger.debug = MagicMock()

service_key = {
'clientid': 'sk-client-id',
'clientsecret': 'sk-client-secret',
'url': 'https://sk-auth-url',
'serviceurls': {'AI_API_URL': 'https://sk-api-url'},
}
with patch.dict(os.environ, {
ENV_VAR_AICORE_SERVICE_KEY: json.dumps(service_key),
f'{AI_CORE_PREFIX}_CLIENT_ID': 'env-client-id',
f'{AI_CORE_PREFIX}_CLIENT_SECRET': 'env-client-secret',
f'{AI_CORE_PREFIX}_AUTH_URL': 'https://env-auth-url',
f'{AI_CORE_PREFIX}_BASE_URL': 'https://env-base-url',
Comment thread
yamaceay marked this conversation as resolved.
}):
credentials = fetch_credentials()

# env vars win
self.assertEqual(credentials['client_id'], 'env-client-id')
self.assertEqual(credentials['client_secret'], 'env-client-secret')
self.assertEqual(credentials['auth_url'], 'https://env-auth-url/oauth/token')
self.assertEqual(credentials['base_url'], 'https://env-base-url/v2')
mock_logger.debug.assert_any_call("Using credentials from: environment variables")

def test_service_key_invalid_json_raises(self):
with patch.dict(os.environ, {ENV_VAR_AICORE_SERVICE_KEY: 'not-valid-json'}):
with self.assertRaises(ValueError) as ctx:
fetch_credentials()
self.assertIn(ENV_VAR_AICORE_SERVICE_KEY, str(ctx.exception))
self.assertIn('invalid JSON', str(ctx.exception))

@patch('ai_core_sdk.credentials.logger')
def test_injecting_credential_values(self, mock_logger):
mock_logger.debug = MagicMock()
Expand Down
Loading
Loading