Shared credential (Azure#10509)

sdk/core/azure-core/CHANGELOG.md

@@ -6,6 +6,7 @@
 ### Features
 
 - Support a default error type in map_error #9773
+- Added `AzureKeyCredential` and its respective policy.
 
 ## 1.3.0 (2020-03-09)
 

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

@@ -4,6 +4,7 @@
 # license information.
 # -------------------------------------------------------------------------
 from typing import TYPE_CHECKING
+import six
 
 
 if TYPE_CHECKING:
@@ -28,3 +29,45 @@ def get_token(self, *scopes, **kwargs):
     from collections import namedtuple
 
     AccessToken = namedtuple("AccessToken", ["token", "expires_on"])
+
+__all__ = ["AzureKeyCredential", "AccessToken"]
+
+
+class AzureKeyCredential(object):
+    """Credential type used for authenticating to an Azure service.
+    It provides the ability to update the key without creating a new client.
+
+    :param str key: The key used to authenticate to an Azure service
+    :raises: TypeError
+    """
+
+    def __init__(self, key):
+        # type: (str) -> None
+        if not isinstance(key, six.string_types):
+            raise TypeError("key must be a string.")
+        self._key = key  # type: str
+
+    @property
+    def key(self):
+        # type () -> str
+        """The value of the configured key.
+
+        :rtype: str
+        """
+        return self._key
+
+    def update(self, key):
+        # type: (str) -> None
+        """Update the key.
+
+        This can be used when you've regenerated your service key and want
+        to update long-lived clients.
+
+        :param str key: The key used to authenticate to an Azure service
+        :raises: ValueError or TypeError
+        """
+        if not key:
+            raise ValueError("The key used for updating can not be None or empty")
+        if not isinstance(key, six.string_types):
+            raise TypeError("The key used for updating must be a string.")
+        self._key = key

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

@@ -25,7 +25,7 @@
 # --------------------------------------------------------------------------
 
 from ._base import HTTPPolicy, SansIOHTTPPolicy, RequestHistory
-from ._authentication import BearerTokenCredentialPolicy
+from ._authentication import BearerTokenCredentialPolicy, AzureKeyCredentialPolicy
 from ._custom_hook import CustomHookPolicy
 from ._redirect import RedirectPolicy
 from ._retry import RetryPolicy, RetryMode
@@ -44,6 +44,7 @@
     'HTTPPolicy',
     'SansIOHTTPPolicy',
     'BearerTokenCredentialPolicy',
+    'AzureKeyCredentialPolicy',
     'HeadersPolicy',
     'UserAgentPolicy',
     'NetworkTraceLoggingPolicy',

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

@@ -4,6 +4,7 @@
 # license information.
 # -------------------------------------------------------------------------
 import time
+import six
 
 from . import SansIOHTTPPolicy
 from ...exceptions import ServiceRequestError
@@ -16,7 +17,7 @@
 if TYPE_CHECKING:
     # pylint:disable=unused-import
     from typing import Any, Dict, Mapping, Optional
-    from azure.core.credentials import AccessToken, TokenCredential
+    from azure.core.credentials import AccessToken, TokenCredential, AzureKeyCredential
     from azure.core.pipeline import PipelineRequest
 
 
@@ -91,3 +92,25 @@ def on_request(self, request):
         if self._need_new_token:
             self._token = self._credential.get_token(*self._scopes)
         self._update_headers(request.http_request.headers, self._token.token)  # type: ignore
+
+
+class AzureKeyCredentialPolicy(SansIOHTTPPolicy):
+    """Adds a key header for the provided credential.
+
+    :param credential: The credential used to authenticate requests.
+    :type credential: ~azure.core.credentials.AzureKeyCredential
+    :param str name: The name of the key header used for the credential.
+    :raises: ValueError or TypeError
+    """
+    def __init__(self, credential, name):
+        # type: (AzureKeyCredential, str) -> None
+        super(AzureKeyCredentialPolicy, self).__init__()
+        self._credential = credential
+        if not name:
+            raise ValueError("name can not be None or empty")
+        if not isinstance(name, six.string_types):
+            raise TypeError("name must be a string.")
+        self._name = name
+
+    def on_request(self, request):
+        request.http_request.headers[self._name] = self._credential.key

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

@@ -6,10 +6,10 @@
 import time
 
 import azure.core
-from azure.core.credentials import AccessToken
+from azure.core.credentials import AccessToken, AzureKeyCredential
 from azure.core.exceptions import ServiceRequestError
 from azure.core.pipeline import Pipeline
-from azure.core.pipeline.policies import BearerTokenCredentialPolicy, SansIOHTTPPolicy
+from azure.core.pipeline.policies import BearerTokenCredentialPolicy, SansIOHTTPPolicy, AzureKeyCredentialPolicy
 from azure.core.pipeline.transport import HttpRequest
 
 import pytest
@@ -146,3 +146,41 @@ def test_key_vault_regression():
     policy._token = AccessToken(token, time.time() + 3600)
     assert not policy._need_new_token
     assert policy._token.token == token
+
+def test_azure_key_credential_policy():
+    """Tests to see if we can create an AzureKeyCredentialPolicy"""
+
+    key_header = "api_key"
+    api_key = "test_key"
+
+    def verify_authorization_header(request):
+        assert request.headers[key_header] == api_key
+
+    transport=Mock(send=verify_authorization_header)
+    credential = AzureKeyCredential(api_key)
+    credential_policy = AzureKeyCredentialPolicy(credential=credential, name=key_header)
+    pipeline = Pipeline(transport=transport, policies=[credential_policy])
+
+    pipeline.run(HttpRequest("GET", "https://test_key_credential"))
+
+def test_azure_key_credential_policy_raises():
+    """Tests AzureKeyCredential and AzureKeyCredentialPolicy raises with non-string input parameters."""
+    api_key = 1234
+    key_header = 5678
+    with pytest.raises(TypeError):
+        credential = AzureKeyCredential(api_key)
+
+    credential = AzureKeyCredential(str(api_key))
+    with pytest.raises(TypeError):
+        credential_policy = AzureKeyCredentialPolicy(credential=credential, name=key_header)
+
+def test_azure_key_credential_updates():
+    """Tests AzureKeyCredential updates"""
+    api_key = "original"
+
+    credential = AzureKeyCredential(api_key)
+    assert credential.key == api_key
+
+    api_key = "new"
+    credential.update(api_key)
+    assert credential.key == api_key