diff --git a/sdk/identity/azure-identity/azure/identity/_credentials/certificate.py b/sdk/identity/azure-identity/azure/identity/_credentials/certificate.py
index 35c81b2e3da5..405a0954e6c0 100644
--- a/sdk/identity/azure-identity/azure/identity/_credentials/certificate.py
+++ b/sdk/identity/azure-identity/azure/identity/_credentials/certificate.py
@@ -2,17 +2,21 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 # ------------------------------------
+from binascii import hexlify
 from typing import TYPE_CHECKING
 
-from .._internal import AadClient, CertificateCredentialBase
-from .._internal.decorators import log_get_token
+from cryptography import x509
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.backends import default_backend
+import six
+
+from .._internal.client_credential_base import ClientCredentialBase
 
 if TYPE_CHECKING:
-    from azure.core.credentials import AccessToken
     from typing import Any
 
 
-class CertificateCredential(CertificateCredentialBase):
+class CertificateCredential(ClientCredentialBase):
     """Authenticates as a service principal using a certificate.
 
     :param str tenant_id: ID of the service principal's tenant. Also called its 'directory' ID.
@@ -31,31 +35,28 @@ class CertificateCredential(CertificateCredentialBase):
           is unavailable. Default to False. Has no effect when `enable_persistent_cache` is False.
     """
 
-    @log_get_token("CertificateCredential")
-    def get_token(self, *scopes, **kwargs):  # pylint:disable=unused-argument
-        # type: (*str, **Any) -> AccessToken
-        """Request an access token for `scopes`.
-
-        .. note:: This method is called by Azure SDK clients. It isn't intended for use in application code.
-
-        :param str scopes: desired scopes for the access token. This method requires at least one scope.
-        :rtype: :class:`azure.core.credentials.AccessToken`
-        :raises ~azure.core.exceptions.ClientAuthenticationError: authentication failed. The error's ``message``
-          attribute gives a reason. Any error response from Azure Active Directory is available as the error's
-          ``response`` attribute.
-        """
-        if not scopes:
-            raise ValueError("'get_token' requires at least one scope")
-
-        token = self._client.get_cached_access_token(scopes, query={"client_id": self._client_id})
-        if not token:
-            token = self._client.obtain_token_by_client_certificate(scopes, self._certificate, **kwargs)
-        elif self._client.should_refresh(token):
-            try:
-                self._client.obtain_token_by_client_certificate(scopes, self._certificate, **kwargs)
-            except Exception:  # pylint: disable=broad-except
-                pass
-        return token
-
-    def _get_auth_client(self, tenant_id, client_id, **kwargs):
-        return AadClient(tenant_id, client_id, **kwargs)
+    def __init__(self, tenant_id, client_id, certificate_path, **kwargs):
+        # type: (str, str, str, **Any) -> None
+        if not certificate_path:
+            raise ValueError(
+                "'certificate_path' must be the path to a PEM file containing an x509 certificate and its private key"
+            )
+
+        password = kwargs.pop("password", None)
+        if isinstance(password, six.text_type):
+            password = password.encode(encoding="utf-8")
+
+        with open(certificate_path, "rb") as f:
+            pem_bytes = f.read()
+
+        cert = x509.load_pem_x509_certificate(pem_bytes, default_backend())
+        fingerprint = cert.fingerprint(hashes.SHA1())  # nosec
+
+        # TODO: msal doesn't formally support passwords (but soon will); the below depends on an implementation detail
+        private_key = serialization.load_pem_private_key(pem_bytes, password=password, backend=default_backend())
+        super(CertificateCredential, self).__init__(
+            client_id=client_id,
+            client_credential={"private_key": private_key, "thumbprint": hexlify(fingerprint).decode("utf-8")},
+            tenant_id=tenant_id,
+            **kwargs
+        )
diff --git a/sdk/identity/azure-identity/azure/identity/_credentials/client_secret.py b/sdk/identity/azure-identity/azure/identity/_credentials/client_secret.py
index a327416cd731..311a6f1ef3e8 100644
--- a/sdk/identity/azure-identity/azure/identity/_credentials/client_secret.py
+++ b/sdk/identity/azure-identity/azure/identity/_credentials/client_secret.py
@@ -2,21 +2,16 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 # ------------------------------------
-from .._internal import AadClient, ClientSecretCredentialBase
-from .._internal.decorators import log_get_token
+from typing import TYPE_CHECKING
 
-try:
-    from typing import TYPE_CHECKING
-except ImportError:
-    TYPE_CHECKING = False
+from .._internal.client_credential_base import ClientCredentialBase
 
 if TYPE_CHECKING:
     # pylint:disable=unused-import,ungrouped-imports
     from typing import Any
-    from azure.core.credentials import AccessToken
 
 
-class ClientSecretCredential(ClientSecretCredentialBase):
+class ClientSecretCredential(ClientCredentialBase):
     """Authenticates as a service principal using a client ID and client secret.
 
     :param str tenant_id: ID of the service principal's tenant. Also called its 'directory' ID.
@@ -32,31 +27,17 @@ class ClientSecretCredential(ClientSecretCredentialBase):
           is unavailable. Default to False. Has no effect when `enable_persistent_cache` is False.
     """
 
-    @log_get_token("ClientSecretCredential")
-    def get_token(self, *scopes, **kwargs):
-        # type: (*str, **Any) -> AccessToken
-        """Request an access token for `scopes`.
-
-        .. note:: This method is called by Azure SDK clients. It isn't intended for use in application code.
-
-        :param str scopes: desired scopes for the access token. This method requires at least one scope.
-        :rtype: :class:`azure.core.credentials.AccessToken`
-        :raises ~azure.core.exceptions.ClientAuthenticationError: authentication failed. The error's ``message``
-          attribute gives a reason. Any error response from Azure Active Directory is available as the error's
-          ``response`` attribute.
-        """
-        if not scopes:
-            raise ValueError("'get_token' requires at least one scope")
-
-        token = self._client.get_cached_access_token(scopes, query={"client_id": self._client_id})
-        if not token:
-            token = self._client.obtain_token_by_client_secret(scopes, self._secret, **kwargs)
-        elif self._client.should_refresh(token):
-            try:
-                self._client.obtain_token_by_client_secret(scopes, self._secret, **kwargs)
-            except Exception:  # pylint: disable=broad-except
-                pass
-        return token
-
-    def _get_auth_client(self, tenant_id, client_id, **kwargs):
-        return AadClient(tenant_id, client_id, **kwargs)
+    def __init__(self, tenant_id, client_id, client_secret, **kwargs):
+        # type: (str, str, str, **Any) -> None
+        if not client_id:
+            raise ValueError("client_id should be the id of an Azure Active Directory application")
+        if not client_secret:
+            raise ValueError("secret should be an Azure Active Directory application's client secret")
+        if not tenant_id:
+            raise ValueError(
+                "tenant_id should be an Azure Active Directory tenant's id (also called its 'directory id')"
+            )
+
+        super(ClientSecretCredential, self).__init__(
+            client_id=client_id, client_credential=client_secret, tenant_id=tenant_id, **kwargs
+        )
diff --git a/sdk/identity/azure-identity/azure/identity/_internal/client_credential_base.py b/sdk/identity/azure-identity/azure/identity/_internal/client_credential_base.py
new file mode 100644
index 000000000000..68fc0df801ea
--- /dev/null
+++ b/sdk/identity/azure-identity/azure/identity/_internal/client_credential_base.py
@@ -0,0 +1,59 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+import time
+from typing import TYPE_CHECKING
+
+import msal
+
+from azure.core.credentials import AccessToken
+from azure.core.exceptions import ClientAuthenticationError
+from .get_token_mixin import GetTokenMixin
+from .persistent_cache import load_service_principal_cache
+
+from . import wrap_exceptions
+from .msal_credentials import MsalCredential
+
+if TYPE_CHECKING:
+    from typing import Any, Optional
+
+
+class ClientCredentialBase(MsalCredential, GetTokenMixin):
+    """Base class for credentials authenticating a service principal with a certificate or secret"""
+
+    def __init__(self, **kwargs):
+        if kwargs.pop("enable_persistent_cache", False):
+            allow_unencrypted = kwargs.pop("allow_unencrypted_cache", False)
+            cache = load_service_principal_cache(allow_unencrypted)
+        else:
+            cache = msal.TokenCache()
+        super(ClientCredentialBase, self).__init__(_cache=cache, **kwargs)
+
+    @wrap_exceptions
+    def _acquire_token_silently(self, *scopes, **kwargs):
+        # type: (*str, **Any) -> Optional[AccessToken]
+        app = self._get_app()
+        request_time = int(time.time())
+        result = app.acquire_token_silent_with_error(list(scopes), account=None, **kwargs)
+        if result and "access_token" in result and "expires_in" in result:
+            return AccessToken(result["access_token"], request_time + int(result["expires_in"]))
+        return None
+
+    @wrap_exceptions
+    def _request_token(self, *scopes, **kwargs):
+        # type: (*str, **Any) -> Optional[AccessToken]
+        app = self._get_app()
+        request_time = int(time.time())
+        result = app.acquire_token_for_client(list(scopes))
+        if "access_token" not in result:
+            message = "Authentication failed: {}".format(result.get("error_description") or result.get("error"))
+            raise ClientAuthenticationError(message=message)
+
+        return AccessToken(result["access_token"], request_time + int(result["expires_in"]))
+
+    def _get_app(self):
+        # type: () -> msal.ConfidentialClientApplication
+        if not self._msal_app:
+            self._msal_app = self._create_app(msal.ConfidentialClientApplication)
+        return self._msal_app
diff --git a/sdk/identity/azure-identity/azure/identity/_internal/msal_credentials.py b/sdk/identity/azure-identity/azure/identity/_internal/msal_credentials.py
index fd5034acd4bb..6860af649cac 100644
--- a/sdk/identity/azure-identity/azure/identity/_internal/msal_credentials.py
+++ b/sdk/identity/azure-identity/azure/identity/_internal/msal_credentials.py
@@ -50,11 +50,7 @@ def __init__(self, client_id, client_credential=None, **kwargs):
 
         # postpone creating the wrapped application because its initializer uses the network
         self._msal_app = None  # type: Optional[msal.ClientApplication]
-
-    @abc.abstractmethod
-    def get_token(self, *scopes, **kwargs):
-        # type: (*str, **Any) -> AccessToken
-        pass
+        super(MsalCredential, self).__init__()
 
     @abc.abstractmethod
     def _get_app(self):
diff --git a/sdk/identity/azure-identity/setup.py b/sdk/identity/azure-identity/setup.py
index 2e6f1ea22af6..2cde1bc35a0c 100644
--- a/sdk/identity/azure-identity/setup.py
+++ b/sdk/identity/azure-identity/setup.py
@@ -73,7 +73,7 @@
     install_requires=[
         "azure-core<2.0.0,>=1.0.0",
         "cryptography>=2.1.4",
-        "msal<2.0.0,>=1.3.0",
+        "msal<1.5.0,>=1.3.0",
         "msal-extensions~=0.2.2",
         "six>=1.6",
     ],
diff --git a/sdk/identity/azure-identity/tests/helpers.py b/sdk/identity/azure-identity/tests/helpers.py
index 2680ea7e574c..692c686df232 100644
--- a/sdk/identity/azure-identity/tests/helpers.py
+++ b/sdk/identity/azure-identity/tests/helpers.py
@@ -156,7 +156,9 @@ def mock_response(status_code=200, headers=None, json_payload=None):
 
 def get_discovery_response(endpoint="https://a/b"):
     aad_metadata_endpoint_names = ("authorization_endpoint", "token_endpoint", "tenant_discovery_endpoint")
-    return mock_response(json_payload={name: endpoint for name in aad_metadata_endpoint_names})
+    payload = {name: endpoint for name in aad_metadata_endpoint_names}
+    payload["metadata"] = ""
+    return mock_response(json_payload=payload)
 
 
 def validating_transport(requests, responses):
@@ -177,6 +179,11 @@ def validate_request(request, **_):
     return mock.Mock(send=mock.Mock(wraps=validate_request))
 
 
+def msal_validating_transport(requests, responses, **kwargs):
+    """a validating transport with default responses to MSAL's discovery requests"""
+    return validating_transport([Request()] * 2 + requests, [get_discovery_response(**kwargs)] * 2 + responses)
+
+
 def urlsafeb64_decode(s):
     if isinstance(s, six.text_type):
         s = s.encode("ascii")
diff --git a/sdk/identity/azure-identity/tests/test_certificate_credential.py b/sdk/identity/azure-identity/tests/test_certificate_credential.py
index af0eee63c580..5f4b49f416e2 100644
--- a/sdk/identity/azure-identity/tests/test_certificate_credential.py
+++ b/sdk/identity/azure-identity/tests/test_certificate_credential.py
@@ -15,9 +15,18 @@
 from cryptography.hazmat.primitives.asymmetric import padding
 from msal import TokenCache
 import pytest
+import six
 from six.moves.urllib_parse import urlparse
 
-from helpers import build_aad_response, urlsafeb64_decode, mock_response, Request, validating_transport
+from helpers import (
+    build_aad_response,
+    get_discovery_response,
+    urlsafeb64_decode,
+    mock_response,
+    msal_validating_transport,
+    Request,
+    validating_transport,
+)
 
 try:
     from unittest.mock import Mock, patch
@@ -41,11 +50,12 @@ def test_no_scopes():
 def test_policies_configurable():
     policy = Mock(spec_set=SansIOHTTPPolicy, on_request=Mock())
 
-    def send(*_, **__):
-        return mock_response(json_payload=build_aad_response(access_token="**"))
+    transport = msal_validating_transport(
+        requests=[Request()], responses=[mock_response(json_payload=build_aad_response(access_token="**"))],
+    )
 
     credential = CertificateCredential(
-        "tenant-id", "client-id", CERT_PATH, policies=[ContentDecodePolicy(), policy], transport=Mock(send=send)
+        "tenant-id", "client-id", CERT_PATH, policies=[ContentDecodePolicy(), policy], transport=transport
     )
 
     credential.get_token("scope")
@@ -54,7 +64,7 @@ def send(*_, **__):
 
 
 def test_user_agent():
-    transport = validating_transport(
+    transport = msal_validating_transport(
         requests=[Request(required_headers={"User-Agent": USER_AGENT})],
         responses=[mock_response(json_payload=build_aad_response(access_token="**"))],
     )
@@ -65,35 +75,37 @@ def test_user_agent():
 
 
 @pytest.mark.parametrize("authority", ("localhost", "https://localhost"))
-@pytest.mark.parametrize("cert_path,cert_password", BOTH_CERTS)
-def test_request_url(cert_path, cert_password, authority):
+def test_authority(authority):
     """the credential should accept an authority, with or without scheme, as an argument or environment variable"""
 
     tenant_id = "expected_tenant"
-    access_token = "***"
     parsed_authority = urlparse(authority)
-    expected_netloc = parsed_authority.netloc or authority  # "localhost" parses to netloc "", path "localhost"
-
-    def mock_send(request, **kwargs):
-        actual = urlparse(request.url)
-        assert actual.scheme == "https"
-        assert actual.netloc == expected_netloc
-        assert actual.path.startswith("/" + tenant_id)
-        return mock_response(json_payload={"token_type": "Bearer", "expires_in": 42, "access_token": access_token})
+    expected_netloc = parsed_authority.netloc or authority
+    expected_authority = "https://{}/{}".format(expected_netloc, tenant_id)
 
-    cred = CertificateCredential(
-        tenant_id, "client-id", cert_path, password=cert_password, transport=Mock(send=mock_send), authority=authority
+    mock_ctor = Mock(
+        return_value=Mock(acquire_token_silent_with_error=lambda *_, **__: {"access_token": "**", "expires_in": 42})
     )
-    token = cred.get_token("scope")
-    assert token.token == access_token
+
+    credential = CertificateCredential(tenant_id, "client-id", CERT_PATH, authority=authority)
+    with patch("msal.ConfidentialClientApplication", mock_ctor):
+        # must call get_token because the credential constructs the MSAL application lazily
+        credential.get_token("scope")
+
+    assert mock_ctor.call_count == 1
+    _, kwargs = mock_ctor.call_args
+    assert kwargs["authority"] == expected_authority
+    mock_ctor.reset_mock()
 
     # authority can be configured via environment variable
     with patch.dict("os.environ", {EnvironmentVariables.AZURE_AUTHORITY_HOST: authority}, clear=True):
-        credential = CertificateCredential(
-            tenant_id, "client-id", cert_path, password=cert_password, transport=Mock(send=mock_send)
-        )
+        credential = CertificateCredential(tenant_id, "client-id", CERT_PATH, authority=authority)
+    with patch("msal.ConfidentialClientApplication", mock_ctor):
         credential.get_token("scope")
-    assert token.token == access_token
+
+    assert mock_ctor.call_count == 1
+    _, kwargs = mock_ctor.call_args
+    assert kwargs["authority"] == expected_authority
 
 
 @pytest.mark.parametrize("cert_path,cert_password", BOTH_CERTS)
@@ -105,6 +117,9 @@ def test_request_body(cert_path, cert_password):
     tenant_id = "tenant"
 
     def mock_send(request, **kwargs):
+        if not request.body:
+            return get_discovery_response()
+
         assert request.body["grant_type"] == "client_credentials"
         assert request.body["scope"] == expected_scope
 
@@ -128,7 +143,7 @@ def validate_jwt(request, client_id, pem_bytes):
     cert = x509.load_pem_x509_certificate(pem_bytes, default_backend())
 
     # jwt is of the form 'header.payload.signature'; 'signature' is 'header.payload' signed with cert's private key
-    jwt = request.body["client_assertion"]
+    jwt = six.ensure_str(request.body["client_assertion"])
     header, payload, signature = (urlsafeb64_decode(s) for s in jwt.split("."))
     signed_part = jwt[: jwt.rfind(".")]
     claims = json.loads(payload.decode("utf-8"))
@@ -211,10 +226,10 @@ def test_persistent_cache_multiple_clients(cert_path, cert_password):
 
     access_token_a = "token a"
     access_token_b = "not " + access_token_a
-    transport_a = validating_transport(
+    transport_a = msal_validating_transport(
         requests=[Request()], responses=[mock_response(json_payload=build_aad_response(access_token=access_token_a))]
     )
-    transport_b = validating_transport(
+    transport_b = msal_validating_transport(
         requests=[Request()], responses=[mock_response(json_payload=build_aad_response(access_token=access_token_b))]
     )
 
@@ -234,9 +249,9 @@ def test_persistent_cache_multiple_clients(cert_path, cert_password):
     scope = "scope"
     token_a = credential_a.get_token(scope)
     assert token_a.token == access_token_a
-    assert transport_a.send.call_count == 1
+    assert transport_a.send.call_count == 3  # two MSAL discovery requests, one token request
 
     # B should get a different token for the same scope
     token_b = credential_b.get_token(scope)
     assert token_b.token == access_token_b
-    assert transport_b.send.call_count == 1
+    assert transport_b.send.call_count == 3
diff --git a/sdk/identity/azure-identity/tests/test_client_secret_credential.py b/sdk/identity/azure-identity/tests/test_client_secret_credential.py
index ea3362a3f0ff..a204c6cf8c6d 100644
--- a/sdk/identity/azure-identity/tests/test_client_secret_credential.py
+++ b/sdk/identity/azure-identity/tests/test_client_secret_credential.py
@@ -2,9 +2,6 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 # ------------------------------------
-import time
-
-from azure.core.credentials import AccessToken
 from azure.core.pipeline.policies import ContentDecodePolicy, SansIOHTTPPolicy
 from azure.identity import ClientSecretCredential
 from azure.identity._constants import EnvironmentVariables
@@ -13,7 +10,7 @@
 import pytest
 from six.moves.urllib_parse import urlparse
 
-from helpers import build_aad_response, mock_response, Request, validating_transport
+from helpers import build_aad_response, mock_response, msal_validating_transport, Request, validating_transport
 
 try:
     from unittest.mock import Mock, patch
@@ -32,11 +29,12 @@ def test_no_scopes():
 def test_policies_configurable():
     policy = Mock(spec_set=SansIOHTTPPolicy, on_request=Mock())
 
-    def send(*_, **__):
-        return mock_response(json_payload=build_aad_response(access_token="**"))
+    transport = msal_validating_transport(
+        requests=[Request()], responses=[mock_response(json_payload=build_aad_response(access_token="**"))],
+    )
 
     credential = ClientSecretCredential(
-        "tenant-id", "client-id", "client-secret", policies=[ContentDecodePolicy(), policy], transport=Mock(send=send)
+        "tenant-id", "client-id", "client-secret", policies=[ContentDecodePolicy(), policy], transport=transport
     )
 
     credential.get_token("scope")
@@ -45,7 +43,7 @@ def send(*_, **__):
 
 
 def test_user_agent():
-    transport = validating_transport(
+    transport = msal_validating_transport(
         requests=[Request(required_headers={"User-Agent": USER_AGENT})],
         responses=[mock_response(json_payload=build_aad_response(access_token="**"))],
     )
@@ -61,89 +59,49 @@ def test_client_secret_credential():
     tenant_id = "fake-tenant-id"
     access_token = "***"
 
-    transport = validating_transport(
+    transport = msal_validating_transport(
+        endpoint="https://localhost/" + tenant_id,
         requests=[Request(url_substring=tenant_id, required_data={"client_id": client_id, "client_secret": secret})],
-        responses=[
-            mock_response(
-                json_payload={
-                    "token_type": "Bearer",
-                    "expires_in": 42,
-                    "ext_expires_in": 42,
-                    "access_token": access_token,
-                }
-            )
-        ],
+        responses=[mock_response(json_payload=build_aad_response(access_token=access_token))],
     )
 
     token = ClientSecretCredential(tenant_id, client_id, secret, transport=transport).get_token("scope")
 
-    # not validating expires_on because doing so requires monkeypatching time, and this is tested elsewhere
     assert token.token == access_token
 
 
 @pytest.mark.parametrize("authority", ("localhost", "https://localhost"))
-def test_request_url(authority):
+def test_authority(authority):
     """the credential should accept an authority, with or without scheme, as an argument or environment variable"""
 
     tenant_id = "expected_tenant"
-    access_token = "***"
     parsed_authority = urlparse(authority)
-    expected_netloc = parsed_authority.netloc or authority  # "localhost" parses to netloc "", path "localhost"
+    expected_netloc = parsed_authority.netloc or authority
+    expected_authority = "https://{}/{}".format(expected_netloc, tenant_id)
 
-    def mock_send(request, **kwargs):
-        actual = urlparse(request.url)
-        assert actual.scheme == "https"
-        assert actual.netloc == expected_netloc
-        assert actual.path.startswith("/" + tenant_id)
-        return mock_response(json_payload={"token_type": "Bearer", "expires_in": 42, "access_token": access_token})
-
-    credential = ClientSecretCredential(
-        tenant_id, "client-id", "secret", transport=Mock(send=mock_send), authority=authority
+    mock_ctor = Mock(
+        return_value=Mock(acquire_token_silent_with_error=lambda *_, **__: {"access_token": "**", "expires_in": 42})
     )
-    token = credential.get_token("scope")
-    assert token.token == access_token
 
-    # authority can be configured via environment variable
-    with patch.dict("os.environ", {EnvironmentVariables.AZURE_AUTHORITY_HOST: authority}, clear=True):
-        credential = ClientSecretCredential(tenant_id, "client-id", "secret", transport=Mock(send=mock_send))
+    credential = ClientSecretCredential(tenant_id, "client-id", "secret", authority=authority)
+    with patch("msal.ConfidentialClientApplication", mock_ctor):
+        # must call get_token because the credential constructs the MSAL application lazily
         credential.get_token("scope")
-    assert token.token == access_token
 
+    assert mock_ctor.call_count == 1
+    _, kwargs = mock_ctor.call_args
+    assert kwargs["authority"] == expected_authority
+    mock_ctor.reset_mock()
 
-def test_cache():
-    expired = "this token's expired"
-    now = int(time.time())
-    expired_on = now - 3600
-    expired_token = AccessToken(expired, expired_on)
-    token_payload = {
-        "access_token": expired,
-        "expires_in": 0,
-        "ext_expires_in": 0,
-        "expires_on": expired_on,
-        "not_before": now,
-        "token_type": "Bearer",
-    }
-    mock_send = Mock(return_value=mock_response(json_payload=token_payload))
-    scope = "scope"
-
-    credential = ClientSecretCredential(
-        tenant_id="some-guid", client_id="client_id", client_secret="secret", transport=Mock(send=mock_send)
-    )
-
-    # get_token initially returns the expired token because the credential
-    # doesn't check whether tokens it receives from the service have expired
-    token = credential.get_token(scope)
-    assert token == expired_token
-
-    access_token = "new token"
-    token_payload["access_token"] = access_token
-    token_payload["expires_on"] = now + 3600
-    valid_token = AccessToken(access_token, now + 3600)
+    # authority can be configured via environment variable
+    with patch.dict("os.environ", {EnvironmentVariables.AZURE_AUTHORITY_HOST: authority}, clear=True):
+        credential = ClientSecretCredential(tenant_id, "client-id", "secret")
+    with patch("msal.ConfidentialClientApplication", mock_ctor):
+        credential.get_token("scope")
 
-    # second call should observe the cached token has expired, and request another
-    token = credential.get_token(scope)
-    assert token == valid_token
-    assert mock_send.call_count == 2
+    assert mock_ctor.call_count == 1
+    _, kwargs = mock_ctor.call_args
+    assert kwargs["authority"] == expected_authority
 
 
 def test_enable_persistent_cache():
@@ -204,10 +162,10 @@ def test_persistent_cache_multiple_clients():
 
     access_token_a = "token a"
     access_token_b = "not " + access_token_a
-    transport_a = validating_transport(
+    transport_a = msal_validating_transport(
         requests=[Request()], responses=[mock_response(json_payload=build_aad_response(access_token=access_token_a))]
     )
-    transport_b = validating_transport(
+    transport_b = msal_validating_transport(
         requests=[Request()], responses=[mock_response(json_payload=build_aad_response(access_token=access_token_b))]
     )
 
@@ -227,9 +185,9 @@ def test_persistent_cache_multiple_clients():
     scope = "scope"
     token_a = credential_a.get_token(scope)
     assert token_a.token == access_token_a
-    assert transport_a.send.call_count == 1
+    assert transport_a.send.call_count == 3  # two MSAL discovery requests, one token request
 
     # B should get a different token for the same scope
     token_b = credential_b.get_token(scope)
     assert token_b.token == access_token_b
-    assert transport_b.send.call_count == 1
+    assert transport_b.send.call_count == 3
diff --git a/sdk/identity/azure-identity/tests/test_environment_credential.py b/sdk/identity/azure-identity/tests/test_environment_credential.py
index 3efb578b7aa9..35ce51045ef7 100644
--- a/sdk/identity/azure-identity/tests/test_environment_credential.py
+++ b/sdk/identity/azure-identity/tests/test_environment_credential.py
@@ -143,34 +143,3 @@ def test_username_password_configuration():
     assert kwargs["password"] == password
     assert kwargs["tenant_id"] == tenant_id
     assert kwargs["foo"] == bar
-
-
-def test_client_secret_credential():
-    client_id = "fake-client-id"
-    secret = "fake-client-secret"
-    tenant_id = "fake-tenant-id"
-    access_token = "***"
-
-    transport = validating_transport(
-        requests=[Request(url_substring=tenant_id, required_data={"client_id": client_id, "client_secret": secret})],
-        responses=[
-            mock_response(
-                json_payload={
-                    "token_type": "Bearer",
-                    "expires_in": 42,
-                    "ext_expires_in": 42,
-                    "access_token": access_token,
-                }
-            )
-        ],
-    )
-
-    environment = {
-        EnvironmentVariables.AZURE_CLIENT_ID: client_id,
-        EnvironmentVariables.AZURE_CLIENT_SECRET: secret,
-        EnvironmentVariables.AZURE_TENANT_ID: tenant_id,
-    }
-    with mock.patch.dict("os.environ", environment, clear=True):
-        token = EnvironmentCredential(transport=transport).get_token("scope")
-
-    assert token.token == access_token
diff --git a/shared_requirements.txt b/shared_requirements.txt
index a8f93695e121..228700cd69d7 100644
--- a/shared_requirements.txt
+++ b/shared_requirements.txt
@@ -97,7 +97,7 @@ futures
 mock
 typing
 typing-extensions
-msal<2.0.0,>=1.3.0
+msal<1.5.0,>=1.3.0
 msal-extensions~=0.2.2
 msrest>=0.5.0
 msrestazure<2.0.0,>=0.4.32