Add AzureNamedKeyCredential

sdk/core/azure-core/CHANGELOG.md

@@ -4,6 +4,7 @@
 
 ### Features
 
+- Added `azure.core.credentials.AzureNamedKeyCredential` credential #17548.
 - Supported adding custom policies  #16519
 
 ### Bug fixes

sdk/core/azure-core/azure/core/credentials.py

@@ -30,7 +30,7 @@ def get_token(self, *scopes, **kwargs):
 
     AccessToken = namedtuple("AccessToken", ["token", "expires_on"])
 
-__all__ = ["AzureKeyCredential", "AzureSasCredential", "AccessToken"]
+__all__ = ["AzureKeyCredential", "AzureSasCredential", "AccessToken", "AzureNamedKeyCredential"]
 
 
 class AzureKeyCredential(object):
@@ -111,3 +111,49 @@ def update(self, signature):
         if not isinstance(signature, six.string_types):
             raise TypeError("The signature used for updating must be a string.")
         self._signature = signature
+
+
+class AzureNamedKeyCredential(object):
+    """Credential type used for working with any service needing a named key that follows patterns
+    established by the other credential types.
+
+    :param str name: The name of the credential used to authenticate to an Azure service.
+    :param str key: The key used to authenticate to an Azure service.
+    :raises: TypeError
+    """
+    def __init__(self, name, key):
+        # type: (str, str) -> None
+        if not isinstance(name, six.string_types) or not isinstance(key, six.string_types):
+            raise TypeError("Both name and key must be Strings.")
+        self._name = name
+        self._key = key
+
+    @property
+    def name(self):
+        # type () -> str
+        """The value of the configured name.
+
+        :rtype: str
+        """
+        return self._name
+
+    @property
+    def key(self):
+        # type () -> str
+        """The value of the configured key.
+
+        :rtype: str
+        """
+        return self._key
+
+    def update(self, name, key):
+        # type: (str, str) -> None
+        """Update the named key credential.
+
+        Both name and key must be provided in order to update the named key credential.
+        Individual attributes cannot be updated.
+        """
+        if not isinstance(name, six.string_types) or not isinstance(key, six.string_types):
+            raise TypeError("Both name and key must be Strings.")
+        self._name = name
+        self._key = key

sdk/core/azure-core/azure/core/pipeline/policies/__init__.py

@@ -25,7 +25,10 @@
 # --------------------------------------------------------------------------
 
 from ._base import HTTPPolicy, SansIOHTTPPolicy, RequestHistory
-from ._authentication import BearerTokenCredentialPolicy, AzureKeyCredentialPolicy, AzureSasCredentialPolicy
+from ._authentication import (
+    BearerTokenCredentialPolicy, AzureKeyCredentialPolicy, AzureSasCredentialPolicy,
+    AzureNamedKeyCredentialPolicy
+    )
 from ._custom_hook import CustomHookPolicy
 from ._redirect import RedirectPolicy
 from ._retry import RetryPolicy, RetryMode
@@ -45,6 +48,7 @@
     'SansIOHTTPPolicy',
     'BearerTokenCredentialPolicy',
     'AzureKeyCredentialPolicy',
+    'AzureNamedKeyCredentialPolicy',
     'AzureSasCredentialPolicy',
     'HeadersPolicy',
     'UserAgentPolicy',

sdk/core/azure-core/azure/core/pipeline/policies/_authentication.py

@@ -17,7 +17,13 @@
 if TYPE_CHECKING:
     # pylint:disable=unused-import
     from typing import Any, Dict, Optional
-    from azure.core.credentials import AccessToken, TokenCredential, AzureKeyCredential, AzureSasCredential
+    from azure.core.credentials import (
+        AccessToken,
+        TokenCredential,
+        AzureKeyCredential,
+        AzureSasCredential,
+        AzureNamedKeyCredential
+    )
     from azure.core.pipeline import PipelineRequest
 
 
@@ -145,3 +151,20 @@ def on_request(self, request):
             else:
                 url = url + "?" + signature
         request.http_request.url = url
+
+
+class AzureNamedKeyCredentialPolicy(SansIOHTTPPolicy):
+    """Adds a key header for the provided credential.
+
+    :param credential: The credential used to authenticate requests.
+    :type credential: ~azure.core.credentials.AzureNamedKeyCredential
+    :raises: ValueError or TypeError
+    """
+    def __init__(self, credential, **kwargs):  # pylint: disable=unused-argument
+        # type: (AzureNamedKeyCredential, **Any) -> None
+        super(AzureNamedKeyCredentialPolicy, self).__init__()
+        self._key = credential.key
+        self._name = credential.name
+
+    def on_request(self, request):
+        request.http_request.headers[self._name] = self._key

sdk/core/azure-core/tests/test_authentication.py

@@ -6,10 +6,13 @@
 import time
 
 import azure.core
-from azure.core.credentials import AccessToken, AzureKeyCredential, AzureSasCredential
+from azure.core.credentials import AccessToken, AzureKeyCredential, AzureSasCredential, AzureNamedKeyCredential
 from azure.core.exceptions import ServiceRequestError
 from azure.core.pipeline import Pipeline
-from azure.core.pipeline.policies import BearerTokenCredentialPolicy, SansIOHTTPPolicy, AzureKeyCredentialPolicy, AzureSasCredentialPolicy
+from azure.core.pipeline.policies import (
+    BearerTokenCredentialPolicy, SansIOHTTPPolicy, AzureKeyCredentialPolicy,
+    AzureSasCredentialPolicy, AzureNamedKeyCredentialPolicy
+)
 from azure.core.pipeline.transport import HttpRequest
 
 import pytest
@@ -230,3 +233,41 @@ def test_azure_sas_credential_policy_raises():
     sas = 1234
     with pytest.raises(TypeError):
         credential = AzureSasCredential(sas)
+
+def test_azure_named_key_credential():
+    cred = AzureNamedKeyCredential("sample_name", "samplekey")
+
+    assert cred.name == "sample_name"
+    assert cred.key == "samplekey"
+
+    cred.update("newname", "newkey")
+    assert cred.name == "newname"
+    assert cred.key == "newkey"
+
+
+def test_azure_named_key_credential_raises():
+    with pytest.raises(TypeError, match="Both name and key must be Strings."):
+        cred = AzureNamedKeyCredential("sample_name", 123345)
+
+    cred = AzureNamedKeyCredential("sample_name", "samplekey")
+    assert cred.name == "sample_name"
+    assert cred.key == "samplekey"
+
+    with pytest.raises(TypeError, match="Both name and key must be Strings."):
+        cred.update(1234, "newkey")
+
+def test_azure_named_key_credential_policy():
+    """Tests to see if we can create an AzureKeyCredentialPolicy"""
+
+    key_name = "api_key"
+    api_key = "test_key"
+
+    def verify_authorization_header(request):
+        assert request.headers[key_name] == api_key
+
+    transport=Mock(send=verify_authorization_header)
+    credential = AzureNamedKeyCredential(key_name, api_key)
+    credential_policy = AzureNamedKeyCredentialPolicy(credential=credential)
+    pipeline = Pipeline(transport=transport, policies=[credential_policy])
+
+    pipeline.run(HttpRequest("GET", "https://test_key_credential"))