Validate tenant IDs used in URLs

sdk/identity/azure-identity/CHANGELOG.md

@@ -10,6 +10,9 @@
    provided, the credential will authenticate users to an Azure development
    application.
    ([#14354](https://github.com/Azure/azure-sdk-for-python/issues/14354))
+- Credentials raise `ValueError` when constructed with tenant IDs containing
+  invalid characters
+  ([#14821](https://github.com/Azure/azure-sdk-for-python/issues/14821))
 - Raised minimum msal version to 1.6.0
 
 ### Fixed

sdk/identity/azure-identity/azure/identity/_credentials/certificate.py

@@ -10,6 +10,7 @@
 from cryptography.hazmat.backends import default_backend
 import six
 
+from .._internal import validate_tenant_id
 from .._internal.client_credential_base import ClientCredentialBase
 
 if TYPE_CHECKING:
@@ -40,6 +41,7 @@ class CertificateCredential(ClientCredentialBase):
 
     def __init__(self, tenant_id, client_id, certificate_path, **kwargs):
         # type: (str, str, str, **Any) -> None
+        validate_tenant_id(tenant_id)
         if not certificate_path:
             raise ValueError(
                 "'certificate_path' must be the path to a PEM file containing an x509 certificate and its private key"

sdk/identity/azure-identity/azure/identity/_credentials/shared_cache.py

@@ -11,7 +11,7 @@
 
 from .. import CredentialUnavailableError
 from .._constants import DEVELOPER_SIGN_ON_CLIENT_ID
-from .._internal import AadClient
+from .._internal import AadClient, validate_tenant_id
 from .._internal.decorators import log_get_token, wrap_exceptions
 from .._internal.msal_client import MsalClient
 from .._internal.shared_token_cache import NO_TOKEN, SharedTokenCacheBase
@@ -53,6 +53,7 @@ def __init__(self, username=None, **kwargs):
         if self._auth_record:
             # authenticate in the tenant that produced the record unless "tenant_id" specifies another
             self._tenant_id = kwargs.pop("tenant_id", None) or self._auth_record.tenant_id
+            validate_tenant_id(self._tenant_id)
             self._cache = kwargs.pop("_cache", None)
             self._app = None
             self._client_kwargs = kwargs

sdk/identity/azure-identity/azure/identity/_credentials/vscode.py

@@ -7,6 +7,7 @@
 
 from .._exceptions import CredentialUnavailableError
 from .._constants import AZURE_VSCODE_CLIENT_ID
+from .._internal import validate_tenant_id
 from .._internal.aad_client import AadClient
 from .._internal.decorators import log_get_token
 
@@ -38,6 +39,7 @@ def __init__(self, **kwargs):
         self._refresh_token = None
         self._client = kwargs.pop("_client", None)
         self._tenant_id = kwargs.pop("tenant_id", None) or "organizations"
+        validate_tenant_id(self._tenant_id)
         if not self._client:
             self._client = AadClient(self._tenant_id, AZURE_VSCODE_CLIENT_ID, **kwargs)
 

sdk/identity/azure-identity/azure/identity/_internal/__init__.py

@@ -29,6 +29,19 @@ def get_default_authority():
     return normalize_authority(authority)
 
 
+VALID_TENANT_ID_CHARACTERS = frozenset("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz" + "0123456789" + "-.")
+
+
+def validate_tenant_id(tenant_id):
+    """Raise ValueError if tenant_id is empty or contains a character invalid for a tenant id"""
+    # type: (str) -> None
+    if not tenant_id or any(c not in VALID_TENANT_ID_CHARACTERS for c in tenant_id):
+        raise ValueError(
+            "Invalid tenant id provided. You can locate your tenant id by following the instructions here: "
+            + "https://docs.microsoft.com/partner-center/find-ids-and-domain-names"
+        )
+
+
 # pylint:disable=wrong-import-position
 from .aad_client import AadClient
 from .aad_client_base import AadClientBase

sdk/identity/azure-identity/azure/identity/_internal/certificate_credential_base.py

@@ -9,6 +9,7 @@
 
 from . import AadClientCertificate
 from .persistent_cache import load_service_principal_cache
+from .._internal import validate_tenant_id
 
 try:
     ABC = abc.ABC
@@ -28,6 +29,7 @@
 class CertificateCredentialBase(ABC):
     def __init__(self, tenant_id, client_id, certificate_path, **kwargs):
         # type: (str, str, str, **Any) -> None
+        validate_tenant_id(tenant_id)
         if not certificate_path:
             raise ValueError(
                 "'certificate_path' must be the path to a PEM file containing an x509 certificate and its private key"

sdk/identity/azure-identity/azure/identity/_internal/client_secret_credential_base.py

@@ -7,6 +7,7 @@
 
 from msal import TokenCache
 
+from . import validate_tenant_id
 from .persistent_cache import load_service_principal_cache
 
 try:
@@ -30,6 +31,7 @@ def __init__(self, tenant_id, client_id, client_secret, **kwargs):
             raise ValueError(
                 "tenant_id should be an Azure Active Directory tenant's id (also called its 'directory id')"
             )
+        validate_tenant_id(tenant_id)
 
         enable_persistent_cache = kwargs.pop("enable_persistent_cache", False)
         if enable_persistent_cache:

sdk/identity/azure-identity/azure/identity/_internal/msal_credentials.py

@@ -5,11 +5,10 @@
 import abc
 
 import msal
-from azure.core.credentials import AccessToken
 
 from .msal_client import MsalClient
 from .persistent_cache import load_user_cache
-from .._internal import get_default_authority, normalize_authority
+from .._internal import get_default_authority, normalize_authority, validate_tenant_id
 
 try:
     ABC = abc.ABC
@@ -34,6 +33,7 @@ def __init__(self, client_id, client_credential=None, **kwargs):
         authority = kwargs.pop("authority", None)
         self._authority = normalize_authority(authority) if authority else get_default_authority()
         self._tenant_id = kwargs.pop("tenant_id", None) or "organizations"
+        validate_tenant_id(self._tenant_id)
 
         self._client_credential = client_credential
         self._client_id = client_id

sdk/identity/azure-identity/azure/identity/aio/_credentials/vscode.py

@@ -10,6 +10,7 @@
 from .._internal.aad_client import AadClient
 from .._internal.decorators import log_get_token_async
 from ..._credentials.vscode import get_credentials
+from ..._internal import validate_tenant_id
 
 if TYPE_CHECKING:
     # pylint:disable=unused-import,ungrouped-imports
@@ -31,6 +32,7 @@ def __init__(self, **kwargs: "Any") -> None:
         self._refresh_token = None
         self._client = kwargs.pop("_client", None)
         self._tenant_id = kwargs.pop("tenant_id", None) or "organizations"
+        validate_tenant_id(self._tenant_id)
         if not self._client:
             self._client = AadClient(self._tenant_id, AZURE_VSCODE_CLIENT_ID, **kwargs)
 

sdk/identity/azure-identity/tests/test_aad_client.py

@@ -75,7 +75,7 @@ def assert_secrets_not_exposed():
 
 @pytest.mark.parametrize("authority", ("localhost", "https://localhost"))
 def test_request_url(authority):
-    tenant_id = "expected_tenant"
+    tenant_id = "expected-tenant"
     parsed_authority = urlparse(authority)
     expected_netloc = parsed_authority.netloc or authority  # "localhost" parses to netloc "", path "localhost"
 

sdk/identity/azure-identity/tests/test_aad_client_async.py

@@ -160,7 +160,7 @@ async def send(request, **_):
 
 @pytest.mark.parametrize("authority", ("localhost", "https://localhost"))
 async def test_request_url(authority):
-    tenant_id = "expected_tenant"
+    tenant_id = "expected-tenant"
     parsed_authority = urlparse(authority)
     expected_netloc = parsed_authority.netloc or authority  # "localhost" parses to netloc "", path "localhost"
 

sdk/identity/azure-identity/tests/test_authn_client.py

@@ -208,7 +208,7 @@ def mock_send(request, **kwargs):
 
 @pytest.mark.parametrize("authority", ("localhost", "https://localhost"))
 def test_request_url(authority):
-    tenant_id = "expected_tenant"
+    tenant_id = "expected-tenant"
     parsed_authority = urlparse(authority)
     expected_netloc = parsed_authority.netloc or authority  # "localhost" parses to netloc "", path "localhost"
 

sdk/identity/azure-identity/tests/test_authn_client_async.py

@@ -18,7 +18,7 @@
 @pytest.mark.asyncio
 @pytest.mark.parametrize("authority", ("localhost", "https://localhost"))
 async def test_request_url(authority):
-    tenant_id = "expected_tenant"
+    tenant_id = "expected-tenant"
     parsed_authority = urlparse(authority)
     expected_netloc = parsed_authority.netloc or authority  # "localhost" parses to netloc "", path "localhost"
 

sdk/identity/azure-identity/tests/test_browser_credential.py

@@ -36,6 +36,19 @@
 WEBBROWSER_OPEN = InteractiveBrowserCredential.__module__ + ".webbrowser.open"
 
 
+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:
+        InteractiveBrowserCredential(tenant_id=tenant)
+
+    invalid_ids = {"my tenant", "my_tenant", "/", "\\", '"my-tenant"', "'my-tenant'"}
+    for tenant in invalid_ids:
+        with pytest.raises(ValueError):
+            InteractiveBrowserCredential(tenant_id=tenant)
+
+
 def test_no_scopes():
     """The credential should raise when get_token is called with no scopes"""
 
@@ -170,7 +183,7 @@ def test_interactive_credential(mock_open, redirect_url):
     expected_token = "access-token"
     expires_in = 3600
     authority = "authority"
-    tenant_id = "tenant_id"
+    tenant_id = "tenant-id"
     endpoint = "https://{}/{}".format(authority, tenant_id)
 
     transport = msal_validating_transport(

sdk/identity/azure-identity/tests/test_certificate_credential.py

@@ -43,6 +43,19 @@
 )
 
 
+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:
+        CertificateCredential(tenant, "client-id", CERT_PATH)
+
+    invalid_ids = {"", "my tenant", "my_tenant", "/", "\\", '"my-tenant"', "'my-tenant'"}
+    for tenant in invalid_ids:
+        with pytest.raises(ValueError):
+            CertificateCredential(tenant, "client-id", CERT_PATH)
+
+
 def test_no_scopes():
     """The credential should raise ValueError when get_token is called with no scopes"""
 
@@ -82,7 +95,7 @@ def test_user_agent():
 def test_authority(authority):
     """the credential should accept an authority, with or without scheme, as an argument or environment variable"""
 
-    tenant_id = "expected_tenant"
+    tenant_id = "expected-tenant"
     parsed_authority = urlparse(authority)
     expected_netloc = parsed_authority.netloc or authority
     expected_authority = "https://{}/{}".format(expected_netloc, tenant_id)

sdk/identity/azure-identity/tests/test_certificate_credential_async.py

@@ -18,6 +18,19 @@
 from test_certificate_credential import BOTH_CERTS, CERT_PATH, validate_jwt
 
 
+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:
+        CertificateCredential(tenant, "client-id", CERT_PATH)
+
+    invalid_ids = {"", "my tenant", "my_tenant", "/", "\\", '"', "'"}
+    for tenant in invalid_ids:
+        with pytest.raises(ValueError):
+            CertificateCredential(tenant, "client-id", CERT_PATH)
+
+
 @pytest.mark.asyncio
 async def test_no_scopes():
     """The credential should raise ValueError when get_token is called with no scopes"""
@@ -83,7 +96,7 @@ async def test_user_agent():
 async def test_request_url(cert_path, cert_password, authority):
     """the credential should accept an authority, with or without scheme, as an argument or environment variable"""
 
-    tenant_id = "expected_tenant"
+    tenant_id = "expected-tenant"
     access_token = "***"
     parsed_authority = urlparse(authority)
     expected_netloc = parsed_authority.netloc or authority  # "localhost" parses to netloc "", path "localhost"

sdk/identity/azure-identity/tests/test_client_secret_credential.py

@@ -18,6 +18,19 @@
     from mock import Mock, patch  # type: ignore
 
 
+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:
+        ClientSecretCredential(tenant, "client-id", "secret")
+
+    invalid_ids = {"", "my tenant", "my_tenant", "/", "\\", '"my-tenant"', "'my-tenant'"}
+    for tenant in invalid_ids:
+        with pytest.raises(ValueError):
+            ClientSecretCredential(tenant, "client-id", "secret")
+
+
 def test_no_scopes():
     """The credential should raise ValueError when get_token is called with no scopes"""
 
@@ -74,7 +87,7 @@ def test_client_secret_credential():
 def test_authority(authority):
     """the credential should accept an authority, with or without scheme, as an argument or environment variable"""
 
-    tenant_id = "expected_tenant"
+    tenant_id = "expected-tenant"
     parsed_authority = urlparse(authority)
     expected_netloc = parsed_authority.netloc or authority
     expected_authority = "https://{}/{}".format(expected_netloc, tenant_id)

sdk/identity/azure-identity/tests/test_client_secret_credential_async.py

@@ -18,6 +18,19 @@
 from helpers_async import async_validating_transport, AsyncMockTransport, wrap_in_future
 
 
+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:
+        ClientSecretCredential(tenant, "client-id", "secret")
+
+    invalid_ids = {"", "my tenant", "my_tenant", "/", "\\", '"my-tenant"', "'my-tenant'"}
+    for tenant in invalid_ids:
+        with pytest.raises(ValueError):
+            ClientSecretCredential(tenant, "client-id", "secret")
+
+
 @pytest.mark.asyncio
 async def test_no_scopes():
     """The credential should raise ValueError when get_token is called with no scopes"""
@@ -111,7 +124,7 @@ async def test_client_secret_credential():
 async def test_request_url(authority):
     """the credential should accept an authority, with or without scheme, as an argument or environment variable"""
 
-    tenant_id = "expected_tenant"
+    tenant_id = "expected-tenant"
     access_token = "***"
     parsed_authority = urlparse(authority)
     expected_netloc = parsed_authority.netloc or authority  # "localhost" parses to netloc "", path "localhost"

sdk/identity/azure-identity/tests/test_device_code_credential.py

@@ -26,6 +26,19 @@
     from mock import Mock  # type: ignore
 
 
+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:
+        DeviceCodeCredential(tenant_id=tenant)
+
+    invalid_ids = {"my tenant", "my_tenant", "/", "\\", '"my-tenant"', "'my-tenant'"}
+    for tenant in invalid_ids:
+        with pytest.raises(ValueError):
+            DeviceCodeCredential(tenant_id=tenant)
+
+
 def test_no_scopes():
     """The credential should raise when get_token is called with no scopes"""
 

sdk/identity/azure-identity/tests/test_shared_cache_credential.py

@@ -39,6 +39,24 @@
 )
 
 
+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:
+        record = AuthenticationRecord(tenant, "client-id", "authority", "home.account.id", "username")
+        SharedTokenCacheCredential(authentication_record=record)
+        SharedTokenCacheCredential(authentication_record=record, tenant_id=tenant)
+
+    invalid_ids = {"", "my tenant", "my_tenant", "/", "\\", '"my-tenant"', "'my-tenant'"}
+    for tenant in invalid_ids:
+        record = AuthenticationRecord(tenant, "client-id", "authority", "home.account.id", "username")
+        with pytest.raises(ValueError):
+            SharedTokenCacheCredential(authentication_record=record)
+        with pytest.raises(ValueError):
+            SharedTokenCacheCredential(authentication_record=record, tenant_id=tenant)
+
+
 def test_supported():
     """the cache is supported on Linux, macOS, Windows, so this should pass unless you're developing on e.g. FreeBSD"""
     assert SharedTokenCacheCredential.supported()
@@ -520,7 +538,7 @@ def test_authority_environment_variable():
 
 
 def test_authentication_record_empty_cache():
-    record = AuthenticationRecord("tenant_id", "client_id", "authority", "home_account_id", "username")
+    record = AuthenticationRecord("tenant-id", "client_id", "authority", "home_account_id", "username")
 
     def send(request, **_):
         # expecting only MSAL discovery requests