diff --git a/sdk/identity/azure-identity/README.md b/sdk/identity/azure-identity/README.md
index 690d317662cb..1132582d925e 100644
--- a/sdk/identity/azure-identity/README.md
+++ b/sdk/identity/azure-identity/README.md
@@ -5,16 +5,57 @@
 # Key concepts
 
 # Examples
-Shortest path to an access token:
+Authenticating as a service principal:
 ```py
+# using a client secret
 from azure.identity import ClientSecretCredential
-
 credential = ClientSecretCredential(client_id, secret, tenant_id)
 
 # all credentials implement get_token
 token = credential.get_token(scopes=["https://vault.azure.net/.default"])
+
+# using a certificate requires a thumbprint and PEM-encoded private key
+from azure.identity import CertificateCredential
+with open("private-key.pem") as f:
+    private_key = f.read()
+credential = CertificateCredential(client_id, tenant_id, private_key, thumbprint)
+```
+
+Authenticating via environment variables:
+```py
+from azure.identity import EnvironmentCredential
+
+# will authenticate with client secret or certificate,
+# depending on which environment variables are set
+# (see constants.py for expected variable names)
+credential = EnvironmentCredential()
+token = credential.get_token(scopes=["https://vault.azure.net/.default"])
+```
+
+Chaining together multiple credentials:
+```py
+from azure.identity import TokenCredentialChain
+
+# default credentials are environment then managed identity
+credential_chain = TokenCredentialChain.default()
+
+scopes = ["https://vault.azure.net/.default"]
+# the chain has a get_token method like all credentials
+token = credential_chain.get_token(scopes)  # try each credential in order, return the first token
 ```
 
+Authenticating from a service client:
+```py
+from azure.core.pipeline import Pipeline
+from azure.core.pipeline.policies import BearerTokenCredentialPolicy
+
+credential_chain = TokenCredentialChain.default()
+scopes = ["https://vault.azure.net/.default"]
+
+# BearerTokenCredentialPolicy gets tokens as necessary, adds appropriate auth headers to requests
+policies = [BearerTokenCredentialPolicy(credential=credential_chain, scopes=scopes)]
+pipeline = Pipeline(transport=some_transport, policies=policies)
+```
 # Troubleshooting
 
 # Next steps
diff --git a/sdk/identity/azure-identity/azure/identity/__init__.py b/sdk/identity/azure-identity/azure/identity/__init__.py
index 180c308436e4..f07942a4b7f8 100644
--- a/sdk/identity/azure-identity/azure/identity/__init__.py
+++ b/sdk/identity/azure-identity/azure/identity/__init__.py
@@ -4,13 +4,40 @@
 # license information.
 # --------------------------------------------------------------------------
 from .exceptions import AuthenticationError
-from .credentials import ClientSecretCredential, TokenCredentialChain
+from .credentials import (
+    CertificateCredential,
+    ClientSecretCredential,
+    EnvironmentCredential,
+    ManagedIdentityCredential,
+    TokenCredentialChain,
+)
 
-__all__ = ["AuthenticationError", "ClientSecretCredential", "TokenCredentialChain"]
+__all__ = [
+    "AuthenticationError",
+    "CertificateCredential",
+    "ClientSecretCredential",
+    "EnvironmentCredential",
+    "ManagedIdentityCredential",
+    "TokenCredentialChain",
+]
 
 try:
-    from .aio import AsyncClientSecretCredential, AsyncTokenCredentialChain
+    from .aio import (
+        AsyncCertificateCredential,
+        AsyncClientSecretCredential,
+        AsyncEnvironmentCredential,
+        AsyncManagedIdentityCredential,
+        AsyncTokenCredentialChain,
+    )
 
-    __all__.extend(["AsyncClientSecretCredential", "AsyncTokenCredentialChain"])
+    __all__.extend(
+        [
+            "AsyncCertificateCredential",
+            "AsyncClientSecretCredential",
+            "AsyncEnvironmentCredential",
+            "AsyncManagedIdentityCredential",
+            "AsyncTokenCredentialChain",
+        ]
+    )
 except SyntaxError:
     pass
diff --git a/sdk/identity/azure-identity/azure/identity/authn_client.py b/sdk/identity/azure-identity/azure/identity/_authn_client.py
similarity index 75%
rename from sdk/identity/azure-identity/azure/identity/authn_client.py
rename to sdk/identity/azure-identity/azure/identity/_authn_client.py
index aca01dd10bdc..c7e81f8c45cd 100644
--- a/sdk/identity/azure-identity/azure/identity/authn_client.py
+++ b/sdk/identity/azure-identity/azure/identity/_authn_client.py
@@ -6,9 +6,9 @@
 from time import time
 
 from azure.core import Configuration, HttpRequest
-from azure.core.pipeline import Pipeline
+from azure.core.pipeline import Pipeline, PipelineRequest
 from azure.core.pipeline.policies import ContentDecodePolicy, NetworkTraceLoggingPolicy, RetryPolicy
-from azure.core.pipeline.transport import RequestsTransport
+from azure.core.pipeline.transport import HttpTransport, RequestsTransport
 from msal import TokenCache
 
 from .exceptions import AuthenticationError
@@ -18,16 +18,21 @@
 except ImportError:
     TYPE_CHECKING = False
 if TYPE_CHECKING:
-    from typing import Any, Iterable, Mapping, Optional
+    from typing import Any, Dict, Iterable, Mapping, Optional
+    from azure.core.pipeline import PipelineResponse
+    from azure.core.pipeline.policies import HTTPPolicy
 
 
-class _AuthnClientBase(object):
+class AuthnClientBase(object):
+    """Sans I/O authentication client methods"""
+
     def __init__(self, auth_url, **kwargs):
+        # type: (str, Mapping[str, Any]) -> None
         if not auth_url:
-            raise ValueError("auth_url")
-        super(_AuthnClientBase, self).__init__(**kwargs)
-        self._cache = TokenCache()
+            raise ValueError("auth_url should be the URL of an OAuth endpoint")
+        super(AuthnClientBase, self).__init__()
         self._auth_url = auth_url
+        self._cache = TokenCache()
 
     def get_cached_token(self, scopes):
         # type: (Iterable[str]) -> Optional[str]
@@ -38,16 +43,8 @@ def get_cached_token(self, scopes):
                     return token["secret"]
         return None
 
-    def _prepare_request(self, method="POST", form_data=None, params=None):
-        request = HttpRequest(method, self._auth_url)
-        if form_data:
-            request.headers["Content-Type"] = "application/x-www-form-urlencoded"
-            request.set_formdata_body(form_data)
-        if params:
-            request.format_parameters(params)
-        return request
-
     def _deserialize_and_cache_token(self, response, scopes):
+        # type: (PipelineResponse, Iterable[str]) -> str
         try:
             if "deserialized_data" in response.context:
                 payload = response.context["deserialized_data"]
@@ -61,11 +58,23 @@ def _deserialize_and_cache_token(self, response, scopes):
         except Exception as ex:
             raise AuthenticationError("Authentication failed: {}".format(str(ex)))
 
+    def _prepare_request(self, method="POST", form_data=None, params=None):
+        # type: (Optional[str], Optional[Mapping[str, str]], Optional[Dict[str, str]]) -> HttpRequest
+        request = HttpRequest(method, self._auth_url)
+        if form_data:
+            request.headers["Content-Type"] = "application/x-www-form-urlencoded"
+            request.set_formdata_body(form_data)
+        if params:
+            request.format_parameters(params)
+        return request
+
+
+class AuthnClient(AuthnClientBase):
+    """Synchronous authentication client"""
 
-class AuthnClient(_AuthnClientBase):
     def __init__(self, auth_url, config=None, policies=None, transport=None, **kwargs):
+        # type: (str, Optional[Configuration], Optional[Iterable[HTTPPolicy]], Optional[HttpTransport], Mapping[str, Any]) -> None
         config = config or self.create_config(**kwargs)
-        # TODO: ContentDecodePolicy doesn't accept kwargs
         policies = policies or [ContentDecodePolicy(), config.logging_policy, config.retry_policy]
         if not transport:
             transport = RequestsTransport(configuration=config)
@@ -73,6 +82,7 @@ def __init__(self, auth_url, config=None, policies=None, transport=None, **kwarg
         super(AuthnClient, self).__init__(auth_url, **kwargs)
 
     def request_token(self, scopes, method="POST", form_data=None, params=None):
+        # type: (Iterable[str], Optional[str], Optional[Mapping[str, str]], Optional[Dict[str, str]]) -> str
         request = self._prepare_request(method, form_data, params)
         response = self._pipeline.run(request, stream=False)
         token = self._deserialize_and_cache_token(response, scopes)
@@ -83,5 +93,5 @@ def create_config(**kwargs):
         # type: (Mapping[str, Any]) -> Configuration
         config = Configuration(**kwargs)
         config.logging_policy = NetworkTraceLoggingPolicy(**kwargs)
-        config.retry_policy = RetryPolicy(retry_on_status_codes=[404, 429] + [x for x in range(500, 600)], **kwargs)
+        config.retry_policy = RetryPolicy(retry_on_status_codes=[404, 429] + list(range(500, 600)), **kwargs)
         return config
diff --git a/sdk/identity/azure-identity/azure/identity/_base.py b/sdk/identity/azure-identity/azure/identity/_base.py
new file mode 100644
index 000000000000..a3221ae34c46
--- /dev/null
+++ b/sdk/identity/azure-identity/azure/identity/_base.py
@@ -0,0 +1,52 @@
+# -------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See LICENSE.txt in the project root for
+# license information.
+# --------------------------------------------------------------------------
+from msal.oauth2cli import JwtSigner
+
+from .constants import OAUTH_ENDPOINT
+
+try:
+    from typing import TYPE_CHECKING
+except ImportError:
+    TYPE_CHECKING = False
+
+if TYPE_CHECKING:
+    # pylint:disable=unused-import
+    from typing import Any, Mapping
+
+
+class ClientSecretCredentialBase(object):
+    def __init__(self, client_id, secret, tenant_id, **kwargs):
+        # type: (str, str, str, Mapping[str, Any]) -> None
+        if not client_id:
+            raise ValueError("client_id should be the id of an Azure Active Directory application")
+        if not secret:
+            raise ValueError("secret should be an Azure Active Directory application's client secret")
+        if not tenant_id:
+            raise ValueError("tenant_id should be an Azure Active Directory tenant's id (also called its 'directory id')")
+        self._form_data = {"client_id": client_id, "client_secret": secret, "grant_type": "client_credentials"}
+        super(ClientSecretCredentialBase, self).__init__()
+
+
+class CertificateCredentialBase(object):
+    def __init__(self, client_id, tenant_id, certificate_path, **kwargs):
+        # type: (str, str, str, Mapping[str, Any]) -> None
+        if not certificate_path:
+            # TODO: support PFX
+            raise ValueError("certificate_path must be the path to a PEM-encoded private key file")
+
+        super(CertificateCredentialBase, self).__init__()
+        auth_url = OAUTH_ENDPOINT.format(tenant_id)
+
+        with open(certificate_path) as pem:
+            private_key = pem.read()
+        signer = JwtSigner(private_key, "RS256")
+        client_assertion = signer.sign_assertion(audience=auth_url, issuer=client_id)
+        self._form_data = {
+            "client_assertion": client_assertion,
+            "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
+            "client_id": client_id,
+            "grant_type": "client_credentials",
+        }
diff --git a/sdk/identity/azure-identity/azure/identity/aio/__init__.py b/sdk/identity/azure-identity/azure/identity/aio/__init__.py
index 476dcc21e9dc..f4f8b6c0e22a 100644
--- a/sdk/identity/azure-identity/azure/identity/aio/__init__.py
+++ b/sdk/identity/azure-identity/azure/identity/aio/__init__.py
@@ -3,6 +3,18 @@
 # Licensed under the MIT License. See LICENSE.txt in the project root for
 # license information.
 # --------------------------------------------------------------------------
-from .credentials import AsyncClientSecretCredential, AsyncTokenCredentialChain
+from .credentials import (
+    AsyncCertificateCredential,
+    AsyncClientSecretCredential,
+    AsyncEnvironmentCredential,
+    AsyncManagedIdentityCredential,
+    AsyncTokenCredentialChain,
+)
 
-__all__ = ["AsyncClientSecretCredential", "AsyncTokenCredentialChain"]
+__all__ = [
+    "AsyncCertificateCredential",
+    "AsyncClientSecretCredential",
+    "AsyncEnvironmentCredential",
+    "AsyncManagedIdentityCredential",
+    "AsyncTokenCredentialChain",
+]
diff --git a/sdk/identity/azure-identity/azure/identity/aio/authn_client.py b/sdk/identity/azure-identity/azure/identity/aio/_authn_client.py
similarity index 85%
rename from sdk/identity/azure-identity/azure/identity/aio/authn_client.py
rename to sdk/identity/azure-identity/azure/identity/aio/_authn_client.py
index 6886b945642d..d42231dcfef5 100644
--- a/sdk/identity/azure-identity/azure/identity/aio/authn_client.py
+++ b/sdk/identity/azure-identity/azure/identity/aio/_authn_client.py
@@ -3,7 +3,7 @@
 # Licensed under the MIT License. See LICENSE.txt in the project root for
 # license information.
 # --------------------------------------------------------------------------
-from typing import Any, Iterable, Mapping, Optional
+from typing import Any, Dict, Iterable, Mapping, Optional
 
 from azure.core import Configuration
 from azure.core.pipeline import AsyncPipeline
@@ -11,10 +11,12 @@
 from azure.core.pipeline.transport import AsyncHttpTransport
 from azure.core.pipeline.transport.requests_asyncio import AsyncioRequestsTransport
 
-from ..authn_client import _AuthnClientBase
+from .._authn_client import AuthnClientBase
 
 
-class AsyncAuthnClient(_AuthnClientBase):
+class AsyncAuthnClient(AuthnClientBase):
+    """Async authentication client"""
+
     def __init__(
         self,
         auth_url: str,
@@ -24,7 +26,6 @@ def __init__(
         **kwargs: Mapping[str, Any]
     ) -> None:
         config = config or self.create_config(**kwargs)
-        # TODO: ContentDecodePolicy doesn't accept kwargs
         policies = policies or [ContentDecodePolicy(), config.logging_policy, config.retry_policy]
         if not transport:
             transport = AsyncioRequestsTransport(configuration=config)
@@ -36,7 +37,7 @@ async def request_token(
         scopes: Iterable[str],
         method: Optional[str] = "POST",
         form_data: Optional[Mapping[str, str]] = None,
-        params: Optional[Mapping[str, str]] = None,
+        params: Optional[Dict[str, str]] = None,
     ) -> str:
         request = self._prepare_request(method, form_data, params)
         response = await self._pipeline.run(request, stream=False)
@@ -48,6 +49,6 @@ def create_config(**kwargs: Mapping[str, Any]) -> Configuration:
         config = Configuration(**kwargs)
         config.logging_policy = NetworkTraceLoggingPolicy(**kwargs)
         config.retry_policy = AsyncRetryPolicy(
-            retry_on_status_codes=[404, 429] + [x for x in range(500, 600)], **kwargs
+            retry_on_status_codes=[404, 429] + list(range(500, 600)), **kwargs
         )
         return config
diff --git a/sdk/identity/azure-identity/azure/identity/aio/credentials.py b/sdk/identity/azure-identity/azure/identity/aio/credentials.py
index 60474f476f95..cc4130efda7b 100644
--- a/sdk/identity/azure-identity/azure/identity/aio/credentials.py
+++ b/sdk/identity/azure-identity/azure/identity/aio/credentials.py
@@ -3,64 +3,125 @@
 # Licensed under the MIT License. See LICENSE.txt in the project root for
 # license information.
 # --------------------------------------------------------------------------
-from typing import Any, Dict, Iterable, Mapping, Optional
+import os
+from typing import Any, Iterable, Mapping, Optional, Union
 
 from azure.core import Configuration
-from azure.core.pipeline.policies import HTTPPolicy
+from azure.core.pipeline.policies import ContentDecodePolicy, HeadersPolicy, NetworkTraceLoggingPolicy, AsyncRetryPolicy
 
-from .authn_client import AsyncAuthnClient
+from ._authn_client import AsyncAuthnClient
+from .._base import ClientSecretCredentialBase, CertificateCredentialBase
+from ..constants import EnvironmentVariables, IMDS_ENDPOINT, OAUTH_ENDPOINT
 from ..credentials import TokenCredentialChain
 from ..exceptions import AuthenticationError
 
-
 # pylint:disable=too-few-public-methods
 
-# TODO: could share more code with sync
-class _AsyncClientCredentialBase(object):
-    _OAUTH_ENDPOINT = "https://login.microsoftonline.com/{}/oauth2/v2.0/token"
 
+class AsyncClientSecretCredential(ClientSecretCredentialBase):
     def __init__(
         self,
         client_id: str,
+        secret: str,
         tenant_id: str,
         config: Optional[Configuration] = None,
-        policies: Optional[Iterable[HTTPPolicy]] = None,
         **kwargs: Mapping[str, Any]
     ) -> None:
-        if not client_id:
-            raise ValueError("client_id")
-        if not tenant_id:
-            raise ValueError("tenant_id")
-        self._client = AsyncAuthnClient(self._OAUTH_ENDPOINT.format(tenant_id), config, policies, **kwargs)
-        self._form_data = {}  # type: Dict[str, str]
+        super(AsyncClientSecretCredential, self).__init__(client_id, secret, tenant_id, **kwargs)
+        self._client = AsyncAuthnClient(OAUTH_ENDPOINT.format(tenant_id), config, **kwargs)
 
     async def get_token(self, scopes: Iterable[str]) -> str:
-        data = self._form_data.copy()
-        data["scope"] = " ".join(scopes)
         token = self._client.get_cached_token(scopes)
         if not token:
+            data = dict(self._form_data, scope=" ".join(scopes))
             token = await self._client.request_token(scopes, form_data=data)
         return token  # type: ignore
 
 
-class AsyncClientSecretCredential(_AsyncClientCredentialBase):
+class AsyncCertificateCredential(CertificateCredentialBase):
     def __init__(
         self,
         client_id: str,
-        secret: str,
         tenant_id: str,
+        certificate_path: str,
         config: Optional[Configuration] = None,
         **kwargs: Mapping[str, Any]
     ) -> None:
-        if not secret:
-            raise ValueError("secret")
-        super(AsyncClientSecretCredential, self).__init__(client_id, tenant_id, config, **kwargs)
-        self._form_data = {"client_id": client_id, "client_secret": secret, "grant_type": "client_credentials"}
+        super(AsyncCertificateCredential, self).__init__(client_id, tenant_id, certificate_path, **kwargs)
+        self._client = AsyncAuthnClient(OAUTH_ENDPOINT.format(tenant_id), config, **kwargs)
+
+    async def get_token(self, scopes: Iterable[str]) -> str:
+        token = self._client.get_cached_token(scopes)
+        if not token:
+            data = dict(self._form_data, scope=" ".join(scopes))
+            token = await self._client.request_token(scopes, form_data=data)
+        return token  # type: ignore
+
+
+class AsyncEnvironmentCredential:
+    def __init__(self, **kwargs: Mapping[str, Any]) -> None:
+        self._credential = None  # type: Optional[Union[AsyncCertificateCredential, AsyncClientSecretCredential]]
+
+        if all(os.environ.get(v) is not None for v in EnvironmentVariables.CLIENT_SECRET_VARS):
+            self._credential = AsyncClientSecretCredential(
+                client_id=os.environ[EnvironmentVariables.AZURE_CLIENT_ID],
+                secret=os.environ[EnvironmentVariables.AZURE_CLIENT_SECRET],
+                tenant_id=os.environ[EnvironmentVariables.AZURE_TENANT_ID],
+                **kwargs
+            )
+        elif all(os.environ.get(v) is not None for v in EnvironmentVariables.CERT_VARS):
+            self._credential = AsyncCertificateCredential(
+                client_id=os.environ[EnvironmentVariables.AZURE_CLIENT_ID],
+                tenant_id=os.environ[EnvironmentVariables.AZURE_TENANT_ID],
+                certificate_path=os.environ[EnvironmentVariables.AZURE_CLIENT_CERTIFICATE_PATH],
+                **kwargs
+            )
+
+    async def get_token(self, scopes: Iterable[str]) -> str:
+        if not self._credential:
+            message = "Missing environment settings. To authenticate with a client secret, set {}. To authenticate with a certificate, set {}.".format(
+                ", ".join(EnvironmentVariables.CLIENT_SECRET_VARS), ", ".join(EnvironmentVariables.CERT_VARS)
+            )
+            raise AuthenticationError(message)
+        return await self._credential.get_token(scopes)
+
+
+class AsyncManagedIdentityCredential:
+    def __init__(self, config: Optional[Configuration] = None, **kwargs: Mapping[str, Any]) -> None:
+        config = config or self.create_config(**kwargs)
+        policies = [config.header_policy, ContentDecodePolicy(), config.logging_policy, config.retry_policy]
+        self._client = AsyncAuthnClient(IMDS_ENDPOINT, config, policies)
+
+    async def get_token(self, scopes: Iterable[str]) -> str:
+        scopes = list(scopes)
+        if len(scopes) != 1:
+            raise ValueError("Managed identity credential supports one scope per request")
+        token = self._client.get_cached_token(scopes)
+        if not token:
+            resource = scopes[0].rstrip("/.default")
+            token = await self._client.request_token(
+                scopes, method="GET", params={"api-version": "2018-02-01", "resource": resource}
+            )
+        return token  # type: ignore
+
+    @staticmethod
+    def create_config(**kwargs: Mapping[str, Any]) -> Configuration:
+        timeout = kwargs.pop("connection_timeout", 2)
+        config = Configuration(connection_timeout=timeout, **kwargs)
+        config.header_policy = HeadersPolicy(base_headers={"Metadata": "true"}, **kwargs)
+        config.logging_policy = NetworkTraceLoggingPolicy(**kwargs)
+        retries = kwargs.pop("retry_total", 5)
+        config.retry_policy = AsyncRetryPolicy(retry_total=retries, retry_on_status_codes=[404, 429] + list(range(500, 600)), **kwargs)
+        return config
 
 
 class AsyncTokenCredentialChain(TokenCredentialChain):
     """A sequence of token credentials"""
 
+    @classmethod
+    def default(cls):
+        return cls([AsyncEnvironmentCredential(), AsyncManagedIdentityCredential()])
+
     async def get_token(self, scopes: Iterable[str]) -> str:
         """Attempts to get a token from each credential, in order, returning the first token.
            If no token is acquired, raises an exception listing error messages.
diff --git a/sdk/identity/azure-identity/azure/identity/constants.py b/sdk/identity/azure-identity/azure/identity/constants.py
new file mode 100644
index 000000000000..0c29cc2559e8
--- /dev/null
+++ b/sdk/identity/azure-identity/azure/identity/constants.py
@@ -0,0 +1,22 @@
+# -------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See LICENSE.txt in the project root for
+# license information.
+# --------------------------------------------------------------------------
+
+
+class EnvironmentVariables:
+    AZURE_CLIENT_ID = "AZURE_CLIENT_ID"
+    AZURE_CLIENT_SECRET = "AZURE_CLIENT_SECRET"
+    AZURE_TENANT_ID = "AZURE_TENANT_ID"
+    CLIENT_SECRET_VARS = (AZURE_CLIENT_ID, AZURE_CLIENT_SECRET, AZURE_TENANT_ID)
+
+    AZURE_CLIENT_CERTIFICATE_PATH = "AZURE_CLIENT_CERTIFICATE_PATH"
+    CERT_VARS = (AZURE_CLIENT_ID, AZURE_CLIENT_CERTIFICATE_PATH, AZURE_TENANT_ID)
+
+
+# https://docs.microsoft.com/en-us/azure/active-directory/managed-identities-azure-resources/how-to-use-vm-token#get-a-token-using-http
+IMDS_ENDPOINT = "http://169.254.169.254/metadata/identity/oauth2/token"
+
+# TODO: other clouds have other endpoints
+OAUTH_ENDPOINT = "https://login.microsoftonline.com/{}/oauth2/v2.0/token"
diff --git a/sdk/identity/azure-identity/azure/identity/credentials.py b/sdk/identity/azure-identity/azure/identity/credentials.py
index 9b71bf3cd2dd..d0293ad7cc7d 100644
--- a/sdk/identity/azure-identity/azure/identity/credentials.py
+++ b/sdk/identity/azure-identity/azure/identity/credentials.py
@@ -3,9 +3,14 @@
 # Licensed under the MIT License. See LICENSE.txt in the project root for
 # license information.
 # --------------------------------------------------------------------------
+import os
+
 from azure.core import Configuration
+from azure.core.pipeline.policies import ContentDecodePolicy, HeadersPolicy, NetworkTraceLoggingPolicy, RetryPolicy
 
-from .authn_client import AuthnClient
+from ._authn_client import AuthnClient
+from ._base import ClientSecretCredentialBase, CertificateCredentialBase
+from .constants import EnvironmentVariables, IMDS_ENDPOINT, OAUTH_ENDPOINT
 from .exceptions import AuthenticationError
 
 try:
@@ -15,42 +20,109 @@
 
 if TYPE_CHECKING:
     # pylint:disable=unused-import
-    from typing import Any, Dict, Iterable, Mapping, Optional
-    from azure.core.pipeline.policies import HTTPPolicy
+    from typing import Any, Iterable, List, Mapping, Optional, Union
     from azure.core.credentials import SupportsGetToken
 
 # pylint:disable=too-few-public-methods
 
 
-class _ClientCredentialBase(object):
-    _OAUTH_ENDPOINT = "https://login.microsoftonline.com/{}/oauth2/v2.0/token"
+class ClientSecretCredential(ClientSecretCredentialBase):
+    """Authenticates with a client secret"""
 
-    def __init__(self, client_id, tenant_id, config=None, policies=None, **kwargs):
-        # type: (str, str, Optional[Configuration], Optional[Iterable[HTTPPolicy]], Mapping[str, Any]) -> None
-        if not client_id:
-            raise ValueError("client_id")
-        if not tenant_id:
-            raise ValueError("tenant_id")
-        self._client = AuthnClient(self._OAUTH_ENDPOINT.format(tenant_id), config, policies, **kwargs)
-        self._form_data = {}  # type: Dict[str, str]
+    def __init__(self, client_id, secret, tenant_id, config=None, **kwargs):
+        # type: (str, str, str, Optional[Configuration], Mapping[str, Any]) -> None
+        super(ClientSecretCredential, self).__init__(client_id, secret, tenant_id, **kwargs)
+        self._client = AuthnClient(OAUTH_ENDPOINT.format(tenant_id), config, **kwargs)
 
     def get_token(self, scopes):
         # type: (Iterable[str]) -> str
-        data = self._form_data.copy()
-        data["scope"] = " ".join(scopes)
         token = self._client.get_cached_token(scopes)
         if not token:
-            return self._client.request_token(scopes, form_data=data)
+            data = dict(self._form_data, scope=" ".join(scopes))
+            token = self._client.request_token(scopes, form_data=data)
         return token
 
 
-class ClientSecretCredential(_ClientCredentialBase):
-    def __init__(self, client_id, secret, tenant_id, config=None, **kwargs):
+class CertificateCredential(CertificateCredentialBase):
+    """Authenticates with a certificate"""
+
+    def __init__(self, client_id, tenant_id, certificate_path, config=None, **kwargs):
         # type: (str, str, str, Optional[Configuration], Mapping[str, Any]) -> None
-        if not secret:
-            raise ValueError("secret")
-        super(ClientSecretCredential, self).__init__(client_id, tenant_id, config, **kwargs)
-        self._form_data = {"client_id": client_id, "client_secret": secret, "grant_type": "client_credentials"}
+        self._client = AuthnClient(OAUTH_ENDPOINT.format(tenant_id), config, **kwargs)
+        super(CertificateCredential, self).__init__(client_id, tenant_id, certificate_path, **kwargs)
+
+    def get_token(self, scopes):
+        # type: (Iterable[str]) -> str
+        token = self._client.get_cached_token(scopes)
+        if not token:
+            data = dict(self._form_data, scope=" ".join(scopes))
+            token = self._client.request_token(scopes, form_data=data)
+        return token
+
+
+class EnvironmentCredential:
+    """Authenticates with a secret or certificate using environment variable settings"""
+
+    def __init__(self, **kwargs):
+        # type: (Mapping[str, Any]) -> None
+        self._credential = None  # type: Optional[Union[CertificateCredential, ClientSecretCredential]]
+
+        if all(os.environ.get(v) is not None for v in EnvironmentVariables.CLIENT_SECRET_VARS):
+            self._credential = ClientSecretCredential(
+                client_id=os.environ[EnvironmentVariables.AZURE_CLIENT_ID],
+                secret=os.environ[EnvironmentVariables.AZURE_CLIENT_SECRET],
+                tenant_id=os.environ[EnvironmentVariables.AZURE_TENANT_ID],
+                **kwargs
+            )
+        elif all(os.environ.get(v) is not None for v in EnvironmentVariables.CERT_VARS):
+            self._credential = CertificateCredential(
+                client_id=os.environ[EnvironmentVariables.AZURE_CLIENT_ID],
+                tenant_id=os.environ[EnvironmentVariables.AZURE_TENANT_ID],
+                certificate_path=os.environ[EnvironmentVariables.AZURE_CLIENT_CERTIFICATE_PATH],
+                **kwargs
+            )
+
+    def get_token(self, scopes):
+        # type: (Iterable[str]) -> str
+        if not self._credential:
+            message = "Missing environment settings. To authenticate with a client secret, set {}. To authenticate with a certificate, set {}.".format(
+                ", ".join(EnvironmentVariables.CLIENT_SECRET_VARS), ", ".join(EnvironmentVariables.CERT_VARS)
+            )
+            raise AuthenticationError(message)
+        return self._credential.get_token(scopes)
+
+
+class ManagedIdentityCredential:
+    """Authenticates with a managed identity"""
+
+    def __init__(self, config=None, **kwargs):
+        # type: (Optional[Configuration], Mapping[str, Any]) -> None
+        config = config or self.create_config(**kwargs)
+        policies = [config.header_policy, ContentDecodePolicy(), config.logging_policy, config.retry_policy]
+        self._client = AuthnClient(IMDS_ENDPOINT, config, policies)
+
+    @staticmethod
+    def create_config(**kwargs):
+        # type: (Mapping[str, str]) -> Configuration
+        timeout = kwargs.pop("connection_timeout", 2)
+        config = Configuration(connection_timeout=timeout, **kwargs)
+        config.header_policy = HeadersPolicy(base_headers={"Metadata": "true"}, **kwargs)
+        config.logging_policy = NetworkTraceLoggingPolicy(**kwargs)
+        retries = kwargs.pop("retry_total", 5)
+        config.retry_policy = RetryPolicy(retry_total=retries, retry_on_status_codes=[404, 429] + list(range(500, 600)), **kwargs)
+        return config
+
+    def get_token(self, scopes):
+        # type: (List[str]) -> str
+        if len(scopes) != 1:
+            raise ValueError("Managed identity credential supports one scope per request")
+        token = self._client.get_cached_token(scopes)
+        if not token:
+            resource = scopes[0].rstrip("/.default")
+            token = self._client.request_token(
+                scopes, method="GET", params={"api-version": "2018-02-01", "resource": resource}
+            )
+        return token
 
 
 class TokenCredentialChain:
@@ -62,6 +134,10 @@ def __init__(self, credentials):
             raise ValueError("at least one credential is required")
         self._credentials = credentials
 
+    @classmethod
+    def default(cls):
+        return cls([EnvironmentCredential(), ManagedIdentityCredential()])
+
     def get_token(self, scopes):
         # type: (Iterable[str]) -> str
         """Attempts to get a token from each credential, in order, returning the first token.
diff --git a/sdk/identity/azure-identity/azure/identity/exceptions.py b/sdk/identity/azure-identity/azure/identity/exceptions.py
index f3edd0b45dee..672feb69be57 100644
--- a/sdk/identity/azure-identity/azure/identity/exceptions.py
+++ b/sdk/identity/azure-identity/azure/identity/exceptions.py
@@ -3,13 +3,8 @@
 # Licensed under the MIT License. See LICENSE.txt in the project root for
 # license information.
 # --------------------------------------------------------------------------
+from azure.core.exceptions import AzureError
 
 
-# TODO: probably a better base for this in azure-core
-class AuthenticationError(Exception):
-    def __init__(self, message):
-        # type: (str) -> None
-        self.message = message
-
-    def __str__(self):
-        return self.message
+class AuthenticationError(AzureError):
+    pass
diff --git a/sdk/identity/azure-identity/azure/identity/version.py b/sdk/identity/azure-identity/azure/identity/version.py
index 937c2e059114..c78e131faf62 100644
--- a/sdk/identity/azure-identity/azure/identity/version.py
+++ b/sdk/identity/azure-identity/azure/identity/version.py
@@ -3,4 +3,4 @@
 # Licensed under the MIT License. See LICENSE.txt in the project root for
 # license information.
 # --------------------------------------------------------------------------
-VERSION = "0.0.1"
+VERSION = "0.1.0"
diff --git a/sdk/identity/azure-identity/dev_requirements.txt b/sdk/identity/azure-identity/dev_requirements.txt
index e30ad303788a..4ac84a358c64 100644
--- a/sdk/identity/azure-identity/dev_requirements.txt
+++ b/sdk/identity/azure-identity/dev_requirements.txt
@@ -1 +1,2 @@
+-e ../../core/azure-core
 typing_extensions>=3.7.2
diff --git a/sdk/identity/azure-identity/setup.py b/sdk/identity/azure-identity/setup.py
index dab4f6d8e050..a76a2f21c32b 100644
--- a/sdk/identity/azure-identity/setup.py
+++ b/sdk/identity/azure-identity/setup.py
@@ -5,7 +5,6 @@
 # Licensed under the MIT License. See License.txt in the project root for
 # license information.
 # --------------------------------------------------------------------------
-
 import re
 import os.path
 from io import open
@@ -37,27 +36,28 @@
 
 # Version extraction inspired from 'requests'
 with open(os.path.join(package_folder_path, "version.py"), "r") as fd:
-    version = re.search(r'^VERSION\s*=\s*[\'"]([^\'"]*)[\'"]', fd.read(), re.MULTILINE).group(1)
+    VERSION = re.search(r'^VERSION\s*=\s*[\'"]([^\'"]*)[\'"]', fd.read(), re.MULTILINE).group(1)
 
-if not version:
+if not VERSION:
     raise RuntimeError("Cannot find version information")
 
-# with open('README.rst', encoding='utf-8') as f:
-#     readme = f.read()
+with open("README.md", encoding="utf-8") as f:
+    README = f.read()
 # with open('HISTORY.rst', encoding='utf-8') as f:
 #     history = f.read()
 
 setup(
     name=PACKAGE_NAME,
-    version=version,
+    version=VERSION,
     description="Microsoft Azure {} Library for Python".format(PACKAGE_PPRINT_NAME),
-    # long_description=readme + '\n\n' + history,
+    long_description=README,
+    long_description_content_type="text/markdown",
     license="MIT License",
     author="Microsoft Corporation",
     # author_email='',
     url="https://github.com/Azure/azure-sdk-for-python",
     classifiers=[
-        "Development Status :: 2 - Pre-Alpha",
+        "Development Status :: 4 - Beta",
         "Programming Language :: Python",
         "Programming Language :: Python :: 2",
         "Programming Language :: Python :: 2.7",
@@ -76,10 +76,9 @@
             "azure",
         ]
     ),
-    install_requires=["azure-core~=1.0.0", "msal>=0.3.1"],
-    extras_require={
-        ":python_version<'3.0'": ["azure-nspkg"],
-        ":python_version<'3.4'": ["enum34>=1.0.4"],
-        ":python_version<'3.5'": ["typing"],
-    },
+    install_requires=[
+        # "azure-core~=1.0.0", TODO: commented until azure-core is published
+        "msal~=0.3.1"
+    ],
+    extras_require={":python_version<'3.0'": ["azure-nspkg"]},
 )
diff --git a/sdk/identity/azure-identity/tests/conftest.py b/sdk/identity/azure-identity/tests/conftest.py
index 01957ae3dd87..c79ec90326a3 100644
--- a/sdk/identity/azure-identity/tests/conftest.py
+++ b/sdk/identity/azure-identity/tests/conftest.py
@@ -6,6 +6,5 @@
 import sys
 
 # Ignore collection of async tests for Python 2
-collect_ignore = []
 if sys.version_info < (3, 5):
-    collect_ignore.append("test_identity_async.py")
+    collect_ignore_glob = ["*_async.py"]
diff --git a/sdk/identity/azure-identity/tests/private-key.pem b/sdk/identity/azure-identity/tests/private-key.pem
new file mode 100644
index 000000000000..636550d834dd
--- /dev/null
+++ b/sdk/identity/azure-identity/tests/private-key.pem
@@ -0,0 +1,35 @@
+Bag Attributes
+    Microsoft Local Key set: <No Values>
+    localKeyID: 01 00 00 00 
+    friendlyName: te-6c7ca2ba-5bec-4802-bde3-1755fb26d651
+    Microsoft CSP Name: Microsoft Software Key Storage Provider
+Key Attributes
+    X509v3 Key Usage: 90 
+-----BEGIN PRIVATE KEY-----
+MIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggSkAgEAAoIBAQDZnnXDSPMKmbA+
+Lx+iZio78D0RS61lP5UfRQJU4NMnZhA1qEdVsvQfR+24eN4U34zXYsHMGP57NVRT
+GYpy842oH6XAJ3uq59FgkO7NB6z1FXqnFbfAese+mssSlXtc91Ach4BpePk2/umO
+eCZp/7l0pQPKMrawY8z7zogTnegAIwSDvVeGU0Zap4WPRMmMDAE38s7i0Dfxa1kh
+9/TXvH/x6/+uNIyTIZ8PMH2uv9w/LNex4vh+Pw8jwo5XdVRXRessdMt2I3dVUtkw
+AGHh1TA9Qkflji9tXjv3r2nRHqMLcqhPeIDJVkqUIqSiINQ61gy7CdkMxalp0MxE
+6Mjo7eBpAgMBAAECggEAdv0XsvGeQnuKTFYD3A40pZVULrLMWoILjY90GOjdS7uY
+vV4HsyooJTp1Ftqvw4YAQnyzLl+0NbYRJ2bdtsDJAdZcENcF3Yrnhv94Mw8xWMin
+ydgsIsh/kw6cXsrxKwHnAdJtOj51Ncbn+YhkqKy0wLzBd7uG/Kd1G3HwIZnDkt6Q
+E6J4G0u9atEX7pvV8/JBK3QxAZoOo1FkOO2IDk92z85G9+7GysDRssu6erzihX/b
+/fG/DVx1vjeZz1jUC3rIbOhEmq1To0FdOlyRT79IW0RZWRNFAEaFE9sNYR4ZSsgZ
+Bwznp1vwtwefPk31WJkGf+WRUERhG8vycOVOBErXCQKBgQD8fgO9JSESdOm8QIHt
+kbqzA+IdyOL7lnK8SdWNiCqdX3e7ZokbYhlUvj6Hf3gGQUTHwTvGa2LIifrSo7jh
+VvO7kIBRg7eMhK6mZMArEINFqEjlqnC5w+Js7hrhjMwPjwIzmcZ3R9GVWl1aq543
+Q8S+0aByxd7HKGG2DOtcdZxW1wKBgQDcpGtKslM3ql7hIMigrwg2vpB7OOj2P4Ja
+cfY/DE4NJ+ZlAzPyDy/Hho7oSxb9ez0xxpOVz5/FpU0Ve1YfRU/41LAndHjudE3z
+sxr7TjGcNS6Ao9790RCqtuE1LipaJ+pKvMS/4J94EM3jbpfZB0Dckmk0gz4lTvUM
+5ZedRpravwKBgQDIfyRm6PnnFxGX3D2QMb1oc7f1YNTlZSV84MCEX9E/IFUKabSM
+Gwz0XxF2NUFQ7jk4yfe2awWJKxASfdHMlmh605cho494NNAe7zgtujITeTtRrFNR
+H/xH9ZdA7bYI0M21vfF8PHpvt88Ttd2wEs9Dm2BmYzuxOB7HGmE3DWl1BwKBgQCJ
+1B+9wpWfYUrxoRQS5CPiZrpEbzF/mf6o1yW3Ds23BCS1FwIdBIWZQyIEU9vhrll0
+vZI19EPfKDp139zVneuuCdacXvKoKnkDce+56oetB7+r1jIXJcEeky0tllAYj3SZ
+CUByiDO1wfGLT+uFRDWtU7xqdE2e6qrDSqyiL5fOawKBgDkkyj0ayZnxcMGgM5W6
+g1bwx5q3Kjc0FzGd8cmO5BLmxW8sCcQykOoZb42qmlGEJU0Q8ppiz82xmI7l+Sml
+QdVTbroI9ECxwRFX97pu8gBR4rI1lyfCllI3Pm2Y6cvapki0eq+1jEReSclsyMK9
+mvgSQ07XbJjrqb6hJJo6iLmh
+-----END PRIVATE KEY-----
diff --git a/sdk/identity/azure-identity/tests/test_identity.py b/sdk/identity/azure-identity/tests/test_identity.py
index a6ef08e2e1b8..6bd626141e1d 100644
--- a/sdk/identity/azure-identity/tests/test_identity.py
+++ b/sdk/identity/azure-identity/tests/test_identity.py
@@ -4,17 +4,26 @@
 # license information.
 # --------------------------------------------------------------------------
 import json
+import os
 import time
 import uuid
 
 try:
-    from unittest.mock import Mock, MagicMock
+    from unittest.mock import Mock
 except ImportError:  # python < 3.3
     from mock import Mock
 
 import pytest
 import requests
-from azure.identity import AuthenticationError, ClientSecretCredential, TokenCredentialChain
+from azure.identity import (
+    AuthenticationError,
+    ClientSecretCredential,
+    EnvironmentCredential,
+    TokenCredentialChain,
+    ManagedIdentityCredential,
+    CertificateCredential,
+)
+from azure.identity.constants import EnvironmentVariables
 
 
 def test_client_secret_credential_cache(monkeypatch):
@@ -45,6 +54,59 @@ def test_client_secret_credential_cache(monkeypatch):
     assert mock_send.call_count == 2
 
 
+def test_client_secret_environment_credential(monkeypatch):
+    client_id = "fake-client-id"
+    secret = "fake-client-secret"
+    tenant_id = "fake-tenant-id"
+
+    monkeypatch.setenv(EnvironmentVariables.AZURE_CLIENT_ID, client_id)
+    monkeypatch.setenv(EnvironmentVariables.AZURE_CLIENT_SECRET, secret)
+    monkeypatch.setenv(EnvironmentVariables.AZURE_TENANT_ID, tenant_id)
+
+    success_message = "request passed validation"
+
+    def validate_request(request, **kwargs):
+        assert tenant_id in request.url
+        assert request.data["client_id"] == client_id
+        assert request.data["client_secret"] == secret
+        # raising here makes mocking a transport response unnecessary
+        raise AuthenticationError(success_message)
+
+    credential = EnvironmentCredential(transport=Mock(send=validate_request))
+    with pytest.raises(AuthenticationError) as ex:
+        credential.get_token(("",))
+    assert str(ex.value) == success_message
+
+
+def test_cert_environment_credential(monkeypatch):
+    client_id = "fake-client-id"
+    pem_path = os.path.join(os.path.dirname(__file__), "private-key.pem")
+    tenant_id = "fake-tenant-id"
+
+    monkeypatch.setenv(EnvironmentVariables.AZURE_CLIENT_ID, client_id)
+    monkeypatch.setenv(EnvironmentVariables.AZURE_CLIENT_CERTIFICATE_PATH, pem_path)
+    monkeypatch.setenv(EnvironmentVariables.AZURE_TENANT_ID, tenant_id)
+
+    success_message = "request passed validation"
+
+    def validate_request(request, **kwargs):
+        assert tenant_id in request.url
+        assert request.data["client_id"] == client_id
+        assert request.data["grant_type"] == "client_credentials"
+        # raising here makes mocking a transport response unnecessary
+        raise AuthenticationError(success_message)
+
+    credential = EnvironmentCredential(transport=Mock(send=validate_request))
+    with pytest.raises(AuthenticationError) as ex:
+        credential.get_token(("",))
+    assert str(ex.value) == success_message
+
+
+def test_environment_credential_error():
+    with pytest.raises(AuthenticationError):
+        EnvironmentCredential().get_token(("",))
+
+
 def test_credential_chain_error_message():
     def raise_authn_error(message):
         raise AuthenticationError(message)
@@ -88,3 +150,62 @@ def test_chain_returns_first_token():
 
     assert credential is expected_token
     assert second_credential.get_token.call_count == 0
+
+
+def test_msi_credential_cache(monkeypatch):
+    scope = "https://foo.bar"
+    expired = "this token's expired"
+    now = int(time.time())
+    token_payload = {
+        "access_token": expired,
+        "refresh_token": "",
+        "expires_in": 0,
+        "expires_on": now - 300,  # expired 5 minutes ago
+        "not_before": now,
+        "resource": scope,
+        "token_type": "Bearer",
+    }
+
+    # monkeypatch requests so we can test pipeline configuration
+    mock_response = Mock(text=json.dumps(token_payload), headers={"content-type": "application/json"}, status_code=200)
+    mock_send = Mock(return_value=mock_response)
+    monkeypatch.setattr(requests.Session, "send", value=mock_send)
+
+    credential = ManagedIdentityCredential()
+    token = credential.get_token((scope,))
+    assert token == expired
+    assert mock_send.call_count == 1
+
+    # calling get_token again should provoke another HTTP request
+    good_for_an_hour = "this token's good for an hour"
+    token_payload["expires_on"] = int(time.time()) + 3600
+    token_payload["expires_in"] = 3600
+    token_payload["access_token"] = good_for_an_hour
+    mock_response.text = json.dumps(token_payload)
+    token = credential.get_token((scope,))
+    assert token == good_for_an_hour
+    assert mock_send.call_count == 2
+
+    # get_token should return the cached token now
+    token = credential.get_token((scope,))
+    assert token == good_for_an_hour
+    assert mock_send.call_count == 2
+
+
+def test_msi_credential_retries(monkeypatch):
+    # monkeypatch requests so we can test pipeline configuration
+    mock_response = Mock(headers={"Retry-After": "0"}, text=b"")
+    mock_send = Mock(return_value=mock_response)
+    monkeypatch.setattr(requests.Session, "send", value=mock_send)
+
+    retry_total = 1
+    credential = ManagedIdentityCredential(retry_total=retry_total)
+
+    for status_code in (404, 429, 500):
+        mock_response.status_code = status_code
+        try:
+            credential.get_token(("",))
+        except AuthenticationError:
+            pass
+        assert mock_send.call_count is 1 + retry_total
+        mock_send.reset_mock()
diff --git a/sdk/identity/azure-identity/tests/test_identity_async.py b/sdk/identity/azure-identity/tests/test_identity_async.py
index 2bf6b867fe29..8414d65e0416 100644
--- a/sdk/identity/azure-identity/tests/test_identity_async.py
+++ b/sdk/identity/azure-identity/tests/test_identity_async.py
@@ -5,13 +5,21 @@
 # --------------------------------------------------------------------------
 import asyncio
 import json
+import os
 import time
 from unittest.mock import Mock
 import uuid
 
 import pytest
 import requests
-from azure.identity import AuthenticationError, AsyncClientSecretCredential, AsyncTokenCredentialChain
+from azure.identity import (
+    AuthenticationError,
+    AsyncClientSecretCredential,
+    AsyncEnvironmentCredential,
+    AsyncTokenCredentialChain,
+    AsyncManagedIdentityCredential,
+)
+from azure.identity.constants import EnvironmentVariables
 
 
 @pytest.mark.asyncio
@@ -43,20 +51,76 @@ async def test_client_secret_credential_cache(monkeypatch):
     assert mock_send.call_count == 2
 
 
+@pytest.mark.asyncio
+async def test_cert_environment_credential(monkeypatch):
+    client_id = "fake-client-id"
+    pem_path = os.path.join(os.path.dirname(__file__), "private-key.pem")
+    tenant_id = "fake-tenant-id"
+
+    monkeypatch.setenv(EnvironmentVariables.AZURE_CLIENT_ID, client_id)
+    monkeypatch.setenv(EnvironmentVariables.AZURE_CLIENT_CERTIFICATE_PATH, pem_path)
+    monkeypatch.setenv(EnvironmentVariables.AZURE_TENANT_ID, tenant_id)
+
+    success_message = "request passed validation"
+
+    def validate_request(request, **kwargs):
+        assert tenant_id in request.url
+        assert request.data["client_id"] == client_id
+        assert request.data["grant_type"] == "client_credentials"
+        # raising here makes mocking a transport response unnecessary
+        raise AuthenticationError(success_message)
+
+    credential = AsyncEnvironmentCredential(transport=Mock(send=validate_request))
+    with pytest.raises(AuthenticationError) as ex:
+        await credential.get_token(("",))
+    assert str(ex.value) == success_message
+
+
+@pytest.mark.asyncio
+async def test_client_secret_environment_credential(monkeypatch):
+    client_id = "fake-client-id"
+    secret = "fake-client-secret"
+    tenant_id = "fake-tenant-id"
+
+    monkeypatch.setenv(EnvironmentVariables.AZURE_CLIENT_ID, client_id)
+    monkeypatch.setenv(EnvironmentVariables.AZURE_CLIENT_SECRET, secret)
+    monkeypatch.setenv(EnvironmentVariables.AZURE_TENANT_ID, tenant_id)
+
+    success_message = "request passed validation"
+
+    def validate_request(request, **kwargs):
+        assert tenant_id in request.url
+        assert request.data["client_id"] == client_id
+        assert request.data["client_secret"] == secret
+        # raising here makes mocking a transport response unnecessary
+        raise AuthenticationError(success_message)
+
+    credential = AsyncEnvironmentCredential(transport=Mock(send=validate_request))
+    with pytest.raises(AuthenticationError) as ex:
+        await credential.get_token(("",))
+    assert str(ex.value) == success_message
+
+
+@pytest.mark.asyncio
+async def test_environment_credential_error():
+    with pytest.raises(AuthenticationError):
+        await AsyncEnvironmentCredential().get_token(("",))
+
+
 @pytest.mark.asyncio
 async def test_credential_chain_error_message():
-    async def raise_authn_error(message):
+    def raise_authn_error(message):
         raise AuthenticationError(message)
 
     first_error = "first_error"
     first_credential = Mock(spec=AsyncClientSecretCredential, get_token=lambda _: raise_authn_error(first_error))
     second_error = "second_error"
-    second_credential = Mock(get_token=lambda _: raise_authn_error(second_error))
+    second_credential = Mock(name="second_credential", get_token=lambda _: raise_authn_error(second_error))
 
     with pytest.raises(AuthenticationError) as ex:
         await AsyncTokenCredentialChain([first_credential, second_credential]).get_token(("scope",))
 
-    assert "AsyncClientSecretCredential" in ex.value.message
+    assert "ClientSecretCredential" in ex.value.message
     assert first_error in ex.value.message
     assert second_error in ex.value.message
 
@@ -91,3 +155,64 @@ async def test_chain_returns_first_token():
 
     assert credential is expected_token
     assert second_credential.get_token.call_count == 0
+
+
+@pytest.mark.asyncio
+async def test_msi_credential_cache(monkeypatch):
+    scope = "https://foo.bar"
+    expired = "this token's expired"
+    now = int(time.time())
+    token_payload = {
+        "access_token": expired,
+        "refresh_token": "",
+        "expires_in": 0,
+        "expires_on": now - 300,  # expired 5 minutes ago
+        "not_before": now,
+        "resource": scope,
+        "token_type": "Bearer",
+    }
+
+    # monkeypatch requests so we can test pipeline configuration
+    mock_response = Mock(text=json.dumps(token_payload), headers={"content-type": "application/json"}, status_code=200)
+    mock_send = Mock(return_value=mock_response)
+    monkeypatch.setattr(requests.Session, "send", value=mock_send)
+
+    credential = AsyncManagedIdentityCredential()
+    token = await credential.get_token((scope,))
+    assert token == expired
+    assert mock_send.call_count == 1
+
+    # calling get_token again should provoke another HTTP request
+    good_for_an_hour = "this token's good for an hour"
+    token_payload["expires_on"] = int(time.time()) + 3600
+    token_payload["expires_in"] = 3600
+    token_payload["access_token"] = good_for_an_hour
+    mock_response.text = json.dumps(token_payload)
+    token = await credential.get_token((scope,))
+    assert token == good_for_an_hour
+    assert mock_send.call_count == 2
+
+    # get_token should return the cached token now
+    token = await credential.get_token((scope,))
+    assert token == good_for_an_hour
+    assert mock_send.call_count == 2
+
+
+@pytest.mark.asyncio
+async def test_msi_credential_retries(monkeypatch):
+    # monkeypatch requests so we can test pipeline configuration
+    mock_response = Mock(headers={"Retry-After": "0"}, text=b"")
+    mock_send = Mock(return_value=mock_response)
+    monkeypatch.setattr(requests.Session, "send", value=mock_send)
+
+    retry_total = 1
+    credential = AsyncManagedIdentityCredential(retry_total=retry_total)
+
+    for status_code in (404, 429, 500):
+        mock_response.status_code = status_code
+        try:
+            await credential.get_token(("",))
+        except AuthenticationError:
+            pass
+        assert mock_send.call_count is 1 + retry_total
+        mock_send.reset_mock()
diff --git a/shared_requirements.txt b/shared_requirements.txt
index d5c48d3227ac..830c1d354648 100644
--- a/shared_requirements.txt
+++ b/shared_requirements.txt
@@ -87,7 +87,7 @@ azure-storage-queue~=1.3
 cryptography>=2.1.4
 futures
 typing
-msal>=0.3.1
+msal~=0.3.1
 msrest>=0.5.0
 msrestazure<2.0.0,>=0.4.32
 requests>=2.18.4