Add Device Code fallback option for when interactive auth isn't avaliable.

msticpy/auth/azure_auth.py

@@ -8,6 +8,7 @@
 from typing import List
 
 from azure.common.exceptions import CloudError
+from azure.identity import DeviceCodeCredential
 from azure.mgmt.subscription import SubscriptionClient
 
 from .._version import VERSION
@@ -23,6 +24,7 @@
     az_connect_core,
     only_interactive_cred,
 )
+from .cred_wrapper import CredentialWrapper
 
 __version__ = VERSION
 __author__ = "Pete Bryan"
@@ -120,3 +122,44 @@ def az_user_connect(tenant_id: str = None, silent: bool = False) -> AzCredential
     return az_connect_core(
         auth_methods=["cli", "interactive"], tenant_id=tenant_id, silent=silent
     )
+
+
+def fallback_devicecode_creds(cloud: str = None, tenant_id: str = None, **kwargs):
+    """
+    Authenticate using device code as a fallback method.
+
+    Parameters
+    ----------
+    cloud : str, optional
+        What Azure cloud to connect to.
+        By default it will attempt to use the cloud setting from config file.
+        If this is not set it will default to Azure Public Cloud
+    tenant_id : str, optional
+        The tenant to authenticate against. If not supplied,
+        the tenant ID is read from configuration, or the default tenant for the identity.
+
+    Returns
+    -------
+    AzCredentials
+                Named tuple of:
+        - legacy (ADAL) credentials
+        - modern (MSAL) credentials
+
+    Raises
+    ------
+    CloudError
+        If chained token credential creation fails.
+
+    """
+    cloud = cloud or kwargs.pop("region", AzureCloudConfig().cloud)
+    az_config = AzureCloudConfig(cloud)
+    aad_uri = az_config.endpoints.active_directory
+    tenant_id = tenant_id or AzureCloudConfig().tenant_id
+    creds = DeviceCodeCredential(authority=aad_uri, tenant_id=tenant_id)
+    legacy_creds = CredentialWrapper(
+        creds, resource_id=AzureCloudConfig(cloud).token_uri
+    )
+    if not creds:
+        raise CloudError("Could not obtain credentials.")
+
+    return AzCredentials(legacy_creds, creds)

msticpy/auth/azure_auth_core.py

@@ -37,16 +37,6 @@
 
 AzCredentials = namedtuple("AzCredentials", ["legacy", "modern"])
 
-_EXCLUDED_AUTH = {
-    "cli": True,
-    "env": True,
-    "msi": True,
-    "vscode": True,
-    "powershell": True,
-    "interactive": True,
-    "cache": True,
-}
-
 
 def get_azure_config_value(key, default):
     """Get a config value from Azure section."""
@@ -208,19 +198,28 @@ def _az_connect_core(
     az_config = AzureCloudConfig(cloud)
     aad_uri = az_config.endpoints.active_directory
     tenant_id = tenant_id or AzureCloudConfig().tenant_id
+    excluded_auth = {
+        "cli": True,
+        "env": True,
+        "msi": True,
+        "vscode": True,
+        "powershell": True,
+        "interactive": True,
+        "cache": True,
+    }
     if auth_methods:
         for method in auth_methods:
-            if method in _EXCLUDED_AUTH:
-                _EXCLUDED_AUTH[method] = False
+            if method in excluded_auth:
+                excluded_auth[method] = False
         creds = DefaultAzureCredential(
             authority=aad_uri,
-            exclude_cli_credential=_EXCLUDED_AUTH["cli"],
-            exclude_environment_credential=_EXCLUDED_AUTH["env"],
-            exclude_managed_identity_credential=_EXCLUDED_AUTH["msi"],
-            exclude_powershell_credential=_EXCLUDED_AUTH["powershell"],
-            exclude_visual_studio_code_credential=_EXCLUDED_AUTH["vscode"],
-            exclude_shared_token_cache_credential=_EXCLUDED_AUTH["cache"],
-            exclude_interactive_browser_credential=_EXCLUDED_AUTH["interactive"],
+            exclude_cli_credential=excluded_auth["cli"],
+            exclude_environment_credential=excluded_auth["env"],
+            exclude_managed_identity_credential=excluded_auth["msi"],
+            exclude_powershell_credential=excluded_auth["powershell"],
+            exclude_visual_studio_code_credential=excluded_auth["vscode"],
+            exclude_shared_token_cache_credential=excluded_auth["cache"],
+            exclude_interactive_browser_credential=excluded_auth["interactive"],
             interactive_browser_tenant_id=tenant_id,
         )
     else:

msticpy/context/azure/azure_data.py

@@ -11,13 +11,15 @@
 import numpy as np
 import pandas as pd
 from azure.common.exceptions import CloudError
+from azure.core.exceptions import ClientAuthenticationError
 from azure.mgmt.resource.subscriptions import SubscriptionClient
 
 from ..._version import VERSION
 from ...auth.azure_auth import (
     AzCredentials,
     AzureCloudConfig,
     az_connect,
+    fallback_devicecode_creds,
     only_interactive_cred,
 )
 from ...auth.cloud_mappings import get_all_endpoints
@@ -898,7 +900,9 @@ def get_api_headers(token: str) -> Dict:
     }
 
 
-def get_token(credential: AzCredentials, tenant_id: str = None) -> str:
+def get_token(
+    credential: AzCredentials, tenant_id: str = None, cloud: str = None
+) -> str:
     """
     Extract token from a azure.identity object.
 
@@ -908,6 +912,8 @@ def get_token(credential: AzCredentials, tenant_id: str = None) -> str:
         Azure OAuth credentials.
     tenant_id : str, optional
         The tenant to connect to if not the users home tenant.
+    cloud: str, optional
+        The Azure cloud to connect to.
 
     Returns
     -------
@@ -916,10 +922,20 @@ def get_token(credential: AzCredentials, tenant_id: str = None) -> str:
 
     """
     if tenant_id:
-        token = credential.modern.get_token(AzureCloudConfig().token_uri)
+        try:
+            token = credential.modern.get_token(AzureCloudConfig().token_uri)
+        except ClientAuthenticationError:
+            credential = fallback_devicecode_creds(cloud=cloud)
+            token = credential.modern.get_token(AzureCloudConfig().token_uri)
     else:
-        token = credential.modern.get_token(
-            AzureCloudConfig().token_uri, tenant_id=tenant_id
-        )
+        try:
+            token = credential.modern.get_token(
+                AzureCloudConfig().token_uri, tenant_id=tenant_id
+            )
+        except ClientAuthenticationError:
+            credential = fallback_devicecode_creds(cloud=cloud, tenant_id=tenant_id)
+            token = credential.modern.get_token(
+                AzureCloudConfig().token_uri, tenant_id=tenant_id
+            )
 
     return token.token

msticpy/context/azure/sentinel_analytics.py

@@ -56,7 +56,9 @@ def list_alert_rules(self) -> pd.DataFrame:
             A table of the workspace's alert rules.
 
         """
-        return self._list_items(item_type="alert_rules")  # type: ignore
+        return self._list_items(  # type: ignore
+            item_type="alert_rules", api_version="2021-10-01"
+        )
 
     def _get_template_id(
         self,

msticpy/context/azure/sentinel_core.py

@@ -65,7 +65,8 @@ def __init__(
             Sentinel Workspace, by default None
 
         """
-        super().__init__(connect=connect, cloud=cloud)
+        self.user_cloud = cloud
+        super().__init__(connect=connect, cloud=self.user_cloud)
         self.config = None  # type: ignore
         self.base_url = self.endpoints.resource_manager
         self.default_subscription: Optional[str] = None
@@ -110,7 +111,9 @@ def connect(
         if "token" in kwargs:
             self.token = kwargs["token"]
         else:
-            self.token = get_token(self.credentials)  # type: ignore
+            self.token = get_token(
+                self.credentials, tenant_id=tenant_id, cloud=self.user_cloud  # type: ignore
+            )
 
         self.res_group_url = None
         self.prov_path = None

msticpy/data/drivers/kql_driver.py

@@ -12,6 +12,7 @@
 from typing import Any, Dict, Iterable, List, Optional, Tuple, Union
 
 import pandas as pd
+from azure.core.exceptions import ClientAuthenticationError
 from IPython import get_ipython
 
 from ...auth.azure_auth import AzureCloudConfig, az_connect, only_interactive_cred
@@ -524,15 +525,20 @@ def _set_az_auth_option(
         endpoint_uri = self._get_endpoint_uri()
         endpoint_token_uri = f"{endpoint_uri}.default"
         # obtain token for the endpoint
-        token = creds.modern.get_token(endpoint_token_uri, tenant_id=mp_az_tenant_id)
-        # set the token values in the namespace
-
-        endpoint_token = {
-            "access_token": token.token,
-            "token_type": "Bearer",
-            "resource": endpoint_uri,
-        }
-        self._set_kql_option("try_token", endpoint_token)
+        try:
+            token = creds.modern.get_token(
+                endpoint_token_uri, tenant_id=mp_az_tenant_id
+            )
+            # set the token values in the namespace
+            endpoint_token = {
+                "access_token": token.token,
+                "token_type": "Bearer",
+                "resource": endpoint_uri,
+            }
+            self._set_kql_option("try_token", endpoint_token)
+        # if the above auth fails fall back to KQLMagics auth method
+        except ClientAuthenticationError:
+            pass
 
     def _get_endpoint_uri(self):
         return _LOGANALYTICS_URL_BY_CLOUD[self.az_cloud]

prospector.yml

@@ -14,13 +14,13 @@ ignore-paths:
 pyroma:
   run: true
 
-pep8:
+pycodestyle:
   full: true
   disable: [
       E501, # Line length handled by Black
     ]
 
-pep257:
+pydocstyle:
   disable: [
       # Disable because not part of PEP257 official convention:
       # see http://pep257.readthedocs.io/en/latest/error_codes.html