[Core] Allow authentication via environment variables

src/azure-cli-core/azure/cli/core/_profile.py

@@ -9,12 +9,14 @@
 from enum import Enum
 
 from azure.cli.core._session import ACCOUNT
+from azure.cli.core.auth.msal_authentication import _TENANT, _CLIENT_ID, _CLIENT_SECRET
 from azure.cli.core.azclierror import AuthenticationError
 from azure.cli.core.cloud import get_active_cloud, set_cloud_subscription
-from azure.cli.core.util import in_cloud_console, can_launch_browser
+from azure.cli.core.util import in_cloud_console, can_launch_browser, is_guid, assert_guid
 from knack.log import get_logger
 from knack.util import CLIError
 
+
 logger = get_logger(__name__)
 
 # Names below are used by azure-xplat-cli to persist account information into
@@ -32,7 +34,6 @@
 _MANAGED_BY_TENANTS = 'managedByTenants'
 _USER_ENTITY = 'user'
 _USER_NAME = 'name'
-_CLIENT_ID = 'clientId'
 _CLOUD_SHELL_ID = 'cloudShellID'
 _SUBSCRIPTIONS = 'subscriptions'
 _INSTALLATION_ID = 'installationId'
@@ -46,13 +47,19 @@
 _TOKEN_ENTRY_TOKEN_TYPE = 'tokenType'
 
 _TENANT_LEVEL_ACCOUNT_NAME = 'N/A(tenant level account)'
+_ENVIRONMENT_VARIABLE_ACCOUNT_NAME = 'Environment Variable Subscription'
 
 _SYSTEM_ASSIGNED_IDENTITY = 'systemAssignedIdentity'
 _USER_ASSIGNED_IDENTITY = 'userAssignedIdentity'
 _ASSIGNED_IDENTITY_INFO = 'assignedIdentityInfo'
 
 _AZ_LOGIN_MESSAGE = "Please run 'az login' to setup account."
 
+_AZURE_CLIENT_ID = 'AZURE_CLIENT_ID'
+_AZURE_CLIENT_SECRET = 'AZURE_CLIENT_SECRET'
+_AZURE_TENANT_ID = 'AZURE_TENANT_ID'
+_AZURE_SUBSCRIPTION_ID = 'AZURE_SUBSCRIPTION_ID'
+
 
 def load_subscriptions(cli_ctx, all_clouds=False, refresh=False):
     profile = Profile(cli_ctx=cli_ctx)
@@ -61,6 +68,51 @@ def load_subscriptions(cli_ctx, all_clouds=False, refresh=False):
     subscriptions = profile.load_cached_subscriptions(all_clouds)
     return subscriptions
 
+def env_var_auth_configured():
+    keys = [_AZURE_CLIENT_ID, _AZURE_CLIENT_SECRET, _AZURE_TENANT_ID]
+    all_provided = all(key in os.environ for key in keys)
+
+    if all_provided:
+        return True
+
+    any_provided = any(key in os.environ for key in keys)
+    if any_provided:
+        raise CLIError("To authenticate using environment variables, "
+                       "all of {}, {}, {} must be specified."
+                       .format(*keys)
+        )
+
+    return False
+
+def load_env_var_credential():
+    if env_var_auth_configured():
+        credential = {
+            _CLIENT_ID: os.environ[_AZURE_CLIENT_ID],
+            _TENANT: os.environ[_AZURE_TENANT_ID],
+            _CLIENT_SECRET: os.environ[_AZURE_CLIENT_SECRET]
+        }
+        return credential
+    return None
+
+def load_env_var_subscription():
+    if env_var_auth_configured():
+        env_var_subscription = {
+            _SUBSCRIPTION_ID: None,
+            _SUBSCRIPTION_NAME: _ENVIRONMENT_VARIABLE_ACCOUNT_NAME,
+            _TENANT_ID: os.environ[_AZURE_TENANT_ID],
+            _USER: {
+                _USER_NAME: os.environ[_AZURE_CLIENT_ID],
+                _USER_TYPE: _SERVICE_PRINCIPAL
+            }
+        }
+
+        if _AZURE_SUBSCRIPTION_ID in os.environ:
+            subscription_id = os.environ[_AZURE_SUBSCRIPTION_ID]
+            assert_guid(subscription_id, _AZURE_SUBSCRIPTION_ID)
+            env_var_subscription[_SUBSCRIPTION_ID] = subscription_id
+        return env_var_subscription
+    return None
+
 
 def _detect_adfs_authority(authority_url, tenant):
     """Prepare authority and tenant for Azure Identity with ADFS support.
@@ -359,7 +411,7 @@ def get_raw_token(self, resource=None, scopes=None, subscription=None, tenant=No
         if subscription and tenant:
             raise CLIError("Please specify only one of subscription and tenant, not both")
 
-        account = self.get_subscription(subscription)
+        account = self.get_subscription(subscription, allow_null_subscription=True)
 
         identity_type, identity_id = Profile._try_parse_msi_account_name(account)
         if identity_type:
@@ -540,24 +592,40 @@ def get_current_account_user(self):
 
         return active_account[_USER_ENTITY][_USER_NAME]
 
-    def get_subscription(self, subscription=None):  # take id or name
+    def get_subscription(self, subscription=None, allow_null_subscription=False):  # take id or name
         subscriptions = self.load_cached_subscriptions()
-        if not subscriptions:
-            raise CLIError(_AZ_LOGIN_MESSAGE)
-
-        result = [x for x in subscriptions if (
-            not subscription and x.get(_IS_DEFAULT_SUBSCRIPTION) or
-            subscription and subscription.lower() in [x[_SUBSCRIPTION_ID].lower(), x[
-                _SUBSCRIPTION_NAME].lower()])]
-        if not result and subscription:
-            raise CLIError("Subscription '{}' not found. "
-                           "Check the spelling and casing and try again.".format(subscription))
-        if not result and not subscription:
-            raise CLIError("No subscription found. Run 'az account set' to select a subscription.")
-        if len(result) > 1:
-            raise CLIError("Multiple subscriptions with the name '{}' found. "
-                           "Specify the subscription ID.".format(subscription))
-        return result[0]
+
+        if subscriptions:
+            result = [x for x in subscriptions if (
+                not subscription and x.get(_IS_DEFAULT_SUBSCRIPTION) or
+                subscription and subscription.lower() in [x[_SUBSCRIPTION_ID].lower(), x[
+                    _SUBSCRIPTION_NAME].lower()])]
+            if not result and subscription:
+                raise CLIError("Subscription '{}' not found. "
+                            "Check the spelling and casing and try again.".format(subscription))
+            if not result and not subscription:
+                raise CLIError("No subscription found. Run 'az account set' to select a subscription.")
+            if len(result) > 1:
+                raise CLIError("Multiple subscriptions with the name '{}' found. "
+                            "Specify the subscription ID.".format(subscription))
+            return result[0]
+
+        # Attempt to use env vars
+        if env_var_auth_configured():
+            logger.warning("Using subscription configured in environment variables.")
+            env_var_sub = load_env_var_subscription()
+            if subscription:
+                # Subscription ID must be a GUID
+                assert_guid(subscription, _AZURE_SUBSCRIPTION_ID)
+                # Overwrite env var subscription if given as argument to get_subscription()
+                env_var_sub[_SUBSCRIPTION_ID] = subscription
+
+            if not env_var_sub[_SUBSCRIPTION_ID] and not allow_null_subscription:
+                error = """Subscription is undefined.
+                        Please specific the subscription ID with either {} or --subscription."""
+                raise CLIError(error.format(_AZURE_SUBSCRIPTION_ID))
+            return env_var_sub
+        raise CLIError(_AZ_LOGIN_MESSAGE)
 
     def get_subscription_id(self, subscription=None):  # take id or name
         return self.get_subscription(subscription)[_SUBSCRIPTION_ID]

src/azure-cli-core/azure/cli/core/auth/identity.py

@@ -307,7 +307,14 @@ def __init__(self, secret_store):
         self._entries = []
 
     def load_entry(self, sp_id, tenant):
+        from azure.cli.core._profile import env_var_auth_configured, load_env_var_credential
         self._load_persistence()
+
+        # If no login data, look for service principal credential in environment variables
+        if not self._entries and env_var_auth_configured():
+            logger.warning("Using service principal credential configured in environment variables.")
+            self._entries = [load_env_var_credential()]
+
         matched = [x for x in self._entries if sp_id == x[_CLIENT_ID]]
         if not matched:
             raise CLIError("Could not retrieve credential from local cache for service principal {}. "

src/azure-cli-core/azure/cli/core/cloud.py

@@ -603,7 +603,8 @@ def set_cloud_subscription(cli_ctx, cloud_name, subscription):
 
 def _set_active_subscription(cli_ctx, cloud_name):
     from azure.cli.core._profile import (Profile, _ENVIRONMENT_NAME, _SUBSCRIPTION_ID,
-                                         _STATE, _SUBSCRIPTION_NAME)
+                                         _STATE, _SUBSCRIPTION_NAME, _AZURE_SUBSCRIPTION_ID,
+                                         env_var_auth_configured, load_env_var_subscription)
     profile = Profile(cli_ctx=cli_ctx)
     subscription_to_use = get_cloud_subscription(cloud_name) or \
                           next((s[_SUBSCRIPTION_ID] for s in profile.load_cached_subscriptions()  # noqa
@@ -619,6 +620,8 @@ def _set_active_subscription(cli_ctx, cloud_name):
             logger.warning(e)
             logger.warning("Unable to automatically switch the active subscription. "
                            "Use 'az account set'.")
+    elif env_var_auth_configured():
+        logger.warning("Using subscription %s configured by environment variable %s. ", load_env_var_subscription(), [_SUBSCRIPTION_ID])
     else:
         logger.warning("Use 'az login' to log in to this cloud.")
         logger.warning("Use 'az account set' to set the active subscription.")

src/azure-cli-core/azure/cli/core/commands/arm.py

@@ -363,7 +363,8 @@ def __call__(self, parser, namespace, value, option_string=None):
                         sub_id = sub['id']
                         break
                 if not sub_id:
-                    logger.warning("Subscription '%s' not recognized.", value)
+                    # User may be authenticating via environment variables
+                    logger.debug("Subscription '%s' not found in local cache.", value)
                     sub_id = value
                 namespace._subscription = sub_id  # pylint: disable=protected-access
 

src/azure-cli-core/azure/cli/core/tests/test_util.py

@@ -16,7 +16,7 @@
     (get_file_json, truncate_text, shell_safe_json_parse, b64_to_hex, hash_string, random_string,
      open_page_in_browser, can_launch_browser, handle_exception, ConfiguredDefaultSetter, send_raw_request,
      should_disable_connection_verify, parse_proxy_resource_id, get_az_user_agent, get_az_rest_user_agent,
-     _get_parent_proc_name, is_wsl)
+     _get_parent_proc_name, is_wsl, is_guid, assert_guid)
 from azure.cli.core.mock import DummyCli
 
 
@@ -421,6 +421,18 @@ def test_get_parent_proc_name(self, mock_process_type):
         parent2.name.return_value = "bash"
         self.assertEqual(_get_parent_proc_name(), "pwsh")
 
+    def test_guid(self):
+        self.assertTrue(is_guid("201ea53e-07b9-4ebf-a85e-5482e48e835c"))
+        self.assertFalse(is_guid(""))
+        self.assertFalse(is_guid(None))
+        self.assertFalse(is_guid("foo"))
+
+        from knack.util import CLIError
+        assert_guid("201ea53e-07b9-4ebf-a85e-5482e48e835c")
+        assert_guid("201ea53e-07b9-4ebf-a85e-5482e48e835c", "named_guid")
+        self.assertRaisesRegex(CLIError, "named_guid must be a GUID.", assert_guid, "foo", "named_guid")
+        self.assertRaisesRegex(CLIError, "foo is not a GUID.", assert_guid, "foo")
+
 
 class TestBase64ToHex(unittest.TestCase):
 

src/azure-cli-core/azure/cli/core/util.py

@@ -1353,3 +1353,17 @@ def should_encrypt_token_cache(cli_ctx):
     encrypt = cli_ctx.config.getboolean('core', 'encrypt_token_cache', fallback=fallback)
 
     return encrypt
+
+def is_guid(guid):
+    import uuid
+    try:
+        uuid.UUID(guid)
+        return True
+    except (ValueError, TypeError):
+        return False
+
+def assert_guid(guid, name=None):
+    if not is_guid(guid):
+        if name:
+            raise CLIError("{} must be a GUID.".format(name))
+        raise CLIError("{} is not a GUID.".format(guid))

src/azure-cli/azure/cli/command_modules/profile/custom.py

@@ -33,24 +33,34 @@
 def list_subscriptions(cmd, all=False, refresh=False):  # pylint: disable=redefined-builtin
     """List the imported subscriptions."""
     from azure.cli.core.api import load_subscriptions
+    from azure.cli.core._profile import load_env_var_subscription, env_var_auth_configured
 
     subscriptions = load_subscriptions(cmd.cli_ctx, all_clouds=all, refresh=refresh)
+
+    if subscriptions:
+        for sub in subscriptions:
+            sub['cloudName'] = sub.pop('environmentName', None)
+        if not all:
+            enabled_ones = [s for s in subscriptions if s.get('state') == 'Enabled']
+            if len(enabled_ones) != len(subscriptions):
+                logger.warning("A few accounts are skipped as they don't have 'Enabled' state. "
+                            "Use '--all' to display them.")
+                subscriptions = enabled_ones
+        return subscriptions
+
+    # If logged out, fetch subscription configured in environment variables
+    if env_var_auth_configured():
+        logger.warning("Fetching subscription configured in environment variables.")
+        subscriptions = [load_env_var_subscription()]
+        return subscriptions
+
     if not subscriptions:
         logger.warning('Please run "az login" to access your accounts.')
-    for sub in subscriptions:
-        sub['cloudName'] = sub.pop('environmentName', None)
-    if not all:
-        enabled_ones = [s for s in subscriptions if s.get('state') == 'Enabled']
-        if len(enabled_ones) != len(subscriptions):
-            logger.warning("A few accounts are skipped as they don't have 'Enabled' state. "
-                           "Use '--all' to display them.")
-            subscriptions = enabled_ones
-    return subscriptions
-
+    return []
 
 def show_subscription(cmd, subscription=None):
     profile = Profile(cli_ctx=cmd.cli_ctx)
-    return profile.get_subscription(subscription)
+    return profile.get_subscription(subscription, allow_null_subscription=True)
 
 
 def get_access_token(cmd, subscription=None, resource=None, scopes=None, resource_type=None, tenant=None):