1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341
|
# ------------------------------------
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
# ------------------------------------
from azure.core.exceptions import ClientAuthenticationError
from azure.identity import (
AuthenticationRequiredError,
AuthenticationRecord,
KnownAuthorities,
CredentialUnavailableError,
)
from azure.identity._internal import InteractiveCredential
from msal import TokenCache
import pytest
try:
from unittest.mock import Mock, patch
except ImportError: # python < 3.3
from mock import Mock, patch # type: ignore
from helpers import build_aad_response, build_id_token, id_token_claims
# fake object for tests which need to exercise request_token but don't care about its return value
REQUEST_TOKEN_RESULT = build_aad_response(
access_token="***",
id_token_claims=id_token_claims(
aud="...",
iss="http://localhost/tenant",
sub="subject",
preferred_username="...",
tenant_id="...",
object_id="...",
),
)
class MockCredential(InteractiveCredential):
"""Test class to drive InteractiveCredential.
Default instances have an empty in-memory cache, and raise rather than send an HTTP request.
"""
def __init__(
self, client_id="...", request_token=None, cache=None, msal_app_factory=None, transport=None, **kwargs
):
self._msal_app_factory = msal_app_factory
self._request_token_impl = request_token or Mock()
transport = transport or Mock(send=Mock(side_effect=Exception("credential shouldn't send a request")))
super(MockCredential, self).__init__(
client_id=client_id, _cache=cache or TokenCache(), transport=transport, **kwargs
)
def _request_token(self, *scopes, **kwargs):
return self._request_token_impl(*scopes, **kwargs)
def _get_app(self):
if self._msal_app_factory:
return self._create_app(self._msal_app_factory)
return super(MockCredential, self)._get_app()
def test_no_scopes():
"""The credential should raise when get_token is called with no scopes"""
request_token = Mock(side_effect=Exception("credential shouldn't begin interactive authentication"))
with pytest.raises(ValueError):
MockCredential(request_token=request_token).get_token()
def test_authentication_record_argument():
"""The credential should initialize its msal.ClientApplication with values from a given record"""
record = AuthenticationRecord("tenant-id", "client-id", "localhost", "object.tenant", "username")
def validate_app_parameters(authority, client_id, **_):
# the 'authority' argument to msal.ClientApplication should be a URL of the form https://authority/tenant
assert authority == "https://{}/{}".format(record.authority, record.tenant_id)
assert client_id == record.client_id
return Mock(get_accounts=Mock(return_value=[]))
app_factory = Mock(wraps=validate_app_parameters)
credential = MockCredential(
authentication_record=record, disable_automatic_authentication=True, msal_app_factory=app_factory,
)
with pytest.raises(AuthenticationRequiredError):
credential.get_token("scope")
assert app_factory.call_count == 1, "credential didn't create an msal application"
def test_tenant_argument_overrides_record():
"""The 'tenant_ic' keyword argument should override a given record's value"""
tenant_id = "some-guid"
authority = "localhost"
record = AuthenticationRecord(tenant_id, "client-id", authority, "object.tenant", "username")
expected_tenant = tenant_id[::-1]
expected_authority = "https://{}/{}".format(authority, expected_tenant)
def validate_authority(authority, **_):
assert authority == expected_authority
return Mock(get_accounts=Mock(return_value=[]))
credential = MockCredential(
authentication_record=record,
tenant_id=expected_tenant,
disable_automatic_authentication=True,
msal_app_factory=validate_authority,
)
with pytest.raises(AuthenticationRequiredError):
credential.get_token("scope")
def test_disable_automatic_authentication():
"""When silent auth fails the credential should raise, if it's configured not to authenticate automatically"""
expected_details = "something went wrong"
record = AuthenticationRecord("tenant-id", "client-id", "localhost", "object.tenant", "username")
msal_app = Mock(
acquire_token_silent_with_error=Mock(return_value={"error_description": expected_details}),
get_accounts=Mock(return_value=[{"home_account_id": record.home_account_id}]),
)
credential = MockCredential(
authentication_record=record,
disable_automatic_authentication=True,
msal_app_factory=lambda *_, **__: msal_app,
request_token=Mock(side_effect=Exception("credential shouldn't begin interactive authentication")),
)
scope = "scope"
with pytest.raises(AuthenticationRequiredError) as ex:
credential.get_token(scope)
# the exception should carry the requested scopes and any error message from AAD
assert ex.value.scopes == (scope,)
assert ex.value.error_details == expected_details
def test_scopes_round_trip():
"""authenticate should accept the value of AuthenticationRequiredError.scopes"""
scope = "scope"
def validate_scopes(*scopes, **_):
assert scopes == (scope,)
return REQUEST_TOKEN_RESULT
request_token = Mock(wraps=validate_scopes)
credential = MockCredential(disable_automatic_authentication=True, request_token=request_token)
with pytest.raises(AuthenticationRequiredError) as ex:
credential.get_token(scope)
credential.authenticate(scopes=ex.value.scopes)
assert request_token.call_count == 1, "validation method wasn't called"
@pytest.mark.parametrize(
"authority,expected_scope",
(
(KnownAuthorities.AZURE_CHINA, "https://management.core.chinacloudapi.cn//.default"),
(KnownAuthorities.AZURE_GERMANY, "https://management.core.cloudapi.de//.default"),
(KnownAuthorities.AZURE_GOVERNMENT, "https://management.core.usgovcloudapi.net//.default"),
(KnownAuthorities.AZURE_PUBLIC_CLOUD, "https://management.core.windows.net//.default"),
),
)
def test_authenticate_default_scopes(authority, expected_scope):
"""when given no scopes, authenticate should default to the ARM scope appropriate for the configured authority"""
def validate_scopes(*scopes):
assert scopes == (expected_scope,)
return REQUEST_TOKEN_RESULT
request_token = Mock(wraps=validate_scopes)
MockCredential(authority=authority, request_token=request_token).authenticate()
assert request_token.call_count == 1
def test_authenticate_unknown_cloud():
"""authenticate should raise when given no scopes in an unknown cloud"""
with pytest.raises(CredentialUnavailableError):
MockCredential(authority="localhost").authenticate()
@pytest.mark.parametrize("option", (True, False))
def test_authenticate_ignores_disable_automatic_authentication(option):
"""authenticate should prompt for authentication regardless of the credential's configuration"""
request_token = Mock(return_value=REQUEST_TOKEN_RESULT)
MockCredential(request_token=request_token, disable_automatic_authentication=option).authenticate()
assert request_token.call_count == 1, "credential didn't begin interactive authentication"
def test_get_token_wraps_exceptions():
"""get_token shouldn't propagate exceptions from MSAL"""
class CustomException(Exception):
pass
expected_message = "something went wrong"
record = AuthenticationRecord("tenant-id", "client-id", "localhost", "object.tenant", "username")
msal_app = Mock(
acquire_token_silent_with_error=Mock(side_effect=CustomException(expected_message)),
get_accounts=Mock(return_value=[{"home_account_id": record.home_account_id}]),
)
credential = MockCredential(msal_app_factory=lambda *_, **__: msal_app, authentication_record=record)
with pytest.raises(ClientAuthenticationError) as ex:
credential.get_token("scope")
assert expected_message in ex.value.message
assert msal_app.acquire_token_silent_with_error.call_count == 1, "credential didn't attempt silent auth"
def test_enable_persistent_cache():
"""the credential should use the persistent cache only when given enable_persistent_cache=True"""
class TestCredential(InteractiveCredential):
def __init__(self, **kwargs):
super(TestCredential, self).__init__(client_id="...", **kwargs)
def _request_token(self, *_, **__):
pass
in_memory_cache = Mock()
persistent_cache = "azure.identity._internal.persistent_cache"
# credential should default to an in memory cache
raise_when_called = Mock(side_effect=Exception("credential shouldn't attempt to load a persistent cache"))
with patch(persistent_cache + "._load_persistent_cache", raise_when_called):
with patch(InteractiveCredential.__module__ + ".msal.TokenCache", lambda: in_memory_cache):
credential = TestCredential()
assert credential._cache is in_memory_cache
# allowing an unencrypted cache doesn't count as opting in to the persistent cache
credential = TestCredential(allow_unencrypted_cache=True)
assert credential._cache is in_memory_cache
# keyword argument opts in to persistent cache
with patch(persistent_cache + ".msal_extensions") as mock_extensions:
TestCredential(enable_persistent_cache=True)
assert mock_extensions.PersistedTokenCache.call_count == 1
# opting in on an unsupported platform raises an exception
with patch(persistent_cache + ".sys.platform", "commodore64"):
with pytest.raises(NotImplementedError):
TestCredential(enable_persistent_cache=True)
with pytest.raises(NotImplementedError):
TestCredential(enable_persistent_cache=True, allow_unencrypted_cache=True)
@patch("azure.identity._internal.persistent_cache.sys.platform", "linux2")
@patch("azure.identity._internal.persistent_cache.msal_extensions")
def test_persistent_cache_linux(mock_extensions):
"""The credential should use an unencrypted cache when encryption is unavailable and the user explicitly opts in.
This test was written when Linux was the only platform on which encryption may not be available.
"""
class TestCredential(InteractiveCredential):
def __init__(self, **kwargs):
super(TestCredential, self).__init__(client_id="...", **kwargs)
def _request_token(self, *_, **__):
pass
# the credential should prefer an encrypted cache even when the user allows an unencrypted one
TestCredential(enable_persistent_cache=True, allow_unencrypted_cache=True)
assert mock_extensions.PersistedTokenCache.called_with(mock_extensions.LibsecretPersistence)
mock_extensions.PersistedTokenCache.reset_mock()
# (when LibsecretPersistence's dependencies aren't available, constructing it raises ImportError)
mock_extensions.LibsecretPersistence = Mock(side_effect=ImportError)
# encryption unavailable, no opt in to unencrypted cache -> credential should raise
with pytest.raises(ValueError):
TestCredential(enable_persistent_cache=True)
TestCredential(enable_persistent_cache=True, allow_unencrypted_cache=True)
assert mock_extensions.PersistedTokenCache.called_with(mock_extensions.FilePersistence)
def test_home_account_id_client_info():
"""when MSAL returns client_info, the credential should decode it to get the home_account_id"""
object_id = "object-id"
home_tenant = "home-tenant-id"
msal_response = build_aad_response(uid=object_id, utid=home_tenant, access_token="***", refresh_token="**")
msal_response["id_token_claims"] = {
"aud": "client-id",
"iss": "https://localhost",
"object_id": object_id,
"tid": home_tenant,
"preferred_username": "me",
"sub": "subject",
}
class TestCredential(InteractiveCredential):
def __init__(self, **kwargs):
super(TestCredential, self).__init__(client_id="...", **kwargs)
def _request_token(self, *_, **__):
return msal_response
record = TestCredential().authenticate()
assert record.home_account_id == "{}.{}".format(object_id, home_tenant)
def test_adfs():
"""the credential should be able to construct an AuthenticationRecord from an ADFS response returned by MSAL"""
authority = "localhost"
subject = "subject"
tenant = "adfs"
username = "username"
msal_response = build_aad_response(access_token="***", refresh_token="**")
msal_response["id_token_claims"] = id_token_claims(
aud="client-id",
iss="https://{}/{}".format(authority, tenant),
sub=subject,
tenant_id=tenant,
object_id="object-id",
upn=username,
)
class TestCredential(InteractiveCredential):
def __init__(self, **kwargs):
super(TestCredential, self).__init__(client_id="...", **kwargs)
def _request_token(self, *_, **__):
return msal_response
record = TestCredential().authenticate()
assert record.authority == authority
assert record.home_account_id == subject
assert record.tenant_id == tenant
assert record.username == username
|