Add AzureNamedKeyCredential

sdk/core/azure-core/CHANGELOG.md

@@ -1,7 +1,10 @@
 # Release History
 
-## 1.13.1 (Unreleased)
+## 1.14.0 (Unreleased)
 
+### New Features
+
+- Added `azure.core.credentials.AzureNamedKeyCredential` credential and its respective policy #17548.
 
 ## 1.13.0 (2021-04-02)
 

sdk/core/azure-core/azure/core/_version.py

@@ -9,4 +9,4 @@
 # regenerated.
 # --------------------------------------------------------------------------
 
-VERSION = "1.13.1"
+VERSION = "1.14.0"

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

@@ -3,10 +3,10 @@
 # Licensed under the MIT License. See LICENSE.txt in the project root for
 # license information.
 # -------------------------------------------------------------------------
+from collections import namedtuple
 from typing import TYPE_CHECKING
 import six
 
-
 if TYPE_CHECKING:
     from typing import Any, NamedTuple
     from typing_extensions import Protocol
@@ -26,11 +26,12 @@ def get_token(self, *scopes, **kwargs):
 
 
 else:
-    from collections import namedtuple
-
     AccessToken = namedtuple("AccessToken", ["token", "expires_on"])
 
-__all__ = ["AzureKeyCredential", "AzureSasCredential", "AccessToken"]
+AzureNamedKey = namedtuple("AzureNamedKey", ["name", "key"])
+
+
+__all__ = ["AzureKeyCredential", "AzureSasCredential", "AccessToken", "AzureNamedKeyCredential"]
 
 
 class AzureKeyCredential(object):
@@ -111,3 +112,38 @@ 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._credential = AzureNamedKey(name, key)
+
+    @property
+    def credential(self):
+        # type () -> str
+        """The value of the configured name.
+
+        :rtype: str
+        """
+        return self._credential
+
+    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._credential = AzureNamedKey(name, 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, name, **kwargs):  # pylint: disable=unused-argument
+        # type: (AzureNamedKeyCredential, str, **Any) -> None
+        super(AzureNamedKeyCredentialPolicy, self).__init__()
+        self._credential = credential
+        self._name = name
+
+    def on_request(self, request):
+        request.http_request.headers[self._name] = self._credential.credential.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.credential.name == "sample_name"
+    assert cred.credential.key == "samplekey"
+
+    cred.update("newname", "newkey")
+    assert cred.credential.name == "newname"
+    assert cred.credential.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.credential.name == "sample_name"
+    assert cred.credential.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, name=key_name)
+    pipeline = Pipeline(transport=transport, policies=[credential_policy])
+
+    pipeline.run(HttpRequest("GET", "https://test_key_credential"))