identity_vscode_credential

sdk/identity/azure-identity/azure/identity/_constants.py

@@ -5,7 +5,8 @@
 
 
 AZURE_CLI_CLIENT_ID = "04b07795-8ddb-461a-bbee-02f9e1bf7b46"
-
+AZURE_VSCODE_CLIENT_ID = "aebc6443-996d-45c2-90f0-388ff96faa56"
+VSCODE_CREDENTIALS_SECTION = "VS Code Azure"
 
 class KnownAuthorities:
     AZURE_CHINA = "login.chinacloudapi.cn"

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

@@ -0,0 +1,64 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+import os
+import json
+from msal_extensions.osx import Keychain
+from .._exceptions import CredentialUnavailableError
+from .._constants import (
+    VSCODE_CREDENTIALS_SECTION,
+    AZURE_VSCODE_CLIENT_ID,
+)
+from .._internal.aad_client import AadClient
+
+
+def _get_user_settings_path():
+    app_data_folder = os.environ['USERPROFILE']
+    return os.path.join(app_data_folder, "Library", "Application Support", "Code", "User", "settings.json")
+
+
+def _get_user_settings():
+    path = _get_user_settings_path()
+    try:
+        with open(path) as file:
+            data = json.load(file)
+            environment_name = data.get("azure.cloud", "Azure")
+            return environment_name
+    except IOError:
+        return "Azure"
+
+
+class MacOSVSCodeCredential(object):
+    """Authenticates by redeeming a refresh token previously saved by VS Code
+
+            """
+    def __init__(self, **kwargs):
+        self._client = kwargs.pop("client", None) or AadClient("organizations", AZURE_VSCODE_CLIENT_ID, **kwargs)
+
+    def get_token(self, *scopes, **kwargs):
+        # 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.
+
+        The first time this method is called, the credential will redeem its authorization code. On subsequent calls
+        the credential will return a cached access token or redeem a refresh token, if it acquired a refresh token upon
+        redeeming the authorization 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.identity.CredentialUnavailableError: fail to get refresh token.
+        """
+        if not scopes:
+            raise ValueError("'get_token' requires at least one scope")
+
+        environment_name = _get_user_settings()
+        key_chain = Keychain()
+        refresh_token = key_chain.get_generic_password(VSCODE_CREDENTIALS_SECTION, environment_name)
+        if not refresh_token:
+            raise CredentialUnavailableError(
+                message="No token available."
+            )
+        token = self._client.obtain_token_by_refresh_token(refresh_token, scopes, **kwargs)
+        return token

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

@@ -0,0 +1,145 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+import os
+import json
+import ctypes as ct
+from .._exceptions import CredentialUnavailableError
+from .._constants import (
+    VSCODE_CREDENTIALS_SECTION,
+    AZURE_VSCODE_CLIENT_ID,
+)
+from .._internal.aad_client import AadClient
+try:
+    import ctypes.wintypes as wt
+except (IOError, ValueError):
+    pass
+
+SUPPORTED_CREDKEYS = set((
+    'Type', 'TargetName', 'Persist',
+    'UserName', 'Comment', 'CredentialBlob'))
+
+
+class _CREDENTIAL(ct.Structure):
+    _fields_ = [
+        ("Flags", wt.DWORD),
+        ("Type", wt.DWORD),
+        ("TargetName", ct.c_wchar_p),
+        ("Comment", ct.c_wchar_p),
+        ("LastWritten", wt.FILETIME),
+        ("CredentialBlobSize", wt.DWORD),
+        ("CredentialBlob", wt.LPBYTE),
+        ("Persist", wt.DWORD),
+        ("AttributeCount", wt.DWORD),
+        ("Attributes", ct.c_void_p),
+        ("TargetAlias", ct.c_wchar_p),
+        ("UserName", ct.c_wchar_p)]
+
+    @classmethod
+    def from_dict(cls, credential):
+        # pylint:disable=attribute-defined-outside-init
+        creds = cls()
+        pcreds = _PCREDENTIAL(creds)
+
+        ct.memset(pcreds, 0, ct.sizeof(creds))
+
+        for key in SUPPORTED_CREDKEYS:
+            if key in credential:
+                if key != 'CredentialBlob':
+                    setattr(creds, key, credential[key])
+                else:
+                    blob = credential['CredentialBlob']
+                    blob_data = ct.create_unicode_buffer(blob)
+                    creds.CredentialBlobSize = \
+                        ct.sizeof(blob_data) - \
+                        ct.sizeof(ct.c_wchar)
+                    creds.CredentialBlob = ct.cast(blob_data, wt.LPBYTE)
+        return creds
+
+
+_PCREDENTIAL = ct.POINTER(_CREDENTIAL)
+
+
+_advapi = ct.WinDLL('advapi32')
+_advapi.CredWriteW.argtypes = [_PCREDENTIAL, wt.DWORD]
+_advapi.CredWriteW.restype = wt.BOOL
+_advapi.CredReadW.argtypes = [wt.LPCWSTR, wt.DWORD, wt.DWORD, ct.POINTER(_PCREDENTIAL)]
+_advapi.CredReadW.restype = wt.BOOL
+_advapi.CredFree.argtypes = [_PCREDENTIAL]
+_advapi.CredDeleteW.restype = wt.BOOL
+_advapi.CredDeleteW.argtypes = [wt.LPCWSTR, wt.DWORD, wt.DWORD]
+
+
+def _cred_write(credential):
+    creds = _CREDENTIAL.from_dict(credential)
+    cred_ptr = _PCREDENTIAL(creds)
+    _advapi.CredWriteW(cred_ptr, 0)
+
+
+def _cred_delete(service_name, account_name):
+    target = u"{}/{}".format(service_name, account_name)
+    _advapi.CredDeleteW(target, 1, 0)
+
+
+def _read_credential(service_name, account_name):
+    target = u"{}/{}".format(service_name, account_name)
+    cred_ptr = _PCREDENTIAL()
+    if _advapi.CredReadW(target, 1, 0, ct.byref(cred_ptr)):
+        cred_blob = cred_ptr.contents.CredentialBlob
+        cred_blob_size = cred_ptr.contents.CredentialBlobSize
+        password_as_list = [int.from_bytes(cred_blob[pos:pos + 2], 'little')
+                            for pos in range(0, cred_blob_size, 2)]
+        cred = ''.join(map(chr, password_as_list))
+        _advapi.CredFree(cred_ptr)
+        return cred
+    return None
+
+
+def _get_user_settings_path():
+    app_data_folder = os.environ['APPDATA']
+    return os.path.join(app_data_folder, "Code", "User", "settings.json")
+
+
+def _get_user_settings():
+    path = _get_user_settings_path()
+    try:
+        with open(path) as file:
+            data = json.load(file)
+            environment_name = data.get("azure.cloud", "Azure")
+            return environment_name
+    except IOError:
+        return "Azure"
+
+
+class WinVSCodeCredential(object):
+    """Authenticates by redeeming a refresh token previously saved by VS Code
+
+        """
+    def __init__(self, **kwargs):
+        self._client = kwargs.pop("client", None) or AadClient("organizations", AZURE_VSCODE_CLIENT_ID, **kwargs)
+
+    def get_token(self, *scopes, **kwargs):
+        # 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.
+
+        When this method is called, the credential will try to get the refresh token saved by VS Code. If a refresh
+        token can be found, it will redeem the refresh token for an access token and return the access token.
+
+        :param str scopes: desired scopes for the access token. This method requires at least one scope.
+        :rtype: :class:`azure.core.credentials.AccessToken`
+        :raises ~azure.identity.CredentialUnavailableError: fail to get refresh token.
+        """
+        if not scopes:
+            raise ValueError("'get_token' requires at least one scope")
+
+        environment_name = _get_user_settings()
+        refresh_token = _read_credential(VSCODE_CREDENTIALS_SECTION, environment_name)
+        if not refresh_token:
+            raise CredentialUnavailableError(
+                message="No token available."
+            )
+        token = self._client.obtain_token_by_refresh_token(refresh_token, scopes, **kwargs)
+        return token

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

@@ -0,0 +1,62 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+import asyncio
+from msal_extensions.osx import Keychain
+from ..._exceptions import CredentialUnavailableError
+from .._credentials.base import AsyncCredentialBase
+from ..._constants import (
+    VSCODE_CREDENTIALS_SECTION,
+    AZURE_VSCODE_CLIENT_ID,
+)
+from .._internal.aad_client import AadClient
+from ..._credentials.macos_vscode_credential import _get_user_settings
+
+
+class MacOSVSCodeCredential(AsyncCredentialBase):
+    """Authenticates by redeeming a refresh token previously saved by VS Code
+
+            """
+    def __init__(self, **kwargs):
+        self._client = kwargs.pop("client", None) or AadClient("organizations", AZURE_VSCODE_CLIENT_ID, **kwargs)
+
+    async def __aenter__(self):
+        if self._client:
+            await self._client.__aenter__()
+        return self
+
+    async def close(self):
+        """Close the credential's transport session."""
+
+        if self._client:
+            await self._client.__aexit__()
+
+    async def get_token(self, *scopes, **kwargs):
+        # 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.
+
+        The first time this method is called, the credential will redeem its authorization code. On subsequent calls
+        the credential will return a cached access token or redeem a refresh token, if it acquired a refresh token upon
+        redeeming the authorization 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.identity.CredentialUnavailableError: fail to get refresh token.
+        """
+        if not scopes:
+            raise ValueError("'get_token' requires at least one scope")
+
+        environment_name = _get_user_settings()
+        key_chain = Keychain()
+        refresh_token = key_chain.get_generic_password(VSCODE_CREDENTIALS_SECTION, environment_name)
+        if not refresh_token:
+            raise CredentialUnavailableError(
+                message="No token available."
+            )
+        loop = kwargs.pop("loop", None) or asyncio.get_event_loop()
+        token = await self._client.obtain_token_by_refresh_token(
+            refresh_token, scopes, loop=loop, **kwargs)
+        return token

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

@@ -0,0 +1,62 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+import asyncio
+from ..._exceptions import CredentialUnavailableError
+from .._credentials.base import AsyncCredentialBase
+from ..._constants import (
+    VSCODE_CREDENTIALS_SECTION,
+    AZURE_VSCODE_CLIENT_ID,
+)
+from .._internal.aad_client import AadClient
+try:
+    from ..._credentials.win_vscode_credential import _read_credential, _get_user_settings
+except ImportError:
+    pass
+
+
+class WinVSCodeCredential(AsyncCredentialBase):
+    """Authenticates by redeeming a refresh token previously saved by VS Code
+
+        """
+    def __init__(self, **kwargs):
+        self._client = kwargs.pop("client", None) or AadClient("organizations", AZURE_VSCODE_CLIENT_ID, **kwargs)
+
+    async def __aenter__(self):
+        if self._client:
+            await self._client.__aenter__()
+        return self
+
+    async def close(self):
+        """Close the credential's transport session."""
+
+        if self._client:
+            await self._client.__aexit__()
+
+    async def get_token(self, *scopes, **kwargs):
+        # 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.
+
+        When this method is called, the credential will try to get the refresh token saved by VS Code. If a refresh
+        token can be found, it will redeem the refresh token for an access token and return the access token.
+
+        :param str scopes: desired scopes for the access token. This method requires at least one scope.
+        :rtype: :class:`azure.core.credentials.AccessToken`
+        :raises ~azure.identity.CredentialUnavailableError: fail to get refresh token.
+        """
+        if not scopes:
+            raise ValueError("'get_token' requires at least one scope")
+
+        environment_name = _get_user_settings()
+        refresh_token = _read_credential(VSCODE_CREDENTIALS_SECTION, environment_name)
+        if not refresh_token:
+            raise CredentialUnavailableError(
+                message="No token available."
+            )
+        loop = kwargs.pop("loop", None) or asyncio.get_event_loop()
+        token = await self._client.obtain_token_by_refresh_token(
+            refresh_token, scopes, loop=loop, **kwargs)
+        return token

sdk/identity/azure-identity/setup.py

@@ -76,6 +76,7 @@
         "msal<2.0.0,>=1.0.0",
         "msal-extensions~=0.1.3",
         "six>=1.6",
+        "pygobject>=3,<4;platform_system=='Linux'",
     ],
     extras_require={
         ":python_version<'3.0'": ["azure-nspkg"],

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

@@ -0,0 +1,31 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+import sys
+import pytest
+from azure.core.credentials import AccessToken
+try:
+    from azure.identity._credentials.macos_vscode_credential import MacOSVSCodeCredential
+except (ImportError, OSError):
+    pass
+try:
+    from unittest.mock import Mock
+except ImportError:  # python < 3.3
+    from mock import Mock  # type: ignore
+
+
+@pytest.mark.skipif(not sys.platform.startswith('darwin'), reason="This test only runs on Windows")
+def test_get_token():
+    expected_token = AccessToken("token", 42)
+
+    mock_client = Mock(spec=object)
+    mock_client.obtain_token_by_refresh_token = Mock(return_value=expected_token)
+
+    credential = MacOSVSCodeCredential(
+        client=mock_client,
+    )
+
+    token = credential.get_token("scope")
+    assert token is expected_token
+    assert mock_client.obtain_token_by_refresh_token.call_count == 1