feat: send user-agent header with auth token requests (#191)

This commit updates our various request-based authenticators
so that the User-Agent header is included with each outbound
token request. The value of the User-Agent header will be
of the form "ibm-python-sdk-core/<authenticator-type>-<core-version> <os-info>".

Signed-off-by: Phil Adams <[email protected]>

Author: padamstx

ibm_cloud_sdk_core/api_exception.py

@@ -54,7 +54,7 @@ def code(self):
         """The old `code` property with a deprecation warning."""
 
         warnings.warn(
-            'Using the `code` attribute on the `ApiException` is deprecated and'
+            'Using the `code` attribute on the `ApiException` is deprecated and '
             'will be removed in the future. Use `status_code` instead.',
             DeprecationWarning,
         )

ibm_cloud_sdk_core/base_service.py

@@ -1,6 +1,6 @@
 # coding: utf-8
 
-# Copyright 2019 IBM All Rights Reserved.
+# Copyright 2019, 2024 IBM All Rights Reserved.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -18,7 +18,6 @@
 import io
 import json as json_import
 import logging
-import platform
 from http.cookiejar import CookieJar
 from os.path import basename
 from typing import Dict, List, Optional, Tuple, Union
@@ -42,7 +41,7 @@
     SSLHTTPAdapter,
     GzipStream,
 )
-from .version import __version__
+from .private_helpers import _build_user_agent
 
 # Uncomment this to enable http debugging
 # import http.client as http_client
@@ -82,7 +81,6 @@ class BaseService:
         ValueError: If Authenticator is not provided or invalid type.
     """
 
-    SDK_NAME = 'ibm-python-sdk-core'
     ERROR_MSG_DISABLE_SSL = (
         'The connection failed because the SSL certificate is not valid. To use a self-signed '
         'certificate, disable verification of the server\'s SSL certificate by invoking the '
@@ -106,7 +104,7 @@ def __init__(
         self.disable_ssl_verification = disable_ssl_verification
         self.default_headers = None
         self.enable_gzip_compression = enable_gzip_compression
-        self._set_user_agent_header(self._build_user_agent())
+        self._set_user_agent_header(_build_user_agent())
         self.retry_config = None
         self.http_adapter = SSLHTTPAdapter(_disable_ssl_verification=self.disable_ssl_verification)
         if not self.authenticator:
@@ -151,15 +149,6 @@ def disable_retries(self):
         self.http_client.mount('http://', self.http_adapter)
         self.http_client.mount('https://', self.http_adapter)
 
-    @staticmethod
-    def _get_system_info() -> str:
-        return '{0} {1} {2}'.format(
-            platform.system(), platform.release(), platform.python_version()  # OS  # OS version  # Python version
-        )
-
-    def _build_user_agent(self) -> str:
-        return '{0}-{1} {2}'.format(self.SDK_NAME, __version__, self._get_system_info())
-
     def configure_service(self, service_name: str) -> None:
         """Look for external configuration of a service. Set service properties.
 

ibm_cloud_sdk_core/private_helpers.py

@@ -0,0 +1,34 @@
+# coding: utf-8
+
+# Copyright 2024 IBM All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# from ibm_cloud_sdk_core.authenticators import Authenticator
+
+import platform
+from .version import __version__
+
+SDK_NAME = 'ibm-python-sdk-core'
+
+
+def _get_system_info() -> str:
+    return 'os.name={0} os.version={1} python.version={2}'.format(
+        platform.system(), platform.release(), platform.python_version()
+    )
+
+
+def _build_user_agent(component: str = None) -> str:
+    sub_component = ""
+    if component is not None:
+        sub_component = '/{0}'.format(component)
+    return '{0}{1}-{2} {3}'.format(SDK_NAME, sub_component, __version__, _get_system_info())

ibm_cloud_sdk_core/token_managers/container_token_manager.py

@@ -1,6 +1,6 @@
 # coding: utf-8
 
-# Copyright 2021 IBM All Rights Reserved.
+# Copyright 2021, 2024 IBM All Rights Reserved.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -18,6 +18,7 @@
 from typing import Dict, Optional
 
 from .iam_request_based_token_manager import IAMRequestBasedTokenManager
+from ..private_helpers import _build_user_agent
 
 
 logger = logging.getLogger(__name__)
@@ -111,6 +112,7 @@ def __init__(
         self.iam_profile_id = iam_profile_id
 
         self.request_payload['grant_type'] = 'urn:ibm:params:oauth:grant-type:cr-token'
+        self._set_user_agent(_build_user_agent('container-authenticator'))
 
     def retrieve_cr_token(self) -> str:
         """Retrieves the CR token for the current compute resource by reading it from the local file system.

ibm_cloud_sdk_core/token_managers/cp4d_token_manager.py

@@ -1,6 +1,6 @@
 # coding: utf-8
 
-# Copyright 2019 IBM All Rights Reserved.
+# Copyright 2019, 2024 IBM All Rights Reserved.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -17,6 +17,7 @@
 import json
 from typing import Dict, Optional
 
+from ..private_helpers import _build_user_agent
 from .jwt_token_manager import JWTTokenManager
 
 
@@ -76,12 +77,21 @@ def __init__(
         self.headers['Content-Type'] = 'application/json'
         self.proxies = proxies
         super().__init__(url, disable_ssl_verification=disable_ssl_verification, token_name=self.TOKEN_NAME)
+        self._set_user_agent(_build_user_agent('cp4d-authenticator'))
 
     def request_token(self) -> dict:
         """Makes a request for a token."""
+        required_headers = {
+            'User-Agent': self.user_agent,
+        }
+        request_headers = {}
+        if self.headers is not None and isinstance(self.headers, dict):
+            request_headers.update(self.headers)
+        request_headers.update(required_headers)
+
         response = self._request(
             method='POST',
-            headers=self.headers,
+            headers=request_headers,
             url=self.url,
             data=json.dumps({"username": self.username, "password": self.password, "api_key": self.apikey}),
             proxies=self.proxies,

ibm_cloud_sdk_core/token_managers/iam_request_based_token_manager.py

@@ -98,9 +98,15 @@ def request_token(self) -> dict:
         Returns:
              A dictionary containing the bearer token to be subsequently used service requests.
         """
-        headers = {'Content-type': 'application/x-www-form-urlencoded', 'Accept': 'application/json'}
+        required_headers = {
+            'Content-Type': 'application/x-www-form-urlencoded',
+            'Accept': 'application/json',
+            'User-Agent': self._get_user_agent(),
+        }
+        request_headers = {}
         if self.headers is not None and isinstance(self.headers, dict):
-            headers.update(self.headers)
+            request_headers.update(self.headers)
+        request_headers.update(required_headers)
 
         data = dict(self.request_payload)
 
@@ -115,7 +121,7 @@ def request_token(self) -> dict:
         response = self._request(
             method='POST',
             url=(self.url + self.OPERATION_PATH) if self.url else self.url,
-            headers=headers,
+            headers=request_headers,
             data=data,
             auth_tuple=auth_tuple,
             proxies=self.proxies,

ibm_cloud_sdk_core/token_managers/iam_token_manager.py

@@ -1,6 +1,6 @@
 # coding: utf-8
 
-# Copyright 2019 IBM All Rights Reserved.
+# Copyright 2019, 2024 IBM All Rights Reserved.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -17,6 +17,7 @@
 from typing import Dict, Optional
 
 from .iam_request_based_token_manager import IAMRequestBasedTokenManager
+from ..private_helpers import _build_user_agent
 
 
 class IAMTokenManager(IAMRequestBasedTokenManager):
@@ -88,3 +89,5 @@ def __init__(
         self.request_payload['grant_type'] = 'urn:ibm:params:oauth:grant-type:apikey'
         self.request_payload['apikey'] = self.apikey
         self.request_payload['response_type'] = 'cloud_iam'
+
+        self._set_user_agent(_build_user_agent('iam-authenticator'))

ibm_cloud_sdk_core/token_managers/mcsp_token_manager.py

@@ -1,6 +1,6 @@
 # coding: utf-8
 
-# Copyright 2023 IBM All Rights Reserved.
+# Copyright 2023, 2024 IBM All Rights Reserved.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -17,6 +17,7 @@
 import json
 from typing import Dict, Optional
 
+from ..private_helpers import _build_user_agent
 from .jwt_token_manager import JWTTokenManager
 
 
@@ -55,12 +56,21 @@ def __init__(
         self.headers['Accept'] = 'application/json'
         self.proxies = proxies
         super().__init__(url, disable_ssl_verification=disable_ssl_verification, token_name=self.TOKEN_NAME)
+        self._set_user_agent(_build_user_agent('mcsp-authenticator'))
 
     def request_token(self) -> dict:
         """Makes a request for a token."""
+        required_headers = {
+            'User-Agent': self.user_agent,
+        }
+        request_headers = {}
+        if self.headers is not None and isinstance(self.headers, dict):
+            request_headers.update(self.headers)
+        request_headers.update(required_headers)
+
         response = self._request(
             method='POST',
-            headers=self.headers,
+            headers=request_headers,
             url=self.url + self.OPERATION_PATH,
             data=json.dumps({"apikey": self.apikey}),
             proxies=self.proxies,

ibm_cloud_sdk_core/token_managers/token_manager.py

@@ -1,6 +1,6 @@
 # coding: utf-8
 
-# Copyright 2020 IBM All Rights Reserved.
+# Copyright 2020, 2024 IBM All Rights Reserved.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -49,6 +49,7 @@ class TokenManager(ABC):
         lock (Lock): Lock variable to serialize access to refresh/request times
         http_config (dict): A dictionary containing values that control the timeout, proxies, and etc of HTTP requests.
         access_token (str): The latest stored access token
+        user_agent (str): The User-Agent header value to be included in each outbound token request
     """
 
     def __init__(self, url: str, *, disable_ssl_verification: bool = False):
@@ -60,6 +61,7 @@ def __init__(self, url: str, *, disable_ssl_verification: bool = False):
         self.lock = Lock()
         self.http_config = {}
         self.access_token = None
+        self.user_agent = None
 
     def get_token(self) -> str:
         """Get a token to be used for authentication.
@@ -95,6 +97,12 @@ def set_disable_ssl_verification(self, status: bool = False) -> None:
         else:
             raise TypeError('status must be a bool')
 
+    def _set_user_agent(self, user_agent: str = None) -> None:
+        self.user_agent = user_agent
+
+    def _get_user_agent(self) -> str:
+        return self.user_agent
+
     def paced_request_token(self) -> None:
         """
         Paces requests to request_token.

ibm_cloud_sdk_core/token_managers/vpc_instance_token_manager.py

@@ -18,6 +18,7 @@
 import logging
 from typing import Optional
 
+from ..private_helpers import _build_user_agent
 from .jwt_token_manager import JWTTokenManager
 
 
@@ -64,6 +65,7 @@ def __init__(
             url = self.DEFAULT_IMS_ENDPOINT
 
         super().__init__(url, token_name=self.TOKEN_NAME)
+        self._set_user_agent(_build_user_agent('vpc-instance-authenticator'))
 
         self.iam_profile_crn = iam_profile_crn
         self.iam_profile_id = iam_profile_id
@@ -92,6 +94,7 @@ def request_token(self) -> dict:
             'Content-Type': 'application/json',
             'Accept': 'application/json',
             'Authorization': 'Bearer ' + instance_identity_token,
+            'User-Agent': self._get_user_agent(),
         }
 
         logger.debug('Invoking VPC \'create_iam_token\' operation: %s', url)
@@ -138,6 +141,7 @@ def retrieve_instance_identity_token(self) -> str:
             'Content-type': 'application/json',
             'Accept': 'application/json',
             'Metadata-Flavor': 'ibm',
+            'User-Agent': self._get_user_agent(),
         }
 
         request_body = {'expires_in': 300}

test/test_base_service.py

@@ -1,5 +1,21 @@
-# coding=utf-8
+# coding: utf-8
+
+# Copyright 2019, 2024 IBM All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
 # pylint: disable=missing-docstring,protected-access,too-few-public-methods,too-many-lines
+
 import gzip
 import json
 import os
@@ -802,7 +818,7 @@ def test_user_agent_header():
     service = AnyServiceV1('2018-11-20', authenticator=NoAuthAuthenticator())
     user_agent_header = service.user_agent_header
     assert user_agent_header is not None
-    assert user_agent_header['User-Agent'] is not None
+    assert user_agent_header['User-Agent'].startswith('ibm-python-sdk-core-')
 
     responses.add(responses.GET, 'https://gateway.watsonplatform.net/test/api', status=200, body='some text')
     prepped = service.prepare_request('GET', url='', headers={'user-agent': 'my_user_agent'})

test/test_container_token_manager.py

@@ -111,6 +111,9 @@ def test_request_token_auth_default():
     assert len(responses.calls) == 1
     assert responses.calls[0].request.url == iam_url
     assert responses.calls[0].request.headers.get('Authorization') is None
+    assert (
+        responses.calls[0].request.headers.get('User-Agent').startswith('ibm-python-sdk-core/container-authenticator')
+    )
     assert json.loads(responses.calls[0].response.text)['access_token'] == TEST_ACCESS_TOKEN_1
 
 

test/test_cp4d_token_manager.py

@@ -1,3 +1,19 @@
+# coding: utf-8
+
+# Copyright 2019, 2024 IBM All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
 # pylint: disable=missing-docstring
 import json
 import time
@@ -38,6 +54,7 @@ def test_request_token():
 
     assert len(responses.calls) == 1
     assert responses.calls[0].request.url == url + '/v1/authorize'
+    assert responses.calls[0].request.headers.get('User-Agent').startswith('ibm-python-sdk-core/cp4d-authenticator')
     assert token == access_token
 
     token_manager = CP4DTokenManager("username", "password", url + '/v1/authorize')