Add user authentication API to UsernamePasswordCredential

sdk/identity/azure-identity/CHANGELOG.md

@@ -1,6 +1,10 @@
 # Release History
 
 ## 1.4.0b4 (Unreleased)
+- The user authentication API added to `DeviceCodeCredential` and
+  `InteractiveBrowserCredential` in 1.4.0b3 is available on
+  `UsernamePasswordCredential` as well.
+  ([#11449](https://github.com/Azure/azure-sdk-for-python/issues/11449))
 - The optional persistent cache for `DeviceCodeCredential` and
   `InteractiveBrowserCredential` added in 1.4.0b3 is now available on Linux and
   macOS as well as Windows.

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

@@ -2,19 +2,15 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 # ------------------------------------
-import time
 from typing import TYPE_CHECKING
 
-from azure.core.credentials import AccessToken
-from azure.core.exceptions import ClientAuthenticationError
-
-from .._internal import wrap_exceptions, PublicClientCredential
+from .._internal import InteractiveCredential, wrap_exceptions
 
 if TYPE_CHECKING:
     from typing import Any
 
 
-class UsernamePasswordCredential(PublicClientCredential):
+class UsernamePasswordCredential(InteractiveCredential):
     """Authenticates a user with a username and password.
 
     In general, Microsoft doesn't recommend this kind of authentication, because it's less secure than other
@@ -37,50 +33,29 @@ class UsernamePasswordCredential(PublicClientCredential):
           defines authorities for other clouds.
     :keyword str tenant_id: tenant ID or a domain associated with a tenant. If not provided, defaults to the
           'organizations' tenant, which supports only Azure Active Directory work or school accounts.
+    :keyword bool enable_persistent_cache: if True, the credential will store tokens in a persistent cache shared by
+         other user credentials. Defaults to False.
+    :keyword bool allow_unencrypted_cache: if True, the credential will fall back to a plaintext cache on platforms
+          where encryption is unavailable. Default to False. Has no effect when `enable_persistent_cache` is False.
     """
 
     def __init__(self, client_id, username, password, **kwargs):
         # type: (str, str, str, Any) -> None
+
+        # The base class will accept an AuthenticationRecord, allowing this credential to authenticate silently the
+        # first time it's asked for a token. However, we want to ensure this first authentication is not silent, to
+        # validate the given password. This class therefore doesn't document the authentication_record argument, and we
+        # discard it here.
+        kwargs.pop("authentication_record", None)
         super(UsernamePasswordCredential, self).__init__(client_id=client_id, **kwargs)
         self._username = username
         self._password = password
 
     @wrap_exceptions
-    def get_token(self, *scopes, **kwargs):  # pylint:disable=unused-argument
-        # type: (*str, **Any) -> AccessToken
-        """Request an access token for `scopes`.
-
-        .. note:: This method is called by Azure SDK clients. It isn't intended for use in application code.
-
-        :param str scopes: desired scopes for the access token. This method requires at least one scope.
-        :rtype: :class:`azure.core.credentials.AccessToken`
-        :raises ~azure.core.exceptions.ClientAuthenticationError: authentication failed. The error's ``message``
-          attribute gives a reason. Any error response from Azure Active Directory is available as the error's
-          ``response`` attribute.
-        """
-        if not scopes:
-            raise ValueError("'get_token' requires at least one scope")
-
-        # MSAL requires scopes be a list
-        scopes = list(scopes)  # type: ignore
-        now = int(time.time())
-
+    def _request_token(self, *scopes, **kwargs):
+        # type: (*str, **Any) -> dict
         app = self._get_app()
-        accounts = app.get_accounts(username=self._username)
-        result = None
-        for account in accounts:
-            result = app.acquire_token_silent(scopes, account=account)
-            if result:
-                break
-
-        if not result:
-            # cache miss -> request a new token
-            with self._adapter:
-                result = app.acquire_token_by_username_password(
-                    username=self._username, password=self._password, scopes=scopes
-                )
-
-        if "access_token" not in result:
-            raise ClientAuthenticationError(message="authentication failed: {}".format(result.get("error_description")))
-
-        return AccessToken(result["access_token"], now + int(result["expires_in"]))
+        with self._adapter:
+            return app.acquire_token_by_username_password(
+                username=self._username, password=self._password, scopes=list(scopes)
+            )

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

@@ -145,6 +145,10 @@ def mock_response(status_code=200, headers=None, json_payload=None):
         response.text = lambda encoding=None: json.dumps(json_payload)
         response.headers["content-type"] = "application/json"
         response.content_type = "application/json"
+    else:
+        response.text = lambda encoding=None: ""
+        response.headers["content-type"] = "text/plain"
+        response.content_type = "text/plain"
     return response
 
 

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

@@ -83,16 +83,17 @@ def test_authenticate():
             )
             record = credential.authenticate(scopes=(scope,))
 
+    for auth_record in (record, credential.authentication_record):
+        assert auth_record.authority == environment
+        assert auth_record.home_account_id == object_id + "." + home_tenant
+        assert auth_record.tenant_id == home_tenant
+        assert auth_record.username == username
+
     # credential should have a cached access token for the scope used in authenticate
     with patch(WEBBROWSER_OPEN, Mock(side_effect=Exception("credential should authenticate silently"))):
         token = credential.get_token(scope)
     assert token.token == access_token
 
-    assert record.authority == environment
-    assert record.home_account_id == object_id + "." + home_tenant
-    assert record.tenant_id == home_tenant
-    assert record.username == username
-
 
 def test_disable_automatic_authentication():
     """When configured for strict silent auth, the credential should raise when silent auth fails"""

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

@@ -78,16 +78,16 @@ def test_authenticate():
         _cache=TokenCache(),
     )
     record = credential.authenticate(scopes=(scope,))
+    for auth_record in (record, credential.authentication_record):
+        assert auth_record.authority == environment
+        assert auth_record.home_account_id == object_id + "." + home_tenant
+        assert auth_record.tenant_id == home_tenant
+        assert auth_record.username == username
 
     # credential should have a cached access token for the scope used in authenticate
     token = credential.get_token(scope)
     assert token.token == access_token
 
-    assert record.authority == environment
-    assert record.home_account_id == object_id + "." + home_tenant
-    assert record.tenant_id == home_tenant
-    assert record.username == username
-
 
 def test_disable_automatic_authentication():
     """When configured for strict silent auth, the credential should raise when silent auth fails"""

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

@@ -2,14 +2,14 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 # ------------------------------------
-from azure.core.exceptions import ClientAuthenticationError
 from azure.core.pipeline.policies import SansIOHTTPPolicy
 from azure.identity import UsernamePasswordCredential
 from azure.identity._internal.user_agent import USER_AGENT
 import pytest
 
 from helpers import (
     build_aad_response,
+    build_id_token,
     get_discovery_response,
     mock_response,
     Request,
@@ -26,7 +26,7 @@ def test_no_scopes():
     """The credential should raise when get_token is called with no scopes"""
 
     credential = UsernamePasswordCredential("client-id", "username", "password")
-    with pytest.raises(ClientAuthenticationError):
+    with pytest.raises(ValueError):
         credential.get_token()
 
 
@@ -88,13 +88,50 @@ def test_username_password_credential():
     assert token.token == expected_token
 
 
-def test_cache_persistence():
-    """The credential should cache only in memory"""
+def test_authenticate():
+    client_id = "client-id"
+    environment = "localhost"
+    issuer = "https://" + environment
+    tenant_id = "some-tenant"
+    authority = issuer + "/" + tenant_id
 
-    expected_cache = Mock()
-    raise_when_called = Mock(side_effect=Exception("credential shouldn't attempt to load a persistent cache"))
-    with patch.multiple("msal_extensions.token_cache", WindowsTokenCache=raise_when_called):
-        with patch("msal.TokenCache", Mock(return_value=expected_cache)):
-            credential = UsernamePasswordCredential("...", "...", "...")
+    access_token = "***"
+    scope = "scope"
 
-    assert credential._cache is expected_cache
+    # mock AAD response with id token
+    object_id = "object-id"
+    home_tenant = "home-tenant-id"
+    username = "[email protected]"
+    id_token = build_id_token(aud=client_id, iss=issuer, object_id=object_id, tenant_id=home_tenant, username=username)
+    auth_response = build_aad_response(
+        uid=object_id, utid=home_tenant, access_token=access_token, refresh_token="**", id_token=id_token
+    )
+
+    transport = validating_transport(
+        requests=[Request(url_substring=issuer)] * 4,
+        responses=[
+            get_discovery_response(authority),  # instance discovery
+            get_discovery_response(authority),  # tenant discovery
+            mock_response(status_code=404),  # user realm discovery
+            mock_response(json_payload=auth_response),  # token request following authenticate()
+        ],
+    )
+
+    credential = UsernamePasswordCredential(
+        username=username,
+        password="1234",
+        authority=environment,
+        client_id=client_id,
+        tenant_id=tenant_id,
+        transport=transport,
+    )
+    record = credential.authenticate(scopes=(scope,))
+    for auth_record in (record, credential.authentication_record):
+        assert auth_record.authority == environment
+        assert auth_record.home_account_id == object_id + "." + home_tenant
+        assert auth_record.tenant_id == home_tenant
+        assert auth_record.username == username
+
+    # credential should have a cached access token for the scope passed to authenticate
+    token = credential.get_token(scope)
+    assert token.token == access_token