diff --git a/sdk/identity/azure-identity/CHANGELOG.md b/sdk/identity/azure-identity/CHANGELOG.md
index 3929c367455e..3bd27002e981 100644
--- a/sdk/identity/azure-identity/CHANGELOG.md
+++ b/sdk/identity/azure-identity/CHANGELOG.md
@@ -5,6 +5,9 @@
 ### Features Added
 - `CertificateCredential` accepts certificates in PKCS12 format
   ([#13540](https://github.com/Azure/azure-sdk-for-python/issues/13540))
+- `OnBehalfOfCredential` supports the on-behalf-of authentication flow for
+  accessing resources on behalf of users
+  ([#19308](https://github.com/Azure/azure-sdk-for-python/issues/19308))
 
 ### Breaking Changes
 
@@ -17,7 +20,6 @@
   ([#18798](https://github.com/Azure/azure-sdk-for-python/issues/18798))
 
 
-
 ## 1.6.1 (2021-08-19)
 
 ### Other Changes
diff --git a/sdk/identity/azure-identity/azure/identity/__init__.py b/sdk/identity/azure-identity/azure/identity/__init__.py
index 4840728113a8..4d7dff365695 100644
--- a/sdk/identity/azure-identity/azure/identity/__init__.py
+++ b/sdk/identity/azure-identity/azure/identity/__init__.py
@@ -21,6 +21,7 @@
     EnvironmentCredential,
     InteractiveBrowserCredential,
     ManagedIdentityCredential,
+    OnBehalfOfCredential,
     SharedTokenCacheCredential,
     UsernamePasswordCredential,
     VisualStudioCodeCredential,
@@ -45,6 +46,7 @@
     "EnvironmentCredential",
     "InteractiveBrowserCredential",
     "KnownAuthorities",
+    "OnBehalfOfCredential",
     "RegionalAuthority",
     "ManagedIdentityCredential",
     "SharedTokenCacheCredential",
diff --git a/sdk/identity/azure-identity/azure/identity/_credentials/__init__.py b/sdk/identity/azure-identity/azure/identity/_credentials/__init__.py
index 05dc788d1bde..11c7b26db428 100644
--- a/sdk/identity/azure-identity/azure/identity/_credentials/__init__.py
+++ b/sdk/identity/azure-identity/azure/identity/_credentials/__init__.py
@@ -12,6 +12,7 @@
 from .default import DefaultAzureCredential
 from .environment import EnvironmentCredential
 from .managed_identity import ManagedIdentityCredential
+from .on_behalf_of import OnBehalfOfCredential
 from .shared_cache import SharedTokenCacheCredential
 from .azure_cli import AzureCliCredential
 from .device_code import DeviceCodeCredential
@@ -32,6 +33,7 @@
     "EnvironmentCredential",
     "InteractiveBrowserCredential",
     "ManagedIdentityCredential",
+    "OnBehalfOfCredential",
     "SharedTokenCacheCredential",
     "AzureCliCredential",
     "UsernamePasswordCredential",
diff --git a/sdk/identity/azure-identity/azure/identity/_credentials/certificate.py b/sdk/identity/azure-identity/azure/identity/_credentials/certificate.py
index 5a9bfa9935df..cdb999d41898 100644
--- a/sdk/identity/azure-identity/azure/identity/_credentials/certificate.py
+++ b/sdk/identity/azure-identity/azure/identity/_credentials/certificate.py
@@ -92,13 +92,16 @@ def load_pkcs12_certificate(certificate_data, password):
     # type: (bytes, Optional[bytes]) -> _Cert
     from cryptography.hazmat.primitives.serialization import Encoding, NoEncryption, pkcs12, PrivateFormat
 
-    private_key, cert, additional_certs = pkcs12.load_key_and_certificates(
-        certificate_data, password, backend=default_backend()
-    )
+    try:
+        private_key, cert, additional_certs = pkcs12.load_key_and_certificates(
+            certificate_data, password, backend=default_backend()
+        )
+    except ValueError as ex:
+        # mentioning PEM here because we raise this error when certificate_data is garbage
+        six.raise_from(ValueError("Failed to deserialize certificate in PEM or PKCS12 format"), ex)
     if not private_key:
         raise ValueError("The certificate must include its private key")
     if not cert:
-        # mentioning PEM here because we raise this error when certificate_data is garbage
         raise ValueError("Failed to deserialize certificate in PEM or PKCS12 format")
 
     # This serializes the private key without any encryption it may have had. Doing so doesn't violate security
@@ -137,7 +140,7 @@ def get_client_credential(certificate_path, password=None, certificate_data=None
         password = None  # load_pkcs12_certificate returns cert.pem_bytes decrypted
 
     if not isinstance(cert.private_key, RSAPrivateKey):
-        raise ValueError("CertificateCredential requires an RSA private key because it uses RS256 for signing")
+        raise ValueError("The certificate must have an RSA private key because RS256 is used for signing")
 
     client_credential = {"private_key": cert.pem_bytes, "thumbprint": hexlify(cert.fingerprint).decode("utf-8")}
     if password:
diff --git a/sdk/identity/azure-identity/azure/identity/_credentials/on_behalf_of.py b/sdk/identity/azure-identity/azure/identity/_credentials/on_behalf_of.py
new file mode 100644
index 000000000000..bc39c7a475e1
--- /dev/null
+++ b/sdk/identity/azure-identity/azure/identity/_credentials/on_behalf_of.py
@@ -0,0 +1,110 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+import time
+from typing import cast, TYPE_CHECKING
+
+import six
+
+from azure.core.credentials import AccessToken
+from azure.core.exceptions import ClientAuthenticationError
+
+from .certificate import get_client_credential
+from .._internal.decorators import wrap_exceptions
+from .._internal.get_token_mixin import GetTokenMixin
+from .._internal.interactive import _build_auth_record
+from .._internal.msal_credentials import MsalCredential
+
+if TYPE_CHECKING:
+    from typing import Any, Dict, Optional, Union
+    import msal
+    from .. import AuthenticationRecord
+
+
+class OnBehalfOfCredential(MsalCredential, GetTokenMixin):
+    """Authenticates a service principal via the on-behalf-of flow.
+
+    This flow is typically used by middle-tier services that authorize requests to other services with a delegated
+    user identity. Because this is not an interactive authentication flow, an application using it must have admin
+    consent for any delegated permissions before requesting tokens for them. See `Azure Active Directory documentation
+    <https://docs.microsoft.com/azure/active-directory/develop/v2-oauth2-on-behalf-of-flow>`_ for a more detailed
+    description of the on-behalf-of flow.
+
+    :param str tenant_id: ID of the service principal's tenant. Also called its "directory" ID.
+    :param str client_id: the service principal's client ID
+    :param client_credential: a credential to authenticate the service principal, either one of its client secrets (a
+        string) or the bytes of a certificate in PEM or PKCS12 format including the private key
+    :type client_credential: str or bytes
+    :param str user_assertion: the access token the credential will use as the user assertion when requesting
+        on-behalf-of tokens
+
+    :keyword bool allow_multitenant_authentication: when True, enables the credential to acquire tokens from any tenant
+        the application is registered in. When False, which is the default, the credential will acquire tokens only
+        from the tenant specified by **tenant_id**.
+    :keyword str authority: Authority of an Azure Active Directory endpoint, for example "login.microsoftonline.com",
+        the authority for Azure Public Cloud (which is the default). :class:`~azure.identity.AzureAuthorityHosts`
+        defines authorities for other clouds.
+    :keyword password: a certificate password. Used only when **client_credential** is certificate bytes. If this value
+        is a unicode string, it will be encoded as UTF-8. If the certificate requires a different encoding, pass
+        appropriately encoded bytes instead.
+    :paramtype password: str or bytes
+    """
+
+    def __init__(self, tenant_id, client_id, client_credential, user_assertion, **kwargs):
+        # type: (str, str, Union[bytes, str], str, **Any) -> None
+        credential = cast("Union[Dict, str]", client_credential)
+        if isinstance(client_credential, six.binary_type):
+            try:
+                credential = get_client_credential(
+                    certificate_path=None, password=kwargs.pop("password", None), certificate_data=client_credential
+                )
+            except ValueError as ex:
+                # client_credential isn't a valid cert. On 2.7 str == bytes and we ignore this exception because we
+                # can't tell whether the caller intended to provide a cert. On Python 3 we can say the caller provided
+                # either an invalid cert, or a client secret as bytes; both are errors.
+                if six.PY3:
+                    message = (
+                        '"client_credential" should be either a client secret (a string)'
+                        + " or the bytes of a certificate in PEM or PKCS12 format"
+                    )
+                    six.raise_from(ValueError(message), ex)
+
+        super(OnBehalfOfCredential, self).__init__(client_id, credential, tenant_id=tenant_id, **kwargs)
+        self._assertion = user_assertion
+        self._auth_record = None  # type: Optional[AuthenticationRecord]
+
+    @wrap_exceptions
+    def _acquire_token_silently(self, *scopes, **kwargs):
+        # type: (*str, **Any) -> Optional[AccessToken]
+        if self._auth_record:
+            claims = kwargs.get("claims")
+            app = self._get_app(**kwargs)
+            for account in app.get_accounts(username=self._auth_record.username):
+                if account.get("home_account_id") != self._auth_record.home_account_id:
+                    continue
+
+                now = int(time.time())
+                result = app.acquire_token_silent_with_error(list(scopes), account=account, claims_challenge=claims)
+                if result and "access_token" in result and "expires_in" in result:
+                    return AccessToken(result["access_token"], now + int(result["expires_in"]))
+
+        return None
+
+    @wrap_exceptions
+    def _request_token(self, *scopes, **kwargs):
+        # type: (*str, **Any) -> AccessToken
+        app = self._get_app(**kwargs)  # type: msal.ConfidentialClientApplication
+        request_time = int(time.time())
+        result = app.acquire_token_on_behalf_of(self._assertion, list(scopes), claims_challenge=kwargs.get("claims"))
+        if "access_token" not in result or "expires_in" not in result:
+            message = "Authentication failed: {}".format(result.get("error_description") or result.get("error"))
+            response = self._client.get_error_response(result)
+            raise ClientAuthenticationError(message=message, response=response)
+
+        try:
+            self._auth_record = _build_auth_record(result)
+        except ClientAuthenticationError:
+            pass  # non-fatal; we'll use the assertion again next time instead of a refresh token
+
+        return AccessToken(result["access_token"], request_time + int(result["expires_in"]))
diff --git a/sdk/identity/azure-identity/azure/identity/_internal/aad_client.py b/sdk/identity/azure-identity/azure/identity/_internal/aad_client.py
index fb27461d8eb4..8b7a3afa9ebd 100644
--- a/sdk/identity/azure-identity/azure/identity/_internal/aad_client.py
+++ b/sdk/identity/azure-identity/azure/identity/_internal/aad_client.py
@@ -10,7 +10,7 @@
 
 if TYPE_CHECKING:
     # pylint:disable=unused-import,ungrouped-imports
-    from typing import Any, Iterable, Optional
+    from typing import Any, Iterable, Optional, Union
     from azure.core.credentials import AccessToken
     from azure.core.pipeline import Pipeline
     from .._internal import AadClientCertificate
@@ -65,6 +65,11 @@ def obtain_token_by_refresh_token(self, scopes, refresh_token, **kwargs):
         response = self._pipeline.run(request, stream=False, retry_on_methods=self._POST, **kwargs)
         return self._process_response(response, now)
 
+    def obtain_token_on_behalf_of(self, scopes, client_credential, user_assertion, **kwargs):
+        # type: (Iterable[str], Union[str, AadClientCertificate], str, **Any) -> AccessToken
+        # no need for an implementation, non-async OnBehalfOfCredential acquires tokens through MSAL
+        raise NotImplementedError()
+
     # pylint:disable=no-self-use
     def _build_pipeline(self, **kwargs):
         # type: (**Any) -> Pipeline
diff --git a/sdk/identity/azure-identity/azure/identity/_internal/aad_client_base.py b/sdk/identity/azure-identity/azure/identity/_internal/aad_client_base.py
index 1bf0b2d02aff..b6a44e8d5681 100644
--- a/sdk/identity/azure-identity/azure/identity/_internal/aad_client_base.py
+++ b/sdk/identity/azure-identity/azure/identity/_internal/aad_client_base.py
@@ -17,6 +17,7 @@
 from azure.core.exceptions import ClientAuthenticationError
 from . import get_default_authority, normalize_authority
 from .._internal import resolve_tenant
+from .._internal.aadclient_certificate import AadClientCertificate
 
 try:
     from typing import TYPE_CHECKING
@@ -34,12 +35,13 @@
     from azure.core.pipeline import AsyncPipeline, Pipeline, PipelineResponse
     from azure.core.pipeline.policies import AsyncHTTPPolicy, HTTPPolicy, SansIOHTTPPolicy
     from azure.core.pipeline.transport import AsyncHttpTransport, HttpTransport
-    from .._internal import AadClientCertificate
 
     PipelineType = Union[AsyncPipeline, Pipeline]
     PolicyType = Union[AsyncHTTPPolicy, HTTPPolicy, SansIOHTTPPolicy]
     TransportType = Union[AsyncHttpTransport, HttpTransport]
 
+JWT_BEARER_ASSERTION = "urn:ietf:params:oauth:client-assertion-type:jwt-bearer"
+
 
 class AadClientBase(ABC):
     _POST = ["POST"]
@@ -96,6 +98,10 @@ def obtain_token_by_client_secret(self, scopes, secret, **kwargs):
     def obtain_token_by_refresh_token(self, scopes, refresh_token, **kwargs):
         pass
 
+    @abc.abstractmethod
+    def obtain_token_on_behalf_of(self, scopes, client_credential, user_assertion, **kwargs):
+        pass
+
     @abc.abstractmethod
     def _build_pipeline(self, **kwargs):
         pass
@@ -173,7 +179,7 @@ def _get_jwt_assertion_request(self, scopes, assertion, **kwargs):
         # type: (Iterable[str], str, **Any) -> HttpRequest
         data = {
             "client_assertion": assertion,
-            "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
+            "client_assertion_type": JWT_BEARER_ASSERTION,
             "client_id": self._client_id,
             "grant_type": "client_credentials",
             "scope": " ".join(scopes),
@@ -182,8 +188,8 @@ def _get_jwt_assertion_request(self, scopes, assertion, **kwargs):
         request = self._post(data, **kwargs)
         return request
 
-    def _get_client_certificate_request(self, scopes, certificate, **kwargs):
-        # type: (Iterable[str], AadClientCertificate, **Any) -> HttpRequest
+    def _get_client_certificate_assertion(self, certificate, **kwargs):
+        # type: (AadClientCertificate, **Any) -> str
         now = int(time.time())
         header = six.ensure_binary(
             json.dumps({"typ": "JWT", "alg": "RS256", "x5t": certificate.thumbprint}), encoding="utf-8"
@@ -204,8 +210,11 @@ def _get_client_certificate_request(self, scopes, certificate, **kwargs):
         jws = base64.urlsafe_b64encode(header) + b"." + base64.urlsafe_b64encode(payload)
         signature = certificate.sign(jws)
         jwt_bytes = jws + b"." + base64.urlsafe_b64encode(signature)
-        assertion = jwt_bytes.decode("utf-8")
+        return jwt_bytes.decode("utf-8")
 
+    def _get_client_certificate_request(self, scopes, certificate, **kwargs):
+        # type: (Iterable[str], AadClientCertificate, **Any) -> HttpRequest
+        assertion = self._get_client_certificate_assertion(certificate, **kwargs)
         return self._get_jwt_assertion_request(scopes, assertion, **kwargs)
 
     def _get_client_secret_request(self, scopes, secret, **kwargs):
@@ -219,6 +228,24 @@ def _get_client_secret_request(self, scopes, secret, **kwargs):
         request = self._post(data, **kwargs)
         return request
 
+    def _get_on_behalf_of_request(self, scopes, client_credential, user_assertion, **kwargs):
+        # type: (Iterable[str], Union[str, AadClientCertificate], str, **Any) -> HttpRequest
+        data = {
+            "assertion": user_assertion,
+            "client_id": self._client_id,
+            "grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
+            "requested_token_use": "on_behalf_of",
+            "scope": " ".join(scopes),
+        }
+        if isinstance(client_credential, AadClientCertificate):
+            data["client_assertion"] = self._get_client_certificate_assertion(client_credential)
+            data["client_assertion_type"] = JWT_BEARER_ASSERTION
+        else:
+            data["client_secret"] = client_credential
+
+        request = self._post(data, **kwargs)
+        return request
+
     def _get_refresh_token_request(self, scopes, refresh_token, **kwargs):
         # type: (Iterable[str], str, **Any) -> HttpRequest
         data = {
diff --git a/sdk/identity/azure-identity/azure/identity/_internal/aadclient_certificate.py b/sdk/identity/azure-identity/azure/identity/_internal/aadclient_certificate.py
index 2b7e09ce453f..1c963f4fb25c 100644
--- a/sdk/identity/azure-identity/azure/identity/_internal/aadclient_certificate.py
+++ b/sdk/identity/azure-identity/azure/identity/_internal/aadclient_certificate.py
@@ -27,7 +27,7 @@ def __init__(self, pem_bytes, password=None):
         # type: (bytes, Optional[bytes]) -> None
         private_key = serialization.load_pem_private_key(pem_bytes, password=password, backend=default_backend())
         if not isinstance(private_key, RSAPrivateKey):
-            raise ValueError("CertificateCredential requires an RSA private key because it uses RS256 for signing")
+            raise ValueError("The certificate must have an RSA private key because RS256 is used for signing")
         self._private_key = private_key
 
         cert = x509.load_pem_x509_certificate(pem_bytes, default_backend())
diff --git a/sdk/identity/azure-identity/azure/identity/aio/__init__.py b/sdk/identity/azure-identity/azure/identity/aio/__init__.py
index 817b87ad06fd..f2de3b977f9b 100644
--- a/sdk/identity/azure-identity/azure/identity/aio/__init__.py
+++ b/sdk/identity/azure-identity/azure/identity/aio/__init__.py
@@ -15,6 +15,7 @@
     DefaultAzureCredential,
     EnvironmentCredential,
     ManagedIdentityCredential,
+    OnBehalfOfCredential,
     SharedTokenCacheCredential,
     VisualStudioCodeCredential,
 )
@@ -30,6 +31,7 @@
     "DefaultAzureCredential",
     "EnvironmentCredential",
     "ManagedIdentityCredential",
+    "OnBehalfOfCredential",
     "ChainedTokenCredential",
     "SharedTokenCacheCredential",
     "VisualStudioCodeCredential",
diff --git a/sdk/identity/azure-identity/azure/identity/aio/_credentials/__init__.py b/sdk/identity/azure-identity/azure/identity/aio/_credentials/__init__.py
index 5ddfe6360133..62aad87e0a63 100644
--- a/sdk/identity/azure-identity/azure/identity/aio/_credentials/__init__.py
+++ b/sdk/identity/azure-identity/azure/identity/aio/_credentials/__init__.py
@@ -9,6 +9,7 @@
 from .default import DefaultAzureCredential
 from .environment import EnvironmentCredential
 from .managed_identity import ManagedIdentityCredential
+from .on_behalf_of import OnBehalfOfCredential
 from .certificate import CertificateCredential
 from .client_secret import ClientSecretCredential
 from .shared_cache import SharedTokenCacheCredential
@@ -27,6 +28,7 @@
     "DefaultAzureCredential",
     "EnvironmentCredential",
     "ManagedIdentityCredential",
+    "OnBehalfOfCredential",
     "SharedTokenCacheCredential",
     "VisualStudioCodeCredential",
 ]
diff --git a/sdk/identity/azure-identity/azure/identity/aio/_credentials/on_behalf_of.py b/sdk/identity/azure-identity/azure/identity/aio/_credentials/on_behalf_of.py
new file mode 100644
index 000000000000..bc7c75f2123a
--- /dev/null
+++ b/sdk/identity/azure-identity/azure/identity/aio/_credentials/on_behalf_of.py
@@ -0,0 +1,107 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+import logging
+from typing import TYPE_CHECKING
+
+from azure.core.exceptions import ClientAuthenticationError
+
+from .._internal import AadClient, AsyncContextManager
+from .._internal.get_token_mixin import GetTokenMixin
+from ..._credentials.certificate import get_client_credential
+from ..._internal import AadClientCertificate, validate_tenant_id
+
+if TYPE_CHECKING:
+    from typing import Any, Optional, Union
+    from azure.core.credentials import AccessToken
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class OnBehalfOfCredential(AsyncContextManager, GetTokenMixin):
+    """Authenticates a service principal via the on-behalf-of flow.
+
+    This flow is typically used by middle-tier services that authorize requests to other services with a delegated
+    user identity. Because this is not an interactive authentication flow, an application using it must have admin
+    consent for any delegated permissions before requesting tokens for them. See `Azure Active Directory documentation
+    <https://docs.microsoft.com/azure/active-directory/develop/v2-oauth2-on-behalf-of-flow>`_ for a more detailed
+    description of the on-behalf-of flow.
+
+    :param str tenant_id: ID of the service principal's tenant. Also called its "directory" ID.
+    :param str client_id: the service principal's client ID
+    :param client_credential: a credential to authenticate the service principal, either one of its client secrets (a
+        string) or the bytes of a certificate in PEM or PKCS12 format including the private key
+    :paramtype client_credential: str or bytes
+    :param str user_assertion: the access token the credential will use as the user assertion when requesting
+        on-behalf-of tokens
+
+    :keyword bool allow_multitenant_authentication: when True, enables the credential to acquire tokens from any tenant
+        the application is registered in. When False, which is the default, the credential will acquire tokens only
+        from the tenant specified by **tenant_id**.
+    :keyword str authority: Authority of an Azure Active Directory endpoint, for example "login.microsoftonline.com",
+        the authority for Azure Public Cloud (which is the default). :class:`~azure.identity.AzureAuthorityHosts`
+        defines authorities for other clouds.
+    :keyword password: a certificate password. Used only when **client_credential** is certificate bytes. If this value
+        is a unicode string, it will be encoded as UTF-8. If the certificate requires a different encoding, pass
+        appropriately encoded bytes instead.
+    :paramtype password: str or bytes
+    """
+
+    def __init__(
+        self,
+        tenant_id: str,
+        client_id: str,
+        client_credential: "Union[bytes, str]",
+        user_assertion: str,
+        **kwargs: "Any"
+    ) -> None:
+        super().__init__()
+        validate_tenant_id(tenant_id)
+
+        if isinstance(client_credential, bytes):
+            try:
+                cert = get_client_credential(None, kwargs.pop("password", None), client_credential)
+            except ValueError as ex:
+                message = (
+                    '"client_credential" should be either a client secret (a string)'
+                    + " or the bytes of a certificate in PEM or PKCS12 format"
+                )
+                raise ValueError(message) from ex
+            self._client_credential = AadClientCertificate(
+                cert["private_key"], password=cert.get("passphrase")
+            )  # type: Union[str, AadClientCertificate]
+        else:
+            self._client_credential = client_credential
+
+        # note AadClient handles "allow_multitenant_authentication", "authority", and any pipeline kwargs
+        self._client = AadClient(tenant_id, client_id, **kwargs)
+        self._assertion = user_assertion
+
+    async def __aenter__(self):
+        await self._client.__aenter__()
+        return self
+
+    async def close(self):
+        await self._client.close()
+
+    async def _acquire_token_silently(self, *scopes: str, **kwargs: "Any") -> "Optional[AccessToken]":
+        return self._client.get_cached_access_token(scopes, **kwargs)
+
+    async def _request_token(self, *scopes: str, **kwargs: "Any") -> "AccessToken":
+        # Note we assume the cache has tokens for one user only. That's okay because each instance of this class is
+        # locked to a single user (assertion). This assumption will become unsafe if this class allows applications
+        # to change an instance's assertion.
+        refresh_tokens = self._client.get_cached_refresh_tokens(scopes)
+        if len(refresh_tokens) == 1:  # there should be only one
+            try:
+                refresh_token = refresh_tokens[0]["secret"]
+                return await self._client.obtain_token_by_refresh_token(scopes, refresh_token, **kwargs)
+            except ClientAuthenticationError as ex:
+                _LOGGER.debug("silent authentication failed: %s", ex, exc_info=True)
+            except (IndexError, KeyError, TypeError) as ex:
+                # this is purely defensive, hasn't been observed in practice
+                _LOGGER.debug("silent authentication failed due to malformed refresh token: %s", ex, exc_info=True)
+
+        # we don't have a refresh token, or silent auth failed: acquire a new token from the assertion
+        return await self._client.obtain_token_on_behalf_of(scopes, self._client_credential, self._assertion, **kwargs)
diff --git a/sdk/identity/azure-identity/azure/identity/aio/_internal/aad_client.py b/sdk/identity/azure-identity/azure/identity/aio/_internal/aad_client.py
index 44123a2c9f69..18252411e069 100644
--- a/sdk/identity/azure-identity/azure/identity/aio/_internal/aad_client.py
+++ b/sdk/identity/azure-identity/azure/identity/aio/_internal/aad_client.py
@@ -80,6 +80,20 @@ async def obtain_token_by_refresh_token(
         response = await self._pipeline.run(request, retry_on_methods=self._POST, **kwargs)
         return self._process_response(response, now)
 
+    async def obtain_token_on_behalf_of(
+        self,
+        scopes: "Iterable[str]",
+        client_credential: "Union[str, AadClientCertificate]",
+        user_assertion: str,
+        **kwargs: "Any"
+    ) -> "AccessToken":
+        request = self._get_on_behalf_of_request(
+            scopes=scopes, client_credential=client_credential, user_assertion=user_assertion, **kwargs
+        )
+        now = int(time.time())
+        response = await self._pipeline.run(request, retry_on_methods=self._POST, **kwargs)
+        return self._process_response(response, now)
+
     # pylint:disable=no-self-use
     def _build_pipeline(self, **kwargs: "Any") -> "AsyncPipeline":
         return build_async_pipeline(**kwargs)
diff --git a/sdk/identity/azure-identity/dev_requirements.txt b/sdk/identity/azure-identity/dev_requirements.txt
index 414d718794ab..71afa54f91c6 100644
--- a/sdk/identity/azure-identity/dev_requirements.txt
+++ b/sdk/identity/azure-identity/dev_requirements.txt
@@ -1,5 +1,6 @@
 ../../core/azure-core
 aiohttp>=3.0; python_version >= '3.5'
+azure-mgmt-resource>=19.0.0
 mock;python_version<"3.3"
 typing_extensions>=3.7.2
 -e ../../../tools/azure-sdk-tools
diff --git a/sdk/identity/azure-identity/samples/custom_credentials.py b/sdk/identity/azure-identity/samples/custom_credentials.py
index ff9a0ce10775..1b1248c60fcb 100644
--- a/sdk/identity/azure-identity/samples/custom_credentials.py
+++ b/sdk/identity/azure-identity/samples/custom_credentials.py
@@ -4,13 +4,9 @@
 # ------------------------------------
 """Demonstrates custom credential implementation"""
 
-import time
 from typing import TYPE_CHECKING
 
 from azure.core.credentials import AccessToken
-from azure.core.exceptions import ClientAuthenticationError
-from azure.identity import AzureAuthorityHosts
-import msal
 
 if TYPE_CHECKING:
     from typing import Any, Union
@@ -36,36 +32,3 @@ def get_token(self, *scopes, **kwargs):
         """get_token is the only method a credential must implement"""
 
         return self._token
-
-
-class OnBehalfOfCredential(object):
-    """Authenticates via the On-Behalf-Of flow using MSAL for Python
-
-    A future version of azure-identity will include a credential supporting the On-Behalf-Of flow. Until then,
-    applications needing to authenticate through that flow can use a custom credential like this one.
-    """
-
-    def __init__(self, tenant_id, client_id, client_secret, user_access_token):
-        # type: (str, str, str, str) -> None
-        self._confidential_client = msal.ConfidentialClientApplication(
-            client_id=client_id,
-            client_credential=client_secret,
-            authority="https://{}/{}".format(AzureAuthorityHosts.AZURE_PUBLIC_CLOUD, tenant_id)
-        )
-        self._user_token = user_access_token
-
-    def get_token(self, *scopes, **kwargs):
-        # type: (*str, **Any) -> AccessToken
-        """get_token is the only method a credential must implement"""
-
-        now = int(time.time())
-        result = self._confidential_client.acquire_token_on_behalf_of(
-            user_assertion=self._user_token, scopes=list(scopes)
-        )
-
-        if result and "access_token" in result and "expires_in" in result:
-            return AccessToken(result["access_token"], now + int(result["expires_in"]))
-
-        raise ClientAuthenticationError(
-            message="Authentication failed: {}".format(result.get("error_description") or result.get("error"))
-        )
diff --git a/sdk/identity/azure-identity/tests/helpers.py b/sdk/identity/azure-identity/tests/helpers.py
index b1be22602c72..8b8c1ecc46da 100644
--- a/sdk/identity/azure-identity/tests/helpers.py
+++ b/sdk/identity/azure-identity/tests/helpers.py
@@ -14,6 +14,9 @@
     import mock  # type: ignore
 
 
+FAKE_CLIENT_ID = "fake-client-id"
+
+
 def build_id_token(
     iss="issuer",
     sub="subject",
diff --git a/sdk/identity/azure-identity/tests/recorded_test_case.py b/sdk/identity/azure-identity/tests/recorded_test_case.py
index 18a7a5182480..e56316a2fac4 100644
--- a/sdk/identity/azure-identity/tests/recorded_test_case.py
+++ b/sdk/identity/azure-identity/tests/recorded_test_case.py
@@ -8,7 +8,7 @@
 from devtools_testutils.azure_testcase import AzureTestCase
 import pytest
 
-from recording_processors import RecordingRedactor
+from recording_processors import IdTokenProcessor, RecordingRedactor
 
 PLAYBACK_CLIENT_ID = "client-id"
 
@@ -19,7 +19,7 @@ def __init__(self, *args, **kwargs):
         super(RecordedTestCase, self).__init__(
             *args,
             recording_processors=[RecordingRedactor(), scrubber],
-            replay_processors=[RequestUrlNormalizer()],
+            replay_processors=[IdTokenProcessor(), RequestUrlNormalizer()],
             **kwargs
         )
         self.scrubber = scrubber
diff --git a/sdk/identity/azure-identity/tests/recording_processors.py b/sdk/identity/azure-identity/tests/recording_processors.py
index e0cd4058606d..3ddba5cbd7d7 100644
--- a/sdk/identity/azure-identity/tests/recording_processors.py
+++ b/sdk/identity/azure-identity/tests/recording_processors.py
@@ -11,6 +11,9 @@
 from azure_devtools.scenario_tests import RecordingProcessor
 import six
 
+from helpers import FAKE_CLIENT_ID
+
+ID_TOKEN_PII_CLAIMS = ("email", "name", "preferred_username", "unique_name")
 
 SECRET_FIELDS = frozenset(
     {
@@ -34,6 +37,19 @@
 )
 
 
+def set_jwt_claims(jwt, claims):
+    header, encoded_payload, signed = jwt.split(".")
+    decoded_payload = base64.b64decode(encoded_payload + "=" * (-len(encoded_payload) % 4))
+
+    payload = json.loads(six.ensure_str(decoded_payload))
+    for name, value in claims:
+        if name in payload:
+            payload[name] = value
+
+    new_payload = six.ensure_binary(json.dumps(payload))
+    return ".".join((header, base64.b64encode(new_payload).decode("utf-8"), signed))
+
+
 class RecordingRedactor(RecordingProcessor):
     """Removes authentication secrets from recordings.
 
@@ -62,7 +78,10 @@ def process_response(self, response):
             return response
 
         for field in body:
-            if field in SECRET_FIELDS:
+            if field == "id_token":
+                scrubbed = set_jwt_claims(body["id_token"], [(claim, "redacted") for claim in ID_TOKEN_PII_CLAIMS])
+                body["id_token"] = scrubbed
+            elif field in SECRET_FIELDS:
                 fake_value = self._get_fake_value(body[field])
                 body[field] = fake_value
 
@@ -79,23 +98,11 @@ def _get_fake_value(self, real_value):
 
 class IdTokenProcessor(RecordingProcessor):
     def process_response(self, response):
-        """Changes the "exp" claim of recorded id tokens to be in the future during playback
-
-        This is necessary because msal always validates id tokens, raising an exception when they've expired.
-        """
+        """Modifies an id token's claims to pass MSAL validation during playback"""
         try:
-            # decode the recorded token
             body = json.loads(six.ensure_str(response["body"]["string"]))
-            header, encoded_payload, signed = body["id_token"].split(".")
-            decoded_payload = base64.b64decode(encoded_payload + "=" * (4 - len(encoded_payload) % 4))
-
-            # set the token's expiry time to one hour from now
-            payload = json.loads(six.ensure_str(decoded_payload))
-            payload["exp"] = int(time.time()) + 3600
-
-            # write the modified token to the response body
-            new_payload = six.ensure_binary(json.dumps(payload))
-            body["id_token"] = ".".join((header, base64.b64encode(new_payload).decode("utf-8"), signed))
+            new_jwt = set_jwt_claims(body["id_token"], [("exp", int(time.time()) + 3600), ("aud", FAKE_CLIENT_ID)])
+            body["id_token"] = new_jwt
             response["body"]["string"] = six.ensure_binary(json.dumps(body))
         except KeyError:
             pass
diff --git a/sdk/identity/azure-identity/tests/recordings/test_obo.test_obo.yaml b/sdk/identity/azure-identity/tests/recordings/test_obo.test_obo.yaml
new file mode 100644
index 000000000000..d604b350b79e
--- /dev/null
+++ b/sdk/identity/azure-identity/tests/recordings/test_obo.test_obo.yaml
@@ -0,0 +1,359 @@
+interactions:
+- request:
+    body: null
+    headers:
+      Accept:
+      - '*/*'
+      Accept-Encoding:
+      - gzip, deflate
+      Connection:
+      - keep-alive
+      User-Agent:
+      - azsdk-python-identity/1.7.0b4 Python/2.7.17 (Linux-5.4.72-microsoft-standard-WSL2-x86_64-with-Ubuntu-20.04-focal)
+    method: GET
+    uri: https://login.microsoftonline.com/tenant/v2.0/.well-known/openid-configuration
+  response:
+    body:
+      string: !!python/unicode '{"msgraph_host": "graph.microsoft.com", "claims_supported":
+        ["sub", "iss", "cloud_instance_name", "cloud_instance_host_name", "cloud_graph_host_name",
+        "msgraph_host", "aud", "exp", "iat", "auth_time", "acr", "nonce", "preferred_username",
+        "name", "tid", "ver", "at_hash", "c_hash", "email"], "subject_types_supported":
+        ["pairwise"], "issuer": "https://login.microsoftonline.com/tenant/v2.0", "cloud_graph_host_name":
+        "graph.windows.net", "token_endpoint": "https://login.microsoftonline.com/tenant/oauth2/v2.0/token",
+        "request_uri_parameter_supported": false, "http_logout_supported": true, "response_modes_supported":
+        ["query", "fragment", "form_post"], "jwks_uri": "https://login.microsoftonline.com/tenant/discovery/v2.0/keys",
+        "scopes_supported": ["openid", "profile", "email", "offline_access"], "token_endpoint_auth_methods_supported":
+        ["client_secret_post", "private_key_jwt", "client_secret_basic"], "id_token_signing_alg_values_supported":
+        ["RS256"], "tenant_region_scope": "NA", "rbac_url": "https://pas.windows.net",
+        "userinfo_endpoint": "https://graph.microsoft.com/oidc/userinfo", "device_authorization_endpoint":
+        "https://login.microsoftonline.com/tenant/oauth2/v2.0/devicecode", "kerberos_endpoint":
+        "https://login.microsoftonline.com/tenant/kerberos", "frontchannel_logout_supported":
+        true, "response_types_supported": ["code", "id_token", "code id_token", "id_token
+        token"], "end_session_endpoint": "https://login.microsoftonline.com/tenant/oauth2/v2.0/logout",
+        "authorization_endpoint": "https://login.microsoftonline.com/tenant/oauth2/v2.0/authorize",
+        "cloud_instance_name": "microsoftonline.com"}'
+    headers:
+      access-control-allow-methods:
+      - GET, OPTIONS
+      access-control-allow-origin:
+      - '*'
+      cache-control:
+      - max-age=86400, private
+      content-length:
+      - '1753'
+      content-type:
+      - application/json; charset=utf-8
+      date:
+      - Tue, 31 Aug 2021 16:25:25 GMT
+      p3p:
+      - CP="DSP CUR OTPi IND OTRi ONL FIN"
+      set-cookie:
+      - fpc=Apf8i1HQc7xLkQJZ0dZ9cOE; expires=Thu, 30-Sep-2021 16:25:26 GMT; path=/;
+        secure; HttpOnly; SameSite=None
+      - esctx=AQABAAAAAAD--DLA3VO7QrddgJg7WevrFtaYmN647hZNghQsL-lUtuj_p_Sr8EotifV35mWTMgAYV0tY4tunU7fRoHpwptzUk4NHMXlPcGXGzqhyqYexqIpmUtQNqtJZtwTjiBtboSM12E2-QQKWlVBnm4BScZMi2mwn238IZiQUIScNU7ifJKqHPDUM5fxkB-YHghWw8lwgAA;
+        domain=.login.microsoftonline.com; path=/; secure; HttpOnly; SameSite=None
+      - x-ms-gateway-slice=estsfd; path=/; secure; samesite=none; httponly
+      - stsservicecookie=estsfd; path=/; secure; samesite=none; httponly
+      strict-transport-security:
+      - max-age=31536000; includeSubDomains
+      x-content-type-options:
+      - nosniff
+      x-ms-ests-server:
+      - 2.1.11961.11 - SCUS ProdSlices
+    status:
+      code: 200
+      message: OK
+- request:
+    body: null
+    headers:
+      Accept:
+      - application/json
+      Accept-Encoding:
+      - gzip, deflate
+      Connection:
+      - keep-alive
+      Cookie:
+      - stsservicecookie=estsfd; fpc=Apf8i1HQc7xLkQJZ0dZ9cOE; x-ms-gateway-slice=estsfd;
+        esctx=AQABAAAAAAD--DLA3VO7QrddgJg7WevrFtaYmN647hZNghQsL-lUtuj_p_Sr8EotifV35mWTMgAYV0tY4tunU7fRoHpwptzUk4NHMXlPcGXGzqhyqYexqIpmUtQNqtJZtwTjiBtboSM12E2-QQKWlVBnm4BScZMi2mwn238IZiQUIScNU7ifJKqHPDUM5fxkB-YHghWw8lwgAA
+      User-Agent:
+      - azsdk-python-identity/1.7.0b4 Python/2.7.17 (Linux-5.4.72-microsoft-standard-WSL2-x86_64-with-Ubuntu-20.04-focal)
+    method: GET
+    uri: https://login.microsoftonline.com/common/userrealm/username?api-version=1.0
+  response:
+    body:
+      string: !!python/unicode '{"account_type": "Managed", "cloud_audience_urn":
+        "urn:federation:MicrosoftOnline", "ver": "1.0", "domain_name": "chlowehotmail.onmicrosoft.com",
+        "cloud_instance_name": "microsoftonline.com"}'
+    headers:
+      cache-control:
+      - no-store, no-cache
+      content-disposition:
+      - inline; filename=userrealm.json
+      content-length:
+      - '182'
+      content-type:
+      - application/json; charset=utf-8
+      date:
+      - Tue, 31 Aug 2021 16:25:26 GMT
+      expires:
+      - '-1'
+      p3p:
+      - CP="DSP CUR OTPi IND OTRi ONL FIN"
+      pragma:
+      - no-cache
+      set-cookie:
+      - fpc=Apf8i1HQc7xLkQJZ0dZ9cOE; expires=Thu, 30-Sep-2021 16:25:26 GMT; path=/;
+        secure; HttpOnly; SameSite=None
+      - x-ms-gateway-slice=estsfd; path=/; secure; samesite=none; httponly
+      - stsservicecookie=estsfd; path=/; secure; samesite=none; httponly
+      strict-transport-security:
+      - max-age=31536000; includeSubDomains
+      x-content-type-options:
+      - nosniff
+      x-ms-ests-server:
+      - 2.1.11984.12 - WUS2 ProdSlices
+    status:
+      code: 200
+      message: OK
+- request:
+    body: null
+    headers:
+      Accept:
+      - application/json
+      Accept-Encoding:
+      - gzip, deflate
+      Connection:
+      - keep-alive
+      Content-Length:
+      - '304'
+      Content-Type:
+      - application/x-www-form-urlencoded
+      Cookie:
+      - stsservicecookie=estsfd; fpc=Apf8i1HQc7xLkQJZ0dZ9cOE; x-ms-gateway-slice=estsfd;
+        esctx=AQABAAAAAAD--DLA3VO7QrddgJg7WevrFtaYmN647hZNghQsL-lUtuj_p_Sr8EotifV35mWTMgAYV0tY4tunU7fRoHpwptzUk4NHMXlPcGXGzqhyqYexqIpmUtQNqtJZtwTjiBtboSM12E2-QQKWlVBnm4BScZMi2mwn238IZiQUIScNU7ifJKqHPDUM5fxkB-YHghWw8lwgAA
+      User-Agent:
+      - azsdk-python-identity/1.7.0b4 Python/2.7.17 (Linux-5.4.72-microsoft-standard-WSL2-x86_64-with-Ubuntu-20.04-focal)
+      x-client-cpu:
+      - x64
+      x-client-current-telemetry:
+      - 4|301,0|
+      x-client-last-telemetry:
+      - 4|0|||
+      x-client-os:
+      - linux2
+      x-client-sku:
+      - MSAL.Python
+      x-client-ver:
+      - 1.12.0
+    method: POST
+    uri: https://login.microsoftonline.com/tenant/oauth2/v2.0/token
+  response:
+    body:
+      string: !!python/unicode '{"access_token": "redacted", "ext_expires_in": 3599,
+        "expires_in": 3599, "client_info": "eyJ1aWQiOiJjZGQ2ZWUyNi1lNGJhLTQyNGUtODJlNS1lN2ZhNzc4M2I1ODAiLCJ1dGlkIjoiM2M2MzFiYjctYTlmNy00MzQzLWE1YmEtYTYxNTkxMzVmMWZjIn0",
+        "token_type": "Bearer", "id_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiIsImtpZCI6Im5PbzNaRHJPRFhFSzFqS1doWHNsSFJfS1hFZyJ9.eyJzdWIiOiAiREVlWDVqRGVVOXQ1NFpCaVZ1SnZLZ3BUUTdHLWFrNi1qYnRmbmh6dURZVSIsICJ1dGkiOiAiWkNUcHdiNldIVUMwUmhTd3ljSXhBQSIsICJ2ZXIiOiAiMi4wIiwgIm5hbWUiOiAicmVkYWN0ZWQiLCAiaXNzIjogImh0dHBzOi8vbG9naW4ubWljcm9zb2Z0b25saW5lLmNvbS8zYzYzMWJiNy1hOWY3LTQzNDMtYTViYS1hNjE1OTEzNWYxZmMvdjIuMCIsICJvaWQiOiAiY2RkNmVlMjYtZTRiYS00MjRlLTgyZTUtZTdmYTc3ODNiNTgwIiwgInByZWZlcnJlZF91c2VybmFtZSI6ICJyZWRhY3RlZCIsICJleHAiOiAxNjMwNDMwNzI2LCAidGlkIjogIjNjNjMxYmI3LWE5ZjctNDM0My1hNWJhLWE2MTU5MTM1ZjFmYyIsICJpYXQiOiAxNjMwNDI2ODI2LCAicmgiOiAiMC5BUmNBdHh0alBQZXBRME9sdXFZVmtUWHhfSzdGRDlma1VwVkl2eHVKUVFQWDd4UVhBUHMuIiwgIm5iZiI6IDE2MzA0MjY4MjYsICJhdWQiOiAiZDcwZmM1YWUtNTJlNC00ODk1LWJmMWItODk0MTAzZDdlZjE0In0=.g3g9G4qF6_wwrHTQfsaLHulF_ok8m3lxBOEFw6LJrq4PsTBSGS7MBVhR9rrJUVmz9wpuou9Uq682ZuEfbvRFDd0g-rzCZ6Jbe1KoRGkjxXm3eUdadyFwa_BzcGmzFpkyiPZPuVja844x7d1z8AVhdV28Sjs5uPDz5ornGW8iMgKKD4kCHmVlFcKrFrDmzWFhWViHUPZDAO7YycCv1apEIcTHC8uNmQm78vyssAF9nv2KIYwLmzCMGk38S7JIEjX5ZbHYjO7XXv8YsPiLlZx1cGytdTdGgtZx6afttHsR6M8y9L_9Bezx3J1SI54UMF7-KRxQBQSVB_6PLUm6azwKtg",
+        "scope": "api://foo/Foo", "refresh_token": "redacted"}'
+    headers:
+      cache-control:
+      - no-store, no-cache
+      content-length:
+      - '3765'
+      content-type:
+      - application/json; charset=utf-8
+      date:
+      - Tue, 31 Aug 2021 16:25:26 GMT
+      expires:
+      - '-1'
+      p3p:
+      - CP="DSP CUR OTPi IND OTRi ONL FIN"
+      pragma:
+      - no-cache
+      set-cookie:
+      - fpc=Apf8i1HQc7xLkQJZ0dZ9cOHArTo4AQAAAPZOwNgOAAAA; expires=Thu, 30-Sep-2021
+        16:25:26 GMT; path=/; secure; HttpOnly; SameSite=None
+      - x-ms-gateway-slice=estsfd; path=/; secure; samesite=none; httponly
+      - stsservicecookie=estsfd; path=/; secure; samesite=none; httponly
+      strict-transport-security:
+      - max-age=31536000; includeSubDomains
+      x-content-type-options:
+      - nosniff
+      x-ms-clitelem:
+      - 1,0,0,,
+      x-ms-ests-server:
+      - 2.1.11984.12 - WUS2 ProdSlices
+    status:
+      code: 200
+      message: OK
+- request:
+    body: null
+    headers:
+      Accept:
+      - '*/*'
+      Accept-Encoding:
+      - gzip, deflate
+      Connection:
+      - keep-alive
+      User-Agent:
+      - azsdk-python-identity/1.7.0b4 Python/2.7.17 (Linux-5.4.72-microsoft-standard-WSL2-x86_64-with-Ubuntu-20.04-focal)
+    method: GET
+    uri: https://login.microsoftonline.com/tenant/v2.0/.well-known/openid-configuration
+  response:
+    body:
+      string: !!python/unicode '{"msgraph_host": "graph.microsoft.com", "claims_supported":
+        ["sub", "iss", "cloud_instance_name", "cloud_instance_host_name", "cloud_graph_host_name",
+        "msgraph_host", "aud", "exp", "iat", "auth_time", "acr", "nonce", "preferred_username",
+        "name", "tid", "ver", "at_hash", "c_hash", "email"], "subject_types_supported":
+        ["pairwise"], "issuer": "https://login.microsoftonline.com/tenant/v2.0", "cloud_graph_host_name":
+        "graph.windows.net", "token_endpoint": "https://login.microsoftonline.com/tenant/oauth2/v2.0/token",
+        "request_uri_parameter_supported": false, "http_logout_supported": true, "response_modes_supported":
+        ["query", "fragment", "form_post"], "jwks_uri": "https://login.microsoftonline.com/tenant/discovery/v2.0/keys",
+        "scopes_supported": ["openid", "profile", "email", "offline_access"], "token_endpoint_auth_methods_supported":
+        ["client_secret_post", "private_key_jwt", "client_secret_basic"], "id_token_signing_alg_values_supported":
+        ["RS256"], "tenant_region_scope": "NA", "rbac_url": "https://pas.windows.net",
+        "userinfo_endpoint": "https://graph.microsoft.com/oidc/userinfo", "device_authorization_endpoint":
+        "https://login.microsoftonline.com/tenant/oauth2/v2.0/devicecode", "kerberos_endpoint":
+        "https://login.microsoftonline.com/tenant/kerberos", "frontchannel_logout_supported":
+        true, "response_types_supported": ["code", "id_token", "code id_token", "id_token
+        token"], "end_session_endpoint": "https://login.microsoftonline.com/tenant/oauth2/v2.0/logout",
+        "authorization_endpoint": "https://login.microsoftonline.com/tenant/oauth2/v2.0/authorize",
+        "cloud_instance_name": "microsoftonline.com"}'
+    headers:
+      access-control-allow-methods:
+      - GET, OPTIONS
+      access-control-allow-origin:
+      - '*'
+      cache-control:
+      - max-age=86400, private
+      content-length:
+      - '1753'
+      content-type:
+      - application/json; charset=utf-8
+      date:
+      - Tue, 31 Aug 2021 16:25:26 GMT
+      p3p:
+      - CP="DSP CUR OTPi IND OTRi ONL FIN"
+      set-cookie:
+      - fpc=At6tqNbf9WJJu6fvgY8Tue4; expires=Thu, 30-Sep-2021 16:25:27 GMT; path=/;
+        secure; HttpOnly; SameSite=None
+      - esctx=AQABAAAAAAD--DLA3VO7QrddgJg7WevrmaDcAL78s8Hk8NXHRi2UdmdsMNOTXjMFrea8E_7rywhRvtVIPMl-5bozcv7j0hOt8rdQRk1iAlhnL_JWFb3B5CvFSN9h-EL5d4LTZpMz1X-dXL1eUKjtTh4m6AAWfuhmBGgjnBVksvRp6TfCPaHgRF15F6uDBkKjhcOE7GYJGcsgAA;
+        domain=.login.microsoftonline.com; path=/; secure; HttpOnly; SameSite=None
+      - x-ms-gateway-slice=estsfd; path=/; secure; samesite=none; httponly
+      - stsservicecookie=estsfd; path=/; secure; samesite=none; httponly
+      strict-transport-security:
+      - max-age=31536000; includeSubDomains
+      x-content-type-options:
+      - nosniff
+      x-ms-ests-server:
+      - 2.1.11961.11 - SCUS ProdSlices
+    status:
+      code: 200
+      message: OK
+- request:
+    body: null
+    headers:
+      Accept:
+      - application/json
+      Accept-Encoding:
+      - gzip, deflate
+      Connection:
+      - keep-alive
+      Content-Length:
+      - '1825'
+      Content-Type:
+      - application/x-www-form-urlencoded
+      Cookie:
+      - stsservicecookie=estsfd; fpc=At6tqNbf9WJJu6fvgY8Tue4; x-ms-gateway-slice=estsfd;
+        esctx=AQABAAAAAAD--DLA3VO7QrddgJg7WevrmaDcAL78s8Hk8NXHRi2UdmdsMNOTXjMFrea8E_7rywhRvtVIPMl-5bozcv7j0hOt8rdQRk1iAlhnL_JWFb3B5CvFSN9h-EL5d4LTZpMz1X-dXL1eUKjtTh4m6AAWfuhmBGgjnBVksvRp6TfCPaHgRF15F6uDBkKjhcOE7GYJGcsgAA
+      User-Agent:
+      - azsdk-python-identity/1.7.0b4 Python/2.7.17 (Linux-5.4.72-microsoft-standard-WSL2-x86_64-with-Ubuntu-20.04-focal)
+      x-client-cpu:
+      - x64
+      x-client-current-telemetry:
+      - 4|523,0|
+      x-client-last-telemetry:
+      - 4|0|||
+      x-client-os:
+      - linux2
+      x-client-sku:
+      - MSAL.Python
+      x-client-ver:
+      - 1.12.0
+    method: POST
+    uri: https://login.microsoftonline.com/tenant/oauth2/v2.0/token
+  response:
+    body:
+      string: !!python/unicode '{"access_token": "redacted", "ext_expires_in": 3597,
+        "expires_in": 3597, "client_info": "eyJ1aWQiOiJjZGQ2ZWUyNi1lNGJhLTQyNGUtODJlNS1lN2ZhNzc4M2I1ODAiLCJ1dGlkIjoiM2M2MzFiYjctYTlmNy00MzQzLWE1YmEtYTYxNTkxMzVmMWZjIn0",
+        "token_type": "Bearer", "id_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiIsImtpZCI6Im5PbzNaRHJPRFhFSzFqS1doWHNsSFJfS1hFZyJ9.eyJzdWIiOiAiREVlWDVqRGVVOXQ1NFpCaVZ1SnZLZ3BUUTdHLWFrNi1qYnRmbmh6dURZVSIsICJ1dGkiOiAiNkNLQVNpVzJpMEtHSWtyOUpvaV9BQSIsICJ2ZXIiOiAiMi4wIiwgIm5hbWUiOiAicmVkYWN0ZWQiLCAiaXNzIjogImh0dHBzOi8vbG9naW4ubWljcm9zb2Z0b25saW5lLmNvbS8zYzYzMWJiNy1hOWY3LTQzNDMtYTViYS1hNjE1OTEzNWYxZmMvdjIuMCIsICJvaWQiOiAiY2RkNmVlMjYtZTRiYS00MjRlLTgyZTUtZTdmYTc3ODNiNTgwIiwgInByZWZlcnJlZF91c2VybmFtZSI6ICJyZWRhY3RlZCIsICJleHAiOiAxNjMwNDMwNzI1LCAidGlkIjogIjNjNjMxYmI3LWE5ZjctNDM0My1hNWJhLWE2MTU5MTM1ZjFmYyIsICJpYXQiOiAxNjMwNDI2ODI3LCAicmgiOiAiMC5BUmNBdHh0alBQZXBRME9sdXFZVmtUWHhfSzdGRDlma1VwVkl2eHVKUVFQWDd4UVhBUHMuIiwgIm5iZiI6IDE2MzA0MjY4MjcsICJhdWQiOiAiZDcwZmM1YWUtNTJlNC00ODk1LWJmMWItODk0MTAzZDdlZjE0In0=.f9i12I0RIoIRG-9D9nKg3nHMM9oFW5a4RypnuUIM7h_wvSHced_8-Oa2sPk8KhvuWYQbkn_bMjoLP7yiAH0T-3WzP3cGIu4xUtt91v9P5oPRXBuA5yZtA5D_785W1L7zGz5k64i1D1190S1v84Hm4PfO9vR9sGSk1WtbTZSG-N5LR9xCMj8LpAMiamlBctVAn4KtoZbEYVOkg8_MWQOIAEUg8b0I22FbzjXNPadnn7vv61AzI1jwNBrEtX2IE7crmLg0LS3BvZ3CDlnZN6bkbojBRfSjFbfToSW98h94lhqjCRTbV4yJNn006edEaQ-xjYJu9B9lvJePpJNzQEawyg",
+        "scope": "https://management.azure.com/user_impersonation https://management.azure.com/.default",
+        "refresh_token": "redacted"}'
+    headers:
+      cache-control:
+      - no-store, no-cache
+      content-length:
+      - '4196'
+      content-type:
+      - application/json; charset=utf-8
+      date:
+      - Tue, 31 Aug 2021 16:25:26 GMT
+      expires:
+      - '-1'
+      p3p:
+      - CP="DSP CUR OTPi IND OTRi ONL FIN"
+      pragma:
+      - no-cache
+      set-cookie:
+      - fpc=At6tqNbf9WJJu6fvgY8Tue70CaNgAQAAAPZOwNgOAAAA; expires=Thu, 30-Sep-2021
+        16:25:27 GMT; path=/; secure; HttpOnly; SameSite=None
+      - x-ms-gateway-slice=estsfd; path=/; secure; samesite=none; httponly
+      - stsservicecookie=estsfd; path=/; secure; samesite=none; httponly
+      strict-transport-security:
+      - max-age=31536000; includeSubDomains
+      x-content-type-options:
+      - nosniff
+      x-ms-clitelem:
+      - 1,0,0,,
+      x-ms-ests-server:
+      - 2.1.11961.11 - SCUS ProdSlices
+    status:
+      code: 200
+      message: OK
+- request:
+    body: null
+    headers:
+      Accept:
+      - application/json
+      Accept-Encoding:
+      - gzip, deflate
+      Connection:
+      - keep-alive
+      User-Agent:
+      - azsdk-python-azure-mgmt-resource/16.1.0 Python/2.7.17 (Linux-5.4.72-microsoft-standard-WSL2-x86_64-with-Ubuntu-20.04-focal)
+    method: GET
+    uri: https://management.azure.com/subscriptions?api-version=2019-11-01
+  response:
+    body:
+      string: !!python/unicode '{"value":[]}'
+    headers:
+      cache-control:
+      - no-cache
+      content-length:
+      - '432'
+      content-type:
+      - application/json; charset=utf-8
+      date:
+      - Tue, 31 Aug 2021 16:25:27 GMT
+      expires:
+      - '-1'
+      pragma:
+      - no-cache
+      strict-transport-security:
+      - max-age=31536000; includeSubDomains
+      vary:
+      - Accept-Encoding
+      x-content-type-options:
+      - nosniff
+    status:
+      code: 200
+      message: OK
+version: 1
diff --git a/sdk/identity/azure-identity/tests/recordings/test_obo.test_obo_cert.yaml b/sdk/identity/azure-identity/tests/recordings/test_obo.test_obo_cert.yaml
new file mode 100644
index 000000000000..96120dad6074
--- /dev/null
+++ b/sdk/identity/azure-identity/tests/recordings/test_obo.test_obo_cert.yaml
@@ -0,0 +1,359 @@
+interactions:
+- request:
+    body: null
+    headers:
+      Accept:
+      - '*/*'
+      Accept-Encoding:
+      - gzip, deflate
+      Connection:
+      - keep-alive
+      User-Agent:
+      - azsdk-python-identity/1.7.0b4 Python/2.7.17 (Linux-5.4.72-microsoft-standard-WSL2-x86_64-with-Ubuntu-20.04-focal)
+    method: GET
+    uri: https://login.microsoftonline.com/tenant/v2.0/.well-known/openid-configuration
+  response:
+    body:
+      string: !!python/unicode '{"msgraph_host": "graph.microsoft.com", "claims_supported":
+        ["sub", "iss", "cloud_instance_name", "cloud_instance_host_name", "cloud_graph_host_name",
+        "msgraph_host", "aud", "exp", "iat", "auth_time", "acr", "nonce", "preferred_username",
+        "name", "tid", "ver", "at_hash", "c_hash", "email"], "subject_types_supported":
+        ["pairwise"], "issuer": "https://login.microsoftonline.com/tenant/v2.0", "cloud_graph_host_name":
+        "graph.windows.net", "token_endpoint": "https://login.microsoftonline.com/tenant/oauth2/v2.0/token",
+        "request_uri_parameter_supported": false, "http_logout_supported": true, "response_modes_supported":
+        ["query", "fragment", "form_post"], "jwks_uri": "https://login.microsoftonline.com/tenant/discovery/v2.0/keys",
+        "scopes_supported": ["openid", "profile", "email", "offline_access"], "token_endpoint_auth_methods_supported":
+        ["client_secret_post", "private_key_jwt", "client_secret_basic"], "id_token_signing_alg_values_supported":
+        ["RS256"], "tenant_region_scope": "NA", "rbac_url": "https://pas.windows.net",
+        "userinfo_endpoint": "https://graph.microsoft.com/oidc/userinfo", "device_authorization_endpoint":
+        "https://login.microsoftonline.com/tenant/oauth2/v2.0/devicecode", "kerberos_endpoint":
+        "https://login.microsoftonline.com/tenant/kerberos", "frontchannel_logout_supported":
+        true, "response_types_supported": ["code", "id_token", "code id_token", "id_token
+        token"], "end_session_endpoint": "https://login.microsoftonline.com/tenant/oauth2/v2.0/logout",
+        "authorization_endpoint": "https://login.microsoftonline.com/tenant/oauth2/v2.0/authorize",
+        "cloud_instance_name": "microsoftonline.com"}'
+    headers:
+      access-control-allow-methods:
+      - GET, OPTIONS
+      access-control-allow-origin:
+      - '*'
+      cache-control:
+      - max-age=86400, private
+      content-length:
+      - '1753'
+      content-type:
+      - application/json; charset=utf-8
+      date:
+      - Tue, 31 Aug 2021 16:25:27 GMT
+      p3p:
+      - CP="DSP CUR OTPi IND OTRi ONL FIN"
+      set-cookie:
+      - fpc=AlRn0BAECQZLn7zGBZWWwxs; expires=Thu, 30-Sep-2021 16:25:27 GMT; path=/;
+        secure; HttpOnly; SameSite=None
+      - esctx=AQABAAAAAAD--DLA3VO7QrddgJg7Wevrth-z-ExnkLrQcl-fQWU-pmCdD2cBu972IX985_iVDbUl1iJm6QIvIgXvkm4qRtkGLHDnfcYq3Ifp63AU6kPqJB39NvlOzWdEFYAf6FQZvevr9C7pq_zHecZiCfubiiZmFoz5qgX4bgZqYqnJwR2dIkqwEY0SeZb-HJR4AIQX8G4gAA;
+        domain=.login.microsoftonline.com; path=/; secure; HttpOnly; SameSite=None
+      - x-ms-gateway-slice=estsfd; path=/; secure; samesite=none; httponly
+      - stsservicecookie=estsfd; path=/; secure; samesite=none; httponly
+      strict-transport-security:
+      - max-age=31536000; includeSubDomains
+      x-content-type-options:
+      - nosniff
+      x-ms-ests-server:
+      - 2.1.11984.12 - WUS2 ProdSlices
+    status:
+      code: 200
+      message: OK
+- request:
+    body: null
+    headers:
+      Accept:
+      - application/json
+      Accept-Encoding:
+      - gzip, deflate
+      Connection:
+      - keep-alive
+      Cookie:
+      - stsservicecookie=estsfd; fpc=AlRn0BAECQZLn7zGBZWWwxs; x-ms-gateway-slice=estsfd;
+        esctx=AQABAAAAAAD--DLA3VO7QrddgJg7Wevrth-z-ExnkLrQcl-fQWU-pmCdD2cBu972IX985_iVDbUl1iJm6QIvIgXvkm4qRtkGLHDnfcYq3Ifp63AU6kPqJB39NvlOzWdEFYAf6FQZvevr9C7pq_zHecZiCfubiiZmFoz5qgX4bgZqYqnJwR2dIkqwEY0SeZb-HJR4AIQX8G4gAA
+      User-Agent:
+      - azsdk-python-identity/1.7.0b4 Python/2.7.17 (Linux-5.4.72-microsoft-standard-WSL2-x86_64-with-Ubuntu-20.04-focal)
+    method: GET
+    uri: https://login.microsoftonline.com/common/userrealm/username?api-version=1.0
+  response:
+    body:
+      string: !!python/unicode '{"account_type": "Managed", "cloud_audience_urn":
+        "urn:federation:MicrosoftOnline", "ver": "1.0", "domain_name": "chlowehotmail.onmicrosoft.com",
+        "cloud_instance_name": "microsoftonline.com"}'
+    headers:
+      cache-control:
+      - no-store, no-cache
+      content-disposition:
+      - inline; filename=userrealm.json
+      content-length:
+      - '182'
+      content-type:
+      - application/json; charset=utf-8
+      date:
+      - Tue, 31 Aug 2021 16:25:27 GMT
+      expires:
+      - '-1'
+      p3p:
+      - CP="DSP CUR OTPi IND OTRi ONL FIN"
+      pragma:
+      - no-cache
+      set-cookie:
+      - fpc=AlRn0BAECQZLn7zGBZWWwxs; expires=Thu, 30-Sep-2021 16:25:28 GMT; path=/;
+        secure; HttpOnly; SameSite=None
+      - x-ms-gateway-slice=estsfd; path=/; secure; samesite=none; httponly
+      - stsservicecookie=estsfd; path=/; secure; samesite=none; httponly
+      strict-transport-security:
+      - max-age=31536000; includeSubDomains
+      x-content-type-options:
+      - nosniff
+      x-ms-ests-server:
+      - 2.1.11984.12 - WUS2 ProdSlices
+    status:
+      code: 200
+      message: OK
+- request:
+    body: null
+    headers:
+      Accept:
+      - application/json
+      Accept-Encoding:
+      - gzip, deflate
+      Connection:
+      - keep-alive
+      Content-Length:
+      - '304'
+      Content-Type:
+      - application/x-www-form-urlencoded
+      Cookie:
+      - stsservicecookie=estsfd; fpc=AlRn0BAECQZLn7zGBZWWwxs; x-ms-gateway-slice=estsfd;
+        esctx=AQABAAAAAAD--DLA3VO7QrddgJg7Wevrth-z-ExnkLrQcl-fQWU-pmCdD2cBu972IX985_iVDbUl1iJm6QIvIgXvkm4qRtkGLHDnfcYq3Ifp63AU6kPqJB39NvlOzWdEFYAf6FQZvevr9C7pq_zHecZiCfubiiZmFoz5qgX4bgZqYqnJwR2dIkqwEY0SeZb-HJR4AIQX8G4gAA
+      User-Agent:
+      - azsdk-python-identity/1.7.0b4 Python/2.7.17 (Linux-5.4.72-microsoft-standard-WSL2-x86_64-with-Ubuntu-20.04-focal)
+      x-client-cpu:
+      - x64
+      x-client-current-telemetry:
+      - 4|301,0|
+      x-client-last-telemetry:
+      - 4|0|||
+      x-client-os:
+      - linux2
+      x-client-sku:
+      - MSAL.Python
+      x-client-ver:
+      - 1.12.0
+    method: POST
+    uri: https://login.microsoftonline.com/tenant/oauth2/v2.0/token
+  response:
+    body:
+      string: !!python/unicode '{"access_token": "redacted", "ext_expires_in": 3599,
+        "expires_in": 3599, "client_info": "eyJ1aWQiOiJjZGQ2ZWUyNi1lNGJhLTQyNGUtODJlNS1lN2ZhNzc4M2I1ODAiLCJ1dGlkIjoiM2M2MzFiYjctYTlmNy00MzQzLWE1YmEtYTYxNTkxMzVmMWZjIn0",
+        "token_type": "Bearer", "id_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiIsImtpZCI6Im5PbzNaRHJPRFhFSzFqS1doWHNsSFJfS1hFZyJ9.eyJzdWIiOiAiREVlWDVqRGVVOXQ1NFpCaVZ1SnZLZ3BUUTdHLWFrNi1qYnRmbmh6dURZVSIsICJ1dGkiOiAiVU1ibTJZNmJma0tOM3NzVVNxRTZBQSIsICJ2ZXIiOiAiMi4wIiwgIm5hbWUiOiAicmVkYWN0ZWQiLCAiaXNzIjogImh0dHBzOi8vbG9naW4ubWljcm9zb2Z0b25saW5lLmNvbS8zYzYzMWJiNy1hOWY3LTQzNDMtYTViYS1hNjE1OTEzNWYxZmMvdjIuMCIsICJvaWQiOiAiY2RkNmVlMjYtZTRiYS00MjRlLTgyZTUtZTdmYTc3ODNiNTgwIiwgInByZWZlcnJlZF91c2VybmFtZSI6ICJyZWRhY3RlZCIsICJleHAiOiAxNjMwNDMwNzI4LCAidGlkIjogIjNjNjMxYmI3LWE5ZjctNDM0My1hNWJhLWE2MTU5MTM1ZjFmYyIsICJpYXQiOiAxNjMwNDI2ODI4LCAicmgiOiAiMC5BUmNBdHh0alBQZXBRME9sdXFZVmtUWHhfSzdGRDlma1VwVkl2eHVKUVFQWDd4UVhBUHMuIiwgIm5iZiI6IDE2MzA0MjY4MjgsICJhdWQiOiAiZDcwZmM1YWUtNTJlNC00ODk1LWJmMWItODk0MTAzZDdlZjE0In0=.lldD44jYmDU2kNBIjdG-XPpztFqJZbPgGc6uUfa985b7OzUHiwlqaELnbKNp7VOc9qL9XvKTAnBCTDHefje2q7Tv_WQQvCjI4Zjz7GO2TaHMs63Tfwa7XBP_YVTBKRPfDEgXtkRwPFaFXPuw81UNo6ZpLMBb1Xovu_IydO7c1DFgOgzAzKH3v9EmBdcCm2x_odia_i8yzqGO59O6U_L0CjVi-AB3E08eyA1uu0qYgxOlYeBQd9u1RksPClGPOaFA0oZX6m3mZfqjX_BdS2KqZ-95gkViEz-7u3BrHSYKtxEtEk_6PESrIJAZ67Y2rtcbtZ-3vqi8_iml94g9FhaK_g",
+        "scope": "api://foo/Foo", "refresh_token": "redacted"}'
+    headers:
+      cache-control:
+      - no-store, no-cache
+      content-length:
+      - '3765'
+      content-type:
+      - application/json; charset=utf-8
+      date:
+      - Tue, 31 Aug 2021 16:25:27 GMT
+      expires:
+      - '-1'
+      p3p:
+      - CP="DSP CUR OTPi IND OTRi ONL FIN"
+      pragma:
+      - no-cache
+      set-cookie:
+      - fpc=AlRn0BAECQZLn7zGBZWWwxvArTo4AQAAAPdOwNgOAAAA; expires=Thu, 30-Sep-2021
+        16:25:28 GMT; path=/; secure; HttpOnly; SameSite=None
+      - x-ms-gateway-slice=estsfd; path=/; secure; samesite=none; httponly
+      - stsservicecookie=estsfd; path=/; secure; samesite=none; httponly
+      strict-transport-security:
+      - max-age=31536000; includeSubDomains
+      x-content-type-options:
+      - nosniff
+      x-ms-clitelem:
+      - 1,0,0,,
+      x-ms-ests-server:
+      - 2.1.11984.12 - WUS2 ProdSlices
+    status:
+      code: 200
+      message: OK
+- request:
+    body: null
+    headers:
+      Accept:
+      - '*/*'
+      Accept-Encoding:
+      - gzip, deflate
+      Connection:
+      - keep-alive
+      User-Agent:
+      - azsdk-python-identity/1.7.0b4 Python/2.7.17 (Linux-5.4.72-microsoft-standard-WSL2-x86_64-with-Ubuntu-20.04-focal)
+    method: GET
+    uri: https://login.microsoftonline.com/tenant/v2.0/.well-known/openid-configuration
+  response:
+    body:
+      string: !!python/unicode '{"msgraph_host": "graph.microsoft.com", "claims_supported":
+        ["sub", "iss", "cloud_instance_name", "cloud_instance_host_name", "cloud_graph_host_name",
+        "msgraph_host", "aud", "exp", "iat", "auth_time", "acr", "nonce", "preferred_username",
+        "name", "tid", "ver", "at_hash", "c_hash", "email"], "subject_types_supported":
+        ["pairwise"], "issuer": "https://login.microsoftonline.com/tenant/v2.0", "cloud_graph_host_name":
+        "graph.windows.net", "token_endpoint": "https://login.microsoftonline.com/tenant/oauth2/v2.0/token",
+        "request_uri_parameter_supported": false, "http_logout_supported": true, "response_modes_supported":
+        ["query", "fragment", "form_post"], "jwks_uri": "https://login.microsoftonline.com/tenant/discovery/v2.0/keys",
+        "scopes_supported": ["openid", "profile", "email", "offline_access"], "token_endpoint_auth_methods_supported":
+        ["client_secret_post", "private_key_jwt", "client_secret_basic"], "id_token_signing_alg_values_supported":
+        ["RS256"], "tenant_region_scope": "NA", "rbac_url": "https://pas.windows.net",
+        "userinfo_endpoint": "https://graph.microsoft.com/oidc/userinfo", "device_authorization_endpoint":
+        "https://login.microsoftonline.com/tenant/oauth2/v2.0/devicecode", "kerberos_endpoint":
+        "https://login.microsoftonline.com/tenant/kerberos", "frontchannel_logout_supported":
+        true, "response_types_supported": ["code", "id_token", "code id_token", "id_token
+        token"], "end_session_endpoint": "https://login.microsoftonline.com/tenant/oauth2/v2.0/logout",
+        "authorization_endpoint": "https://login.microsoftonline.com/tenant/oauth2/v2.0/authorize",
+        "cloud_instance_name": "microsoftonline.com"}'
+    headers:
+      access-control-allow-methods:
+      - GET, OPTIONS
+      access-control-allow-origin:
+      - '*'
+      cache-control:
+      - max-age=86400, private
+      content-length:
+      - '1753'
+      content-type:
+      - application/json; charset=utf-8
+      date:
+      - Tue, 31 Aug 2021 16:25:28 GMT
+      p3p:
+      - CP="DSP CUR OTPi IND OTRi ONL FIN"
+      set-cookie:
+      - fpc=AhpCwg7nXolElOYV8ZYRD4M; expires=Thu, 30-Sep-2021 16:25:28 GMT; path=/;
+        secure; HttpOnly; SameSite=None
+      - esctx=AQABAAAAAAD--DLA3VO7QrddgJg7WevrszouIFQe6sBE5mJYc6-KDI4WcIOvAcgDid2UotGPc7ln1DxPrqob_6EZotD0nTG_5XW2m_kDu5u0POfZGWLxI5tNCDB0BE1MNzw5dfv7AKzuXMl2JwKD3YdgBBBJt3lKmCDNvEG04WExEP7HoDTahTAqzbG0agT5cjb-IajkK2QgAA;
+        domain=.login.microsoftonline.com; path=/; secure; HttpOnly; SameSite=None
+      - x-ms-gateway-slice=estsfd; path=/; secure; samesite=none; httponly
+      - stsservicecookie=estsfd; path=/; secure; samesite=none; httponly
+      strict-transport-security:
+      - max-age=31536000; includeSubDomains
+      x-content-type-options:
+      - nosniff
+      x-ms-ests-server:
+      - 2.1.11984.12 - WUS2 ProdSlices
+    status:
+      code: 200
+      message: OK
+- request:
+    body: null
+    headers:
+      Accept:
+      - application/json
+      Accept-Encoding:
+      - gzip, deflate
+      Connection:
+      - keep-alive
+      Content-Length:
+      - '2682'
+      Content-Type:
+      - application/x-www-form-urlencoded
+      Cookie:
+      - stsservicecookie=estsfd; fpc=AhpCwg7nXolElOYV8ZYRD4M; x-ms-gateway-slice=estsfd;
+        esctx=AQABAAAAAAD--DLA3VO7QrddgJg7WevrszouIFQe6sBE5mJYc6-KDI4WcIOvAcgDid2UotGPc7ln1DxPrqob_6EZotD0nTG_5XW2m_kDu5u0POfZGWLxI5tNCDB0BE1MNzw5dfv7AKzuXMl2JwKD3YdgBBBJt3lKmCDNvEG04WExEP7HoDTahTAqzbG0agT5cjb-IajkK2QgAA
+      User-Agent:
+      - azsdk-python-identity/1.7.0b4 Python/2.7.17 (Linux-5.4.72-microsoft-standard-WSL2-x86_64-with-Ubuntu-20.04-focal)
+      x-client-cpu:
+      - x64
+      x-client-current-telemetry:
+      - 4|523,0|
+      x-client-last-telemetry:
+      - 4|0|||
+      x-client-os:
+      - linux2
+      x-client-sku:
+      - MSAL.Python
+      x-client-ver:
+      - 1.12.0
+    method: POST
+    uri: https://login.microsoftonline.com/tenant/oauth2/v2.0/token
+  response:
+    body:
+      string: !!python/unicode '{"access_token": "redacted", "ext_expires_in": 3597,
+        "expires_in": 3597, "client_info": "eyJ1aWQiOiJjZGQ2ZWUyNi1lNGJhLTQyNGUtODJlNS1lN2ZhNzc4M2I1ODAiLCJ1dGlkIjoiM2M2MzFiYjctYTlmNy00MzQzLWE1YmEtYTYxNTkxMzVmMWZjIn0",
+        "token_type": "Bearer", "id_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiIsImtpZCI6Im5PbzNaRHJPRFhFSzFqS1doWHNsSFJfS1hFZyJ9.eyJzdWIiOiAiREVlWDVqRGVVOXQ1NFpCaVZ1SnZLZ3BUUTdHLWFrNi1qYnRmbmh6dURZVSIsICJ1dGkiOiAiSnpLNmhFUUhmRS1SNXowVVhoWE5BQSIsICJ2ZXIiOiAiMi4wIiwgIm5hbWUiOiAicmVkYWN0ZWQiLCAiaXNzIjogImh0dHBzOi8vbG9naW4ubWljcm9zb2Z0b25saW5lLmNvbS8zYzYzMWJiNy1hOWY3LTQzNDMtYTViYS1hNjE1OTEzNWYxZmMvdjIuMCIsICJvaWQiOiAiY2RkNmVlMjYtZTRiYS00MjRlLTgyZTUtZTdmYTc3ODNiNTgwIiwgInByZWZlcnJlZF91c2VybmFtZSI6ICJyZWRhY3RlZCIsICJleHAiOiAxNjMwNDMwNzI2LCAidGlkIjogIjNjNjMxYmI3LWE5ZjctNDM0My1hNWJhLWE2MTU5MTM1ZjFmYyIsICJpYXQiOiAxNjMwNDI2ODI4LCAicmgiOiAiMC5BUmNBdHh0alBQZXBRME9sdXFZVmtUWHhfSzdGRDlma1VwVkl2eHVKUVFQWDd4UVhBUHMuIiwgIm5iZiI6IDE2MzA0MjY4MjgsICJhdWQiOiAiZDcwZmM1YWUtNTJlNC00ODk1LWJmMWItODk0MTAzZDdlZjE0In0=.P-tjPtkwwW6wfgF5-UWdMMW5spPEXWcDUNqGrDkAJViWBFGwaZjIHtsx-zTLY9rZrGTp4crl54_kLfnuC0ylpEVVdo-x7CGvfR72lysNJGXvHnqd9MnZQ3xDq5qcc_zv4215A-CH4wq1hgrKeTggOKVUxCandrmsQld25fd2Fpi4anxfZ40zC6vDueOQ249TcOTUaSeS40bWWM3Ecfzk_LpcGhYCJJtN6NAbEJqYLUGUOpnUuhsFU1FIwhJ7WAvSchBqw6I_qIqU-Bmhble2hW57svFS688_LWNv2GZDLi1fC34_YYMxffSffqJ62FkrmSgc0owNIl0dIPQjTfYH_A",
+        "scope": "https://management.azure.com/user_impersonation https://management.azure.com/.default",
+        "refresh_token": "redacted"}'
+    headers:
+      cache-control:
+      - no-store, no-cache
+      content-length:
+      - '4196'
+      content-type:
+      - application/json; charset=utf-8
+      date:
+      - Tue, 31 Aug 2021 16:25:28 GMT
+      expires:
+      - '-1'
+      p3p:
+      - CP="DSP CUR OTPi IND OTRi ONL FIN"
+      pragma:
+      - no-cache
+      set-cookie:
+      - fpc=AhpCwg7nXolElOYV8ZYRD4P0CaNgAQAAAPhOwNgOAAAA; expires=Thu, 30-Sep-2021
+        16:25:29 GMT; path=/; secure; HttpOnly; SameSite=None
+      - x-ms-gateway-slice=estsfd; path=/; secure; samesite=none; httponly
+      - stsservicecookie=estsfd; path=/; secure; samesite=none; httponly
+      strict-transport-security:
+      - max-age=31536000; includeSubDomains
+      x-content-type-options:
+      - nosniff
+      x-ms-clitelem:
+      - 1,0,0,,
+      x-ms-ests-server:
+      - 2.1.11961.11 - EUS ProdSlices
+    status:
+      code: 200
+      message: OK
+- request:
+    body: null
+    headers:
+      Accept:
+      - application/json
+      Accept-Encoding:
+      - gzip, deflate
+      Connection:
+      - keep-alive
+      User-Agent:
+      - azsdk-python-azure-mgmt-resource/16.1.0 Python/2.7.17 (Linux-5.4.72-microsoft-standard-WSL2-x86_64-with-Ubuntu-20.04-focal)
+    method: GET
+    uri: https://management.azure.com/subscriptions?api-version=2019-11-01
+  response:
+    body:
+      string: !!python/unicode '{"value":[]}'
+    headers:
+      cache-control:
+      - no-cache
+      content-length:
+      - '432'
+      content-type:
+      - application/json; charset=utf-8
+      date:
+      - Tue, 31 Aug 2021 16:25:28 GMT
+      expires:
+      - '-1'
+      pragma:
+      - no-cache
+      strict-transport-security:
+      - max-age=31536000; includeSubDomains
+      vary:
+      - Accept-Encoding
+      x-content-type-options:
+      - nosniff
+    status:
+      code: 200
+      message: OK
+version: 1
diff --git a/sdk/identity/azure-identity/tests/recordings/test_obo_async.test_obo.yaml b/sdk/identity/azure-identity/tests/recordings/test_obo_async.test_obo.yaml
new file mode 100644
index 000000000000..f7cd94b03c80
--- /dev/null
+++ b/sdk/identity/azure-identity/tests/recordings/test_obo_async.test_obo.yaml
@@ -0,0 +1,244 @@
+interactions:
+- request:
+    body: null
+    headers:
+      Accept:
+      - '*/*'
+      Accept-Encoding:
+      - gzip, deflate
+      Connection:
+      - keep-alive
+      User-Agent:
+      - azsdk-python-identity/1.7.0b4 Python/3.8.10 (Linux-5.4.72-microsoft-standard-WSL2-x86_64-with-glibc2.29)
+    method: GET
+    uri: https://login.microsoftonline.com/tenant/v2.0/.well-known/openid-configuration
+  response:
+    body:
+      string: '{"token_endpoint": "https://login.microsoftonline.com/tenant/oauth2/v2.0/token",
+        "token_endpoint_auth_methods_supported": ["client_secret_post", "private_key_jwt",
+        "client_secret_basic"], "jwks_uri": "https://login.microsoftonline.com/tenant/discovery/v2.0/keys",
+        "response_modes_supported": ["query", "fragment", "form_post"], "subject_types_supported":
+        ["pairwise"], "id_token_signing_alg_values_supported": ["RS256"], "response_types_supported":
+        ["code", "id_token", "code id_token", "id_token token"], "scopes_supported":
+        ["openid", "profile", "email", "offline_access"], "issuer": "https://login.microsoftonline.com/tenant/v2.0",
+        "request_uri_parameter_supported": false, "userinfo_endpoint": "https://graph.microsoft.com/oidc/userinfo",
+        "authorization_endpoint": "https://login.microsoftonline.com/tenant/oauth2/v2.0/authorize",
+        "device_authorization_endpoint": "https://login.microsoftonline.com/tenant/oauth2/v2.0/devicecode",
+        "http_logout_supported": true, "frontchannel_logout_supported": true, "end_session_endpoint":
+        "https://login.microsoftonline.com/tenant/oauth2/v2.0/logout", "claims_supported":
+        ["sub", "iss", "cloud_instance_name", "cloud_instance_host_name", "cloud_graph_host_name",
+        "msgraph_host", "aud", "exp", "iat", "auth_time", "acr", "nonce", "preferred_username",
+        "name", "tid", "ver", "at_hash", "c_hash", "email"], "kerberos_endpoint":
+        "https://login.microsoftonline.com/tenant/kerberos", "tenant_region_scope":
+        "NA", "cloud_instance_name": "microsoftonline.com", "cloud_graph_host_name":
+        "graph.windows.net", "msgraph_host": "graph.microsoft.com", "rbac_url": "https://pas.windows.net"}'
+    headers:
+      access-control-allow-methods:
+      - GET, OPTIONS
+      access-control-allow-origin:
+      - '*'
+      cache-control:
+      - max-age=86400, private
+      content-length:
+      - '1753'
+      content-type:
+      - application/json; charset=utf-8
+      date:
+      - Tue, 31 Aug 2021 16:25:23 GMT
+      p3p:
+      - CP="DSP CUR OTPi IND OTRi ONL FIN"
+      set-cookie:
+      - fpc=An9LD1p-9hxKvTPdtSdShxg; expires=Thu, 30-Sep-2021 16:25:23 GMT; path=/;
+        secure; HttpOnly; SameSite=None
+      - esctx=AQABAAAAAAD--DLA3VO7QrddgJg7WevrzgjxTgvtVEmRB8rlvEBNYDaJhozZAYjvymTUgItB_EnEG02mnlrc-OBmFolSgBoJCsHV0ceHe8FD6fs9o9qfpbwEDP6wEPwL2gntGGUglE-MTaMEsvB7w5bJtHMfvSloOAczVkVm_67lQIRELy4sXV3ZB1Yd0YhPew_P-bXGjzkgAA;
+        domain=.login.microsoftonline.com; path=/; secure; HttpOnly; SameSite=None
+      - x-ms-gateway-slice=estsfd; path=/; secure; samesite=none; httponly
+      - stsservicecookie=estsfd; path=/; secure; samesite=none; httponly
+      strict-transport-security:
+      - max-age=31536000; includeSubDomains
+      x-content-type-options:
+      - nosniff
+      x-ms-ests-server:
+      - 2.1.11961.11 - SCUS ProdSlices
+    status:
+      code: 200
+      message: OK
+- request:
+    body: null
+    headers:
+      Accept:
+      - application/json
+      Accept-Encoding:
+      - gzip, deflate
+      Connection:
+      - keep-alive
+      Cookie:
+      - esctx=AQABAAAAAAD--DLA3VO7QrddgJg7WevrzgjxTgvtVEmRB8rlvEBNYDaJhozZAYjvymTUgItB_EnEG02mnlrc-OBmFolSgBoJCsHV0ceHe8FD6fs9o9qfpbwEDP6wEPwL2gntGGUglE-MTaMEsvB7w5bJtHMfvSloOAczVkVm_67lQIRELy4sXV3ZB1Yd0YhPew_P-bXGjzkgAA;
+        fpc=An9LD1p-9hxKvTPdtSdShxg; stsservicecookie=estsfd; x-ms-gateway-slice=estsfd
+      User-Agent:
+      - azsdk-python-identity/1.7.0b4 Python/3.8.10 (Linux-5.4.72-microsoft-standard-WSL2-x86_64-with-glibc2.29)
+    method: GET
+    uri: https://login.microsoftonline.com/common/userrealm/username?api-version=1.0
+  response:
+    body:
+      string: '{"ver": "1.0", "account_type": "Managed", "domain_name": "chlowehotmail.onmicrosoft.com",
+        "cloud_instance_name": "microsoftonline.com", "cloud_audience_urn": "urn:federation:MicrosoftOnline"}'
+    headers:
+      cache-control:
+      - no-store, no-cache
+      content-disposition:
+      - inline; filename=userrealm.json
+      content-length:
+      - '182'
+      content-type:
+      - application/json; charset=utf-8
+      date:
+      - Tue, 31 Aug 2021 16:25:23 GMT
+      expires:
+      - '-1'
+      p3p:
+      - CP="DSP CUR OTPi IND OTRi ONL FIN"
+      pragma:
+      - no-cache
+      set-cookie:
+      - fpc=An9LD1p-9hxKvTPdtSdShxg; expires=Thu, 30-Sep-2021 16:25:24 GMT; path=/;
+        secure; HttpOnly; SameSite=None
+      - x-ms-gateway-slice=estsfd; path=/; secure; samesite=none; httponly
+      - stsservicecookie=estsfd; path=/; secure; samesite=none; httponly
+      strict-transport-security:
+      - max-age=31536000; includeSubDomains
+      x-content-type-options:
+      - nosniff
+      x-ms-ests-server:
+      - 2.1.11961.11 - EUS ProdSlices
+    status:
+      code: 200
+      message: OK
+- request:
+    body: null
+    headers:
+      Accept:
+      - application/json
+      Accept-Encoding:
+      - gzip, deflate
+      Connection:
+      - keep-alive
+      Content-Length:
+      - '304'
+      Content-Type:
+      - application/x-www-form-urlencoded
+      Cookie:
+      - esctx=AQABAAAAAAD--DLA3VO7QrddgJg7WevrzgjxTgvtVEmRB8rlvEBNYDaJhozZAYjvymTUgItB_EnEG02mnlrc-OBmFolSgBoJCsHV0ceHe8FD6fs9o9qfpbwEDP6wEPwL2gntGGUglE-MTaMEsvB7w5bJtHMfvSloOAczVkVm_67lQIRELy4sXV3ZB1Yd0YhPew_P-bXGjzkgAA;
+        fpc=An9LD1p-9hxKvTPdtSdShxg; stsservicecookie=estsfd; x-ms-gateway-slice=estsfd
+      User-Agent:
+      - azsdk-python-identity/1.7.0b4 Python/3.8.10 (Linux-5.4.72-microsoft-standard-WSL2-x86_64-with-glibc2.29)
+      x-client-cpu:
+      - x64
+      x-client-current-telemetry:
+      - 4|301,0|
+      x-client-last-telemetry:
+      - 4|0|||
+      x-client-os:
+      - linux
+      x-client-sku:
+      - MSAL.Python
+      x-client-ver:
+      - 1.14.0
+      x-ms-lib-capability:
+      - retry-after, h429
+    method: POST
+    uri: https://login.microsoftonline.com/tenant/oauth2/v2.0/token
+  response:
+    body:
+      string: '{"token_type": "Bearer", "scope": "api://foo/Foo", "expires_in": 3599,
+        "ext_expires_in": 3599, "access_token": "redacted", "refresh_token": "redacted",
+        "id_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiIsImtpZCI6Im5PbzNaRHJPRFhFSzFqS1doWHNsSFJfS1hFZyJ9.eyJhdWQiOiAiZDcwZmM1YWUtNTJlNC00ODk1LWJmMWItODk0MTAzZDdlZjE0IiwgImlzcyI6ICJodHRwczovL2xvZ2luLm1pY3Jvc29mdG9ubGluZS5jb20vM2M2MzFiYjctYTlmNy00MzQzLWE1YmEtYTYxNTkxMzVmMWZjL3YyLjAiLCAiaWF0IjogMTYzMDQyNjgyNCwgIm5iZiI6IDE2MzA0MjY4MjQsICJleHAiOiAxNjMwNDMwNzI0LCAibmFtZSI6ICJyZWRhY3RlZCIsICJvaWQiOiAiY2RkNmVlMjYtZTRiYS00MjRlLTgyZTUtZTdmYTc3ODNiNTgwIiwgInByZWZlcnJlZF91c2VybmFtZSI6ICJyZWRhY3RlZCIsICJyaCI6ICIwLkFSY0F0eHRqUFBlcFEwT2x1cVlWa1RYeF9LN0ZEOWZrVXBWSXZ4dUpRUVBYN3hRWEFQcy4iLCAic3ViIjogIkRFZVg1akRlVTl0NTRaQmlWdUp2S2dwVFE3Ry1hazYtamJ0Zm5oenVEWVUiLCAidGlkIjogIjNjNjMxYmI3LWE5ZjctNDM0My1hNWJhLWE2MTU5MTM1ZjFmYyIsICJ1dGkiOiAiZlBlejItNTY1RXlHTE82aW1kbXFBQSIsICJ2ZXIiOiAiMi4wIn0=.kmTL0QldU04sl7NKDsGxj9xd_DyLPZdEuFSqsqH9Ej5sd-sEyRAKNUyRT96vLn4yOsGScz1tJE6839TXuevD7kwInjakcmT_nh3xcrB1ZjntGDnDiMD_DDV7MwZz7b9O0s7-NKCXT6cmt6TUO1KADsqA1V9o5yzGtcQnaCXuRDONFFP3yv16HVeGVQ5CX6V51S2uyTEPahHGM9BAg9PqvqIOSyJ1f4QIgf5w32AY1-sHfGGapUXMYAQX9GZEbor53WJoBwvRkJBoXWekx-noM3YuHCASTffBR07ztQAWzUFDnlN9-WyBf9mpkjvfeqGV82CkwXhnRgfDnbYuxopJew",
+        "client_info": "eyJ1aWQiOiJjZGQ2ZWUyNi1lNGJhLTQyNGUtODJlNS1lN2ZhNzc4M2I1ODAiLCJ1dGlkIjoiM2M2MzFiYjctYTlmNy00MzQzLWE1YmEtYTYxNTkxMzVmMWZjIn0"}'
+    headers:
+      cache-control:
+      - no-store, no-cache
+      content-length:
+      - '3765'
+      content-type:
+      - application/json; charset=utf-8
+      date:
+      - Tue, 31 Aug 2021 16:25:23 GMT
+      expires:
+      - '-1'
+      p3p:
+      - CP="DSP CUR OTPi IND OTRi ONL FIN"
+      pragma:
+      - no-cache
+      set-cookie:
+      - fpc=An9LD1p-9hxKvTPdtSdShxjArTo4AQAAAPNOwNgOAAAA; expires=Thu, 30-Sep-2021
+        16:25:24 GMT; path=/; secure; HttpOnly; SameSite=None
+      - x-ms-gateway-slice=estsfd; path=/; secure; samesite=none; httponly
+      - stsservicecookie=estsfd; path=/; secure; samesite=none; httponly
+      strict-transport-security:
+      - max-age=31536000; includeSubDomains
+      x-content-type-options:
+      - nosniff
+      x-ms-clitelem:
+      - 1,0,0,,
+      x-ms-ests-server:
+      - 2.1.11961.11 - NCUS ProdSlices
+    status:
+      code: 200
+      message: OK
+- request:
+    body: null
+    headers:
+      Content-Type:
+      - application/x-www-form-urlencoded
+      User-Agent:
+      - azsdk-python-identity/1.7.0b4 Python/3.8.10 (Linux-5.4.72-microsoft-standard-WSL2-x86_64-with-glibc2.29)
+    method: POST
+    uri: https://login.microsoftonline.com/tenant/oauth2/v2.0/token
+  response:
+    body:
+      string: '{"token_type": "Bearer", "scope": "https://management.azure.com/user_impersonation
+        https://management.azure.com/.default", "expires_in": 3598, "ext_expires_in":
+        3598, "access_token": "redacted", "refresh_token": "redacted"}'
+    headers:
+      cache-control: no-store, no-cache
+      content-length: '2906'
+      content-type: application/json; charset=utf-8
+      date: Tue, 31 Aug 2021 16:25:24 GMT
+      expires: '-1'
+      p3p: CP="DSP CUR OTPi IND OTRi ONL FIN"
+      pragma: no-cache
+      set-cookie: stsservicecookie=estsfd; path=/; secure; samesite=none; httponly
+      strict-transport-security: max-age=31536000; includeSubDomains
+      x-content-type-options: nosniff
+      x-ms-ests-server: 2.1.11961.11 - EUS ProdSlices
+    status:
+      code: 200
+      message: OK
+    url: https://login.microsoftonline.com/3c631bb7-a9f7-4343-a5ba-a6159135f1fc/oauth2/v2.0/token
+- request:
+    body: null
+    headers:
+      Accept:
+      - application/json
+      User-Agent:
+      - azsdk-python-azure-mgmt-resource/18.1.0 Python/3.8.10 (Linux-5.4.72-microsoft-standard-WSL2-x86_64-with-glibc2.29)
+    method: GET
+    uri: https://management.azure.com/subscriptions?api-version=2019-11-01
+  response:
+    body:
+      string: '{"value":[]}'
+    headers:
+      cache-control: no-cache
+      content-length: '416'
+      content-type: application/json; charset=utf-8
+      date: Tue, 31 Aug 2021 16:25:24 GMT
+      expires: '-1'
+      pragma: no-cache
+      strict-transport-security: max-age=31536000; includeSubDomains
+      vary: Accept-Encoding
+      x-content-type-options: nosniff
+    status:
+      code: 200
+      message: OK
+    url: https://management.azure.com/subscriptions?api-version=2019-11-01
+version: 1
diff --git a/sdk/identity/azure-identity/tests/recordings/test_obo_async.test_obo_cert.yaml b/sdk/identity/azure-identity/tests/recordings/test_obo_async.test_obo_cert.yaml
new file mode 100644
index 000000000000..8ea6c9194e37
--- /dev/null
+++ b/sdk/identity/azure-identity/tests/recordings/test_obo_async.test_obo_cert.yaml
@@ -0,0 +1,244 @@
+interactions:
+- request:
+    body: null
+    headers:
+      Accept:
+      - '*/*'
+      Accept-Encoding:
+      - gzip, deflate
+      Connection:
+      - keep-alive
+      User-Agent:
+      - azsdk-python-identity/1.7.0b4 Python/3.8.10 (Linux-5.4.72-microsoft-standard-WSL2-x86_64-with-glibc2.29)
+    method: GET
+    uri: https://login.microsoftonline.com/tenant/v2.0/.well-known/openid-configuration
+  response:
+    body:
+      string: '{"token_endpoint": "https://login.microsoftonline.com/tenant/oauth2/v2.0/token",
+        "token_endpoint_auth_methods_supported": ["client_secret_post", "private_key_jwt",
+        "client_secret_basic"], "jwks_uri": "https://login.microsoftonline.com/tenant/discovery/v2.0/keys",
+        "response_modes_supported": ["query", "fragment", "form_post"], "subject_types_supported":
+        ["pairwise"], "id_token_signing_alg_values_supported": ["RS256"], "response_types_supported":
+        ["code", "id_token", "code id_token", "id_token token"], "scopes_supported":
+        ["openid", "profile", "email", "offline_access"], "issuer": "https://login.microsoftonline.com/tenant/v2.0",
+        "request_uri_parameter_supported": false, "userinfo_endpoint": "https://graph.microsoft.com/oidc/userinfo",
+        "authorization_endpoint": "https://login.microsoftonline.com/tenant/oauth2/v2.0/authorize",
+        "device_authorization_endpoint": "https://login.microsoftonline.com/tenant/oauth2/v2.0/devicecode",
+        "http_logout_supported": true, "frontchannel_logout_supported": true, "end_session_endpoint":
+        "https://login.microsoftonline.com/tenant/oauth2/v2.0/logout", "claims_supported":
+        ["sub", "iss", "cloud_instance_name", "cloud_instance_host_name", "cloud_graph_host_name",
+        "msgraph_host", "aud", "exp", "iat", "auth_time", "acr", "nonce", "preferred_username",
+        "name", "tid", "ver", "at_hash", "c_hash", "email"], "kerberos_endpoint":
+        "https://login.microsoftonline.com/tenant/kerberos", "tenant_region_scope":
+        "NA", "cloud_instance_name": "microsoftonline.com", "cloud_graph_host_name":
+        "graph.windows.net", "msgraph_host": "graph.microsoft.com", "rbac_url": "https://pas.windows.net"}'
+    headers:
+      access-control-allow-methods:
+      - GET, OPTIONS
+      access-control-allow-origin:
+      - '*'
+      cache-control:
+      - max-age=86400, private
+      content-length:
+      - '1753'
+      content-type:
+      - application/json; charset=utf-8
+      date:
+      - Tue, 31 Aug 2021 16:25:24 GMT
+      p3p:
+      - CP="DSP CUR OTPi IND OTRi ONL FIN"
+      set-cookie:
+      - fpc=AposoGDZdFhIiZzprtzz93c; expires=Thu, 30-Sep-2021 16:25:25 GMT; path=/;
+        secure; HttpOnly; SameSite=None
+      - esctx=AQABAAAAAAD--DLA3VO7QrddgJg7WevrtbE2f5B3vanGCwsniSd9hZj9nmk9O9F9ihMc5SeyN6FT7EZjOyj5heOGFo-vKwJabfpy7CMbjJ6ifwQipMpHAZ9SyjxRdcJgkX6nuu3f0sqXtH8hESNXtzURDyB4lXIlJ6IWa6ObFNVYlJx39jWhSPG34YXdIf1qJWnBXF-cX8kgAA;
+        domain=.login.microsoftonline.com; path=/; secure; HttpOnly; SameSite=None
+      - x-ms-gateway-slice=estsfd; path=/; secure; samesite=none; httponly
+      - stsservicecookie=estsfd; path=/; secure; samesite=none; httponly
+      strict-transport-security:
+      - max-age=31536000; includeSubDomains
+      x-content-type-options:
+      - nosniff
+      x-ms-ests-server:
+      - 2.1.11984.12 - WUS2 ProdSlices
+    status:
+      code: 200
+      message: OK
+- request:
+    body: null
+    headers:
+      Accept:
+      - application/json
+      Accept-Encoding:
+      - gzip, deflate
+      Connection:
+      - keep-alive
+      Cookie:
+      - esctx=AQABAAAAAAD--DLA3VO7QrddgJg7WevrtbE2f5B3vanGCwsniSd9hZj9nmk9O9F9ihMc5SeyN6FT7EZjOyj5heOGFo-vKwJabfpy7CMbjJ6ifwQipMpHAZ9SyjxRdcJgkX6nuu3f0sqXtH8hESNXtzURDyB4lXIlJ6IWa6ObFNVYlJx39jWhSPG34YXdIf1qJWnBXF-cX8kgAA;
+        fpc=AposoGDZdFhIiZzprtzz93c; stsservicecookie=estsfd; x-ms-gateway-slice=estsfd
+      User-Agent:
+      - azsdk-python-identity/1.7.0b4 Python/3.8.10 (Linux-5.4.72-microsoft-standard-WSL2-x86_64-with-glibc2.29)
+    method: GET
+    uri: https://login.microsoftonline.com/common/userrealm/username?api-version=1.0
+  response:
+    body:
+      string: '{"ver": "1.0", "account_type": "Managed", "domain_name": "chlowehotmail.onmicrosoft.com",
+        "cloud_instance_name": "microsoftonline.com", "cloud_audience_urn": "urn:federation:MicrosoftOnline"}'
+    headers:
+      cache-control:
+      - no-store, no-cache
+      content-disposition:
+      - inline; filename=userrealm.json
+      content-length:
+      - '182'
+      content-type:
+      - application/json; charset=utf-8
+      date:
+      - Tue, 31 Aug 2021 16:25:25 GMT
+      expires:
+      - '-1'
+      p3p:
+      - CP="DSP CUR OTPi IND OTRi ONL FIN"
+      pragma:
+      - no-cache
+      set-cookie:
+      - fpc=AposoGDZdFhIiZzprtzz93c; expires=Thu, 30-Sep-2021 16:25:25 GMT; path=/;
+        secure; HttpOnly; SameSite=None
+      - x-ms-gateway-slice=estsfd; path=/; secure; samesite=none; httponly
+      - stsservicecookie=estsfd; path=/; secure; samesite=none; httponly
+      strict-transport-security:
+      - max-age=31536000; includeSubDomains
+      x-content-type-options:
+      - nosniff
+      x-ms-ests-server:
+      - 2.1.11961.11 - SCUS ProdSlices
+    status:
+      code: 200
+      message: OK
+- request:
+    body: null
+    headers:
+      Accept:
+      - application/json
+      Accept-Encoding:
+      - gzip, deflate
+      Connection:
+      - keep-alive
+      Content-Length:
+      - '304'
+      Content-Type:
+      - application/x-www-form-urlencoded
+      Cookie:
+      - esctx=AQABAAAAAAD--DLA3VO7QrddgJg7WevrtbE2f5B3vanGCwsniSd9hZj9nmk9O9F9ihMc5SeyN6FT7EZjOyj5heOGFo-vKwJabfpy7CMbjJ6ifwQipMpHAZ9SyjxRdcJgkX6nuu3f0sqXtH8hESNXtzURDyB4lXIlJ6IWa6ObFNVYlJx39jWhSPG34YXdIf1qJWnBXF-cX8kgAA;
+        fpc=AposoGDZdFhIiZzprtzz93c; stsservicecookie=estsfd; x-ms-gateway-slice=estsfd
+      User-Agent:
+      - azsdk-python-identity/1.7.0b4 Python/3.8.10 (Linux-5.4.72-microsoft-standard-WSL2-x86_64-with-glibc2.29)
+      x-client-cpu:
+      - x64
+      x-client-current-telemetry:
+      - 4|301,0|
+      x-client-last-telemetry:
+      - 4|0|||
+      x-client-os:
+      - linux
+      x-client-sku:
+      - MSAL.Python
+      x-client-ver:
+      - 1.14.0
+      x-ms-lib-capability:
+      - retry-after, h429
+    method: POST
+    uri: https://login.microsoftonline.com/tenant/oauth2/v2.0/token
+  response:
+    body:
+      string: '{"token_type": "Bearer", "scope": "api://foo/Foo", "expires_in": 3599,
+        "ext_expires_in": 3599, "access_token": "redacted", "refresh_token": "redacted",
+        "id_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiIsImtpZCI6Im5PbzNaRHJPRFhFSzFqS1doWHNsSFJfS1hFZyJ9.eyJhdWQiOiAiZDcwZmM1YWUtNTJlNC00ODk1LWJmMWItODk0MTAzZDdlZjE0IiwgImlzcyI6ICJodHRwczovL2xvZ2luLm1pY3Jvc29mdG9ubGluZS5jb20vM2M2MzFiYjctYTlmNy00MzQzLWE1YmEtYTYxNTkxMzVmMWZjL3YyLjAiLCAiaWF0IjogMTYzMDQyNjgyNSwgIm5iZiI6IDE2MzA0MjY4MjUsICJleHAiOiAxNjMwNDMwNzI1LCAibmFtZSI6ICJyZWRhY3RlZCIsICJvaWQiOiAiY2RkNmVlMjYtZTRiYS00MjRlLTgyZTUtZTdmYTc3ODNiNTgwIiwgInByZWZlcnJlZF91c2VybmFtZSI6ICJyZWRhY3RlZCIsICJyaCI6ICIwLkFSY0F0eHRqUFBlcFEwT2x1cVlWa1RYeF9LN0ZEOWZrVXBWSXZ4dUpRUVBYN3hRWEFQcy4iLCAic3ViIjogIkRFZVg1akRlVTl0NTRaQmlWdUp2S2dwVFE3Ry1hazYtamJ0Zm5oenVEWVUiLCAidGlkIjogIjNjNjMxYmI3LWE5ZjctNDM0My1hNWJhLWE2MTU5MTM1ZjFmYyIsICJ1dGkiOiAiSGtMT0w3ZkVHRTZBcFFYa0RVZlNBQSIsICJ2ZXIiOiAiMi4wIn0=.DKoQSY24Ie24IetMLuzntnKMncaPl5atPW5wEJAC291VCOe2hM8RtgYGskh4QmosI-GqQ8_7wdic12RUCqz6QZNGzAHAmjG_QVWwtjE8Bd1YsvjiafFW91g2BpCJJZi9oLg24sYrax2XyBgeJC7KP4KzSrsoeOFoJ3Q-pVTF6l0YsPp_z08kEzhBJj1POuCwkHPZmioa5Tq17CnnSMGXCDs_7CXuRqKF2RV6CMw7_2I1pUkCOk1xnTTwBfjDvP96tnot1euKbV39kvx4d71-br2znoGIxyn3vqWWnejdj4vPpdKJ9AYvOxRp1WDd5TBuM_7_SoOpu3JhxO2wM_y79A",
+        "client_info": "eyJ1aWQiOiJjZGQ2ZWUyNi1lNGJhLTQyNGUtODJlNS1lN2ZhNzc4M2I1ODAiLCJ1dGlkIjoiM2M2MzFiYjctYTlmNy00MzQzLWE1YmEtYTYxNTkxMzVmMWZjIn0"}'
+    headers:
+      cache-control:
+      - no-store, no-cache
+      content-length:
+      - '3765'
+      content-type:
+      - application/json; charset=utf-8
+      date:
+      - Tue, 31 Aug 2021 16:25:25 GMT
+      expires:
+      - '-1'
+      p3p:
+      - CP="DSP CUR OTPi IND OTRi ONL FIN"
+      pragma:
+      - no-cache
+      set-cookie:
+      - fpc=AposoGDZdFhIiZzprtzz93fArTo4AQAAAPVOwNgOAAAA; expires=Thu, 30-Sep-2021
+        16:25:25 GMT; path=/; secure; HttpOnly; SameSite=None
+      - x-ms-gateway-slice=estsfd; path=/; secure; samesite=none; httponly
+      - stsservicecookie=estsfd; path=/; secure; samesite=none; httponly
+      strict-transport-security:
+      - max-age=31536000; includeSubDomains
+      x-content-type-options:
+      - nosniff
+      x-ms-clitelem:
+      - 1,0,0,,
+      x-ms-ests-server:
+      - 2.1.11961.11 - SCUS ProdSlices
+    status:
+      code: 200
+      message: OK
+- request:
+    body: null
+    headers:
+      Content-Type:
+      - application/x-www-form-urlencoded
+      User-Agent:
+      - azsdk-python-identity/1.7.0b4 Python/3.8.10 (Linux-5.4.72-microsoft-standard-WSL2-x86_64-with-glibc2.29)
+    method: POST
+    uri: https://login.microsoftonline.com/tenant/oauth2/v2.0/token
+  response:
+    body:
+      string: '{"token_type": "Bearer", "scope": "https://management.azure.com/user_impersonation
+        https://management.azure.com/.default", "expires_in": 3597, "ext_expires_in":
+        3597, "access_token": "redacted", "refresh_token": "redacted"}'
+    headers:
+      cache-control: no-store, no-cache
+      content-length: '2906'
+      content-type: application/json; charset=utf-8
+      date: Tue, 31 Aug 2021 16:25:25 GMT
+      expires: '-1'
+      p3p: CP="DSP CUR OTPi IND OTRi ONL FIN"
+      pragma: no-cache
+      set-cookie: stsservicecookie=estsfd; path=/; secure; samesite=none; httponly
+      strict-transport-security: max-age=31536000; includeSubDomains
+      x-content-type-options: nosniff
+      x-ms-ests-server: 2.1.11961.11 - EUS ProdSlices
+    status:
+      code: 200
+      message: OK
+    url: https://login.microsoftonline.com/3c631bb7-a9f7-4343-a5ba-a6159135f1fc/oauth2/v2.0/token
+- request:
+    body: null
+    headers:
+      Accept:
+      - application/json
+      User-Agent:
+      - azsdk-python-azure-mgmt-resource/18.1.0 Python/3.8.10 (Linux-5.4.72-microsoft-standard-WSL2-x86_64-with-glibc2.29)
+    method: GET
+    uri: https://management.azure.com/subscriptions?api-version=2019-11-01
+  response:
+    body:
+      string: '{"value":[]}'
+    headers:
+      cache-control: no-cache
+      content-length: '416'
+      content-type: application/json; charset=utf-8
+      date: Tue, 31 Aug 2021 16:25:26 GMT
+      expires: '-1'
+      pragma: no-cache
+      strict-transport-security: max-age=31536000; includeSubDomains
+      vary: Accept-Encoding
+      x-content-type-options: nosniff
+    status:
+      code: 200
+      message: OK
+    url: https://management.azure.com/subscriptions?api-version=2019-11-01
+version: 1
diff --git a/sdk/identity/azure-identity/tests/test_context_manager.py b/sdk/identity/azure-identity/tests/test_context_manager.py
index 2466ec43e3fe..2292578a398d 100644
--- a/sdk/identity/azure-identity/tests/test_context_manager.py
+++ b/sdk/identity/azure-identity/tests/test_context_manager.py
@@ -17,6 +17,7 @@
     DeviceCodeCredential,
     EnvironmentCredential,
     InteractiveBrowserCredential,
+    OnBehalfOfCredential,
     SharedTokenCacheCredential,
     UsernamePasswordCredential,
     VisualStudioCodeCredential,
@@ -59,6 +60,10 @@ def get_credential(self, **kwargs):
         ),
     ),
     CredentialFixture(InteractiveBrowserCredential),
+    CredentialFixture(
+        OnBehalfOfCredential,
+        {kwarg: "..." for kwarg in ("tenant_id", "client_id", "client_credential", "user_assertion")},
+    ),
     CredentialFixture(UsernamePasswordCredential, {"client_id": "...", "username": "...", "password": "..."}),
     CredentialFixture(VisualStudioCodeCredential, ctor_patch_factory=lambda: patch(GET_USER_SETTINGS, lambda: {})),
 )
diff --git a/sdk/identity/azure-identity/tests/test_obo.py b/sdk/identity/azure-identity/tests/test_obo.py
new file mode 100644
index 000000000000..0bf2075e3f2c
--- /dev/null
+++ b/sdk/identity/azure-identity/tests/test_obo.py
@@ -0,0 +1,215 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+import os
+
+try:
+    from unittest.mock import Mock, patch
+except ImportError:
+    from mock import Mock, patch  # type: ignore
+
+from azure_devtools.scenario_tests import RecordingProcessor
+from azure.core.pipeline.policies import ContentDecodePolicy, SansIOHTTPPolicy
+from azure.identity import OnBehalfOfCredential, UsernamePasswordCredential
+from azure.identity._constants import EnvironmentVariables
+from azure.identity._internal.user_agent import USER_AGENT
+from azure.mgmt.resource import SubscriptionClient
+import pytest
+import six
+from six.moves.urllib_parse import urlparse
+
+from helpers import build_aad_response, FAKE_CLIENT_ID, get_discovery_response, mock_response
+from recorded_test_case import RecordedTestCase
+from test_certificate_credential import PEM_CERT_PATH
+
+
+class SubscriptionListRemover(RecordingProcessor):
+    def process_response(self, response):
+        if "/subscriptions/" in response["body"]["string"]:
+            response["body"]["string"] = '{"value":[]}'
+        return response
+
+
+class OboRecordedTestCase(RecordedTestCase):
+    def __init__(self, *args, **kwargs):
+        super(OboRecordedTestCase, self).__init__(*args, **kwargs)
+
+        if self.is_live:
+            missing_variables = [
+                var
+                for var in (
+                    "OBO_CERT_BYTES",
+                    "OBO_CLIENT_ID",
+                    "OBO_CLIENT_SECRET",
+                    "OBO_PASSWORD",
+                    "OBO_SCOPE",
+                    "OBO_TENANT_ID",
+                    "OBO_USERNAME",
+                )
+                if var not in os.environ
+            ]
+            if any(missing_variables):
+                pytest.skip("No value for environment variables: " + ", ".join(missing_variables))
+
+            self.obo_settings = {
+                "cert_bytes": six.ensure_binary(os.environ["OBO_CERT_BYTES"]),
+                "client_id": os.environ["OBO_CLIENT_ID"],
+                "client_secret": os.environ["OBO_CLIENT_SECRET"],
+                "password": os.environ["OBO_PASSWORD"],
+                "scope": os.environ["OBO_SCOPE"],
+                "tenant_id": os.environ["OBO_TENANT_ID"],
+                "username": os.environ["OBO_USERNAME"],
+            }
+
+            self.recording_processors.append(SubscriptionListRemover())
+            self.scrubber.register_name_pair(self.obo_settings["tenant_id"], "tenant")
+            self.scrubber.register_name_pair(self.obo_settings["username"], "username")
+
+        else:
+            self.obo_settings = {
+                "cert_bytes": open(PEM_CERT_PATH, "rb").read(),
+                "client_id": FAKE_CLIENT_ID,
+                "client_secret": "secret",
+                "password": "fake-password",
+                "scope": "api://scope",
+                "tenant_id": "tenant",
+                "username": "username",
+            }
+
+
+class RecordedTests(OboRecordedTestCase):
+    def test_obo(self):
+        client_id = self.obo_settings["client_id"]
+        tenant_id = self.obo_settings["tenant_id"]
+
+        user_credential = UsernamePasswordCredential(
+            client_id, self.obo_settings["username"], self.obo_settings["password"], tenant_id=tenant_id
+        )
+        assertion = user_credential.get_token(self.obo_settings["scope"]).token
+        credential = OnBehalfOfCredential(tenant_id, client_id, self.obo_settings["client_secret"], assertion)
+        client = SubscriptionClient(credential)
+        list(client.subscriptions.list())
+
+    def test_obo_cert(self):
+        client_id = self.obo_settings["client_id"]
+        tenant_id = self.obo_settings["tenant_id"]
+
+        user_credential = UsernamePasswordCredential(
+            client_id, self.obo_settings["username"], self.obo_settings["password"], tenant_id=tenant_id
+        )
+        assertion = user_credential.get_token(self.obo_settings["scope"]).token
+        credential = OnBehalfOfCredential(tenant_id, client_id, self.obo_settings["cert_bytes"], assertion)
+        client = SubscriptionClient(credential)
+        list(client.subscriptions.list())
+
+
+def test_allow_multitenant_authentication():
+    """When allow_multitenant_authentication is True, the credential should respect get_token(tenant_id=...)"""
+
+    first_tenant = "first-tenant"
+    first_token = "***"
+    second_tenant = "second-tenant"
+    second_token = first_token * 2
+
+    def send(request, **_):
+        assert request.headers["User-Agent"].startswith(USER_AGENT)
+        parsed = urlparse(request.url)
+        tenant = parsed.path.split("/")[1]
+        assert tenant in (first_tenant, second_tenant), 'unexpected tenant "{}"'.format(tenant)
+        if "/oauth2/v2.0/token" not in parsed.path:
+            return get_discovery_response("https://{}/{}".format(parsed.netloc, tenant))
+
+        token = first_token if tenant == first_tenant else second_token
+        return mock_response(json_payload=build_aad_response(access_token=token))
+
+    transport = Mock(send=Mock(wraps=send))
+    credential = OnBehalfOfCredential(
+        first_tenant, "client-id", "secret", "assertion", allow_multitenant_authentication=True, transport=transport
+    )
+    token = credential.get_token("scope")
+    assert token.token == first_token
+
+    token = credential.get_token("scope", tenant_id=first_tenant)
+    assert token.token == first_token
+
+    token = credential.get_token("scope", tenant_id=second_tenant)
+    assert token.token == second_token
+
+    # should still default to the first tenant
+    token = credential.get_token("scope")
+    assert token.token == first_token
+
+
+@pytest.mark.parametrize("authority", ("localhost", "https://localhost"))
+def test_authority(authority):
+    """the credential should accept an authority, with or without scheme, as an argument or environment variable"""
+
+    tenant_id = "expected-tenant"
+    parsed_authority = urlparse(authority)
+    expected_netloc = parsed_authority.netloc or authority
+    expected_authority = "https://{}/{}".format(expected_netloc, tenant_id)
+
+    mock_ctor = Mock(
+        return_value=Mock(acquire_token_on_behalf_of=lambda *_, **__: {"access_token": "**", "expires_in": 42})
+    )
+
+    credential = OnBehalfOfCredential(tenant_id, "client-id", "secret", "assertion", 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 = OnBehalfOfCredential(tenant_id, "client-id", "secret", "assertion")
+    with patch("msal.ConfidentialClientApplication", mock_ctor):
+        credential.get_token("scope")
+
+    assert mock_ctor.call_count == 1
+    _, kwargs = mock_ctor.call_args
+    assert kwargs["authority"] == expected_authority
+
+
+def test_tenant_id_validation():
+    """The credential should raise ValueError when given an invalid tenant_id"""
+    valid_ids = {"c878a2ab-8ef4-413b-83a0-199afb84d7fb", "contoso.onmicrosoft.com", "organizations", "common"}
+    for tenant in valid_ids:
+        OnBehalfOfCredential(tenant, "client-id", "secret", "assertion")
+    invalid_ids = {"my tenant", "my_tenant", "/", "\\", '"my-tenant"', "'my-tenant'"}
+    for tenant in invalid_ids:
+        with pytest.raises(ValueError):
+            OnBehalfOfCredential(tenant, "client-id", "secret", "assertion")
+
+
+def test_no_scopes():
+    """The credential should raise ValueError when get_token is called with no scopes"""
+    credential = OnBehalfOfCredential("tenant-id", "client-id", "client-secret", "assertion")
+    with pytest.raises(ValueError):
+        credential.get_token()
+
+
+def test_policies_configurable():
+    policy = Mock(spec_set=SansIOHTTPPolicy, on_request=Mock(), on_exception=lambda _: False)
+
+    def send(request, **_):
+        parsed = urlparse(request.url)
+        tenant = parsed.path.split("/")[1]
+        if "/oauth2/v2.0/token" not in parsed.path:
+            return get_discovery_response("https://{}/{}".format(parsed.netloc, tenant))
+        return mock_response(json_payload=build_aad_response(access_token="***"))
+
+    credential = OnBehalfOfCredential(
+        "tenant-id",
+        "client-id",
+        "client-secret",
+        "assertion",
+        policies=[ContentDecodePolicy(), policy],
+        transport=Mock(send=send),
+    )
+    credential.get_token("scope")
+    assert policy.on_request.called
diff --git a/sdk/identity/azure-identity/tests/test_obo_async.py b/sdk/identity/azure-identity/tests/test_obo_async.py
new file mode 100644
index 000000000000..c8ab933f0185
--- /dev/null
+++ b/sdk/identity/azure-identity/tests/test_obo_async.py
@@ -0,0 +1,219 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+from urllib.parse import urlparse
+from unittest.mock import Mock, patch
+
+from azure.core.pipeline.policies import ContentDecodePolicy, SansIOHTTPPolicy
+from azure.identity import UsernamePasswordCredential
+from azure.identity._constants import EnvironmentVariables
+from azure.identity._internal.user_agent import USER_AGENT
+from azure.identity.aio import OnBehalfOfCredential
+from azure.mgmt.resource.subscriptions.aio import SubscriptionClient
+import pytest
+
+from helpers import build_aad_response, get_discovery_response, mock_response
+from helpers_async import AsyncMockTransport
+from recorded_test_case import RecordedTestCase
+from test_obo import OboRecordedTestCase
+
+
+class RecordedTests(OboRecordedTestCase):
+    @RecordedTestCase.await_prepared_test
+    async def test_obo(self):
+        client_id = self.obo_settings["client_id"]
+        client_secret = self.obo_settings["client_secret"]
+        tenant_id = self.obo_settings["tenant_id"]
+
+        user_credential = UsernamePasswordCredential(
+            client_id, self.obo_settings["username"], self.obo_settings["password"], tenant_id=tenant_id
+        )
+        assertion = user_credential.get_token(self.obo_settings["scope"]).token
+        credential = OnBehalfOfCredential(tenant_id, client_id, client_secret, assertion)
+        client = SubscriptionClient(credential)
+        async for _ in client.subscriptions.list():
+            pass
+
+    @RecordedTestCase.await_prepared_test
+    async def test_obo_cert(self):
+        client_id = self.obo_settings["client_id"]
+        tenant_id = self.obo_settings["tenant_id"]
+
+        user_credential = UsernamePasswordCredential(
+            client_id, self.obo_settings["username"], self.obo_settings["password"], tenant_id=tenant_id
+        )
+        assertion = user_credential.get_token(self.obo_settings["scope"]).token
+        credential = OnBehalfOfCredential(tenant_id, client_id, self.obo_settings["cert_bytes"], assertion)
+        client = SubscriptionClient(credential)
+        async for _ in client.subscriptions.list():
+            pass
+
+
+@pytest.mark.asyncio
+async def test_close():
+    transport = AsyncMockTransport()
+    credential = OnBehalfOfCredential("tenant-id", "client-id", "client-secret", "assertion", transport=transport)
+
+    await credential.close()
+
+    assert transport.__aexit__.call_count == 1
+
+
+@pytest.mark.asyncio
+async def test_context_manager():
+    transport = AsyncMockTransport()
+    credential = OnBehalfOfCredential("tenant-id", "client-id", "client-secret", "assertion", transport=transport)
+
+    async with credential:
+        assert transport.__aenter__.call_count == 1
+        assert not transport.__aexit__.called
+
+    assert transport.__aenter__.call_count == 1
+    assert transport.__aexit__.call_count == 1
+
+
+@pytest.mark.asyncio
+async def test_allow_multitenant_authentication():
+    """When allow_multitenant_authentication is True, the credential should respect get_token(tenant_id=...)"""
+
+    first_tenant = "first-tenant"
+    first_token = "***"
+    second_tenant = "second-tenant"
+    second_token = first_token * 2
+
+    async def send(request, **_):
+        assert request.headers["User-Agent"].startswith(USER_AGENT)
+        parsed = urlparse(request.url)
+        tenant = parsed.path.split("/")[1]
+        assert tenant in (first_tenant, second_tenant), 'unexpected tenant "{}"'.format(tenant)
+        token = first_token if tenant == first_tenant else second_token
+        return mock_response(json_payload=build_aad_response(access_token=token))
+
+    transport = Mock(send=Mock(wraps=send))
+    credential = OnBehalfOfCredential(
+        first_tenant, "client-id", "secret", "assertion", allow_multitenant_authentication=True, transport=transport
+    )
+    token = await credential.get_token("scope")
+    assert token.token == first_token
+    assert transport.send.call_count == 1
+
+    token = await credential.get_token("scope", tenant_id=first_tenant)
+    assert token.token == first_token
+    assert transport.send.call_count == 1  # should be a cached token
+
+    token = await credential.get_token("scope", tenant_id=second_tenant)
+    assert token.token == second_token
+    assert transport.send.call_count == 2
+
+    # should still default to the first tenant
+    token = await credential.get_token("scope")
+    assert token.token == first_token
+    assert transport.send.call_count == 2  # should be a cached token
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("authority", ("localhost", "https://localhost"))
+async def test_authority(authority):
+    """the credential should accept an authority, with or without scheme, as an argument or environment variable"""
+
+    tenant_id = "expected-tenant"
+    parsed_authority = urlparse(authority)
+    expected_netloc = parsed_authority.netloc or authority
+    expected_authority = "https://{}/{}".format(expected_netloc, tenant_id)
+    expected_token = "***"
+
+    async def send(request, **_):
+        assert request.url.startswith(expected_authority)
+        return mock_response(json_payload=build_aad_response(access_token=expected_token))
+
+    transport = Mock(send=send)
+    credential = OnBehalfOfCredential(
+        tenant_id, "client-id", "secret", "assertion", authority=authority, transport=transport
+    )
+    token = await credential.get_token("scope")
+    assert token.token == expected_token
+
+    # authority can be configured via environment variable
+    with patch.dict("os.environ", {EnvironmentVariables.AZURE_AUTHORITY_HOST: authority}, clear=True):
+        credential = OnBehalfOfCredential(tenant_id, "client-id", "secret", "assertion", transport=transport)
+    token = await credential.get_token("scope")
+    assert token.token == expected_token
+
+
+@pytest.mark.asyncio
+async def test_policies_configurable():
+    policy = Mock(spec_set=SansIOHTTPPolicy, on_request=Mock(), on_exception=lambda _: False)
+
+    async def send(request, **_):
+        parsed = urlparse(request.url)
+        tenant = parsed.path.split("/")[1]
+        if "/oauth2/v2.0/token" not in parsed.path:
+            return get_discovery_response("https://{}/{}".format(parsed.netloc, tenant))
+        return mock_response(json_payload=build_aad_response(access_token="***"))
+
+    credential = OnBehalfOfCredential(
+        "tenant-id",
+        "client-id",
+        "client-secret",
+        "assertion",
+        policies=[ContentDecodePolicy(), policy],
+        transport=Mock(send=send),
+    )
+    await credential.get_token("scope")
+    assert policy.on_request.called
+
+
+def test_invalid_cert():
+    """The credential should raise ValueError when given invalid cert bytes"""
+    with pytest.raises(ValueError):
+        OnBehalfOfCredential("tenant-id", "client-id", b"not a cert", "assertion")
+
+
+@pytest.mark.asyncio
+async def test_refresh_token():
+    first_token = "***"
+    second_token = first_token * 2
+    refresh_token = "refresh-token"
+    requests = 0
+
+    async def send(request, **_):
+        nonlocal requests
+        assert requests < 3, "unexpected request"
+        requests += 1
+        if requests == 1:
+            assert "refresh_token" not in request.body
+            return mock_response(
+                json_payload=build_aad_response(access_token=first_token, refresh_token=refresh_token, expires_in=0)
+            )
+        if requests == 2:
+            assert request.body["refresh_token"] == refresh_token
+            return mock_response(json_payload=build_aad_response(access_token=second_token))
+
+    credential = OnBehalfOfCredential("tenant-id", "client-id", "secret", "assertion", transport=Mock(send=send))
+    token = await credential.get_token("scope")
+    assert token.token == first_token
+
+    token = await credential.get_token("scope")
+    assert token.token == second_token
+
+    assert requests == 2
+
+
+def test_tenant_id_validation():
+    """The credential should raise ValueError when given an invalid tenant_id"""
+    valid_ids = {"c878a2ab-8ef4-413b-83a0-199afb84d7fb", "contoso.onmicrosoft.com", "organizations", "common"}
+    for tenant in valid_ids:
+        OnBehalfOfCredential(tenant, "client-id", "secret", "assertion")
+    invalid_ids = {"my tenant", "my_tenant", "/", "\\", '"my-tenant"', "'my-tenant'"}
+    for tenant in invalid_ids:
+        with pytest.raises(ValueError):
+            OnBehalfOfCredential(tenant, "client-id", "secret", "assertion")
+
+
+@pytest.mark.asyncio
+async def test_no_scopes():
+    """The credential should raise ValueError when get_token is called with no scopes"""
+    credential = OnBehalfOfCredential("tenant-id", "client-id", "client-secret", "assertion")
+    with pytest.raises(ValueError):
+        await credential.get_token()