File: test_managed_identity.py

package info (click to toggle)
python-azure 20250603%2Bgit-1
links: PTS, VCS
area: main
in suites: forky, sid, trixie
size: 851,724 kB
sloc: python: 7,362,925; ansic: 804; javascript: 287; makefile: 195; sh: 145; xml: 109
file content (1094 lines) | stat: -rw-r--r-- 41,092 bytes
# ------------------------------------
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
# ------------------------------------
from itertools import product
import time
import logging
from unittest import mock

from azure.core.exceptions import ClientAuthenticationError
from azure.identity import ManagedIdentityCredential, CredentialUnavailableError
from azure.identity._constants import EnvironmentVariables
from azure.identity._credentials.imds import IMDS_AUTHORITY, IMDS_TOKEN_PATH
from azure.identity._credentials.managed_identity import validate_identity_config
from azure.identity._internal.user_agent import USER_AGENT
from azure.identity._internal import within_credential_chain
import pytest

from helpers import build_aad_response, validating_transport, mock_response, Request, GET_TOKEN_METHODS

MANAGED_IDENTITY_ENVIRON = "azure.identity._credentials.managed_identity.os.environ"
ALL_ENVIRONMENTS = (
    {EnvironmentVariables.IDENTITY_ENDPOINT: "...", EnvironmentVariables.IDENTITY_HEADER: "..."},  # App Service
    {EnvironmentVariables.MSI_ENDPOINT: "..."},  # Cloud Shell
    {  # Service Fabric
        EnvironmentVariables.IDENTITY_ENDPOINT: "...",
        EnvironmentVariables.IDENTITY_HEADER: "...",
        EnvironmentVariables.IDENTITY_SERVER_THUMBPRINT: "...",
    },
    {EnvironmentVariables.IDENTITY_ENDPOINT: "...", EnvironmentVariables.IMDS_ENDPOINT: "..."},  # Arc
    {  # token exchange
        EnvironmentVariables.AZURE_AUTHORITY_HOST: "https://localhost",
        EnvironmentVariables.AZURE_CLIENT_ID: "...",
        EnvironmentVariables.AZURE_TENANT_ID: "...",
        EnvironmentVariables.AZURE_FEDERATED_TOKEN_FILE: __file__,
    },
    {},  # IMDS
    {EnvironmentVariables.MSI_ENDPOINT: "...", EnvironmentVariables.MSI_SECRET: "..."},  # Azure ML
)


@pytest.mark.parametrize("environ", ALL_ENVIRONMENTS)
def test_close(environ):
    transport = mock.MagicMock()
    with mock.patch.dict("os.environ", environ, clear=True):
        credential = ManagedIdentityCredential(transport=transport)
    assert transport.__exit__.call_count == 0

    credential.close()
    assert transport.__exit__.call_count == 1


@pytest.mark.parametrize("environ", ALL_ENVIRONMENTS)
def test_context_manager(environ):
    transport = mock.MagicMock()
    with mock.patch.dict("os.environ", environ, clear=True):
        credential = ManagedIdentityCredential(transport=transport)

    with credential:
        assert transport.__enter__.call_count == 1
        assert transport.__exit__.call_count == 0

    assert transport.__enter__.call_count == 1
    assert transport.__exit__.call_count == 1


def test_close_incomplete_configuration():
    ManagedIdentityCredential().close()


def test_context_manager_incomplete_configuration():
    with ManagedIdentityCredential():
        pass


@pytest.mark.parametrize("environ,get_token_method", product(ALL_ENVIRONMENTS, GET_TOKEN_METHODS))
def test_custom_hooks(environ, get_token_method):
    """The credential's pipeline should include azure-core's CustomHookPolicy"""

    scope = "scope"
    expected_token = "***"
    request_hook = mock.Mock()
    response_hook = mock.Mock()
    now = int(time.time())
    expected_response = mock_response(
        json_payload={
            "access_token": expected_token,
            "expires_in": 3600,
            "expires_on": now + 3600,
            "ext_expires_in": 3600,
            "not_before": now,
            "resource": scope,
            "token_type": "Bearer",
        }
    )
    transport = validating_transport(requests=[Request()] * 2, responses=[expected_response] * 2)

    with mock.patch.dict(MANAGED_IDENTITY_ENVIRON, environ, clear=True):
        credential = ManagedIdentityCredential(
            transport=transport, raw_request_hook=request_hook, raw_response_hook=response_hook
        )
    getattr(credential, get_token_method)(scope)

    assert request_hook.call_count == 1
    assert response_hook.call_count == 1
    args, kwargs = response_hook.call_args
    pipeline_response = args[0]
    assert pipeline_response.http_response == expected_response


@pytest.mark.parametrize("environ,get_token_method", product(ALL_ENVIRONMENTS, GET_TOKEN_METHODS))
def test_tenant_id(environ, get_token_method):
    scope = "scope"
    expected_token = "***"
    request_hook = mock.Mock()
    response_hook = mock.Mock()
    now = int(time.time())
    expected_response = mock_response(
        json_payload={
            "access_token": expected_token,
            "expires_in": 3600,
            "expires_on": now + 3600,
            "ext_expires_in": 3600,
            "not_before": now,
            "resource": scope,
            "token_type": "Bearer",
        }
    )
    transport = validating_transport(requests=[Request()] * 2, responses=[expected_response] * 2)

    with mock.patch.dict(MANAGED_IDENTITY_ENVIRON, environ, clear=True):
        credential = ManagedIdentityCredential(
            transport=transport, raw_request_hook=request_hook, raw_response_hook=response_hook
        )
    getattr(credential, get_token_method)(scope)

    assert request_hook.call_count == 1
    assert response_hook.call_count == 1
    args, kwargs = response_hook.call_args
    pipeline_response = args[0]
    assert pipeline_response.http_response == expected_response


@pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
def test_cloud_shell(get_token_method):
    """Cloud Shell environment: only MSI_ENDPOINT set"""

    access_token = "****"
    expires_on = 42
    expected_token = access_token
    endpoint = "http://localhost:42/token"
    scope = "scope"
    transport = validating_transport(
        requests=[
            Request(
                base_url=endpoint,
                method="POST",
                required_headers={"Metadata": "true", "User-Agent": USER_AGENT},
                required_data={"resource": scope},
            )
        ],
        responses=[
            mock_response(
                json_payload={
                    "access_token": access_token,
                    "expires_in": 0,
                    "expires_on": expires_on,
                    "not_before": int(time.time()),
                    "resource": scope,
                    "token_type": "Bearer",
                }
            )
        ],
    )

    with mock.patch("os.environ", {EnvironmentVariables.MSI_ENDPOINT: endpoint}):
        token = getattr(ManagedIdentityCredential(transport=transport), get_token_method)(scope)
        assert token.token == expected_token
        assert abs(token.expires_on - expires_on) <= 1


@pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
def test_cloud_shell_tenant_id(get_token_method):
    access_token = "****"
    expires_on = 42
    expected_token = access_token
    endpoint = "http://localhost:42/token"
    scope = "scope"
    transport = validating_transport(
        requests=[
            Request(
                base_url=endpoint,
                method="POST",
                required_headers={"Metadata": "true", "User-Agent": USER_AGENT},
                required_data={"resource": scope},
            )
        ],
        responses=[
            mock_response(
                json_payload={
                    "access_token": access_token,
                    "expires_in": 0,
                    "expires_on": expires_on,
                    "not_before": int(time.time()),
                    "resource": scope,
                    "token_type": "Bearer",
                }
            )
        ],
    )

    with mock.patch("os.environ", {EnvironmentVariables.MSI_ENDPOINT: endpoint}):
        kwargs = {"tenant_id": "tenant_id"}
        if get_token_method == "get_token_info":
            kwargs = {"options": kwargs}
        token = getattr(ManagedIdentityCredential(transport=transport), get_token_method)(scope, **kwargs)
        assert token.token == expected_token
        assert abs(token.expires_on - expires_on) <= 1


@pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
def test_azure_ml(get_token_method):
    """Azure ML: MSI_ENDPOINT, MSI_SECRET set (like App Service 2017-09-01 but with a different response format)"""

    expected_token = "****"
    expires_on = int(time.time()) + 3600
    url = "http://localhost:42/token"
    secret = "expected-secret"
    scope = "scope"
    client_id = "client"

    transport = validating_transport(
        requests=[
            Request(
                url,
                method="GET",
                required_headers={"secret": secret, "User-Agent": USER_AGENT},
                required_params={"api-version": "2017-09-01", "resource": scope},
            ),
            Request(
                url,
                method="GET",
                required_headers={"secret": secret, "User-Agent": USER_AGENT},
                required_params={"api-version": "2017-09-01", "resource": scope, "clientid": client_id},
            ),
        ],
        responses=[
            mock_response(
                json_payload={
                    "access_token": expected_token,
                    "expires_in": 3600,
                    "expires_on": expires_on,
                    "resource": scope,
                    "token_type": "Bearer",
                }
            )
        ]
        * 2,
    )

    with mock.patch.dict(
        MANAGED_IDENTITY_ENVIRON,
        {EnvironmentVariables.MSI_ENDPOINT: url, EnvironmentVariables.MSI_SECRET: secret},
        clear=True,
    ):
        token = getattr(ManagedIdentityCredential(transport=transport), get_token_method)(scope)
        assert token.token == expected_token
        assert abs(token.expires_on - expires_on) <= 1

        token = getattr(ManagedIdentityCredential(transport=transport, client_id=client_id), get_token_method)(scope)
        assert token.token == expected_token
        assert abs(token.expires_on - expires_on) <= 1


@pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
def test_azure_ml_tenant_id(get_token_method):
    expected_token = "****"
    expires_on = int(time.time()) + 3600
    url = "http://localhost:42/token"
    secret = "expected-secret"
    scope = "scope"
    client_id = "client"

    transport = validating_transport(
        requests=[
            Request(
                url,
                method="GET",
                required_headers={"secret": secret, "User-Agent": USER_AGENT},
                required_params={"api-version": "2017-09-01", "resource": scope},
            ),
            Request(
                url,
                method="GET",
                required_headers={"secret": secret, "User-Agent": USER_AGENT},
                required_params={"api-version": "2017-09-01", "resource": scope, "clientid": client_id},
            ),
        ],
        responses=[
            mock_response(
                json_payload={
                    "access_token": expected_token,
                    "expires_in": 3600,
                    "expires_on": expires_on,
                    "resource": scope,
                    "token_type": "Bearer",
                }
            )
        ]
        * 2,
    )

    with mock.patch.dict(
        MANAGED_IDENTITY_ENVIRON,
        {EnvironmentVariables.MSI_ENDPOINT: url, EnvironmentVariables.MSI_SECRET: secret},
        clear=True,
    ):
        kwargs = {"tenant_id": "tenant_id"}
        if get_token_method == "get_token_info":
            kwargs = {"options": kwargs}
        token = getattr(ManagedIdentityCredential(transport=transport), get_token_method)(scope, **kwargs)
        assert token.token == expected_token
        assert abs(token.expires_on - expires_on) <= 1


@pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
def test_cloud_shell_identity_config(get_token_method):
    """Cloud Shell environment: only MSI_ENDPOINT set"""

    expected_token = "****"
    expires_on = 42
    endpoint = "http://localhost:42/token"
    scope = "scope"
    param_name, param_value = "foo", "bar"

    transport = validating_transport(
        requests=[
            Request(
                base_url=endpoint,
                method="POST",
                required_headers={"Metadata": "true", "User-Agent": USER_AGENT},
                required_data={"resource": scope},
            ),
            Request(
                base_url=endpoint,
                method="POST",
                required_headers={"Metadata": "true", "User-Agent": USER_AGENT},
                required_data={"resource": scope, param_name: param_value},
            ),
        ],
        responses=[
            mock_response(
                json_payload={
                    "access_token": expected_token,
                    "expires_in": 0,
                    "expires_on": expires_on,
                    "not_before": int(time.time()),
                    "resource": scope,
                    "token_type": "Bearer",
                }
            )
        ]
        * 2,
    )

    with mock.patch.dict(MANAGED_IDENTITY_ENVIRON, {EnvironmentVariables.MSI_ENDPOINT: endpoint}, clear=True):
        token = getattr(ManagedIdentityCredential(transport=transport), get_token_method)(scope)
        assert token.token == expected_token
        assert abs(token.expires_on - expires_on) <= 1

        credential = ManagedIdentityCredential(transport=transport, identity_config={param_name: param_value})
        token = getattr(credential, get_token_method)(scope)
        assert token.token == expected_token
        assert abs(token.expires_on - expires_on) <= 1


@pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
def test_prefers_app_service_2019_08_01(get_token_method):
    """When the environment is configured for both App Service versions, the credential should prefer the most recent"""

    access_token = "****"
    expires_on = 42
    endpoint = "http://localhost:42/token"
    secret = "expected-secret"
    scope = "scope"
    transport = validating_transport(
        requests=[
            Request(
                base_url=endpoint,
                method="GET",
                required_headers={"X-IDENTITY-HEADER": secret, "User-Agent": USER_AGENT},
                required_params={"api-version": "2019-08-01", "resource": scope},
            )
        ],
        responses=[
            mock_response(
                json_payload={
                    "access_token": access_token,
                    "expires_on": str(expires_on),
                    "resource": scope,
                    "token_type": "Bearer",
                }
            )
        ],
    )

    environ = {
        EnvironmentVariables.IDENTITY_ENDPOINT: endpoint,
        EnvironmentVariables.IDENTITY_HEADER: secret,
        EnvironmentVariables.MSI_ENDPOINT: endpoint,
        EnvironmentVariables.MSI_SECRET: secret,
    }
    with mock.patch.dict("os.environ", environ, clear=True):
        token = getattr(ManagedIdentityCredential(transport=transport), get_token_method)(scope)
    assert token.token == access_token
    assert abs(token.expires_on - expires_on) <= 1


@pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
def test_app_service_2019_08_01(get_token_method):
    """App Service 2019-08-01: IDENTITY_ENDPOINT, IDENTITY_HEADER set"""

    access_token = "****"
    expires_on = 42
    endpoint = "http://localhost:42/token"
    new_endpoint = "http://localhost:42/new-token"
    secret = "expected-secret"
    new_secret = "new-expected-secret"
    scope = "scope"

    def send(request, **kwargs):
        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
        assert "claims" not in kwargs
        assert "tenant_id" not in kwargs
        assert request.url.startswith(new_endpoint)
        assert request.method == "GET"
        assert request.headers["X-IDENTITY-HEADER"] == new_secret
        assert request.headers["User-Agent"] == USER_AGENT
        assert request.query["api-version"] == "2019-08-01"
        assert request.query["resource"] == scope

        return mock_response(
            json_payload={
                "access_token": access_token,
                "expires_on": str(expires_on),
                "resource": scope,
                "token_type": "Bearer",
            }
        )

    # when configuration for both API versions is present, the credential should prefer the most recent
    with mock.patch.dict(
        MANAGED_IDENTITY_ENVIRON,
        {
            EnvironmentVariables.IDENTITY_ENDPOINT: new_endpoint,
            EnvironmentVariables.IDENTITY_HEADER: new_secret,
            EnvironmentVariables.MSI_ENDPOINT: endpoint,
            EnvironmentVariables.MSI_SECRET: secret,
        },
        clear=True,
    ):
        token = getattr(ManagedIdentityCredential(transport=mock.Mock(send=send)), get_token_method)(scope)
        assert token.token == access_token
        assert abs(token.expires_on - expires_on) <= 1


@pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
def test_app_service_2019_08_01_tenant_id(get_token_method):
    """App Service 2019-08-01: IDENTITY_ENDPOINT, IDENTITY_HEADER set"""

    access_token = "****"
    expires_on = 42
    endpoint = "http://localhost:42/token"
    new_endpoint = "http://localhost:42/new-token"
    secret = "expected-secret"
    new_secret = "new-expected-secret"
    scope = "scope"

    def send(request, **kwargs):
        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
        assert "claims" not in kwargs
        assert "tenant_id" not in kwargs
        assert request.url.startswith(new_endpoint)
        assert request.method == "GET"
        assert request.headers["X-IDENTITY-HEADER"] == new_secret
        assert request.headers["User-Agent"] == USER_AGENT
        assert request.query["api-version"] == "2019-08-01"
        assert request.query["resource"] == scope

        return mock_response(
            json_payload={
                "access_token": access_token,
                "expires_on": str(expires_on),
                "resource": scope,
                "token_type": "Bearer",
            }
        )

    # when configuration for both API versions is present, the credential should prefer the most recent
    with mock.patch.dict(
        MANAGED_IDENTITY_ENVIRON,
        {
            EnvironmentVariables.IDENTITY_ENDPOINT: new_endpoint,
            EnvironmentVariables.IDENTITY_HEADER: new_secret,
            EnvironmentVariables.MSI_ENDPOINT: endpoint,
            EnvironmentVariables.MSI_SECRET: secret,
        },
        clear=True,
    ):
        kwargs = {"tenant_id": "tenant_id"}
        if get_token_method == "get_token_info":
            kwargs = {"options": kwargs}
        token = getattr(ManagedIdentityCredential(transport=mock.Mock(send=send)), get_token_method)(scope, **kwargs)
        assert token.token == access_token
        assert abs(token.expires_on - expires_on) <= 1


@pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
def test_app_service_user_assigned_identity(get_token_method):
    """App Service 2019-08-01: IDENTITY_ENDPOINT, IDENTITY_HEADER set"""

    expected_token = "****"
    expires_on = 42
    client_id = "some-guid"
    endpoint = "http://localhost:42/token"
    secret = "expected-secret"
    scope = "scope"

    transport = validating_transport(
        requests=[
            Request(
                base_url=endpoint,
                method="GET",
                required_headers={"X-IDENTITY-HEADER": secret, "User-Agent": USER_AGENT},
                required_params={"api-version": "2019-08-01", "client_id": client_id, "resource": scope},
            ),
            Request(
                base_url=endpoint,
                method="GET",
                required_headers={"X-IDENTITY-HEADER": secret, "User-Agent": USER_AGENT},
                required_params={
                    "api-version": "2019-08-01",
                    "client_id": client_id,
                    "resource": scope,
                },
            ),
        ],
        responses=[
            mock_response(
                json_payload={
                    "access_token": expected_token,
                    "expires_on": expires_on,
                    "resource": scope,
                    "token_type": "Bearer",
                }
            )
        ]
        * 2,
    )

    with mock.patch.dict(
        MANAGED_IDENTITY_ENVIRON,
        {EnvironmentVariables.IDENTITY_ENDPOINT: endpoint, EnvironmentVariables.IDENTITY_HEADER: secret},
        clear=True,
    ):
        token = getattr(ManagedIdentityCredential(client_id=client_id, transport=transport), get_token_method)(scope)
        assert token.token == expected_token
        assert abs(token.expires_on - expires_on) <= 1

        credential = ManagedIdentityCredential(client_id=client_id, transport=transport)
        token = getattr(credential, get_token_method)(scope)
        assert token.token == expected_token
        assert abs(token.expires_on - expires_on) <= 1


@pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
def test_imds(get_token_method):
    access_token = "****"
    expires_on = 42
    expected_token = access_token
    scope = "scope"
    transport = validating_transport(
        requests=[
            Request(
                base_url=IMDS_AUTHORITY + IMDS_TOKEN_PATH,
                method="GET",
                required_headers={"Metadata": "true", "User-Agent": USER_AGENT},
                required_params={"api-version": "2018-02-01", "resource": scope},
            ),
        ],
        responses=[
            mock_response(
                json_payload={
                    "access_token": access_token,
                    "expires_in": 42,
                    "expires_on": expires_on,
                    "ext_expires_in": 42,
                    "not_before": int(time.time()),
                    "resource": scope,
                    "token_type": "Bearer",
                }
            ),
        ],
    )

    # ensure e.g. $MSI_ENDPOINT isn't set, so we get ImdsCredential
    with mock.patch.dict("os.environ", clear=True):
        token = getattr(ManagedIdentityCredential(transport=transport), get_token_method)(scope)
    assert token.token == expected_token


@pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
def test_imds_tenant_id(get_token_method):
    access_token = "****"
    expires_on = 42
    expected_token = access_token
    scope = "scope"
    transport = validating_transport(
        requests=[
            Request(
                base_url=IMDS_AUTHORITY + IMDS_TOKEN_PATH,
                method="GET",
                required_headers={"Metadata": "true", "User-Agent": USER_AGENT},
                required_params={"api-version": "2018-02-01", "resource": scope},
            ),
        ],
        responses=[
            mock_response(
                json_payload={
                    "access_token": access_token,
                    "expires_in": 42,
                    "expires_on": expires_on,
                    "ext_expires_in": 42,
                    "not_before": int(time.time()),
                    "resource": scope,
                    "token_type": "Bearer",
                }
            ),
        ],
    )

    # ensure e.g. $MSI_ENDPOINT isn't set, so we get ImdsCredential
    with mock.patch.dict("os.environ", clear=True):
        kwargs = {"tenant_id": "tenant_id"}
        if get_token_method == "get_token_info":
            kwargs = {"options": kwargs}
        token = getattr(ManagedIdentityCredential(transport=transport), get_token_method)(scope, **kwargs)
    assert token.token == expected_token


@pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
def test_imds_text_response(get_token_method):
    within_credential_chain.set(True)
    response = mock.Mock(
        text=lambda encoding=None: b"{This is a text response}",
        headers={"content-type": "text/html; charset=UTF-8"},
        content_type="text/html; charset=UTF-8",
        status_code=200,
    )
    mock_send = mock.Mock(return_value=response)
    credential = ManagedIdentityCredential(transport=mock.Mock(send=mock_send))
    with pytest.raises(CredentialUnavailableError):
        token = getattr(credential, get_token_method)("scope")
    within_credential_chain.set(False)


@pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
def test_client_id_none(get_token_method):
    """the credential should ignore client_id=None"""

    expected_access_token = "****"
    scope = "scope"

    def send(request, **kwargs):
        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
        assert "claims" not in kwargs
        assert "tenant_id" not in kwargs
        assert "client_id" not in request.query
        if request.data:
            assert "client_id" not in request.body  # Cloud Shell
        return mock_response(
            json_payload=(build_aad_response(access_token=expected_access_token, expires_on="42", resource=scope))
        )

    # IMDS
    credential = ManagedIdentityCredential(client_id=None, transport=mock.Mock(send=send))
    token = getattr(credential, get_token_method)(scope)
    assert token.token == expected_access_token

    # Cloud Shell
    with mock.patch.dict(
        MANAGED_IDENTITY_ENVIRON, {EnvironmentVariables.MSI_ENDPOINT: "https://localhost"}, clear=True
    ):
        credential = ManagedIdentityCredential(client_id=None, transport=mock.Mock(send=send))
        token = getattr(credential, get_token_method)(scope)
    assert token.token == expected_access_token


@pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
def test_imds_user_assigned_identity(get_token_method):
    access_token = "****"
    expires_on = 42
    expected_token = access_token
    endpoint = IMDS_AUTHORITY + IMDS_TOKEN_PATH
    scope = "scope"
    client_id = "some-guid"
    transport = validating_transport(
        requests=[
            Request(
                base_url=endpoint,
                method="GET",
                required_headers={"Metadata": "true", "User-Agent": USER_AGENT},
                required_params={"api-version": "2018-02-01", "client_id": client_id, "resource": scope},
            ),
        ],
        responses=[
            mock_response(
                json_payload={
                    "access_token": access_token,
                    "client_id": client_id,
                    "expires_in": 42,
                    "expires_on": expires_on,
                    "ext_expires_in": 42,
                    "not_before": int(time.time()),
                    "resource": scope,
                    "token_type": "Bearer",
                }
            ),
        ],
    )

    # ensure e.g. $MSI_ENDPOINT isn't set, so we get ImdsCredential
    with mock.patch.dict("os.environ", clear=True):
        token = getattr(ManagedIdentityCredential(client_id=client_id, transport=transport), get_token_method)(scope)
    assert token.token == expected_token


@pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
def test_service_fabric(get_token_method):
    """Service Fabric 2019-07-01-preview"""
    access_token = "****"
    expires_on = 42
    endpoint = "http://localhost:42/token"
    secret = "expected-secret"
    thumbprint = "SHA1HEX"
    scope = "scope"

    def send(request, **kwargs):
        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
        assert "claims" not in kwargs
        assert "tenant_id" not in kwargs
        assert request.url.startswith(endpoint)
        assert request.method == "GET"
        assert request.headers["Secret"] == secret
        assert request.query["api-version"] == "2019-07-01-preview"
        assert request.query["resource"] == scope

        return mock_response(
            json_payload={
                "access_token": access_token,
                "expires_on": str(expires_on),
                "resource": scope,
                "token_type": "Bearer",
            }
        )

    with mock.patch(
        "os.environ",
        {
            EnvironmentVariables.IDENTITY_ENDPOINT: endpoint,
            EnvironmentVariables.IDENTITY_HEADER: secret,
            EnvironmentVariables.IDENTITY_SERVER_THUMBPRINT: thumbprint,
        },
    ):
        token = getattr(ManagedIdentityCredential(transport=mock.Mock(send=send)), get_token_method)(scope)
        assert token.token == access_token
        assert abs(token.expires_on - expires_on) <= 1


@pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
def test_service_fabric_tenant_id(get_token_method):
    access_token = "****"
    expires_on = 42
    endpoint = "http://localhost:42/token"
    secret = "expected-secret"
    thumbprint = "SHA1HEX"
    scope = "scope"

    def send(request, **kwargs):
        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
        assert "claims" not in kwargs
        assert "tenant_id" not in kwargs
        assert request.url.startswith(endpoint)
        assert request.method == "GET"
        assert request.headers["Secret"] == secret
        assert request.query["api-version"] == "2019-07-01-preview"
        assert request.query["resource"] == scope

        return mock_response(
            json_payload={
                "access_token": access_token,
                "expires_on": str(expires_on),
                "resource": scope,
                "token_type": "Bearer",
            }
        )

    with mock.patch(
        "os.environ",
        {
            EnvironmentVariables.IDENTITY_ENDPOINT: endpoint,
            EnvironmentVariables.IDENTITY_HEADER: secret,
            EnvironmentVariables.IDENTITY_SERVER_THUMBPRINT: thumbprint,
        },
    ):
        kwargs = {"tenant_id": "tenant_id"}
        if get_token_method == "get_token_info":
            kwargs = {"options": kwargs}
        token = getattr(ManagedIdentityCredential(transport=mock.Mock(send=send)), get_token_method)(scope, **kwargs)
        assert token.token == access_token
        assert abs(token.expires_on - expires_on) <= 1


@pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
def test_service_fabric_with_client_id_error(get_token_method):
    """ManagedIdentityCredential should raise an error if a user identity is provided."""
    endpoint = "http://localhost:42"
    with mock.patch(
        "os.environ",
        {
            EnvironmentVariables.IDENTITY_ENDPOINT: endpoint,
            EnvironmentVariables.IDENTITY_HEADER: "secret",
            EnvironmentVariables.IDENTITY_SERVER_THUMBPRINT: "thumbprint",
        },
    ):
        cred = ManagedIdentityCredential(client_id="client_id")
        with pytest.raises(ClientAuthenticationError):
            getattr(cred, get_token_method)("scope")

        cred = ManagedIdentityCredential(identity_config={"resource_id": "resource_id"})
        with pytest.raises(ClientAuthenticationError):
            getattr(cred, get_token_method)("scope")


@pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
def test_token_exchange(tmpdir, get_token_method):
    exchange_token = "exchange-token"
    token_file = tmpdir.join("token")
    token_file.write(exchange_token)
    access_token = "***"
    authority = "https://localhost"
    default_client_id = "default_client_id"
    tenant = "tenant_id"
    scope = "scope"

    success_response = mock_response(
        json_payload={
            "access_token": access_token,
            "expires_in": 3600,
            "ext_expires_in": 3600,
            "expires_on": int(time.time()) + 3600,
            "not_before": int(time.time()),
            "resource": scope,
            "token_type": "Bearer",
        }
    )
    transport = validating_transport(
        requests=[
            Request(
                base_url=authority,
                method="POST",
                required_data={
                    "client_assertion": exchange_token,
                    "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
                    "client_id": default_client_id,
                    "grant_type": "client_credentials",
                    "scope": scope,
                },
            )
        ],
        responses=[success_response],
    )

    mock_environ = {
        EnvironmentVariables.AZURE_AUTHORITY_HOST: authority,
        EnvironmentVariables.AZURE_CLIENT_ID: default_client_id,
        EnvironmentVariables.AZURE_TENANT_ID: tenant,
        EnvironmentVariables.AZURE_FEDERATED_TOKEN_FILE: token_file.strpath,
    }
    # credential should default to AZURE_CLIENT_ID
    with mock.patch.dict("os.environ", mock_environ, clear=True):
        credential = ManagedIdentityCredential(transport=transport)
        token = getattr(credential, get_token_method)(scope)
        assert token.token == access_token

    # client_id kwarg should override AZURE_CLIENT_ID
    nondefault_client_id = "non" + default_client_id
    transport = validating_transport(
        requests=[
            Request(
                base_url=authority,
                method="POST",
                required_data={
                    "client_assertion": exchange_token,
                    "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
                    "client_id": nondefault_client_id,
                    "grant_type": "client_credentials",
                    "scope": scope,
                },
            )
        ],
        responses=[success_response],
    )

    with mock.patch.dict("os.environ", mock_environ, clear=True):
        credential = ManagedIdentityCredential(client_id=nondefault_client_id, transport=transport)
        token = getattr(credential, get_token_method)(scope)
    assert token.token == access_token

    # AZURE_CLIENT_ID may not have a value, in which case client_id is required
    transport = validating_transport(
        requests=[
            Request(
                base_url=authority,
                method="POST",
                required_data={
                    "client_assertion": exchange_token,
                    "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
                    "client_id": nondefault_client_id,
                    "grant_type": "client_credentials",
                    "scope": scope,
                },
            )
        ],
        responses=[success_response],
    )

    with mock.patch.dict(
        "os.environ",
        {
            EnvironmentVariables.AZURE_AUTHORITY_HOST: authority,
            EnvironmentVariables.AZURE_TENANT_ID: tenant,
            EnvironmentVariables.AZURE_FEDERATED_TOKEN_FILE: token_file.strpath,
        },
        clear=True,
    ):
        with pytest.raises(ValueError):
            ManagedIdentityCredential()

        credential = ManagedIdentityCredential(client_id=nondefault_client_id, transport=transport)
        token = getattr(credential, get_token_method)(scope)
    assert token.token == access_token


@pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
def test_token_exchange_tenant_id(tmpdir, get_token_method):
    exchange_token = "exchange-token"
    token_file = tmpdir.join("token")
    token_file.write(exchange_token)
    access_token = "***"
    authority = "https://localhost"
    default_client_id = "default_client_id"
    tenant = "tenant_id"
    scope = "scope"

    success_response = mock_response(
        json_payload={
            "access_token": access_token,
            "expires_in": 3600,
            "ext_expires_in": 3600,
            "expires_on": int(time.time()) + 3600,
            "not_before": int(time.time()),
            "resource": scope,
            "token_type": "Bearer",
        }
    )
    transport = validating_transport(
        requests=[
            Request(
                base_url=authority,
                method="POST",
                required_data={
                    "client_assertion": exchange_token,
                    "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
                    "client_id": default_client_id,
                    "grant_type": "client_credentials",
                    "scope": scope,
                },
            )
        ],
        responses=[success_response],
    )

    mock_environ = {
        EnvironmentVariables.AZURE_AUTHORITY_HOST: authority,
        EnvironmentVariables.AZURE_CLIENT_ID: default_client_id,
        EnvironmentVariables.AZURE_TENANT_ID: tenant,
        EnvironmentVariables.AZURE_FEDERATED_TOKEN_FILE: token_file.strpath,
    }
    with mock.patch.dict("os.environ", mock_environ, clear=True):
        credential = ManagedIdentityCredential(transport=transport)
        kwargs = {"tenant_id": "tenant_id"}
        if get_token_method == "get_token_info":
            kwargs = {"options": kwargs}
        token = getattr(credential, get_token_method)(scope, **kwargs)
        assert token.token == access_token


def test_validate_identity_config():
    ManagedIdentityCredential()
    ManagedIdentityCredential(client_id="foo")
    ManagedIdentityCredential(identity_config={"foo": "bar"})
    ManagedIdentityCredential(identity_config={"client_id": "foo"})
    ManagedIdentityCredential(identity_config={"object_id": "foo"})
    ManagedIdentityCredential(identity_config={"resource_id": "foo"})
    ManagedIdentityCredential(identity_config={"foo": "bar"}, client_id="foo")

    with pytest.raises(ValueError):
        ManagedIdentityCredential(identity_config={"client_id": "foo"}, client_id="foo")
    with pytest.raises(ValueError):
        ManagedIdentityCredential(identity_config={"object_id": "bar"}, client_id="bar")
    with pytest.raises(ValueError):
        ManagedIdentityCredential(identity_config={"resource_id": "bar"}, client_id="bar")
    with pytest.raises(ValueError):
        ManagedIdentityCredential(identity_config={"object_id": "bar", "resource_id": "foo"})
    with pytest.raises(ValueError):
        ManagedIdentityCredential(identity_config={"object_id": "bar", "client_id": "foo"})


def test_validate_identity_config_output():
    output = validate_identity_config(None, {"client_id": "foo"})
    assert output == ("client_id", "foo")

    output = validate_identity_config("foo", None)
    assert output == ("client_id", "foo")

    output = validate_identity_config(None, {"object_id": "bar"})
    assert output == ("object_id", "bar")

    output = validate_identity_config(None, {"resource_id": "biz"})
    assert output == ("resource_id", "biz")


def test_validate_cloud_shell_credential():
    with mock.patch.dict(
        MANAGED_IDENTITY_ENVIRON, {EnvironmentVariables.MSI_ENDPOINT: "https://localhost"}, clear=True
    ):
        ManagedIdentityCredential()
        with pytest.raises(ValueError):
            ManagedIdentityCredential(client_id="foo")
        with pytest.raises(ValueError):
            ManagedIdentityCredential(identity_config={"client_id": "foo"})
        with pytest.raises(ValueError):
            ManagedIdentityCredential(identity_config={"object_id": "foo"})
        with pytest.raises(ValueError):
            ManagedIdentityCredential(identity_config={"resource_id": "foo"})


def test_log(caplog):
    with caplog.at_level(logging.INFO, logger="azure.identity._credentials.managed_identity"):
        ManagedIdentityCredential()
        assert "ManagedIdentityCredential will use IMDS" in caplog.text

        caplog.clear()
        with mock.patch.dict(
            MANAGED_IDENTITY_ENVIRON,
            {
                EnvironmentVariables.IDENTITY_ENDPOINT: "new_endpoint",
                EnvironmentVariables.IDENTITY_HEADER: "new_secret",
                EnvironmentVariables.MSI_ENDPOINT: "endpoint",
                EnvironmentVariables.MSI_SECRET: "secret",
            },
            clear=True,
        ):
            ManagedIdentityCredential()
            assert "App Service managed identity" in caplog.text

            caplog.clear()
            ManagedIdentityCredential(client_id="foo")
            assert "App Service managed identity with client_id: foo" in caplog.text

            caplog.clear()
            ManagedIdentityCredential(identity_config={"object_id": "bar"})
            assert "App Service managed identity with object_id: bar" in caplog.text

        caplog.clear()
        mock_environ = {
            EnvironmentVariables.AZURE_AUTHORITY_HOST: "authority",
            EnvironmentVariables.AZURE_TENANT_ID: "tenant",
            EnvironmentVariables.AZURE_FEDERATED_TOKEN_FILE: "token_file",
        }
        with mock.patch.dict("os.environ", mock_environ, clear=True):
            ManagedIdentityCredential(client_id="foo")
            assert "workload identity with client_id: foo" in caplog.text