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/linux_vscode_adapter.py

@@ -0,0 +1,63 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+import os
+import json
+import ctypes as ct
+from .._constants import VSCODE_CREDENTIALS_SECTION
+
+
+def _c_str(string):
+    return ct.c_char_p(string.encode('utf-8'))
+
+
+try:
+    _libsecret = ct.cdll.LoadLibrary('libsecret-1.so.0')
+    _libsecret.secret_schema_new.argtypes = \
+        [ct.c_char_p, ct.c_uint, ct.c_char_p, ct.c_uint, ct.c_char_p, ct.c_uint, ct.c_void_p]
+    _libsecret.secret_password_lookup_sync.argtypes = \
+        [ct.c_void_p, ct.c_void_p, ct.c_void_p, ct.c_char_p, ct.c_char_p, ct.c_char_p, ct.c_char_p, ct.c_void_p]
+except OSError:
+    _libsecret = None
+
+
+def _get_user_settings_path():
+    app_data_folder = os.environ['HOME']
+    return os.path.join(app_data_folder, ".config", "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"
+
+
+def _get_refresh_token(service_name, account_name):
+    # _libsecret.secret_password_lookup_sync raises segment fault on Python 2.7
+    # temporarily disable it on 2.7
+    import sys
+    if sys.version_info[0] < 3:
+        return None
+
+    if not _libsecret:
+        return None
+    schema = _libsecret.secret_schema_new(_c_str("org.freedesktop.Secret.Generic"), 2,
+                                          _c_str("service"), 0, _c_str("account"), 0, None)
+    p_str = _libsecret.secret_password_lookup_sync(schema, None, None, _c_str("service"), _c_str(service_name),
+                                                   _c_str("account"), _c_str(account_name), None)
+    return ct.c_char_p(p_str).value.decode('utf-8')
+
+
+def get_credentials():
+    try:
+        environment_name = _get_user_settings()
+        credentials = _get_refresh_token(VSCODE_CREDENTIALS_SECTION, environment_name)
+        return credentials
+    except Exception: #pylint: disable=broad-except
+        return None

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

@@ -0,0 +1,41 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+import os
+import json
+from msal_extensions.osx import Keychain, KeychainError
+from .._constants import VSCODE_CREDENTIALS_SECTION
+
+
+def _get_user_settings_path():
+    app_data_folder = os.environ['USER']
+    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"
+
+
+def _get_refresh_token(service_name, account_name):
+    key_chain = Keychain()
+    try:
+        return key_chain.get_generic_password(service_name, account_name)
+    except KeychainError:
+        return None
+
+
+def get_credentials():
+    try:
+        environment_name = _get_user_settings()
+        credentials = _get_refresh_token(VSCODE_CREDENTIALS_SECTION, environment_name)
+        return credentials
+    except Exception: #pylint: disable=broad-except
+        return None

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

@@ -0,0 +1,53 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+import sys
+from typing import TYPE_CHECKING
+from .._exceptions import CredentialUnavailableError
+from .._constants import AZURE_VSCODE_CLIENT_ID
+from .._internal.aad_client import AadClient
+if sys.platform.startswith('win'):
+    from .win_vscode_adapter import get_credentials
+elif sys.platform.startswith('darwin'):
+    from .macos_vscode_adapter import get_credentials
+else:
+    from .linux_vscode_adapter import get_credentials
+
+if TYPE_CHECKING:
+    # pylint:disable=unused-import,ungrouped-imports
+    from typing import Any, Iterable, Optional
+    from azure.core.credentials import AccessToken
+
+
+class VSCodeCredential(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")
+
+        refresh_token = get_credentials()
+        if not refresh_token:
+            raise CredentialUnavailableError(
+                message="No Azure user is logged in to Visual Studio Code."
+            )
+        token = self._client.get_cached_access_token(scopes) \
+                or self._client.obtain_token_by_refresh_token(refresh_token, scopes, **kwargs)
+        return token

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

@@ -0,0 +1,84 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+import os
+import json
+import ctypes as ct
+from .._constants import VSCODE_CREDENTIALS_SECTION
+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)]
+
+
+_PCREDENTIAL = ct.POINTER(_CREDENTIAL)
+
+_advapi = ct.WinDLL('advapi32')
+_advapi.CredReadW.argtypes = [wt.LPCWSTR, wt.DWORD, wt.DWORD, ct.POINTER(_PCREDENTIAL)]
+_advapi.CredReadW.restype = wt.BOOL
+_advapi.CredFree.argtypes = [_PCREDENTIAL]
+
+
+def _read_credential(service_name, account_name):
+    target = "{}/{}".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 + 1], 'little')
+                            for pos in range(0, cred_blob_size)]
+        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"
+
+
+def _get_refresh_token(service_name, account_name):
+    return _read_credential(service_name, account_name)
+
+
+def get_credentials():
+    try:
+        environment_name = _get_user_settings()
+        credentials = _get_refresh_token(VSCODE_CREDENTIALS_SECTION, environment_name)
+        return credentials
+    except Exception: #pylint: disable=broad-except
+        return None

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

@@ -0,0 +1,58 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+from typing import TYPE_CHECKING
+from ..._exceptions import CredentialUnavailableError
+from .._credentials.base import AsyncCredentialBase
+from ..._constants import AZURE_VSCODE_CLIENT_ID
+from .._internal.aad_client import AadClient
+from ..._credentials.vscode_credential import get_credentials
+if TYPE_CHECKING:
+    # pylint:disable=unused-import,ungrouped-imports
+    from typing import Any, Iterable, Optional
+    from azure.core.credentials import AccessToken
+
+class VSCodeCredential(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")
+
+        refresh_token = get_credentials()
+        if not refresh_token:
+            raise CredentialUnavailableError(
+                message="No Azure user is logged in to Visual Studio Code."
+            )
+        token = self._client.get_cached_access_token(scopes)
+        if not token:
+            token = await self._client.obtain_token_by_refresh_token(refresh_token, scopes, **kwargs)
+        return token