diff --git a/src/azure-cli-core/azure/cli/core/adal_authentication.py b/src/azure-cli-core/azure/cli/core/adal_authentication.py index 3bc28b0d1d6..d267ac4fdce 100644 --- a/src/azure-cli-core/azure/cli/core/adal_authentication.py +++ b/src/azure-cli-core/azure/cli/core/adal_authentication.py @@ -3,8 +3,6 @@ # Licensed under the MIT License. See License.txt in the project root for license information. # -------------------------------------------------------------------------------------------- -import datetime -import time import requests import adal @@ -77,17 +75,56 @@ def get_token(self, *scopes, **kwargs): # pylint:disable=unused-argument _, token, full_token, _ = self._get_token(_try_scopes_to_resource(scopes)) - try: - expires_on = full_token['expiresOn'] - return AccessToken(token, int(datetime.datetime.strptime(expires_on, '%Y-%m-%d %H:%M:%S.%f').timestamp())) - except: # pylint: disable=bare-except - pass # To avoid crashes due to some unexpected token formats - - try: - return AccessToken(token, int(full_token['expiresIn'] + time.time())) - except KeyError: # needed to deal with differing unserialized MSI token payload + # NEVER use expiresIn (expires_in) as the token is cached and expiresIn will be already out-of date + # when being retrieved. + + # User token entry sample: + # { + # "tokenType": "Bearer", + # "expiresOn": "2020-11-13 14:44:42.492318", + # "resource": "https://management.core.windows.net/", + # "userId": "test@azuresdkteam.onmicrosoft.com", + # "accessToken": "eyJ0eXAiOiJKV...", + # "refreshToken": "0.ATcAImuCVN...", + # "_clientId": "04b07795-8ddb-461a-bbee-02f9e1bf7b46", + # "_authority": "https://login.microsoftonline.com/54826b22-38d6-4fb2-bad9-b7b93a3e9c5a", + # "isMRRT": True, + # "expiresIn": 3599 + # } + + # Service Principal token entry sample: + # { + # "tokenType": "Bearer", + # "expiresIn": 3599, + # "expiresOn": "2020-11-12 13:50:47.114324", + # "resource": "https://management.core.windows.net/", + # "accessToken": "eyJ0eXAiOiJKV...", + # "isMRRT": True, + # "_clientId": "22800c35-46c2-4210-b8a7-d8c3ec3b526f", + # "_authority": "https://login.microsoftonline.com/54826b22-38d6-4fb2-bad9-b7b93a3e9c5a" + # } + if 'expiresOn' in full_token: + import datetime + expires_on_timestamp = int(_timestamp( + datetime.datetime.strptime(full_token['expiresOn'], '%Y-%m-%d %H:%M:%S.%f'))) + return AccessToken(token, expires_on_timestamp) + + # Cloud Shell (Managed Identity) token entry sample: + # { + # "access_token": "eyJ0eXAiOiJKV...", + # "refresh_token": "", + # "expires_in": "2106", + # "expires_on": "1605686811", + # "not_before": "1605682911", + # "resource": "https://management.core.windows.net/", + # "token_type": "Bearer" + # } + if 'expires_on' in full_token: return AccessToken(token, int(full_token['expires_on'])) + from azure.cli.core.azclierror import CLIInternalError + raise CLIInternalError("No expiresOn or expires_on is available in the token entry.") + # This method is exposed for msrest. def signed_session(self, session=None): # pylint: disable=arguments-differ logger.debug("AdalAuthentication.signed_session invoked by Track 1 SDK") @@ -118,6 +155,17 @@ def get_token(self, *scopes, **kwargs): # pylint:disable=unused-argument # If available, use resource provided by SDK self.resource = resource self.set_token() + # Managed Identity token entry sample: + # { + # "access_token": "eyJ0eXAiOiJKV...", + # "client_id": "da95e381-d7ab-4fdc-8047-2457909c723b", + # "expires_in": "86386", + # "expires_on": "1605238724", + # "ext_expires_in": "86399", + # "not_before": "1605152024", + # "resource": "https://management.azure.com/", + # "token_type": "Bearer" + # } return AccessToken(self.token['access_token'], int(self.token['expires_on'])) def set_token(self): @@ -183,4 +231,14 @@ def __init__(self, access_token): def get_token(self, *scopes, **kwargs): # pylint:disable=unused-argument # Because get_token can't refresh the access token, always mark the token as unexpired + import time return AccessToken(self.access_token, int(time.time() + 3600)) + + +def _timestamp(dt): + # datetime.datetime can't be patched: + # TypeError: can't set attributes of built-in/extension type 'datetime.datetime' + # So we wrap datetime.datetime.timestamp with this function. + # https://docs.python.org/3/library/unittest.mock-examples.html#partial-mocking + # https://williambert.online/2011/07/how-to-unit-testing-in-django-with-mocking-and-patching/ + return dt.timestamp() diff --git a/src/azure-cli-core/azure/cli/core/tests/test_adal_authentication.py b/src/azure-cli-core/azure/cli/core/tests/test_adal_authentication.py index be660f2e751..2d9bbf84e13 100644 --- a/src/azure-cli-core/azure/cli/core/tests/test_adal_authentication.py +++ b/src/azure-cli-core/azure/cli/core/tests/test_adal_authentication.py @@ -4,8 +4,12 @@ # -------------------------------------------------------------------------------------------- # pylint: disable=line-too-long +import datetime import unittest -from azure.cli.core.adal_authentication import _try_scopes_to_resource +import unittest.mock as mock +from unittest.mock import MagicMock + +from azure.cli.core.adal_authentication import AdalAuthentication, _try_scopes_to_resource class TestUtils(unittest.TestCase): @@ -26,5 +30,58 @@ def test_try_scopes_to_resource(self): self.assertEqual(resource, "https://management.core.chinacloudapi.cn/") +class TestAdalAuthentication(unittest.TestCase): + + def test_get_token(self): + user_full_token = ( + 'Bearer', + 'access_token_user_mock', + { + 'tokenType': 'Bearer', + 'expiresIn': 3599, + 'expiresOn': '2020-11-18 15:35:17.512862', # Local time + 'resource': 'https://management.core.windows.net/', + 'accessToken': 'access_token_user_mock', + 'refreshToken': 'refresh_token_user_mock', + 'oid': '6d97229a-391f-473a-893f-f0608b592d7b', 'userId': 'rolelivetest@azuresdkteam.onmicrosoft.com', + 'isMRRT': True, '_clientId': '04b07795-8ddb-461a-bbee-02f9e1bf7b46', + '_authority': 'https://login.microsoftonline.com/54826b22-38d6-4fb2-bad9-b7b93a3e9c5a' + }) + cloud_shell_full_token = ( + 'Bearer', + 'access_token_cloud_shell_mock', + { + 'access_token': 'access_token_cloud_shell_mock', + 'refresh_token': '', + 'expires_in': '2732', + 'expires_on': '1605683384', + 'not_before': '1605679484', + 'resource': 'https://management.core.windows.net/', + 'token_type': 'Bearer' + }) + token_retriever = MagicMock() + cred = AdalAuthentication(token_retriever) + + def utc_to_timestamp(dt): + # Obtain the POSIX timestamp from a naive datetime instance representing UTC time + # https://docs.python.org/3/library/datetime.html#datetime.datetime.timestamp + return dt.replace(tzinfo=datetime.timezone.utc).timestamp() + + # Test expiresOn is used and converted to epoch time + # Force expiresOn to be treated as UTC to make the test pass on both local machine (such as UTC+8) + # and CI (UTC). + with mock.patch("azure.cli.core.adal_authentication._timestamp", utc_to_timestamp): + token_retriever.return_value = user_full_token + access_token = cred.get_token("https://management.core.windows.net//.default") + self.assertEqual(access_token.token, "access_token_user_mock") + self.assertEqual(access_token.expires_on, 1605713717) + + # Test expires_on is used as epoch directly + token_retriever.return_value = cloud_shell_full_token + access_token = cred.get_token("https://management.core.windows.net//.default") + self.assertEqual(access_token.token, "access_token_cloud_shell_mock") + self.assertEqual(access_token.expires_on, 1605683384) + + if __name__ == '__main__': unittest.main()